fix(worker): close orchestrator generators when a dispatch ends (#300)

Give each replay dispatch explicit ownership of its orchestrator generator: close it once in an outer finally after the response is built, so cleanup in finally blocks runs deterministically and cannot alter the returned actions or custom status. Ordinary teardown failures are logged without replacing dispatch results or abandonment.

Async bridges such as Pydantic AI otherwise retain AnyIO task scopes, event loops, and orchestration contexts indefinitely. Also relevant to the async/await work in #293 and #294.

Fixes #299

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 35371373-86cf-4897-bf7c-460fa3c88d1f

Author: tamirdresher

CHANGELOG.md

@@ -7,6 +7,12 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
 
+FIXED
+
+- Orchestrator generators are now closed when a dispatch ends, so cleanup in
+`finally` blocks runs deterministically, including for async bridges that would
+otherwise retain tasks, event loops, and orchestration contexts.
+
 ## v1.12.0
 
 ADDED

azure-functions-durable/CHANGELOG.md

@@ -19,6 +19,12 @@ CHANGED
 
 - Updated the minimum `azure-functions` dependency to v2.4.0b1.
 
+FIXED
+
+- With the corresponding core `durabletask` SDK fix, orchestrator generators
+are closed when a dispatch ends, so cleanup in `finally` blocks runs
+deterministically, including for async bridges.
+
 ## v2.0.0rc3
 
 ADDED

durabletask-azuremanaged/CHANGELOG.md

@@ -7,6 +7,12 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
 
+FIXED
+
+- With the corresponding core `durabletask` SDK fix, orchestrator generators
+are closed when a dispatch ends, so cleanup in `finally` blocks runs
+deterministically, including for async bridges.
+
 ## v1.12.0
 
 ADDED

durabletask/worker.py

@@ -2289,64 +2289,80 @@ def execute(
             logger=self._logger,
         )
         try:
-            # Rebuild local state by replaying old history into the orchestrator function
-            self._logger.debug(
-                f"{instance_id}: Rebuilding local state with {len(old_events)} history event..."
-            )
-            ctx._is_replaying = True  # pyright: ignore[reportPrivateUsage]
-            for old_event in old_events:
-                self.process_event(ctx, old_event)
-
-            # Get new actions by executing newly received events into the orchestrator function
-            if self._logger.level <= logging.DEBUG:
-                summary = _get_new_event_summary(new_events)
+            try:
+                # Rebuild local state by replaying old history into the orchestrator function
                 self._logger.debug(
-                    f"{instance_id}: Processing {len(new_events)} new event(s): {summary}"
+                    f"{instance_id}: Rebuilding local state with {len(old_events)} history event..."
                 )
-            ctx._is_replaying = False  # pyright: ignore[reportPrivateUsage]
-            for new_event in new_events:
-                self.process_event(ctx, new_event)
-
-        except pe.VersionFailureException as ex:
-            if self._registry.versioning and self._registry.versioning.failure_strategy == VersionFailureStrategy.FAIL:
-                if ex.error_details:
-                    ctx.set_failed(ex.error_details)
-                else:
-                    ctx.set_failed(ex)
-            elif self._registry.versioning and self._registry.versioning.failure_strategy == VersionFailureStrategy.REJECT:
-                raise pe.AbandonOrchestrationError
+                ctx._is_replaying = True  # pyright: ignore[reportPrivateUsage]
+                for old_event in old_events:
+                    self.process_event(ctx, old_event)
+
+                # Get new actions by executing newly received events into the orchestrator function
+                if self._logger.level <= logging.DEBUG:
+                    summary = _get_new_event_summary(new_events)
+                    self._logger.debug(
+                        f"{instance_id}: Processing {len(new_events)} new event(s): {summary}"
+                    )
+                ctx._is_replaying = False  # pyright: ignore[reportPrivateUsage]
+                for new_event in new_events:
+                    self.process_event(ctx, new_event)
+
+            except pe.VersionFailureException as ex:
+                if self._registry.versioning and self._registry.versioning.failure_strategy == VersionFailureStrategy.FAIL:
+                    if ex.error_details:
+                        ctx.set_failed(ex.error_details)
+                    else:
+                        ctx.set_failed(ex)
+                elif self._registry.versioning and self._registry.versioning.failure_strategy == VersionFailureStrategy.REJECT:
+                    raise pe.AbandonOrchestrationError
 
-        except Exception as ex:
-            # Unhandled exceptions fail the orchestration
-            self._logger.debug(f"{instance_id}: Orchestration {orchestration_name} failed")
-            ctx.set_failed(ex)
+            except Exception as ex:
+                # Unhandled exceptions fail the orchestration
+                self._logger.debug(f"{instance_id}: Orchestration {orchestration_name} failed")
+                ctx.set_failed(ex)
 
-        if not ctx._is_complete:  # pyright: ignore[reportPrivateUsage]
-            task_count = len(ctx._pending_tasks)  # pyright: ignore[reportPrivateUsage]
-            event_count = len(ctx._pending_events)  # pyright: ignore[reportPrivateUsage]
-            self._logger.info(
-                f"{instance_id}: Orchestrator {orchestration_name} yielded with {task_count} task(s) "
-                f"and {event_count} event(s) outstanding."
-            )
-        elif (
-                ctx._completion_status and ctx._completion_status is not pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW  # pyright: ignore[reportPrivateUsage]
-        ):
-            completion_status_str = ph.get_orchestration_status_str(
-                ctx._completion_status  # pyright: ignore[reportPrivateUsage]
-            )
-            self._logger.info(
-                f"{instance_id}: Orchestration {orchestration_name} completed with status: {completion_status_str}"
-            )
+            if not ctx._is_complete:  # pyright: ignore[reportPrivateUsage]
+                task_count = len(ctx._pending_tasks)  # pyright: ignore[reportPrivateUsage]
+                event_count = len(ctx._pending_events)  # pyright: ignore[reportPrivateUsage]
+                self._logger.info(
+                    f"{instance_id}: Orchestrator {orchestration_name} yielded with {task_count} task(s) "
+                    f"and {event_count} event(s) outstanding."
+                )
+            elif (
+                    ctx._completion_status and ctx._completion_status is not pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW  # pyright: ignore[reportPrivateUsage]
+            ):
+                completion_status_str = ph.get_orchestration_status_str(
+                    ctx._completion_status  # pyright: ignore[reportPrivateUsage]
+                )
+                self._logger.info(
+                    f"{instance_id}: Orchestration {orchestration_name} completed with status: {completion_status_str}"
+                )
 
-        actions = ctx.get_actions()
-        if self._logger.level <= logging.DEBUG:
-            self._logger.debug(
-                f"{instance_id}: Returning {len(actions)} action(s): {_get_action_summary(actions)}"
+            actions = ctx.get_actions()
+            if self._logger.level <= logging.DEBUG:
+                self._logger.debug(
+                    f"{instance_id}: Returning {len(actions)} action(s): {_get_action_summary(actions)}"
+                )
+            return ExecutionResults(
+                actions=actions, encoded_custom_status=ctx._encoded_custom_status,  # pyright: ignore[reportPrivateUsage]
+                orchestration_trace_context=ctx._orchestration_trace_context,  # pyright: ignore[reportPrivateUsage]
             )
-        return ExecutionResults(
-            actions=actions, encoded_custom_status=ctx._encoded_custom_status,  # pyright: ignore[reportPrivateUsage]
-            orchestration_trace_context=ctx._orchestration_trace_context,  # pyright: ignore[reportPrivateUsage]
-        )
+        finally:
+            # Close after building the response, so the orchestrator's finally
+            # blocks do not change the actions or custom status we return.
+            if ctx._generator is not None:  # pyright: ignore[reportPrivateUsage]
+                try:
+                    ctx._generator.close()  # pyright: ignore[reportPrivateUsage]
+                except Exception:
+                    cleanup_name = next(
+                        (e.executionStarted.name for e in new_events if e.HasField("executionStarted")),
+                        orchestration_name,
+                    )
+                    self._logger.warning(
+                        f"{instance_id}: Failed to close orchestrator {cleanup_name} generator.",
+                        exc_info=True,
+                    )
 
     def _build_rewind_result(
             self,

requirements.txt

@@ -3,6 +3,7 @@ grpcio>=1.60.0 # 1.60.0 is the version introducing protobuf 1.25.X support, newe
 protobuf
 pytest
 pytest-asyncio
+anyio
 pytest-cov
 azure-identity
 azure-functions