From 6577584aa6c2dee41365e084d29bbe2ecfc13e34 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 22:25:39 -0600 Subject: [PATCH] Replay a condition wait whose orphaned duplicate's timer fires Server 3.0.x can record a second open of a condition wait while the first is still open (server#1015). When the first open's timer fired later, replay ended the logical wait with that timeout while history continued with the reopened wait, raising NonDeterministicReplayError. A wait now counts as reopened when the next same-logical open precedes the earlier wait's terminal event; a wait that timed out before the next open stays finished (the grace vector). sample-app's sdk-php#283 shape is a corpus fixture. Fixes #253. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 4 + src/durable_workflow/workflow.py | 17 ++ ...n-wait-orphaned-duplicate-timer-fired.json | 153 ++++++++++++++++++ tests/test_replay_regression_corpus.py | 7 +- tests/test_wait_condition.py | 74 +++++++++ 5 files changed, 254 insertions(+), 1 deletion(-) create mode 100644 tests/fixtures/replay_regressions/condition-wait-orphaned-duplicate-timer-fired.json diff --git a/CHANGELOG.md b/CHANGELOG.md index 9aed71b6..36a90cc7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Opt-in typed activity values (#184). `@activity.defn(..., typed=True)` decodes an activity's positional arguments to its parameter annotations, and `ctx.schedule_activity(..., result_type=T)` decodes the completed result in single calls, parallel groups and selections. Supported types are dataclasses, `Enum` (including `IntEnum` and `StrEnum`), `Optional[T]`, `list[T]`, `dict[str, T]` and scalars; other annotations raise `TypeError` up front. Undeclared keys are ignored and missing fields take their defaults. A mismatch raises the new `TypedValueError`, which names the field path but never the value. It is non-retryable in a typed activity and is raised at the `yield` for `result_type`. Dataclass and `Enum` activity arguments and results are now sent as their neutral values, the same bytes as `dataclasses.asdict()` with each enum's `.value`; these values were refused before. Untyped activities and calls receive plain values as before, so in-flight histories replay unchanged. +### Fixed + +- Replay no longer fails when a condition wait has an orphaned duplicate open whose timer fires later. Server 3.0.x can record a second open of the same wait while the first is still open (server#1015). The first open's timeout used to end the logical wait in replay, while history continued with the reopened wait, so replay raised `NonDeterministicReplayError`. A wait now counts as reopened when the next open of the same wait was recorded before the earlier wait's own end. A wait that timed out before the next open is still finished, and the next open is a separate call. Every earlier release has this gap; it is not a 2.13.2 regression. (#253) + ## [2.13.2] - 2026-11-04 ### Fixed diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 86fb9157..70d47567 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -5157,6 +5157,8 @@ def _state( # The recorded time of each ``wait_opened`` row, or None when history # carries no parseable time. A reopened wait keeps its first open's deadline. wait_opened_times: list[datetime | None] = [] + # The history index of each ``wait_opened`` row. + wait_opened_event_indexes: list[int] = [] # Map condition_wait_id → resolution: 'satisfied' (from ConditionWaitSatisfied # in history, future server-recorded) or 'timed_out' (from a matching # condition_timeout TimerFired event). @@ -5705,6 +5707,7 @@ def _receiver_condition_wait_bindings() -> dict[int, str | None]: recorded_wait_steps.append(_recorded_step("condition wait", ev)) wait_opened.append(dict(payload)) wait_opened_times.append(_history_event_time(ev)) + wait_opened_event_indexes.append(event_index) elif etype == "ConditionWaitSatisfied": wait_id = payload.get("condition_wait_id") if isinstance(wait_id, str) and wait_id: @@ -7020,6 +7023,12 @@ def _consume_cancellation(command: Any, boundary: CancellationDelivery) -> _Repl except StopIteration as stop: return _terminal_state(stop.value, include_pending=True) + def _opened_before_terminal(next_index: int, opened: Mapping[str, Any] | None) -> bool: + """Whether the next open was recorded while *opened* was still open.""" + wait_id = opened.get("condition_wait_id") if opened is not None else None + terminal = condition_wait_terminal_indexes.get(wait_id) if isinstance(wait_id, str) else None + return terminal is not None and wait_opened_event_indexes[next_index] < terminal + def _reopened_with_remaining_timeout(command: WaitCondition, first_index: int) -> WaitCondition: """Reopen a logical wait with the time left before its first deadline. @@ -7701,6 +7710,14 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None ) if resolution == "timed_out": + if has_reopened_same_wait and _opened_before_terminal(next_wait_index, opened): + # The same wait was opened again while this one was + # still open (a duplicate open, server#1015), so this + # physical wait's later timeout does not end the + # logical wait (#253, sdk-php#283). + consumed_reopen_for_current_wait = True + wait_yield_count = next_wait_index + continue terminal_condition_reopen_cmd = None next_value = False wait_yield_count += 1 diff --git a/tests/fixtures/replay_regressions/condition-wait-orphaned-duplicate-timer-fired.json b/tests/fixtures/replay_regressions/condition-wait-orphaned-duplicate-timer-fired.json new file mode 100644 index 00000000..a9ad36ab --- /dev/null +++ b/tests/fixtures/replay_regressions/condition-wait-orphaned-duplicate-timer-fired.json @@ -0,0 +1,153 @@ +{ + "$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-timer-fired", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "source": { + "runtime": "server", + "package": "durableworkflow/server", + "version": "3.0.5" + }, + "workflow": { + "type": "tests.replay.condition-orphaned-duplicate", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-04T04:07:36.890000+00:00", + "payload": { + "workflow_type": "tests.replay.condition-orphaned-duplicate" + }, + "sequence": 1 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T04:07:36.890000+00:00", + "payload": { + "condition_wait_id": "w3", + "condition_key": "decision-1", + "sequence": 1, + "timeout_seconds": 300 + }, + "sequence": 2 + }, + { + "event_type": "TimerScheduled", + "recorded_at": "2026-11-04T04:07:36.890000+00:00", + "payload": { + "timer_kind": "condition_timeout", + "condition_wait_id": "w3", + "condition_key": "decision-1", + "sequence": 1, + "delay_seconds": 300 + }, + "sequence": 3 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T04:07:36.908000+00:00", + "payload": { + "condition_wait_id": "w4", + "condition_key": "decision-1", + "sequence": 2, + "timeout_seconds": 300 + }, + "sequence": 4 + }, + { + "event_type": "TimerScheduled", + "recorded_at": "2026-11-04T04:07:36.908000+00:00", + "payload": { + "timer_kind": "condition_timeout", + "condition_wait_id": "w4", + "condition_key": "decision-1", + "sequence": 2, + "delay_seconds": 300 + }, + "sequence": 5 + }, + { + "event_type": "SignalReceived", + "recorded_at": "2026-11-04T04:09:36.979000+00:00", + "payload": { + "signal_name": "decision", + "value": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwCAgAA" + }, + "payload_codec": "avro", + "workflow_sequence": 2 + }, + "sequence": 6 + }, + { + "event_type": "ConditionWaitSatisfied", + "recorded_at": "2026-11-04T04:09:37.008000+00:00", + "payload": { + "condition_wait_id": "w4", + "condition_key": "decision-1", + "sequence": 2 + }, + "sequence": 7 + }, + { + "event_type": "TimerCancelled", + "recorded_at": "2026-11-04T04:09:37.008000+00:00", + "payload": { + "timer_kind": "condition_timeout", + "condition_wait_id": "w4", + "sequence": 2 + }, + "sequence": 8 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-04T04:09:37.010000+00:00", + "payload": { + "condition_wait_id": "w5", + "condition_key": "decision-1", + "sequence": 3, + "timeout_seconds": 300 + }, + "sequence": 9 + }, + { + "event_type": "TimerScheduled", + "recorded_at": "2026-11-04T04:09:37.010000+00:00", + "payload": { + "timer_kind": "condition_timeout", + "condition_wait_id": "w5", + "condition_key": "decision-1", + "sequence": 3, + "delay_seconds": 300 + }, + "sequence": 10 + }, + { + "event_type": "TimerFired", + "recorded_at": "2026-11-04T04:12:36.956000+00:00", + "payload": { + "timer_kind": "condition_timeout", + "condition_wait_id": "w3", + "condition_key": "decision-1", + "sequence": 1 + }, + "sequence": 11 + } + ], + "expected": { + "command_sequence": [ + { + "type": "open_condition_wait", + "command_type": "WaitCondition", + "condition_key": "decision-1", + "timeout_seconds": 0 + } + ] + } +} diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index fe741ddf..95e30596 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -60,7 +60,11 @@ PostConditionReceiversWorkflow, UpdateSignalConditionTimerWorkflow, ) -from tests.test_wait_condition import RemainingTimeoutApproval, SignalCounterUntilFinished +from tests.test_wait_condition import ( + OrphanedDuplicateDecision, + RemainingTimeoutApproval, + SignalCounterUntilFinished, +) from tests.test_worker import compatible_cluster_info from tests.test_worker_status import RecoveryWorkflow as StatusRecoveryWorkflow from tests.test_workflow_identity import IdentityAcrossReplay, IdentityContinueAsNew @@ -416,6 +420,7 @@ def run(self, ctx: WorkflowContext, message: str) -> None: YieldedContinueAfterMetadataWorkflow, StatusRecoveryWorkflow, RemainingTimeoutApproval, + OrphanedDuplicateDecision, TypedActivityResultsWorkflow, ] WORKFLOW_TYPES = {str(getattr(workflow, "__workflow_name__", workflow.__name__)): workflow for workflow in WORKFLOWS} diff --git a/tests/test_wait_condition.py b/tests/test_wait_condition.py index 609ad0c8..937e67fe 100644 --- a/tests/test_wait_condition.py +++ b/tests/test_wait_condition.py @@ -6,6 +6,7 @@ from durable_workflow.errors import NonDeterministicReplayError, QueryFailed from durable_workflow.workflow import ( CompleteWorkflow, + ScheduleActivity, WaitCondition, WorkflowContext, query_state, @@ -1394,3 +1395,76 @@ def test_a_selection_condition_member_is_not_reopened_by_a_wrong_signal() -> Non "payload_codec": serializer.AVRO_CODEC}} outcome = replay(DurableSelectionConditionSignalReplayWorkflow, [START, *opens, wrong], []) assert list(outcome.commands) == [] + + +# #253 (sdk-php#283): an orphaned duplicate open whose timer fires later does +# not end the logical wait. Server 3.0.x can record a second open of the same +# wait while the first is still open (server#1015); the first's timer stays live. +@workflow.defn(name="tests.replay.condition-orphaned-duplicate") +class OrphanedDuplicateDecision: + def __init__(self) -> None: + self.decided = False + + @workflow.signal("decision") + def decision(self, approved: bool) -> None: + self.decided = bool(approved) + + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + decided = yield ctx.wait_condition(lambda: self.decided, key="decision-1", timeout=300) + if not decided: + yield ctx.schedule_activity("expire-proposal", []) + return "expired" + return "decided" + + +def _orphaned_duplicate_history(reopen: int, orphan_fired: bool) -> list[dict]: + def open_wait(wait_id: str, sequence: int, at: str, timeout: int) -> list[dict]: + return [ + {"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 = [ + {"event_type": "WorkflowStarted", "recorded_at": "2026-11-04T04:07:36.890000+00:00", "payload": {}}, + *open_wait("w3", 1, "2026-11-04T04:07:36.890000+00:00", 300), + *open_wait("w4", 2, "2026-11-04T04:07:36.908000+00:00", 300), + {"event_type": "SignalReceived", "recorded_at": "2026-11-04T04:09:36.979000+00:00", "payload": { + "signal_name": "decision", "value": serializer.envelope([False]), "payload_codec": serializer.AVRO_CODEC, + "workflow_sequence": 2}}, + {"event_type": "ConditionWaitSatisfied", "recorded_at": "2026-11-04T04:09:37.008000+00:00", "payload": { + "condition_wait_id": "w4", "condition_key": "decision-1", "sequence": 2}}, + {"event_type": "TimerCancelled", "recorded_at": "2026-11-04T04:09:37.008000+00:00", "payload": { + "timer_kind": "condition_timeout", "condition_wait_id": "w4", "sequence": 2}}, + *open_wait("w5", 3, "2026-11-04T04:09:37.010000+00:00", reopen), + ] + if orphan_fired: + history.append({"event_type": "TimerFired", "recorded_at": "2026-11-04T04:12:36.956000+00:00", "payload": { + "timer_kind": "condition_timeout", "condition_wait_id": "w3", "condition_key": "decision-1", + "sequence": 1}}) + return history + + +class TestOrphanedDuplicateWait: + @pytest.mark.parametrize("reopen", [300, 180]) + def test_the_orphans_timer_does_not_end_the_reopened_wait(self, reopen: int) -> None: + (command,) = replay(OrphanedDuplicateDecision, _orphaned_duplicate_history(reopen, True), []).commands + # The logical wait is still the reopened one; its first deadline has passed. + assert isinstance(command, WaitCondition) and command.timeout_seconds == 0 + + @pytest.mark.parametrize("reopen", [300, 180]) + def test_before_the_orphan_fires_the_wait_reopens_with_the_remaining_time(self, reopen: int) -> None: + (command,) = replay(OrphanedDuplicateDecision, _orphaned_duplicate_history(reopen, False), []).commands + assert isinstance(command, WaitCondition) and command.timeout_seconds == 180 + + def test_the_reopened_waits_own_timeout_expires_the_decision(self) -> None: + history = _orphaned_duplicate_history(180, True) + [ + {"event_type": "TimerFired", "recorded_at": "2026-11-04T04:12:37.216000+00:00", "payload": { + "timer_kind": "condition_timeout", "condition_wait_id": "w5", "condition_key": "decision-1", + "sequence": 3}}, + ] + (command,) = replay(OrphanedDuplicateDecision, history, []).commands + assert isinstance(command, ScheduleActivity) and command.activity_type == "expire-proposal"