From 4a5e6b35c8b83bda79acc88dee79a06006f75002 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Tue, 29 Sep 2026 11:56:57 -0600 Subject: [PATCH] Prevent timer callbacks after orchestration completion Guard timer continuation at the executor callback boundary while preserving pre-terminal actions and trailing history processing. Cover terminal states, replay, native/chunked retries, cancellation, and Functions compatibility. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 7 + azure-functions-durable/CHANGELOG.md | 7 + durabletask-azuremanaged/CHANGELOG.md | 8 + durabletask/worker.py | 4 + .../test_worker_compat.py | 42 +++- .../test_orchestration_executor.py | 224 +++++++++++++++++- 6 files changed, 288 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9a320ea7..0cf01641 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 + +- 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. + ## v1.11.0 FIXED diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index 5944f4bb..476b92a7 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +FIXED + +- 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. + ## v2.0.0rc2 ADDED diff --git a/durabletask-azuremanaged/CHANGELOG.md b/durabletask-azuremanaged/CHANGELOG.md index d26c18ff..73ea2d3e 100644 --- a/durabletask-azuremanaged/CHANGELOG.md +++ b/durabletask-azuremanaged/CHANGELOG.md @@ -7,6 +7,14 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## Unreleased +FIXED + +- 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 +long timers without chunking; the core long-timer chunking correction does not +change that behavior. + ## v1.11.0 ADDED diff --git a/durabletask/worker.py b/durabletask/worker.py index f4f31a32..c04e40e6 100644 --- a/durabletask/worker.py +++ b/durabletask/worker.py @@ -2571,6 +2571,10 @@ def process_event( scheduled_time_ns=created_ns, parent_trace_context=ctx._orchestration_trace_context or ctx._parent_trace_context, # pyright: ignore[reportPrivateUsage] ) + # A prior event in this batch may have ended the orchestration. + # Do not schedule another chunk, retry work, or resume user code. + if ctx._is_complete: # pyright: ignore[reportPrivateUsage] + return next_fire_at = timer_task._handle_timer_fired(event.timerFired.fireAt.ToDatetime()) # pyright: ignore[reportPrivateUsage] if next_fire_at is not None: id = ctx.next_sequence_number() diff --git a/tests/azure-functions-durable/test_worker_compat.py b/tests/azure-functions-durable/test_worker_compat.py index dbf7d29c..d39c2c38 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 @@ -433,6 +434,45 @@ def orchestrator(context): assert json.loads(completion.result.value) == {"echo": {"n": 5}} +@pytest.mark.parametrize("native", [False, True]) +def test_long_timer_does_not_schedule_chunk_after_completion(native): + def compatible_orchestrator(context): + done = context.wait_for_external_event("done") + timeout = context.create_timer(context.current_utc_datetime + timedelta(days=10)) + yield context.task_any([done, timeout]) + return "done" + + def native_orchestrator(context: task.OrchestrationContext, _): + done = context.wait_for_external_event("done") + timeout = context.create_timer(timedelta(days=10)) + yield task.when_any([done, timeout]) + return "done" + + start = datetime(2020, 1, 1) + fire_at = start + timedelta(days=3) + request = pb.OrchestratorRequest( + instanceId=TEST_INSTANCE_ID, + pastEvents=[ + helpers.new_orchestrator_started_event(start), + helpers.new_execution_started_event("timer-race", TEST_INSTANCE_ID), + helpers.new_timer_created_event(1, fire_at), + ], + newEvents=[ + helpers.new_event_raised_event("done", json.dumps(True)), + helpers.new_timer_fired_event(1, fire_at), + ], + ) + encoded = base64.b64encode(request.SerializeToString()).decode("utf-8") + orchestrator = native_orchestrator if native else compatible_orchestrator + result = DurableFunctionsWorker().execute_orchestration_request(orchestrator, encoded) + + response = _decode_orchestrator_response(result) + assert len(response.actions) == 1 + completion = _get_completion_action(response) + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED + assert json.loads(completion.result.value) == "done" + + def test_execute_orchestration_request_registers_under_event_name(): """The orchestrator is registered under the name from the ExecutionStarted event.""" def orchestrator(context): diff --git a/tests/durabletask/test_orchestration_executor.py b/tests/durabletask/test_orchestration_executor.py index 4c213403..7f34597c 100644 --- a/tests/durabletask/test_orchestration_executor.py +++ b/tests/durabletask/test_orchestration_executor.py @@ -201,7 +201,8 @@ def delay_orchestrator(ctx: task.OrchestrationContext, _): assert actions[0].createTimer.fireAt.ToDatetime() == expected_fire_at -def test_timer_fired_completion(): +@pytest.mark.parametrize("maximum_timer_interval", [timedelta(days=3), None]) +def test_timer_fired_completion(maximum_timer_interval): """Tests the resumption of task using a timer_fired event""" def delay_orchestrator(ctx: task.OrchestrationContext, _): @@ -222,7 +223,10 @@ def delay_orchestrator(ctx: task.OrchestrationContext, _): new_events = [ helpers.new_timer_fired_event(1, expected_fire_at)] - executor = worker._OrchestrationExecutor(registry, TEST_LOGGER, JsonDataConverter()) + executor = worker._OrchestrationExecutor( + registry, TEST_LOGGER, JsonDataConverter(), + maximum_timer_interval=maximum_timer_interval, + ) result = executor.execute(TEST_INSTANCE_ID, old_events, new_events) actions = result.actions @@ -352,6 +356,143 @@ def orchestrator(ctx: task.OrchestrationContext, _): assert complete_action.result.value == '"done"' +@pytest.mark.parametrize("terminal_state", [ + "completed", "failed", "invalid_result", "terminated", "continued_as_new", +]) +@pytest.mark.parametrize("replay_terminal_event", [False, True]) +def test_long_timer_does_not_continue_after_terminal_state(terminal_state, replay_terminal_event): + def orchestrator(ctx: task.OrchestrationContext, _): + done = ctx.wait_for_external_event("done") + timeout = ctx.create_timer(timedelta(days=10)) + yield task.when_any([done, timeout]) + if terminal_state == "failed": + raise ValueError("orchestrator failed") + if terminal_state == "invalid_result": + return object() + if terminal_state == "continued_as_new": + ctx.continue_as_new("next", save_events=True) + return "done" + + registry = worker._Registry() + name = registry.add_orchestrator(orchestrator) + executor = worker._OrchestrationExecutor(registry, TEST_LOGGER, JsonDataConverter()) + start = datetime(2020, 1, 1) + fire_at = start + timedelta(days=3) + old_events = [ + helpers.new_orchestrator_started_event(start), + helpers.new_execution_started_event(name, TEST_INSTANCE_ID, encoded_input=None), + helpers.new_timer_created_event(1, fire_at), + ] + terminal_event = ( + helpers.new_terminated_event(encoded_output=json.dumps("terminated")) + if terminal_state == "terminated" + else helpers.new_event_raised_event("done", json.dumps(True)) + ) + new_events = [helpers.new_timer_fired_event(1, fire_at)] + if replay_terminal_event: + old_events.append(terminal_event) + else: + new_events.insert(0, terminal_event) + new_events.append(helpers.new_event_raised_event("carryover", json.dumps("saved"))) + + result = executor.execute(TEST_INSTANCE_ID, old_events, new_events) + + completion = get_and_validate_complete_orchestration_action_list(1, result.actions) + if terminal_state in ("failed", "invalid_result"): + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_FAILED + assert completion.HasField("failureDetails") + elif terminal_state == "terminated": + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_TERMINATED + assert completion.result.value == json.dumps("terminated") + elif terminal_state == "continued_as_new": + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW + assert completion.result.value == json.dumps("next") + assert len(completion.carryoverEvents) == 1 + assert completion.carryoverEvents[0].eventRaised.name == "carryover" + assert completion.carryoverEvents[0].eventRaised.input.value == json.dumps("saved") + else: + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED + assert completion.result.value == json.dumps("done") + + +@pytest.mark.parametrize("terminal_state", ["terminated", "continued_as_new"]) +def test_final_timer_callback_does_not_resume_terminal_orchestrator(terminal_state): + def orchestrator(ctx: task.OrchestrationContext, _): + done = ctx.wait_for_external_event("done") + timeout = ctx.create_timer(timedelta(hours=1)) + yield task.when_any([done, timeout]) + if terminal_state == "continued_as_new": + ctx.continue_as_new("next") + yield timeout + ctx.send_event("target", "unexpected") + + registry = worker._Registry() + name = registry.add_orchestrator(orchestrator) + executor = worker._OrchestrationExecutor(registry, TEST_LOGGER, JsonDataConverter()) + start = datetime(2020, 1, 1) + fire_at = start + timedelta(hours=1) + old_events = [ + helpers.new_orchestrator_started_event(start), + helpers.new_execution_started_event(name, TEST_INSTANCE_ID, encoded_input=None), + helpers.new_timer_created_event(1, fire_at), + ] + new_events = [ + helpers.new_terminated_event(encoded_output=None) + if terminal_state == "terminated" + else helpers.new_event_raised_event("done", json.dumps(True)), + helpers.new_timer_fired_event(1, fire_at), + ] + + result = executor.execute(TEST_INSTANCE_ID, old_events, new_events) + + completion = get_and_validate_complete_orchestration_action_list(1, result.actions) + expected_status = ( + pb.ORCHESTRATION_STATUS_TERMINATED + if terminal_state == "terminated" + else pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW + ) + assert completion.orchestrationStatus == expected_status + + +@pytest.mark.parametrize("cancel_timer", [False, True]) +def test_long_timer_preserves_work_scheduled_before_completion(cancel_timer): + def orchestrator(ctx: task.OrchestrationContext, _): + done = ctx.wait_for_external_event("done") + timeout = ctx.create_timer(timedelta(days=10)) + yield task.when_any([done, timeout]) + if cancel_timer: + assert timeout.cancel() + ctx.send_event("target", "done", data=True) + return "done" + + registry = worker._Registry() + name = registry.add_orchestrator(orchestrator) + executor = worker._OrchestrationExecutor(registry, TEST_LOGGER, JsonDataConverter()) + start = datetime(2020, 1, 1) + fire_at = start + timedelta(days=3) + old_events = [ + helpers.new_orchestrator_started_event(start), + helpers.new_execution_started_event(name, TEST_INSTANCE_ID, encoded_input=None), + helpers.new_timer_created_event(1, fire_at), + ] + new_events = [ + helpers.new_timer_fired_event(1, fire_at), + helpers.new_event_raised_event("done", json.dumps(True)), + ] + + result = executor.execute(TEST_INSTANCE_ID, old_events, new_events) + + expected_actions = ["sendEvent", "completeOrchestration"] + if not cancel_timer: + expected_actions.insert(0, "createTimer") + assert result.actions[0].id == 2 + assert result.actions[0].createTimer.fireAt.ToDatetime() == start + timedelta(days=6) + assert [action.WhichOneof("orchestratorActionType") for action in result.actions] == expected_actions + completion = result.actions[-1].completeOrchestration + assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED + assert completion.result.value == json.dumps("done") + + def test_long_timer_can_be_cancelled_after_when_any_winner(): """Tests cancellation of a long timer after an external event wins when_any.""" @@ -393,7 +534,10 @@ def orchestrator(ctx: task.OrchestrationContext, _): second = executor.execute( TEST_INSTANCE_ID, old_events, - [helpers.new_event_raised_event("approval", json.dumps(True))], + [ + helpers.new_event_raised_event("approval", json.dumps(True)), + helpers.new_timer_fired_event(1, first_chunk_fire_at), + ], ) complete_action = get_and_validate_complete_orchestration_action_list(1, second.actions) assert complete_action.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED @@ -1246,6 +1390,80 @@ def orchestrator(ctx: task.OrchestrationContext, orchestrator_input): assert actions[-1].id == 1 +@pytest.mark.parametrize("terminal_state", ["active", "completed", "terminated", "continued_as_new"]) +@pytest.mark.parametrize("is_sub_orchestration", [False, True]) +@pytest.mark.parametrize("retry_delay", [timedelta(hours=1), timedelta(days=10)]) +@pytest.mark.parametrize("maximum_timer_interval", [timedelta(days=3), None]) +def test_retry_timer_callback_respects_terminal_state( + terminal_state, is_sub_orchestration, retry_delay, maximum_timer_interval, +): + def orchestrator(ctx: task.OrchestrationContext, _): + retry_policy = task.RetryPolicy( + first_retry_interval=retry_delay, max_number_of_attempts=2, + ) + if is_sub_orchestration: + work = ctx.call_sub_orchestrator("child", instance_id="child-instance", retry_policy=retry_policy) + else: + work = ctx.call_activity("activity", retry_policy=retry_policy) + done = ctx.wait_for_external_event("done") + yield task.when_any([done, work]) + if terminal_state == "continued_as_new": + ctx.continue_as_new("next") + return "done" + + registry = worker._Registry() + name = registry.add_orchestrator(orchestrator) + executor = worker._OrchestrationExecutor( + registry, TEST_LOGGER, JsonDataConverter(), + maximum_timer_interval=maximum_timer_interval, + ) + start = datetime(2020, 1, 1) + interval = min(retry_delay, maximum_timer_interval) if maximum_timer_interval else retry_delay + fire_at = start + interval + old_events = [ + helpers.new_orchestrator_started_event(start), + helpers.new_execution_started_event(name, TEST_INSTANCE_ID, encoded_input=None), + ] + if is_sub_orchestration: + old_events.extend([ + helpers.new_sub_orchestration_created_event(1, "child", "child-instance"), + helpers.new_sub_orchestration_failed_event(1, ValueError("retry me")), + ]) + else: + old_events.extend([ + helpers.new_task_scheduled_event(1, "activity"), + helpers.new_task_failed_event(1, ValueError("retry me")), + ]) + old_events.append(helpers.new_timer_created_event(2, fire_at)) + new_events = [] + if terminal_state == "terminated": + new_events.append(helpers.new_terminated_event(encoded_output=None)) + elif terminal_state != "active": + new_events.append(helpers.new_event_raised_event("done", json.dumps(True))) + new_events.append(helpers.new_timer_fired_event(2, fire_at)) + + result = executor.execute(TEST_INSTANCE_ID, old_events, new_events) + + assert len(result.actions) == 1 + if terminal_state == "active": + action = result.actions[0] + if interval < retry_delay: + assert action.HasField("createTimer") + assert action.id == 3 + assert action.createTimer.fireAt.ToDatetime() == start + interval * 2 + else: + assert action.HasField("createSubOrchestration" if is_sub_orchestration else "scheduleTask") + assert action.id == 1 + else: + completion = get_and_validate_complete_orchestration_action_list(1, result.actions) + expected_status = { + "completed": pb.ORCHESTRATION_STATUS_COMPLETED, + "terminated": pb.ORCHESTRATION_STATUS_TERMINATED, + "continued_as_new": pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW, + }[terminal_state] + assert completion.orchestrationStatus == expected_status + + def test_nondeterminism_expected_timer(): """Tests the non-determinism detection logic when call_timer is expected but some other method (call_activity) is called instead""" def dummy_activity(ctx, _):