Merge branch 'main' into dependabot/github_actions/github-actions-4b2c77c676

Author: andystaples

CHANGELOG.md

@@ -9,13 +9,19 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 FIXED
 
+- Single-instance `purge_orchestration()` requests now explicitly target
+orchestrations rather than entities in both synchronous and asynchronous clients.
 - Timer callbacks no longer schedule additional long-timer chunks, retry
 activities or sub-orchestrations, or resume orchestrator code after completion,
 failure, termination, or continue-as-new. Work scheduled before the terminal
 state is preserved.
 - `continue_as_new(..., save_events=True)` now preserves the global arrival
 order of unconsumed buffered external events across different event names,
 instead of grouping carryover events by name.
+- Fixed external events arriving after `continue_as_new(..., save_events=True)`
+being lost to abandoned waits instead of carried into the next execution.
+Events already delivered to live waits are not carried over, and
+`save_events=False` still discards unprocessed events.
 
 ## v1.11.0
 

azure-functions-durable/CHANGELOG.md

@@ -9,13 +9,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 FIXED
 
+- With the corresponding core `durabletask` SDK fix, single-instance
+`purge_orchestration()` requests from `DurableFunctionsClient` and
+`SyncDurableFunctionsClient` now explicitly target orchestrations rather than
+entities. This also applies to the `purge_instance_history()` compatibility alias.
 - With the corresponding core `durabletask` SDK fix, timer callbacks no longer
 schedule additional long-timer chunks, retry activities or sub-orchestrations,
 or resume orchestrator code after completion, failure, termination, or
 continue-as-new. This applies to both native and compatibility orchestration APIs.
 - With the corresponding core `durabletask` SDK fix,
 `continue_as_new(..., save_events=True)` preserves the global arrival order
 of unconsumed buffered external events across different event names.
+- With a corrected core `durabletask` SDK, native durabletask orchestrators
+preserve external events arriving after `continue_as_new(..., save_events=True)`
+instead of losing them to abandoned waits.
 
 ## v2.0.0rc2
 

durabletask-azuremanaged/CHANGELOG.md

@@ -9,6 +9,9 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 FIXED
 
+- With the corresponding core `durabletask` SDK fix, single-instance
+`purge_orchestration()` requests from synchronous and asynchronous Azure Managed
+clients now explicitly target orchestrations rather than entities.
 - With the corresponding core `durabletask` SDK fix, timer callbacks no longer
 retry activities or sub-orchestrations, or resume orchestrator code after
 completion, failure, termination, or continue-as-new. Azure Managed uses native
@@ -17,6 +20,9 @@ change that behavior.
 - With the corresponding core `durabletask` SDK fix,
 `continue_as_new(..., save_events=True)` preserves the global arrival order
 of unconsumed buffered external events across different event names.
+- With a corrected core `durabletask` SDK, external events arriving after
+`continue_as_new(..., save_events=True)` are no longer lost to abandoned waits
+and are carried into the next execution.
 
 ## v1.11.0
 

durabletask/client.py

