From 7a0bd6c3125fca9262244b406f40b91f0e50b2fd Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Wed, 2 Sep 2026 22:42:58 +0000 Subject: [PATCH 1/2] Keep post-condition receivers on their replay boundary --- CHANGELOG.md | 4 + src/durable_workflow/workflow.py | 5 + tests/test_update_signal_condition_replay.py | 129 +++++++++++++++++++ 3 files changed, 138 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index a30eac8..0eae35c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -91,6 +91,10 @@ Earlier SDK versions remain historical releases and are not alternate supported `NamespaceDescription.deleted`. ### Fixed +- Cold replay now treats an intervening durable step as the authoritative + boundary for later signals and updates, even when they retain an older + condition sequence. Post-condition activities can therefore complete before + those receivers advance the workflow without rebinding them to a closed wait. - The README and PyPI project description now require Server worker protocol `>=1.19,<2.0`. Package and API-reference release validation derive that interval from the shipped SDK protocol and reject semantic guidance drift. diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index d7366c3..75f9937 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -3925,6 +3925,11 @@ def _receiver_condition_wait_bindings() -> dict[int, str | None]: if current_wait_id is None: if prefix_can_bind_to_first_wait or explicit_sequence is None: prefix_receivers.append(index) + else: + # A durable step separated this receiver from the last + # condition. Record that boundary so a stale sequence + # cannot bind it back to the completed wait. + bindings[index] = None continue receivers_since_wait.append(index) diff --git a/tests/test_update_signal_condition_replay.py b/tests/test_update_signal_condition_replay.py index aa63ba8..b65431a 100644 --- a/tests/test_update_signal_condition_replay.py +++ b/tests/test_update_signal_condition_replay.py @@ -92,6 +92,40 @@ def run(self, ctx: WorkflowContext) -> Generator[Any, Any, dict[str, Any]]: } +@workflow.defn(name="tests.replay.post-condition-receivers") +class PostConditionReceiversWorkflow: + def __init__(self) -> None: + self.gate_open = False + self.note: str | None = None + self.finished = False + + @workflow.signal("open-gate") + def open_gate(self) -> None: + self.gate_open = True + + @workflow.update("set-note") + def set_note(self, note: str) -> str: + self.note = note + return note + + @workflow.signal("finish") + def finish(self) -> None: + self.finished = True + + def run(self, ctx: WorkflowContext) -> Generator[Any, Any, dict[str, Any]]: + yield ctx.wait_condition(lambda: self.gate_open, key="gate") + activity_result = yield ctx.schedule_activity("after-condition", []) + yield ctx.wait_condition( + lambda: self.note is not None and self.finished, + key="post-activity-receivers", + ) + return { + "activity_result": activity_result, + "note": self.note, + "finished": self.finished, + } + + def _event(event_type: str, payload: dict[str, Any]) -> dict[str, Any]: return {"event_type": event_type, "payload": payload} @@ -273,7 +307,102 @@ def _legacy_history() -> list[dict[str, Any]]: return history +def _post_condition_receiver_history() -> list[dict[str, Any]]: + return [ + _event("WorkflowStarted", {"workflow_type": "tests.replay.post-condition-receivers"}), + _event( + "ConditionWaitOpened", + { + "sequence": 1, + "condition_wait_id": "condition:1", + "condition_key": "gate", + }, + ), + _event( + "SignalReceived", + { + "workflow_sequence": 1, + "signal_name": "open-gate", + "value": _payload([]), + }, + ), + _event( + "ConditionWaitSatisfied", + { + "sequence": 1, + "condition_wait_id": "condition:1", + "condition_key": "gate", + }, + ), + _event( + "ActivityScheduled", + {"sequence": 2, "activity_type": "after-condition"}, + ), + _event( + "ActivityStarted", + {"sequence": 2, "activity_type": "after-condition"}, + ), + _event( + "UpdateApplied", + { + "sequence": 1, + "update_id": "update-after-condition", + "update_name": "set-note", + "arguments": _payload(["accepted"]), + }, + ), + _event( + "UpdateCompleted", + { + "sequence": 1, + "update_id": "update-after-condition", + "update_name": "set-note", + "result": _payload("accepted"), + }, + ), + _event( + "SignalReceived", + { + "workflow_sequence": 1, + "signal_name": "finish", + "value": _payload([]), + }, + ), + _event( + "ActivityCompleted", + { + "sequence": 2, + "activity_type": "after-condition", + "result": _payload("completed"), + }, + ), + _event( + "WorkflowCompleted", + { + "result": _payload( + { + "activity_result": "completed", + "note": "accepted", + "finished": True, + } + ) + }, + ), + ] + + class TestUpdateSignalConditionReplay: + def test_post_condition_step_boundary_overrides_stale_receiver_sequences(self) -> None: + outcome = replay(PostConditionReceiversWorkflow, _post_condition_receiver_history(), []) + + assert len(outcome.commands) == 1 + assert isinstance(outcome.commands[0], CompleteWorkflow) + assert outcome.commands[0].result == { + "activity_result": "completed", + "note": "accepted", + "finished": True, + } + def test_current_history_uses_explicit_timeout_identity_without_legacy_classification(self) -> None: with patch( "durable_workflow.workflow._legacy_condition_timeout_timer_sequence_aliases", From ca67e249574af1333d7a988b99c79485bf712706 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Wed, 2 Sep 2026 22:45:25 +0000 Subject: [PATCH 2/2] Add post-condition receiver replay corpus evidence --- ...ndition-receivers-stale-sequence-avro.json | 113 ++++++++++++++++++ tests/test_replay_regression_corpus.py | 6 +- 2 files changed, 118 insertions(+), 1 deletion(-) create mode 100644 tests/fixtures/replay_regressions/post-condition-receivers-stale-sequence-avro.json diff --git a/tests/fixtures/replay_regressions/post-condition-receivers-stale-sequence-avro.json b/tests/fixtures/replay_regressions/post-condition-receivers-stale-sequence-avro.json new file mode 100644 index 0000000..6499f99 --- /dev/null +++ b/tests/fixtures/replay_regressions/post-condition-receivers-stale-sequence-avro.json @@ -0,0 +1,113 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "post-condition-receivers-stale-sequence-avro", + "protocol_version": "1.19", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.post-condition-receivers", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "WorkflowStarted", + "payload": { + "workflow_type": "tests.replay.post-condition-receivers", + "payload_codec": "avro" + } + }, + { + "event_type": "ConditionWaitOpened", + "payload": { + "sequence": 1, + "condition_wait_id": "condition:1", + "condition_key": "gate" + } + }, + { + "event_type": "SignalReceived", + "payload": { + "workflow_sequence": 1, + "signal_name": "open-gate", + "value": "wwHioz3/VYAiNwwA", + "payload_codec": "avro" + } + }, + { + "event_type": "ConditionWaitSatisfied", + "payload": { + "sequence": 1, + "condition_wait_id": "condition:1", + "condition_key": "gate" + } + }, + { + "event_type": "ActivityScheduled", + "payload": { + "sequence": 2, + "activity_type": "after-condition" + } + }, + { + "event_type": "ActivityStarted", + "payload": { + "sequence": 2, + "activity_type": "after-condition" + } + }, + { + "event_type": "UpdateApplied", + "payload": { + "sequence": 1, + "update_id": "update-after-condition", + "update_name": "set-note", + "arguments": "wwHioz3/VYAiNwwCChBhY2NlcHRlZAA=", + "payload_codec": "avro" + } + }, + { + "event_type": "UpdateCompleted", + "payload": { + "sequence": 1, + "update_id": "update-after-condition", + "update_name": "set-note", + "result": "wwHioz3/VYAiNwoQYWNjZXB0ZWQ=", + "payload_codec": "avro" + } + }, + { + "event_type": "SignalReceived", + "payload": { + "workflow_sequence": 1, + "signal_name": "finish", + "value": "wwHioz3/VYAiNwwA", + "payload_codec": "avro" + } + }, + { + "event_type": "ActivityCompleted", + "payload": { + "sequence": 2, + "activity_type": "after-condition", + "result": "wwHioz3/VYAiNwoSY29tcGxldGVk" + } + }, + { + "event_type": "WorkflowCompleted", + "payload": { + "result": "wwHioz3/VYAiNw4GHmFjdGl2aXR5X3Jlc3VsdAoSY29tcGxldGVkCG5vdGUKEGFjY2VwdGVkEGZpbmlzaGVkAgEA" + } + } + ], + "expected": { + "command_type": "CompleteWorkflow", + "result": { + "activity_result": "completed", + "note": "accepted", + "finished": true + } + } +} diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index ed8e321..61abba5 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -18,7 +18,10 @@ GoldenTimeoutWaitWorkflow, GoldenVersionMarkerWorkflow, ) -from tests.test_update_signal_condition_replay import UpdateSignalConditionTimerWorkflow +from tests.test_update_signal_condition_replay import ( + PostConditionReceiversWorkflow, + UpdateSignalConditionTimerWorkflow, +) FIXTURE_SCHEMA = "durable-workflow.replay-regression/v1" FIXTURE_DIR = Path(__file__).parent / "fixtures" / "replay_regressions" @@ -193,6 +196,7 @@ def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] NestedParallelPathWorkflow, ParallelMetadataProducerWorkflow, ParallelResultBindingWorkflow, + PostConditionReceiversWorkflow, SelectionAwaitMarkerWorkflow, SelectionCancellationQueryWorkflow, SelectionMissingNestedOpeningWorkflow,