From 6e05e9856fc23ebcee5cf79e19245ad10a0a58be Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 14:58:55 -0600 Subject: [PATCH 1/2] Consume an already-open duplicate after orphan satisfaction --- CHANGELOG.md | 2 + src/durable_workflow/workflow.py | 11 +- ...ion-wait-orphaned-duplicate-satisfied.json | 109 ++++++++++++++++++ tests/test_wait_condition.py | 62 ++++++++++ 4 files changed, 180 insertions(+), 4 deletions(-) create mode 100644 tests/fixtures/replay_regressions/condition-wait-orphaned-duplicate-satisfied.json diff --git a/CHANGELOG.md b/CHANGELOG.md index b708f01..2bbf7d5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Replay consumes an already-open duplicate of a condition wait when the older physical wait is satisfied with a true predicate. A wait opened after that satisfaction remains a separate authored call with its own timeout. (#256) + - `wait_condition` predicate fingerprints no longer change across Python versions or worker processes. Every earlier release hashed the predicate's bytecode and the `repr` of its constants. So upgrading Python (for example 3.12 to 3.13) failed replay of every in-flight run at its first wait, and a predicate containing a nested function (a generator expression, a nested lambda, or a comprehension on 3.10 and 3.11) got a new fingerprint in every process, so a restart or cache loss failed replay with `condition_fingerprint_changed`. New waits record a v2 fingerprint (`dwpy2:sha256:...`) computed from the predicate's module, qualified name and a canonical form of its source AST, with no bytecode, addresses, paths or line numbers. Without source (for example `eval` or a REPL) no fingerprint is sent, and the wait is identified by its key and call order. A v1 value recorded by an earlier release still matches when this process computes the same v1; otherwise it is accepted as unverifiable instead of failing replay, so drift in those predicates is no longer detected. A reopen sends the fingerprint its first open recorded, so a wait opened before the upgrade keeps its v1 value, and a keyless wait links a v1 value only within its own chain. A keyless wait without source links its false-wake reopens within its chain, as before. **Rollback and mixed fleets:** a run that opened a wait under this release records a v2 fingerprint, which 2.14.0 and earlier refuse on replay. Don't roll a task queue back below this release once such waits exist, and upgrade every worker on a queue together. (#258) - Same-line `wait_condition` lambdas whose bodies contain list, dict or set comprehensions share one combined fingerprint on every Python version. Put those predicates on separate lines to detect a change from one candidate to another. (#258) diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index aba9ac8..438019a 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -7855,10 +7855,13 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None prefix="wait_condition predicate raised", )]) if resolution == "satisfied": - if has_reopened_same_wait and not satisfied: - # Only an open consumed as a reopen is bounded (#248); - # a later same-key wait after a true predicate is a - # separate call with its own timeout. + if has_reopened_same_wait and ( + not satisfied or _opened_before_terminal(next_wait_index, opened) + ): + # An already-open duplicate belongs to this wait + # even when the older open satisfied it (#256). + # An open after satisfaction can be a separate + # authored call, with its own timeout (#248). _assert_reopen_within_first_timeout(first_open_index, next_wait_index) consumed_reopen_for_current_wait = True wait_yield_count = next_wait_index diff --git a/tests/fixtures/replay_regressions/condition-wait-orphaned-duplicate-satisfied.json b/tests/fixtures/replay_regressions/condition-wait-orphaned-duplicate-satisfied.json new file mode 100644 index 0000000..2b052ca --- /dev/null +++ b/tests/fixtures/replay_regressions/condition-wait-orphaned-duplicate-satisfied.json @@ -0,0 +1,109 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "condition-wait-orphaned-duplicate-satisfied", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "source": { + "runtime": "synthetic-python-replay", + "version": "unreleased", + "issue": "https://github.com/durable-workflow/sdk-python/issues/256", + "note": "Synthetic row-order regression; no published Server producer or runtime observation is claimed." + }, + "workflow": { + "type": "tests.replay.condition-orphaned-duplicate", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-04T00:00:00+00:00", + "payload": { + "workflow_type": "tests.replay.condition-orphaned-duplicate" + }, + "sequence": 1 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T00:00:00+00:00", + "payload": { + "condition_wait_id": "older", + "condition_key": "decision-1", + "sequence": 1, + "timeout_seconds": 300 + }, + "sequence": 2 + }, + { + "event_type": "TimerScheduled", + "recorded_at": "2026-11-04T00:00:00+00:00", + "payload": { + "timer_kind": "condition_timeout", + "condition_wait_id": "older", + "condition_key": "decision-1", + "sequence": 1, + "delay_seconds": 300 + }, + "sequence": 3 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T00:00:10+00:00", + "payload": { + "condition_wait_id": "duplicate", + "condition_key": "decision-1", + "sequence": 2, + "timeout_seconds": 180 + }, + "sequence": 4 + }, + { + "event_type": "TimerScheduled", + "recorded_at": "2026-11-04T00:00:10+00:00", + "payload": { + "timer_kind": "condition_timeout", + "condition_wait_id": "duplicate", + "condition_key": "decision-1", + "sequence": 2, + "delay_seconds": 180 + }, + "sequence": 5 + }, + { + "event_type": "SignalReceived", + "payload": { + "signal_name": "decision", + "value": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwCAgEA" + }, + "payload_codec": "avro", + "workflow_sequence": 1 + }, + "recorded_at": "2026-11-04T00:00:10+00:00", + "sequence": 6 + }, + { + "event_type": "ConditionWaitSatisfied", + "recorded_at": "2026-11-04T00:00:10+00:00", + "payload": { + "condition_wait_id": "older", + "condition_key": "decision-1", + "sequence": 1 + }, + "sequence": 7 + } + ], + "expected": { + "command_sequence": [ + { + "type": "complete_workflow", + "command_type": "CompleteWorkflow", + "result": "decided" + } + ] + } +} diff --git a/tests/test_wait_condition.py b/tests/test_wait_condition.py index 00391aa..288449e 100644 --- a/tests/test_wait_condition.py +++ b/tests/test_wait_condition.py @@ -1561,6 +1561,68 @@ def test_the_reopened_waits_own_timeout_expires_the_decision(self) -> None: assert isinstance(command, ScheduleActivity) and command.activity_type == "expire-proposal" +def _satisfied_orphan_history(duplicate_timeout: int = 180, *, equal_times: bool = False) -> list[dict]: + """Synthetic #256 history: the older open is satisfied after a duplicate.""" + first_at = "2026-11-04T00:00:00+00:00" + end_at = "2026-11-04T00:00:10+00:00" + duplicate_at = end_at if equal_times else "2026-11-04T00:00:09+00:00" + history = [{"event_type": "WorkflowStarted", "recorded_at": first_at, "payload": {}}] + for wait_id, sequence, at, timeout in (("older", 1, first_at, 300), + ("duplicate", 2, duplicate_at, duplicate_timeout)): + history.extend([ + {"event_type": "ConditionWaitOpened", "recorded_at": at, "payload": { + "condition_wait_id": wait_id, "condition_key": "decision-1", "sequence": sequence, + "timeout_seconds": timeout}}, + {"event_type": "TimerScheduled", "recorded_at": at, "payload": { + "timer_kind": "condition_timeout", "condition_wait_id": wait_id, + "condition_key": "decision-1", "sequence": sequence, "delay_seconds": timeout}}, + ]) + history.extend([ + {**_signal_received_event("decision", [True], workflow_sequence=1), "recorded_at": end_at}, + {"event_type": "ConditionWaitSatisfied", "recorded_at": end_at, "payload": { + "condition_wait_id": "older", "condition_key": "decision-1", "sequence": 1}}, + ]) + return history + + +@pytest.mark.parametrize("duplicate_timeout", [0, 180, 300]) +@pytest.mark.parametrize("equal_times", [False, True], ids=["distinct-times", "equal-times"]) +def test_a_satisfied_orphan_consumes_its_already_open_duplicate( + duplicate_timeout: int, equal_times: bool, +) -> None: + outcome = replay(OrphanedDuplicateDecision, _satisfied_orphan_history( + duplicate_timeout, equal_times=equal_times, + ), []) + assert outcome.commands == [CompleteWorkflow("decided")] + + +@pytest.mark.parametrize("duplicate_timeout", [600, None], ids=["longer", "dropped"]) +def test_a_satisfied_orphans_duplicate_keeps_the_reopen_timeout_bound(duplicate_timeout: int | None) -> None: + history = _satisfied_orphan_history() + duplicate = history[3]["payload"] + if duplicate_timeout is None: + del duplicate["timeout_seconds"] + else: + duplicate["timeout_seconds"] = duplicate_timeout + with pytest.raises(NonDeterministicReplayError, match="condition_wait_reopen_timeout_exceeded"): + replay(OrphanedDuplicateDecision, history, []) + + +def test_a_call_after_the_satisfied_orphan_chain_keeps_its_own_timeout() -> None: + history = _satisfied_orphan_history() + for event in history: + payload = event["payload"] + if "condition_key" in payload: + payload["condition_key"] = "approval" + if event["event_type"] == "SignalReceived": + payload.update(signal_name="approve", value=serializer.envelope([])) + history.append({"event_type": "ConditionWaitOpened", "recorded_at": "2026-11-04T00:00:11+00:00", "payload": { + "condition_wait_id": "next-call", "condition_key": "approval", "sequence": 3, "timeout_seconds": 400, + }}) + (command,) = replay(TwoApprovals, history, []).commands + assert isinstance(command, WaitCondition) and command.timeout_seconds == 400 + + @pytest.mark.parametrize("duplicate_timeout", [600, None], ids=["longer", "dropped"]) @pytest.mark.parametrize("orphan_fired", [False, True], ids=["before-orphan-fires", "after-orphan-fires"]) def test_an_orphans_duplicate_open_is_bounded_like_any_reopen( From ecadcb5aaa0e6bc9bee8094e7c6ec2d384cddbe3 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 15:07:35 -0600 Subject: [PATCH 2/2] Split the new replay control assertions --- tests/test_wait_condition.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/test_wait_condition.py b/tests/test_wait_condition.py index 288449e..ab2583d 100644 --- a/tests/test_wait_condition.py +++ b/tests/test_wait_condition.py @@ -1620,7 +1620,8 @@ def test_a_call_after_the_satisfied_orphan_chain_keeps_its_own_timeout() -> None "condition_wait_id": "next-call", "condition_key": "approval", "sequence": 3, "timeout_seconds": 400, }}) (command,) = replay(TwoApprovals, history, []).commands - assert isinstance(command, WaitCondition) and command.timeout_seconds == 400 + assert isinstance(command, WaitCondition) + assert command.timeout_seconds == 400 @pytest.mark.parametrize("duplicate_timeout", [600, None], ids=["longer", "dropped"])