From b5752cd9ce41860896ca45bd329c28f7568b56b1 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Tue, 29 Sep 2026 11:56:58 -0600 Subject: [PATCH] Preserve trailing external events after continue-as-new Gate ordinary external-event delivery on the existing terminal context boundary while retaining entity routing and raw carryover buffering. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 7 + azure-functions-durable/CHANGELOG.md | 6 + durabletask-azuremanaged/CHANGELOG.md | 6 + durabletask/task.py | 6 + durabletask/worker.py | 6 +- .../test_worker_compat.py | 34 ++- .../test_orchestration_executor.py | 233 ++++++++++++++++++ 7 files changed, 295 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9a320ea7..780e70d1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,13 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## Unreleased +FIXED + +- 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 FIXED diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index 5944f4bb..3f768d10 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +FIXED + +- 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 ADDED diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md index d26c18ff..d5664fc7 100644 --- a/durabletask-azuremanaged/CHANGELOG.md +++ b/durabletask-azuremanaged/CHANGELOG.md @@ -7,6 +7,12 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## Unreleased +FIXED + +- 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 ADDED diff --git a/durabletask/task.py b/durabletask/task.py index 68ae0944..a7b6c5fb 100644 --- a/durabletask/task.py +++ b/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. """ diff --git a/durabletask/worker.py b/durabletask/worker.py index f4f31a32..9985d508 100644 --- a/durabletask/worker.py +++ b/durabletask/worker.py @@ -2843,7 +2843,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( @@ -2869,7 +2871,7 @@ def _cancel_timer() -> None: event_list.append(buffered_payload) 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: diff --git a/tests/azure-functions-durable/test_worker_compat.py b/tests/azure-functions-durable/test_worker_compat.py index dbf7d29c..2c164a37 100644 --- a/tests/azure-functions-durable/test_worker_compat.py +++ b/tests/azure-functions-durable/test_worker_compat.py @@ -15,7 +15,7 @@ import json import threading from concurrent.futures import ThreadPoolExecutor -from datetime import datetime +from datetime import datetime, timedelta from types import SimpleNamespace from unittest.mock import AsyncMock, Mock @@ -27,6 +27,7 @@ import azure.durable_functions as df from azure.durable_functions.internal import invocation, payloads from azure.durable_functions.worker import DurableFunctionsWorker +from durabletask import task from durabletask.entities import EntityInstanceId from durabletask.payload import PayloadStore @@ -502,6 +503,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): diff --git a/tests/durabletask/test_orchestration_executor.py b/tests/durabletask/test_orchestration_executor.py index 4c213403..d754a3c5 100644 --- a/tests/durabletask/test_orchestration_executor.py +++ b/tests/durabletask/test_orchestration_executor.py @@ -1857,6 +1857,239 @@ def orchestrator(ctx: task.OrchestrationContext, input: int): assert event.eventRaised.input.value == json.dumps(42 + i) +@pytest.mark.parametrize("save_events", [True, False]) +@pytest.mark.parametrize("replay_completion", [True, False]) +@pytest.mark.parametrize("event_wins", [True, False]) +@pytest.mark.parametrize("cancel_waits", [True, False]) +def test_continue_as_new_trailing_external_events( + save_events: bool, replay_completion: bool, event_wins: bool, cancel_waits: bool): + waits: list[task.CancellableTask[Any]] = [] + + def orchestrator(ctx: task.OrchestrationContext, _): + waits.extend([ctx.wait_for_external_event("EVENT"), ctx.wait_for_external_event("event")]) + timer = ctx.create_timer(timedelta(seconds=1)) + yield task.when_any([waits[0], timer]) + if cancel_waits: + for event_task in waits: + if not event_task.is_complete: + event_task.cancel() + ctx.continue_as_new(None, save_events=save_events) + + registry = worker._Registry() + name = registry.add_orchestrator(orchestrator) + start_time = datetime(2026, 1, 1) + fire_at = start_time + timedelta(seconds=1) + old_events = [ + helpers.new_orchestrator_started_event(start_time), + helpers.new_execution_started_event(name, TEST_INSTANCE_ID), + helpers.new_timer_created_event(1, fire_at), + ] + payloads = ["1", "2", None, '{ "value": 3 }'] + new_events = [ + helpers.new_event_raised_event("eVeNt", "0") if event_wins + else helpers.new_timer_fired_event(1, fire_at), + *[helpers.new_event_raised_event("eVeNt", payload) for payload in payloads], + ] + if replay_completion: + old_events.extend(new_events[:2]) + new_events = new_events[2:] + + executor = worker._OrchestrationExecutor(registry, TEST_LOGGER, JsonDataConverter()) + result = executor.execute(TEST_INSTANCE_ID, old_events, new_events) + completion = get_and_validate_complete_orchestration_action_list(1, result.actions) + + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW + assert not completion.HasField("result") + assert [ + event.eventRaised.input.value if event.eventRaised.HasField("input") else None + for event in completion.carryoverEvents + ] == (payloads if save_events else []) + assert all(event.eventRaised.name == "event" for event in completion.carryoverEvents) + if event_wins: + assert waits[0].get_result() == 0 + for event_task in waits[1:] if event_wins else waits: + assert event_task.is_cancelled == cancel_waits + assert event_task.is_complete == cancel_waits + + +@pytest.mark.parametrize("save_events", [True, False]) +def test_continue_as_new_does_not_restore_events_delivered_while_live(save_events: bool): + waits: list[task.CancellableTask[Any]] = [] + + def orchestrator(ctx: task.OrchestrationContext, _): + waits.extend([ctx.wait_for_external_event("event"), ctx.wait_for_external_event("event")]) + yield ctx.create_timer(timedelta(seconds=1)) + ctx.continue_as_new(None, save_events=save_events) + + registry = worker._Registry() + name = registry.add_orchestrator(orchestrator) + start_time = datetime(2026, 1, 1) + fire_at = start_time + timedelta(seconds=1) + old_events = [ + helpers.new_orchestrator_started_event(start_time), + helpers.new_execution_started_event(name, TEST_INSTANCE_ID), + helpers.new_timer_created_event(1, fire_at), + helpers.new_event_raised_event("event", "0"), + ] + new_events = [ + helpers.new_event_raised_event("event", "1"), + helpers.new_timer_fired_event(1, fire_at), + helpers.new_event_raised_event("event", "2"), + ] + executor = worker._OrchestrationExecutor(registry, TEST_LOGGER, JsonDataConverter()) + result = executor.execute(TEST_INSTANCE_ID, old_events, new_events) + completion = get_and_validate_complete_orchestration_action_list(1, result.actions) + + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW + # Delivery to a live wait consumes the event, even if the generator never yielded that wait. + assert [event_task.get_result() for event_task in waits] == [0, 1] + assert [event.eventRaised.input.value for event in completion.carryoverEvents] == ( + ["2"] if save_events else []) + + +@pytest.mark.parametrize("save_events", [True, False]) +def test_continue_as_new_is_terminal_before_generator_returns(save_events: bool): + resumed: list[bool] = [] + + def orchestrator(ctx: task.OrchestrationContext, _): + event_task = ctx.wait_for_external_event("event") + ctx.continue_as_new(None, save_events=save_events) + yield event_task + resumed.append(True) + + registry = worker._Registry() + name = registry.add_orchestrator(orchestrator) + events = [ + helpers.new_execution_started_event(name, TEST_INSTANCE_ID), + helpers.new_event_raised_event("event", "1"), + helpers.new_event_raised_event("event", "2"), + ] + executor = worker._OrchestrationExecutor(registry, TEST_LOGGER, JsonDataConverter()) + result = executor.execute(TEST_INSTANCE_ID, [], events) + completion = get_and_validate_complete_orchestration_action_list(1, result.actions) + + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW + assert resumed == [] + assert [event.eventRaised.input.value for event in completion.carryoverEvents] == ( + ["1", "2"] if save_events else []) + + +@pytest.mark.parametrize("terminal", ["completed", "failed", "terminated"]) +def test_terminal_external_events_do_not_complete_abandoned_waits(terminal: str): + waits: list[task.CancellableTask[Any]] = [] + resumed: list[bool] = [] + + def orchestrator(ctx: task.OrchestrationContext, _): + event_task = ctx.wait_for_external_event("event") + waits.append(event_task) + timer = ctx.create_timer(timedelta(seconds=1)) + yield task.when_any([event_task, timer]) + resumed.append(True) + if terminal == "failed": + raise ValueError("orchestrator failed") + return "done" + + registry = worker._Registry() + name = registry.add_orchestrator(orchestrator) + start_time = datetime(2026, 1, 1) + fire_at = start_time + timedelta(seconds=1) + old_events = [ + helpers.new_orchestrator_started_event(start_time), + helpers.new_execution_started_event(name, TEST_INSTANCE_ID), + helpers.new_timer_created_event(1, fire_at), + ] + new_events = [ + helpers.new_terminated_event(encoded_output=json.dumps("terminated")) if terminal == "terminated" + else helpers.new_timer_fired_event(1, fire_at), + helpers.new_event_raised_event("event", "1"), + helpers.new_event_raised_event("event", "2"), + ] + executor = worker._OrchestrationExecutor(registry, TEST_LOGGER, JsonDataConverter()) + result = executor.execute(TEST_INSTANCE_ID, old_events, new_events) + completion = get_and_validate_complete_orchestration_action_list(1, result.actions) + + assert completion.orchestrationStatus == { + "completed": pb.ORCHESTRATION_STATUS_COMPLETED, + "failed": pb.ORCHESTRATION_STATUS_FAILED, + "terminated": pb.ORCHESTRATION_STATUS_TERMINATED, + }[terminal] + assert not completion.carryoverEvents + assert not waits[0].is_complete + assert not waits[0].is_cancelled + assert resumed == ([] if terminal == "terminated" else [True]) + if terminal == "failed": + assert completion.failureDetails.errorMessage == "orchestrator failed" + else: + assert completion.result.value == json.dumps("terminated" if terminal == "terminated" else "done") + + +@pytest.mark.parametrize("continue_before_failure", [True, False]) +def test_continue_as_new_failure_keeps_existing_completion_boundary(continue_before_failure: bool): + def orchestrator(ctx: task.OrchestrationContext, _): + if continue_before_failure: + ctx.continue_as_new(None, save_events=True) + raise ValueError("orchestrator failed") + + registry = worker._Registry() + name = registry.add_orchestrator(orchestrator) + executor = worker._OrchestrationExecutor(registry, TEST_LOGGER, JsonDataConverter()) + result = executor.execute( + TEST_INSTANCE_ID, [], [helpers.new_execution_started_event(name, TEST_INSTANCE_ID)]) + completion = get_and_validate_complete_orchestration_action_list(1, result.actions) + + assert completion.orchestrationStatus == ( + pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW if continue_before_failure + else pb.ORCHESTRATION_STATUS_FAILED) + assert not completion.carryoverEvents + + +@pytest.mark.parametrize("entity_response", ["call", "lock"]) +def test_continue_as_new_entity_responses_are_not_external_events(entity_response: str): + entity_id = entities.EntityInstanceId("Counter", "key") + entity_tasks: list[task.Task[Any]] = [] + + def orchestrator(ctx: task.OrchestrationContext, _): + entity_tasks.append( + ctx.call_entity(entity_id, "get") if entity_response == "call" + else ctx.lock_entities([entity_id])) + yield ctx.create_timer(timedelta(seconds=1)) + ctx.continue_as_new(None, save_events=True) + + registry = worker._Registry() + name = registry.add_orchestrator(orchestrator) + start_time = datetime(2026, 1, 1) + fire_at = start_time + timedelta(seconds=1) + started_events = [ + helpers.new_orchestrator_started_event(start_time), + helpers.new_execution_started_event(name, TEST_INSTANCE_ID), + ] + executor = worker._OrchestrationExecutor(registry, TEST_LOGGER, JsonDataConverter()) + initial_result = executor.execute(TEST_INSTANCE_ID, [], started_events) + message = initial_result.actions[0].sendEntityMessage + request_id = (message.entityOperationCalled.requestId if entity_response == "call" + else message.entityLockRequested.criticalSectionId) + old_events = started_events + [ + helpers.new_event_sent_event(1, str(entity_id), json.dumps({"id": request_id})), + helpers.new_timer_created_event(2, fire_at), + ] + response_event = helpers.new_event_raised_event( + request_id, json.dumps({"result": "42"}) if entity_response == "call" else None) + timer_event = helpers.new_timer_fired_event(2, fire_at) + new_events = ([timer_event, response_event] if entity_response == "call" + else [response_event, timer_event]) + new_events.append(helpers.new_event_raised_event("event", "1")) + result = executor.execute(TEST_INSTANCE_ID, old_events, new_events) + completion = get_and_validate_complete_orchestration_action_list( + 1 if entity_response == "call" else 2, result.actions) + + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW + assert entity_tasks[-1].is_complete + if entity_response == "lock": + assert result.actions[0].sendEntityMessage.entityUnlockSent.criticalSectionId == request_id + assert [(event.eventRaised.name, event.eventRaised.input.value) + for event in completion.carryoverEvents] == [("event", "1")] + + def test_fan_out(): """Tests that a fan-out pattern correctly schedules N tasks""" def hello(_, name: str):