From c209731093ef8aff51e099ccd635ea5d6248eb33 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Wed, 7 Oct 2026 06:41:55 +0000 Subject: [PATCH] Preserve pending scalar operations during signal replay --- CHANGELOG.md | 10 ++ docs/sdk-reference.md | 15 ++ pyproject.toml | 6 +- src/durable_workflow/workflow.py | 13 ++ .../pending-timer-after-signal-avro.json | 52 ++++++ tests/test_child_workflow_policies.py | 7 +- ...ommitted_cancellation_scope_descendants.py | 3 +- ...est_committed_cancellation_scope_groups.py | 3 +- ...est_committed_cancellation_scope_replay.py | 5 +- tests/test_pending_operation_replay.py | 154 ++++++++++++++++++ tests/test_replay_regression_corpus.py | 2 + tests/test_update_signal_condition_replay.py | 17 +- 12 files changed, 270 insertions(+), 17 deletions(-) create mode 100644 tests/fixtures/replay_regressions/pending-timer-after-signal-avro.json create mode 100644 tests/test_pending_operation_replay.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 470d012..f11db86 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [2.4.1] - 2026-10-07 + +### Fixed +- Keep already scheduled scalar timers, remote activities and child workflows + pending during signal wakeups and cold worker replay. Replay no longer emits + another scheduling command for the same recorded operation, including + shielded cancellation cleanup timers. +- Reject changed timer delays when history records the original delay, while + preserving compatibility with older histories that omit it. + ## [2.3.9] - 2026-10-04 ### Fixed diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index ae538aa..c2cafae 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -506,6 +506,21 @@ constraining throughput. ## Replay captured histories +Signals can update workflow state while a timer, remote activity or child is +pending. Replaying that state waits for the recorded operation. It keeps the +original timer deadline and does not schedule another activity or child. + +Python SDK 2.4.1 fixes duplicate scheduling during these wakeups. Upgrade every +Python worker on an affected queue. For a run created by an older worker, +inspect its history for extra `TimerScheduled`, `ActivityScheduled` or +`ChildWorkflowScheduled` events before the original operation completed. +The fix prevents new duplicates and does not rewrite existing history. An +extra completed timer can still produce `NonDeterministicReplayError` when +replay reaches the next authored operation. Capture that history with the +workflow code and reconcile any activity or child side effects before +terminating and replacing an affected run. Do not delete history rows or skip +recorded operations to force replay to succeed. + Use `Replayer` to debug a captured history without connecting to a live server: ```python diff --git a/pyproject.toml b/pyproject.toml index 58cc843..0347da4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "durable-workflow" -version = "2.4.0" +version = "2.4.1" description = "Python client and worker SDK for Durable Workflow Cloud and self-hosted Server" readme = "README.md" requires-python = ">=3.10" @@ -71,8 +71,8 @@ durable-workflow-replay-conformance = "durable_workflow.replay_conformance:main" durable-workflow-workflow-updates-conformance = "durable_workflow.workflow_updates_conformance:main" [tool.durable-workflow] -product-train = "2.4.0" -registry-version = "2.4.0" +product-train = "2.4.1" +registry-version = "2.4.1" supported-server-versions = "2.5.0" worker-protocol-version = "1.19" control-plane-version = "2" diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 4dee23e..8a590d2 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -3624,6 +3624,8 @@ def _activity_type_from_payload(payload: Mapping[str, Any]) -> str | None: def _recorded_step_details(payload: Mapping[str, Any]) -> dict[str, Any]: details: dict[str, Any] = {} + if "delay_seconds" in payload: + details["delay_seconds"] = payload["delay_seconds"] if "cancellation_cleanup" in payload: details["cancellation_cleanup"] = payload["cancellation_cleanup"] local_preparation = payload.get("local_preparation") @@ -3846,6 +3848,12 @@ def _recorded_detail_mismatch(command: Any, step: _RecordedStep) -> str | None: f"child_workflow_policy_changed: recorded {field} {recorded_policy!r}, " f"but current workflow requested {actual_policy!r}." ) + elif isinstance(command, StartTimer): + if "delay_seconds" in step.details and step.details["delay_seconds"] != command.delay_seconds: + return ( + f"Recorded delay_seconds {step.details['delay_seconds']!r}, but current workflow " + f"requested {command.delay_seconds!r}." + ) elif isinstance(command, RecordVersionMarker): recorded = step.details.get("change_id") if isinstance(recorded, str) and recorded != command.change_id: @@ -6667,6 +6675,11 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None continue ctx.logger._set_replaying(False) _assert_pending_step_matches(cmd) + if not isinstance(cmd, RecordLocalActivity) and _next_unconsumed_recorded_step() is not None: + # A signal may wake a workflow while its existing operation is + # still pending. Reissuing it allocates a new durable sequence + # and can move a timer's deadline or duplicate external work. + return _state(pending) if isinstance(cmd, RecordLocalActivity) and prepare_local_activities: return _state(pending, _prepared_call(cmd, current_call_sequence)) if isinstance(cmd, RecordLocalActivity) and local_activity_executor is not None: diff --git a/tests/fixtures/replay_regressions/pending-timer-after-signal-avro.json b/tests/fixtures/replay_regressions/pending-timer-after-signal-avro.json new file mode 100644 index 0000000..0c539b2 --- /dev/null +++ b/tests/fixtures/replay_regressions/pending-timer-after-signal-avro.json @@ -0,0 +1,52 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "pending-timer-after-signal-avro", + "protocol_version": "1.19", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.pending-operation", + "input": [ + "timer" + ], + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "ActivityCompleted", + "payload": { + "sequence": 1, + "activity_type": "echo", + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoMc2FtcGxl" + } + } + }, + { + "event_type": "TimerScheduled", + "payload": { + "sequence": 2, + "timer_id": "original-timer", + "delay_seconds": 120, + "fire_at": "2026-10-07T06:02:00Z" + } + }, + { + "event_type": "SignalReceived", + "payload": { + "signal_name": "increment", + "arguments": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwCBA4A" + }, + "payload_codec": "avro" + } + } + ], + "expected": { + "command_sequence": [] + } +} diff --git a/tests/test_child_workflow_policies.py b/tests/test_child_workflow_policies.py index 8fb1250..b07db0a 100644 --- a/tests/test_child_workflow_policies.py +++ b/tests/test_child_workflow_policies.py @@ -104,9 +104,8 @@ def test_existing_positional_constructor_and_omitted_wire_fields_are_preserved() result = replay(workflow("sequential", { "parent_close_policy": ParentClosePolicy.ABANDON, "cancellation_policy": CancellationPolicy.ABANDON, }), scheduled([StartChildWorkflow("child", ["argument"])]), []) - assert len(result.commands) == 1 - assert isinstance(result.commands[0], StartChildWorkflow) - assert result.commands[0].cancellation_policy == CancellationPolicy.ABANDON + # Omitted historical fields retain their defaults, without starting another child. + assert result.commands == [] @pytest.mark.parametrize("mode", ["sequential", "parallel", "selection"]) @@ -120,7 +119,7 @@ def test_changed_policies_fail_cold_replay_and_cancellation_delivery(mode: str, assert replay(workflow(mode, policies), history, [], run_id="run-1").commands == [CompleteWorkflow("request-1")] else: cold = replay(workflow(mode, policies), history, []) - if mode == "selection": + if mode in ("sequential", "selection"): assert cold.commands == [] else: assert scheduled(cold.commands) == history diff --git a/tests/test_committed_cancellation_scope_descendants.py b/tests/test_committed_cancellation_scope_descendants.py index f8e28c9..e7e69ca 100644 --- a/tests/test_committed_cancellation_scope_descendants.py +++ b/tests/test_committed_cancellation_scope_descendants.py @@ -227,7 +227,8 @@ def test_cleanup_timer_retains_original_ancestor_receipt_and_each_scope_authorit "2026-10-04T00:00:10.123456Z", ) value["task"].update({"lease_owner": "replacement", "workflow_task_attempt": 17}) - assert workflow.commands_to_server_commands(run(cls, value).commands, "queue") == fresh + assert run(cls, value).commands == [] + assert run(cls, value).commands == [] append_cleanup_event( value, "TimerFired", diff --git a/tests/test_committed_cancellation_scope_groups.py b/tests/test_committed_cancellation_scope_groups.py index 3f4a4e2..a9b1437 100644 --- a/tests/test_committed_cancellation_scope_groups.py +++ b/tests/test_committed_cancellation_scope_groups.py @@ -217,7 +217,8 @@ def test_group_cleanup_timer_replays_original_receipt_and_narrower_authority(lay "cancellation_cleanup": snapshot, }, "2026-10-04T00:00:09.123456Z") value["task"].update({"workflow_task_attempt": 23, "lease_owner": "replacement"}) - assert workflow.commands_to_server_commands(run(cls, value).commands, "queue") == first + assert run(cls, value).commands == [] + assert run(cls, value).commands == [] append_cleanup_event(value, "TimerFired", { "sequence": 8, "timer_id": "cleanup-timer", "delay_seconds": 1, "cancellation_scope_id": original.scope_id, diff --git a/tests/test_committed_cancellation_scope_replay.py b/tests/test_committed_cancellation_scope_replay.py index 1f11b8e..aae70dc 100644 --- a/tests/test_committed_cancellation_scope_replay.py +++ b/tests/test_committed_cancellation_scope_replay.py @@ -76,7 +76,10 @@ def outer(): # type: ignore[no-untyped-def] field: snapshot[field] for field in ("scope_id", "request_id", "delivery_history_event_id") } assert wire[0]["cancellation_scope_id"] == snapshot["scope_id"] - assert workflow.commands_to_server_commands(run(Probe, value).commands, "queue") == wire + # The original cleanup timer and its authority are already durable. + # Cold replay waits for it rather than starting a second timer. + assert run(Probe, value).commands == [] + assert run(Probe, value).commands == [] append_cleanup_event(value, "TimerFired", { "sequence": 4, "timer_id": "cleanup-timer", "delay_seconds": 1, "cancellation_scope_id": snapshot["scope_id"], diff --git a/tests/test_pending_operation_replay.py b/tests/test_pending_operation_replay.py new file mode 100644 index 0000000..ba9353e --- /dev/null +++ b/tests/test_pending_operation_replay.py @@ -0,0 +1,154 @@ +"""Scheduled scalar work stays the same operation when signals wake replay.""" +from __future__ import annotations + +from typing import Any + +import pytest + +from durable_workflow import serializer, workflow +from durable_workflow.errors import NonDeterministicReplayError +from durable_workflow.workflow import CompleteWorkflow, WorkflowContext, query_state, replay + + +@workflow.defn(name="tests.replay.pending-operation") +class PendingOperation: + def __init__(self) -> None: + self.total = 0 + self.released = False + + @workflow.signal("increment") + def increment(self, amount: int) -> None: + self.total += amount + + @workflow.signal("release") + def release(self) -> None: + self.released = True + + @workflow.query("total") + def current(self) -> int: + return self.total + + def run(self, ctx: WorkflowContext, kind: str): # type: ignore[no-untyped-def] + first = yield ctx.schedule_activity("echo", ["sample"]) + if kind == "timer": + yield ctx.sleep(120) + elif kind == "activity": + yield ctx.schedule_activity("work", [first]) + else: + yield ctx.start_child_workflow("child", [first]) + yield ctx.wait_condition(lambda: self.released, key="release") + return self.total + + +def initial_history() -> list[dict[str, Any]]: + return [ + {"event_type": "ActivityCompleted", "payload": { + "sequence": 1, "activity_type": "echo", "result": serializer.envelope("sample"), + }}, + ] + + +OPENINGS = { + "timer": {"event_type": "TimerScheduled", "payload": { + "sequence": 2, "timer_id": "original-timer", "delay_seconds": 120, + "fire_at": "2026-10-07T06:02:00Z", + }}, + "activity": {"event_type": "ActivityScheduled", "payload": { + "sequence": 2, "activity_type": "work", "activity_execution_id": "original-activity", + }}, + "child": {"event_type": "ChildWorkflowScheduled", "payload": { + "sequence": 2, "workflow_type": "child", "child_workflow_instance_id": "original-child", + }}, +} + + +@pytest.mark.parametrize("kind", OPENINGS) +def test_signal_replay_does_not_reschedule_pending_operation(kind: str) -> None: + history = initial_history() + [OPENINGS[kind]] + for amount in (7, 11): + history.append({"event_type": "SignalReceived", "payload": { + "signal_name": "increment", "arguments": serializer.envelope([amount]), + }}) + # Each call constructs a fresh workflow instance, as a replacement worker does. + outcome = replay(PendingOperation, history, [kind]) + assert outcome.commands == [] + assert query_state(PendingOperation, history, [kind], "total") == sum( + serializer.decode_envelope(event["payload"]["arguments"])[0] + for event in history if event["event_type"] == "SignalReceived" + ) + + +@pytest.mark.parametrize("kind", OPENINGS) +def test_new_operation_is_still_scheduled(kind: str) -> None: + outcome = replay(PendingOperation, initial_history(), [kind]) + assert len(outcome.commands) == 1 + assert outcome.commands[0].to_server_command("queue")["type"] == { + "timer": "start_timer", "activity": "schedule_activity", "child": "start_child_workflow", + }[kind] + + +def test_original_timer_completion_consumes_signals_once() -> None: + history = initial_history() + [OPENINGS["timer"], + {"event_type": "SignalReceived", "payload": { + "signal_name": "increment", "arguments": serializer.envelope([7]), + }}, + {"event_type": "TimerFired", "payload": {"sequence": 2, "timer_id": "original-timer"}}, + {"event_type": "SignalReceived", "payload": { + "signal_name": "release", "arguments": serializer.envelope([]), + }}, + ] + outcome = replay(PendingOperation, history, ["timer"]) + assert len(outcome.commands) == 1 + assert isinstance(outcome.commands[0], CompleteWorkflow) + assert outcome.commands[0].result == 7 + + +def test_pending_timer_still_rejects_changed_delay() -> None: + changed = {"event_type": "TimerScheduled", "payload": {**OPENINGS["timer"]["payload"], "delay_seconds": 121}} + with pytest.raises(NonDeterministicReplayError) as captured: + replay(PendingOperation, initial_history() + [changed], ["timer"]) + assert captured.value.workflow_sequence == 2 + assert captured.value.recorded_event_types == ["TimerScheduled"] + + +@pytest.mark.parametrize("kind", ["activity", "child"]) +def test_original_external_operation_completion_consumes_signals_once(kind: str) -> None: + terminal = { + "event_type": "ActivityCompleted" if kind == "activity" else "ChildRunCompleted", + "payload": {**OPENINGS[kind]["payload"], "result": serializer.envelope("original-result")}, + } + history = initial_history() + [OPENINGS[kind], + {"event_type": "SignalReceived", "payload": { + "signal_name": "increment", "arguments": serializer.envelope([7]), + }}, terminal, + {"event_type": "SignalReceived", "payload": { + "signal_name": "release", "arguments": serializer.envelope([]), + }}, + ] + assert replay(PendingOperation, history, [kind]).commands == [CompleteWorkflow(7)] + + +@pytest.mark.parametrize("kind", OPENINGS) +def test_recorded_operation_cannot_be_replaced_by_a_different_kind(kind: str) -> None: + different = "activity" if kind == "timer" else "timer" + with pytest.raises(NonDeterministicReplayError) as captured: + replay(PendingOperation, initial_history() + [OPENINGS[kind]], [different]) + assert captured.value.workflow_sequence == 2 + + +def test_legacy_timer_without_recorded_delay_still_waits_for_original_operation() -> None: + legacy = {"event_type": "TimerScheduled", "payload": {"sequence": 2, "timer_id": "original-timer"}} + assert replay(PendingOperation, initial_history() + [legacy], ["timer"]).commands == [] + + +def test_already_duplicated_timers_are_not_silently_skipped() -> None: + duplicate = {"event_type": "TimerScheduled", "payload": { + **OPENINGS["timer"]["payload"], "sequence": 3, "timer_id": "duplicate-timer", + }} + history = initial_history() + [OPENINGS["timer"], duplicate, + {"event_type": "TimerFired", "payload": {"sequence": 2, "timer_id": "original-timer"}}, + {"event_type": "TimerFired", "payload": {"sequence": 3, "timer_id": "duplicate-timer"}}, + ] + with pytest.raises(NonDeterministicReplayError) as captured: + replay(PendingOperation, history, ["timer"]) + assert captured.value.workflow_sequence == 3 diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index 6e39eef..92731ef 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -24,6 +24,7 @@ GoldenTimeoutWaitWorkflow, GoldenVersionMarkerWorkflow, ) +from tests.test_pending_operation_replay import PendingOperation from tests.test_update_signal_condition_replay import ( PostConditionReceiversWorkflow, UpdateSignalConditionTimerWorkflow, @@ -265,6 +266,7 @@ def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] NestedParallelPathWorkflow, ParallelMetadataProducerWorkflow, ParallelResultBindingWorkflow, + PendingOperation, PagedRecordedSideEffectsWorkflow, PostConditionReceiversWorkflow, RecordedSideEffectWorkflow, diff --git a/tests/test_update_signal_condition_replay.py b/tests/test_update_signal_condition_replay.py index b65431a..450243a 100644 --- a/tests/test_update_signal_condition_replay.py +++ b/tests/test_update_signal_condition_replay.py @@ -460,13 +460,13 @@ def test_every_history_prefix_cold_replays_each_workflow_sequence_once(self) -> expected_commands = [ StartTimer, StartTimer, - StartTimer, - ScheduleActivity, - ScheduleActivity, - ScheduleActivity, - ScheduleActivity, + None, ScheduleActivity, + None, + None, ScheduleActivity, + None, + None, WaitCondition, WaitCondition, WaitCondition, @@ -475,8 +475,8 @@ def test_every_history_prefix_cold_replays_each_workflow_sequence_once(self) -> WaitCondition, ScheduleActivity, ScheduleActivity, - ScheduleActivity, - ScheduleActivity, + None, + None, WaitCondition, WaitCondition, WaitCondition, @@ -492,6 +492,9 @@ def test_every_history_prefix_cold_replays_each_workflow_sequence_once(self) -> [], ) + if expected_command is None: + assert outcome.commands == [], prefix_length + continue assert len(outcome.commands) == 1, prefix_length assert isinstance(outcome.commands[0], expected_command), prefix_length if prefix_length in (15, 16, 17, 18):