Replay a condition wait whose orphaned duplicate's timer fires

CHANGELOG.md

@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Fixed
 
 - Replay now refuses a recorded condition-wait reopen that records a longer timeout than the wait's first open, or that adds or drops the timeout, with a `NonDeterministicReplayError` whose detail starts `condition_wait_reopen_timeout_exceeded`. Earlier versions linked such a reopen silently. Reopens with the full timeout (written before 2.13.2) or with the time remaining (2.13.2 and later) still replay. This matches sdk-php#271. (#248)
+- 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
 

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:
@@ -7044,6 +7047,12 @@ def _assert_reopen_within_first_timeout(first_index: int, reopen_index: int) ->
             "first open, or adds or drops it",
         )
 
+    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.
 
@@ -7724,6 +7733,16 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None
                         and _same_logical_condition_wait(wait_opened[next_wait_index], cmd)
                     )
                     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). It is consumed
+                            # as a reopen, so the reopen bound applies (#248).
+                            _assert_reopen_within_first_timeout(first_open_index, next_wait_index)
+                            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

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
+      }
+    ]
+  }
+}

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}

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,
@@ -1484,3 +1485,92 @@
         TwoApprovals.first_timeout, TwoApprovals.second_timeout = 300, 400
     assert isinstance(command, WaitCondition) and command.condition_key == "approval"
     assert command.timeout_seconds == second
+
+
+# #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"
+
+
+@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(
+    duplicate_timeout: int | None, orphan_fired: bool,
+) -> None:
+    history = _orphaned_duplicate_history(180, orphan_fired)
+    duplicate = history[3]["payload"]
+    assert duplicate["condition_wait_id"] == "w4"
+    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, [])