fix(insight): release displaced records outside every lock

`_emit` hands a record to the scheduler while holding this instance's lock,
and that hand-off displaces the record already pending for the execution.
Releasing the displaced one can run a customer `__del__`, because the record
is a snapshot of customer data, and a finalizer that reaches another
execution's plugin blocks on that instance's lock. Two threads doing that to
each other at once hang both invocations, and the invocation end is awaited
before the response.

The reentrant lock does not help: it covers re-entry on the same instance,
which is why customer code inside a build can call this execution's hooks
safely, and says nothing about a second instance.

`schedule()` now returns what it displaced and the hook frame releases it
once the outermost hook returns, with no lock held. Releases run before the
deferred drains, so a finalizer that schedules a record is covered by the
drain that follows it.

Author: ParidelPooya

packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/_export_scheduler.py

@@ -144,14 +144,28 @@ def __init__(self, exporters: list[InsightExporter]) -> None:
         # a worker that keeps making progress never approaches the bound.
         self._worker_faults = 0
 
-    def schedule(self, execution: _ExportState, record: dict[str, Any]) -> None:
-        """Replace this execution's pending snapshot; never runs exporters inline."""
+    def schedule(self, execution: _ExportState, record: dict[str, Any]) -> list[Any]:
+        """Replace this execution's pending snapshot; never runs exporters inline.
+
+        Returns the records this call displaced, for the caller to release once it
+        holds no lock. They are handed back rather than dropped here because
+        releasing one can run a customer ``__del__``, and this method is called
+        from inside the calling plugin's own lock. A finalizer that re-entered a
+        *different* execution's plugin would then block on that instance's lock
+        while holding this one, which deadlocks both invocations if the mirror
+        image happens on another thread at the same time. The plugin's hook frame
+        drops them when the outermost hook returns; see
+        ``WorkflowInsightPlugin._hook_frame``.
+
+        A caller with no frame to defer to may drop the returned list
+        immediately: doing so is only unsafe while a plugin lock is held.
+        """
         displaced: dict[str, Any] | None = None
         failed_pending: _Dropped | None = None
         start_error: Exception | None = None
         with self._condition:
             if self._disabled:
-                return
+                return []
             self._seq += 1
             execution.scheduled_seq = self._seq
             displaced = execution.pending_record
@@ -161,14 +175,13 @@ def schedule(self, execution: _ExportState, record: dict[str, Any]) -> None:
             self._pending[execution] = None
             failed_pending, start_error = self._ensure_worker_locked()
             self._condition.notify_all()
-        # Releasing either record may run custom finalizers, so do it unlocked.
-        del displaced, failed_pending
         if start_error is not None:
             _logger.warning(
                 "workflow-insight: could not start export worker; disabling "
                 "asynchronous export: %s",
                 start_error,
             )
