diff --git a/CHANGELOG.md b/CHANGELOG.md index 7495f174..de1f9418 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,10 @@ 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 diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index 5bc3ccda..76626cb8 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -20,6 +20,9 @@ continue-as-new. This applies to both native and compatibility orchestration API - 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 diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md index ae70066d..425792fb 100644 --- a/durabletask-azuremanaged/CHANGELOG.md +++ b/durabletask-azuremanaged/CHANGELOG.md @@ -20,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 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 752a6585..25010a17 100644 --- a/durabletask/worker.py +++ b/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: diff --git a/tests/azure-functions-durable/test_worker_compat.py b/tests/azure-functions-durable/test_worker_compat.py index d39c2c38..5e20f3a2 100644 --- a/tests/azure-functions-durable/test_worker_compat.py +++ b/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): diff --git a/tests/durabletask/test_orchestration_executor.py b/tests/durabletask/test_orchestration_executor.py index 697006fc..29c49fc3 100644 --- a/tests/durabletask/test_orchestration_executor.py +++ b/tests/durabletask/test_orchestration_executor.py @@ -2227,6 +2227,300 @@ def orchestrator(ctx: task.OrchestrationContext, _): assert converter.serialized == [json.loads(consumed_payload) if consumed_payload else None] +@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]) +@pytest.mark.parametrize("replay_completion", [True, False]) +def test_continue_as_new_trailing_events_preserve_global_order( + save_events: bool, replay_completion: bool): + waits: list[task.CancellableTask[Any]] = [] + + def orchestrator(ctx: task.OrchestrationContext, input: str | None): + if input == "next": + received: list[Any] = [] + for event_name, _ in expected: + received.append((yield ctx.wait_for_external_event(event_name))) + return received + waits.extend([ctx.wait_for_external_event("A"), ctx.wait_for_external_event("b")]) + yield ctx.create_timer(timedelta(seconds=1)) + ctx.continue_as_new("next", 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) + history = [ + 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("C", "0"), + helpers.new_timer_fired_event(1, fire_at), + ] + expected = [("c", "0"), ("a", "1"), ("b", None), ("a", "null"), ("c", '{ "value": 2 }')] + trailing_events = [ + helpers.new_event_raised_event(event_name.upper(), payload) + for event_name, payload in expected[1:] + ] + old_events = history + trailing_events[:1] if replay_completion else history[:-1] + new_events = trailing_events[1:] if replay_completion else history[-1:] + trailing_events + 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 [ + (event.eventRaised.name, + event.eventRaised.input.value if event.eventRaised.HasField("input") else None) + for event in completion.carryoverEvents + ] == (expected if save_events else []) + assert all(not event_task.is_complete for event_task in waits) + if save_events: + next_execution = [ + helpers.new_orchestrator_started_event(start_time), + helpers.new_execution_started_event( + name, TEST_INSTANCE_ID, encoded_input=completion.result.value), + *completion.carryoverEvents, + ] + next_result = executor.execute(TEST_INSTANCE_ID, [], next_execution) + next_completion = get_and_validate_complete_orchestration_action_list(1, next_result.actions) + assert next_completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED + assert json.loads(next_completion.result.value) == [ + json.loads(payload) if payload is not None else None for _, payload in expected + ] + assert not next_completion.carryoverEvents + + +@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):