Prevent timer callbacks after orchestration completion (#280)

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>

Author: andystaples

CHANGELOG.md

@@ -9,6 +9,10 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 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.
 - `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.

azure-functions-durable/CHANGELOG.md

@@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 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.
 - 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.

durabletask-azuremanaged/CHANGELOG.md

@@ -9,6 +9,11 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 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.
 - 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.

durabletask/worker.py

@@ -2576,6 +2576,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()

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):

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, _):