+        return [item for item in (displaced, failed_pending) if item is not None]
 
     def drain(self, execution: _ExportState) -> None:
         """Wait until this execution's latest record is exported and exporters flush.

packages/aws-durable-execution-sdk-python-insight/src/aws_durable_execution_sdk_python_insight/plugin.py

@@ -191,6 +191,11 @@ class _HookFrames(threading.local):
     def __init__(self) -> None:
         self.depth = 0
         self.pending: list[WorkflowInsightPlugin] = []
+        # Records displaced from the scheduler's pending slots, held until every
+        # plugin lock on this thread is released. Releasing one can run a customer
+        # finalizer, and a finalizer that reaches another execution's plugin must
+        # not do so while this thread holds a plugin lock.
+        self.releases: list[Any] = []
 
 
 _hook_frames = _HookFrames()
@@ -341,10 +346,16 @@ def _hook_frame(self) -> Iterator[None]:
             yield
         finally:
             state.depth -= 1
-            if state.depth == 0 and state.pending:
-                owed, state.pending = state.pending, []
-                for plugin in owed:
-                    plugin._drain()
+            if state.depth == 0:
+                # Released before the drains, and with no lock held: a finalizer
+                # that schedules a record is then covered by the drain that
+                # follows it.
+                if state.releases:
+                    state.releases.clear()
+                if state.pending:
+                    owed, state.pending = state.pending, []
+                    for plugin in owed:
+                        plugin._drain()
 
     def _request_drain(self) -> None:
         """Ask for a drain once the outermost hook frame on this thread unwinds."""
@@ -645,7 +656,13 @@ def _emit(
         # every later non-terminal record is rejected above.
         if not closing and (self._closed or revision != self._build_revision):
             return
-        self._shared._scheduler.schedule(self, record)
+        # The records this hand-off displaces are released by the hook frame, not
+        # here: this runs inside `_lock`, and a displaced record can carry a
+        # customer object whose `__del__` re-enters a hook. Re-entering *this*
+        # execution's hook is safe because `_lock` is reentrant; re-entering
+        # another execution's is not, and two threads doing it to each other at
+        # once would hang both invocations. See `_hook_frame`.
+        _hook_frames.releases.extend(self._shared._scheduler.schedule(self, record))
 
 
 class _WorkflowInsightFactory:

packages/aws-durable-execution-sdk-python-insight/tests/test_plugin.py

@@ -1240,6 +1240,62 @@ def failing_on_the_second_call(value: Any) -> Any:
     assert _wait_until(lambda: not factory._scheduler._worker_alive())
 
 
+def test_a_displaced_records_finalizer_runs_with_no_plugin_lock_held():
+    # `_emit` hands the record to the scheduler while holding this instance's
+    # `_lock`, and that hand-off displaces the record already pending for this
+    # execution. Releasing the displaced one can run a customer `__del__` -- it is
+    # a snapshot of customer data -- and a finalizer that reaches ANOTHER
+    # execution's plugin blocks on that instance's lock. Two threads doing that to
+    # each other at once hang both invocations, which the reentrant lock does not
+    # help with: it only covers re-entry on the same instance. The hook frame
+    # therefore releases displaced records after every lock is dropped.
+    exporter = ConcurrentCaptureExporter()
+    observed: list[bool] = []
+    holder: dict[str, Any] = {}
+
+    class _FinalizerProbe:
+        def __del__(self) -> None:
+            plugin = holder.get("plugin")
+            if plugin is None:
+                return
+            # `_is_owned()` answers "does the calling thread hold this lock",
+            # which is the question here. Acquiring it would not: an RLock lets
+            # its owner acquire it again.
+            observed.append(not plugin._lock._is_owned())
+
+    # A transform returning a fresh object per emit is what makes the record the
+    # only reference to it, so releasing the displaced record is what collects it.
+    factory = workflow_insight(
+        WorkflowInsightConfig(
+            exporters=[exporter],
+            emit_mode="on-change",
+            content=ContentConfig(input=lambda value: _FinalizerProbe()),
+        )
+    )
+    plugin = factory.create_plugin(_start(operations={}))
+    holder["plugin"] = plugin
+
+    # The first emit queues a record holding a probe; the next displaces it while
+    # the worker is still parked, so the release happens on this thread.
+    plugin.on_invocation_start(_start(operations={}))
+    plugin.on_operation_change(
+        OperationChangeInfo(
+            execution_arn=ARN,
+            updated_operations=_ops(_step("s")),
+            operations=_ops(_step("s")),
+        )
+    )
+    plugin.on_invocation_end(_end(operations=_ops(_step("s"))))
+
+    assert observed, "the displaced record's finalizer must have run"
+    assert all(observed), (
+        "a displaced record was released while this thread still held the "
+        "plugin's lock, which is what deadlocks two invocations whose finalizers "
+        "reach each other"
+    )
+    assert _wait_until(lambda: not factory._scheduler._worker_alive())
+
+
 def _free_for_another_thread(lock: Any) -> bool:
     """Report whether a lock is unheld, as seen from a thread that never took it.