@@ -857,7 +857,7 @@ def restart_orchestration(self, instance_id: str, *,
         return res.instanceId
 
     def purge_orchestration(self, instance_id: str, recursive: bool = True) -> PurgeInstancesResult:
-        req = pb.PurgeInstancesRequest(instanceId=instance_id, recursive=recursive)
+        req = pb.PurgeInstancesRequest(instanceId=instance_id, recursive=recursive, isOrchestration=True)
         self._logger.info(f"Purging instance '{instance_id}'.")
         resp: pb.PurgeInstancesResponse = self._stub.PurgeInstances(req)
         return new_purge_instances_result(resp)
@@ -1402,7 +1402,7 @@ async def restart_orchestration(self, instance_id: str, *,
         return res.instanceId
 
     async def purge_orchestration(self, instance_id: str, recursive: bool = True) -> PurgeInstancesResult:
-        req = pb.PurgeInstancesRequest(instanceId=instance_id, recursive=recursive)
+        req = pb.PurgeInstancesRequest(instanceId=instance_id, recursive=recursive, isOrchestration=True)
         self._logger.info(f"Purging instance '{instance_id}'.")
         resp: pb.PurgeInstancesResponse = await self._get_stub().PurgeInstances(req)
         return new_purge_instances_result(resp)

durabletask/task.py

@@ -393,12 +393,18 @@ def continue_as_new(self, new_input: Any, *, save_events: bool = False,
                         new_version: str | None = None) -> None:
         """Continue the orchestration execution as a new instance.
 
+        Orchestrators should return immediately after calling this method.
+        Subsequent external events are no longer delivered to pending waits
+        in the current execution.
+
         Parameters
         ----------
         new_input : Any
             The new input to use for the new orchestration instance.
         save_events : bool
             A flag indicating whether to add any unprocessed external events in the new orchestration history.
+            Events already delivered to a waiting task are not saved, even if
+            the orchestrator has not yielded that task.
         new_version : str | None
             An optional version to assign to the new orchestration instance.
         """

durabletask/worker.py

@@ -2852,7 +2852,9 @@ def _cancel_timer() -> None:
                         self._logger.info(f"{ctx.instance_id} Event raised: {event_name}")
                     task_list = ctx._pending_events.get(event_name, None)  # pyright: ignore[reportPrivateUsage]
                     decoded_result: Any | None = None
-                    if task_list:
+                    # Completed executions leave abandoned waits behind. Buffer
+                    # trailing events instead, so continue-as-new can carry them over.
+                    if task_list and not ctx._is_complete:  # pyright: ignore[reportPrivateUsage]
                         event_task = task_list.pop(0)
                         if not ph.is_empty(event.eventRaised.input):
                             decoded_result = self._data_converter.deserialize(
@@ -2879,7 +2881,7 @@ def _cancel_timer() -> None:
                         ctx._received_event_sequence += 1  # pyright: ignore[reportPrivateUsage]
                         if not ctx.is_replaying:
                             self._logger.info(
-                                f"{ctx.instance_id}: Event '{event_name}' has been buffered as there are no tasks waiting for it."
+                                f"{ctx.instance_id}: Event '{event_name}' has been buffered as there are no active tasks waiting for it."
                             )
             elif event.HasField("executionSuspended"):
                 if not self._is_suspended and not ctx.is_replaying:

tests/azure-functions-durable/test_client_compat.py

@@ -20,7 +20,7 @@
     replace_url_origin,
 )
 from durabletask import history as dt_history, task as dt_task
-from durabletask.client import AsyncTaskHubGrpcClient, OrchestrationStatus
+from durabletask.client import AsyncTaskHubGrpcClient, OrchestrationStatus, PurgeInstancesResult
 from durabletask.entities import EntityInstanceId
 from durabletask.internal import orchestrator_service_pb2 as pb
 from durabletask.task import RetryPolicy
@@ -1288,14 +1288,54 @@ async def test_get_status_by_returns_wrapped_list():
 # Return-type shims: PurgeHistoryResult
 # ---------------------------------------------------------------------------
 
+@pytest.mark.parametrize("recursive", [None, True, False])
+def test_sync_purge_orchestration_request(recursive: bool | None) -> None:
+    client = df.SyncDurableFunctionsClient(_CLIENT_CONFIG)
+    stub = Mock()
+    stub.PurgeInstances.return_value = pb.PurgeInstancesResponse(deletedInstanceCount=3)
+    try:
+        with patch.object(client, "_stub", stub):
+            if recursive is None:
+                result = client.purge_orchestration("abc")
+            else:
+                result = client.purge_orchestration("abc", recursive=recursive)
+        stub.PurgeInstances.assert_called_once_with(pb.PurgeInstancesRequest(
+            instanceId="abc", recursive=True if recursive is None else recursive,
+            isOrchestration=True))
+        assert result == PurgeInstancesResult(deleted_instance_count=3, is_complete=None)
+    finally:
+        client.close()
+
+
+@pytest.mark.parametrize("recursive", [None, True, False])
+async def test_async_purge_orchestration_request(recursive: bool | None) -> None:
+    client = _make_client()
+    stub = Mock()
+    stub.PurgeInstances = AsyncMock(return_value=pb.PurgeInstancesResponse(deletedInstanceCount=3))
+    try:
+        with patch.object(client, "_get_stub", return_value=stub):
+            if recursive is None:
+                result = await client.purge_orchestration("abc")
+            else:
+                result = await client.purge_orchestration("abc", recursive=recursive)
+        stub.PurgeInstances.assert_awaited_once_with(pb.PurgeInstancesRequest(
+            instanceId="abc", recursive=True if recursive is None else recursive,
+            isOrchestration=True))
+        assert result == PurgeInstancesResult(deleted_instance_count=3, is_complete=None)
+    finally:
+        await client.close()
+
+
 async def test_purge_instance_history_returns_purge_history_result():
     client = _make_client()
+    stub = Mock()
+    stub.PurgeInstances = AsyncMock(return_value=pb.PurgeInstancesResponse(deletedInstanceCount=3))
     try:
-        result = SimpleNamespace(deleted_instance_count=3, is_complete=True)
-        with patch.object(client, "purge_orchestration",
-                          new=AsyncMock(return_value=result)):
+        with patch.object(client, "_get_stub", return_value=stub):
             with pytest.warns(DeprecationWarning):
                 purge = await client.purge_instance_history("abc")
+        stub.PurgeInstances.assert_awaited_once_with(pb.PurgeInstancesRequest(
+            instanceId="abc", recursive=True, isOrchestration=True))
         assert purge.instances_deleted == 3
     finally:
         await client.close()

tests/azure-functions-durable/test_worker_compat.py

@@ -542,6 +542,37 @@ def orchestrator(context):
     assert "boom" in completion.failureDetails.errorMessage
 
 
+@pytest.mark.parametrize("save_events", [True, False])
+def test_continue_as_new_preserves_trailing_events_in_worker_response(save_events: bool):
+    def orchestrator(ctx: task.OrchestrationContext, _):
+        event_task = ctx.wait_for_external_event("event")
+        timer = ctx.create_timer(timedelta(seconds=1))
+        yield task.when_any([event_task, timer])
+        ctx.continue_as_new(None, save_events=save_events)
+
+    started_at = datetime(2026, 1, 1)
+    fire_at = started_at + timedelta(seconds=1)
+    request = pb.OrchestratorRequest(instanceId=TEST_INSTANCE_ID)
+    request.pastEvents.extend([
+        helpers.new_orchestrator_started_event(started_at),
+        helpers.new_execution_started_event("continue-events", TEST_INSTANCE_ID),
+        helpers.new_timer_created_event(1, fire_at),
+    ])
+    request.newEvents.extend([
+        helpers.new_timer_fired_event(1, fire_at),
+        helpers.new_event_raised_event("event", "1"),
+        helpers.new_event_raised_event("event", "2"),
+    ])
+    encoded = base64.b64encode(request.SerializeToString()).decode("utf-8")
+    response = _decode_orchestrator_response(
+        DurableFunctionsWorker().execute_orchestration_request(orchestrator, encoded))
+    completion = _get_completion_action(response)
+
+    assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW
+    assert [event.eventRaised.input.value for event in completion.carryoverEvents] == (
+        ["1", "2"] if save_events else [])
+
+
 def test_activity_retry_then_fan_out_uses_distinct_task_ids():
     """Regression test for Azure/azure-functions-durable-python#603."""
     def orchestrator(context):

tests/durabletask-azuremanaged/test_dts_purge.py

@@ -0,0 +1,54 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from durabletask.azuremanaged.client import (
+    AsyncDurableTaskSchedulerClient,
+    DurableTaskSchedulerClient,
+)
+from durabletask.client import PurgeInstancesResult
+from durabletask.internal import orchestrator_service_pb2 as pb
+
+
+@pytest.mark.parametrize("recursive", [None, True, False])
+def test_dts_purge_orchestration_request(recursive: bool | None) -> None:
+    stub = MagicMock()
+    stub.PurgeInstances.return_value = pb.PurgeInstancesResponse(deletedInstanceCount=3)
+
+    with patch("durabletask.client.stubs.TaskHubSidecarServiceStub", return_value=stub):
+        with DurableTaskSchedulerClient(
+                host_address="localhost:4001", taskhub="hub",
+                token_credential=None, channel=MagicMock()) as client:
+            if recursive is None:
+                result = client.purge_orchestration("instance")
+            else:
+                result = client.purge_orchestration("instance", recursive=recursive)
+
+    stub.PurgeInstances.assert_called_once_with(pb.PurgeInstancesRequest(
+        instanceId="instance", recursive=True if recursive is None else recursive,
+        isOrchestration=True))
+    assert result == PurgeInstancesResult(deleted_instance_count=3, is_complete=None)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("recursive", [None, True, False])
+async def test_async_dts_purge_orchestration_request(recursive: bool | None) -> None:
+    stub = MagicMock()
+    stub.PurgeInstances = AsyncMock(return_value=pb.PurgeInstancesResponse(deletedInstanceCount=3))
+
+    with patch("durabletask.client.stubs.TaskHubSidecarServiceStub", return_value=stub):
+        async with AsyncDurableTaskSchedulerClient(
+                host_address="localhost:4001", taskhub="hub",
+                token_credential=None, channel=MagicMock()) as client:
+            if recursive is None:
+                result = await client.purge_orchestration("instance")
+            else:
+                result = await client.purge_orchestration("instance", recursive=recursive)
+
+    stub.PurgeInstances.assert_awaited_once_with(pb.PurgeInstancesRequest(
+        instanceId="instance", recursive=True if recursive is None else recursive,
+        isOrchestration=True))
+    assert result == PurgeInstancesResult(deleted_instance_count=3, is_complete=None)