Conformance finding: signals/queries smoke hits Python nondeterministic replay (#121)

Author: durable-workflow-ops

CHANGELOG.md

@@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Fixed
+- Condition-wait replay now binds signals that arrive during a leased
+  workflow task to the next recorded wait when the server history records
+  those signals before the task's `ConditionWaitOpened` row. This avoids
+  applying rapid signal batches to the previous wait and prevents replay from
+  completing with a later wait/timer history step left unconsumed.
 - Workflow workers now report unhandled workflow-task execution errors back to
   the server instead of leaving the leased task pending until the lease or CLI
   wait times out. This lets the server observe and retry or fail the task

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "durable-workflow"
-version = "0.4.55"
+version = "0.4.56"
 description = "Python SDK for the Durable Workflow server (language-neutral HTTP protocol)"
 readme = "README.md"
 requires-python = ">=3.10"

src/durable_workflow/workflow.py

@@ -2112,8 +2112,6 @@ def _state(commands: list[Command]) -> _ReplayState:
     # in history, future server-recorded) or 'timed_out' (from a matching
     # condition_timeout TimerFired event).
     wait_resolutions: dict[str, str] = {}
-    open_condition_wait_ids: list[str] = []
-
     def _append_resolved_result(value: Any, shape: str, event: Mapping[str, Any]) -> None:
         payload = event.get("payload") or {}
         if not isinstance(payload, Mapping):
@@ -2227,16 +2225,113 @@ def _assert_no_unconsumed_history(terminal_shape: str) -> None:
             step.event_types,
         )
 
-    def _current_condition_wait_id() -> str | None:
-        return open_condition_wait_ids[-1] if open_condition_wait_ids else None
+    def _is_external_receiver_event(event_type: str | None) -> bool:
+        return event_type in ("SignalReceived", "UpdateApplied")
 
-    def _close_condition_wait(wait_id: Any) -> None:
-        if not isinstance(wait_id, str) or not wait_id:
-            return
-        with contextlib.suppress(ValueError):
-            open_condition_wait_ids.remove(wait_id)
+    def _receiver_binding_boundary_kind(event_type: str | None, payload: Mapping[str, Any]) -> str | None:
+        if event_type in (
+            "ConditionWaitSatisfied",
+            "ConditionWaitTimedOut",
+        ):
+            return "condition"
+        if (
+            event_type in ("TimerFired", "TimerCancelled")
+            and payload.get("timer_kind") == "condition_timeout"
+        ):
+            return "condition"
+        if event_type in (
+            "WorkflowCompleted",
+            "WorkflowFailed",
+            "WorkflowContinuedAsNew",
+            "ActivityScheduled",
+            "ActivityStarted",
+            "ChildWorkflowScheduled",
+            "ChildRunStarted",
+        ):
+            return "step"
+        if event_type == "TimerScheduled":
+            return "step" if payload.get("timer_kind") not in ("condition_timeout", "signal_timeout") else None
+        return "step" if _is_resolved_step_event(event_type, payload) else None
+
+    def _receiver_condition_wait_bindings() -> dict[int, str | None]:
+        bindings: dict[int, str | None] = {}
+        prefix_receivers: list[int] = []
+        prefix_can_bind_to_first_wait = True
+        current_wait_id: str | None = None
+        receivers_since_wait: list[int | None] = []
+
+        for index, event in enumerate(events):
+            event_type = _history_event_type(event)
+            payload = event.get("payload") or {}
+            if not isinstance(payload, Mapping):
+                payload = {}
+
+            if event_type == "ConditionWaitOpened":
+                wait_id = payload.get("condition_wait_id")
+                if not isinstance(wait_id, str) or not wait_id:
+                    continue
 
-    for ev in events:
+                if current_wait_id is None:
+                    for receiver_index in prefix_receivers:
+                        bindings[receiver_index] = wait_id if prefix_can_bind_to_first_wait else None
+                else:
+                    # When multiple signals arrive while the task woken by the
+                    # first signal is still leased, the server records those
+                    # later SignalReceived rows before the task's next
+                    # ConditionWaitOpened row. Replay them at that next wait.
+                    for receiver_index in receivers_since_wait[1:]:
+                        if receiver_index is not None:
+                            bindings[receiver_index] = wait_id
+
+                prefix_receivers = []
+                prefix_can_bind_to_first_wait = True
+                current_wait_id = wait_id
+                receivers_since_wait = []
+                continue
+
+            if _is_external_receiver_event(event_type):
+                explicit_sequence = _workflow_sequence(payload) is not None
+                if current_wait_id is None:
+                    if not explicit_sequence:
+                        prefix_receivers.append(index)
+                    continue
+
+                receivers_since_wait.append(None if explicit_sequence else index)
+                if len(receivers_since_wait) == 1 and not explicit_sequence:
+                    bindings[index] = current_wait_id
+                continue
+
+            boundary_kind = _receiver_binding_boundary_kind(event_type, payload)
+            if boundary_kind is None:
+                continue
+
+            if current_wait_id is None:
+                if boundary_kind == "step":
+                    for receiver_index in prefix_receivers:
+                        bindings[receiver_index] = None
+                    prefix_receivers = []
+                    prefix_can_bind_to_first_wait = False
+                continue
+
+            for receiver_index in receivers_since_wait:
+                if receiver_index is not None and receiver_index not in bindings:
+                    bindings[receiver_index] = current_wait_id
+            current_wait_id = None
+            receivers_since_wait = []
+            prefix_can_bind_to_first_wait = boundary_kind == "condition"
+
+        for receiver_index in prefix_receivers:
+            bindings[receiver_index] = None
+        if current_wait_id is not None:
+            for receiver_index in receivers_since_wait:
+                if receiver_index is not None and receiver_index not in bindings:
+                    bindings[receiver_index] = current_wait_id
+
+        return bindings
+
+    receiver_condition_wait_ids = _receiver_condition_wait_bindings()
+
+    for event_index, ev in enumerate(events):
         etype = _history_event_type(ev)
         payload = ev.get("payload") or {}
         if etype == "ActivityCompleted":
@@ -2260,7 +2355,6 @@ def _close_condition_wait(wait_id: Any) -> None:
                 wait_id = payload.get("condition_wait_id")
                 if isinstance(wait_id, str) and wait_id:
                     wait_resolutions[wait_id] = "timed_out"
-                    _close_condition_wait(wait_id)
                 continue
             if timer_kind == "signal_timeout":
                 continue
@@ -2274,17 +2368,14 @@ def _close_condition_wait(wait_id: Any) -> None:
             if isinstance(wait_id, str) and wait_id:
                 recorded_wait_steps.append(_recorded_step("condition wait", ev))
                 wait_opened.append(dict(payload))
-                open_condition_wait_ids.append(wait_id)
         elif etype == "ConditionWaitSatisfied":
             wait_id = payload.get("condition_wait_id")
             if isinstance(wait_id, str) and wait_id:
                 wait_resolutions[wait_id] = "satisfied"
-                _close_condition_wait(wait_id)
         elif etype == "ConditionWaitTimedOut":
             wait_id = payload.get("condition_wait_id")
             if isinstance(wait_id, str) and wait_id:
                 wait_resolutions[wait_id] = "timed_out"
-                _close_condition_wait(wait_id)
         elif etype in ("SideEffectRecorded", "ChildRunCompleted"):
             shape = "side effect" if etype == "SideEffectRecorded" else "child workflow"
             _append_resolved_result(
@@ -2316,7 +2407,7 @@ def _close_condition_wait(wait_id: Any) -> None:
                 condition_wait_id = (
                     condition_wait_ids_by_sequence.get(workflow_sequence)
                     if workflow_sequence is not None
-                    else None
+                    else receiver_condition_wait_ids.get(event_index)
                 )
                 pending_receivers.append(_PendingReceiver(
                     result_index=len(resolved_results),
@@ -2332,11 +2423,17 @@ def _close_condition_wait(wait_id: Any) -> None:
                         external_storage=external_storage,
                         external_storage_cache=external_storage_cache,
                     ),
-                    condition_wait_id=condition_wait_id or _current_condition_wait_id(),
+                    condition_wait_id=condition_wait_id,
                 ))
         elif etype == "UpdateApplied":
             update_name = payload.get("update_name")
             if isinstance(update_name, str) and update_name:
+                workflow_sequence = _workflow_sequence(payload)
+                condition_wait_id = (
+                    condition_wait_ids_by_sequence.get(workflow_sequence)
+                    if workflow_sequence is not None
+                    else receiver_condition_wait_ids.get(event_index)
+                )
                 pending_receivers.append(_PendingReceiver(
                     result_index=len(resolved_results),
                     kind="update",
@@ -2351,7 +2448,7 @@ def _close_condition_wait(wait_id: Any) -> None:
                         external_storage=external_storage,
                         external_storage_cache=external_storage_cache,
                     ),
-                    condition_wait_id=_current_condition_wait_id(),
+                    condition_wait_id=condition_wait_id,
                 ))
 
     signal_registry: dict[str, str] = getattr(workflow_cls, "__workflow_signals__", {}) or {}

tests/test_wait_condition.py

@@ -464,6 +464,55 @@ def test_signal_received_before_next_wait_uses_declared_workflow_sequence(self)
         assert outcome.commands[0].result == expected
         assert query_state(SignalCounterUntilFinished, history, [], "current") == expected
 
+    def test_signal_received_before_reopened_wait_uses_next_wait_when_sequence_absent(self) -> None:
+        history = [
+            {
+                "event_type": "ConditionWaitOpened",
+                "payload": {
+                    "condition_wait_id": "wait-count-3",
+                    "condition_key": "done",
+                    "sequence": 21,
+                },
+            },
+            _signal_received_event("increment", [3]),
+            _signal_received_event("finish", []),
+            {
+                "event_type": "ConditionWaitOpened",
+                "payload": {
+                    "condition_wait_id": "wait-finish",
+                    "condition_key": "done",
+                    "sequence": 22,
+                    "timeout_seconds": 30,
+                },
+            },
+            {
+                "event_type": "TimerScheduled",
+                "payload": {
+                    "timer_kind": "condition_timeout",
+                    "condition_wait_id": "wait-finish",
+                    "condition_key": "done",
+                    "sequence": 22,
+                    "delay_seconds": 30,
+                },
+            },
+        ]
+
+        expected = {
+            "count": 3,
+            "done": True,
+            "events": [
+                {"signal": "increment", "amount": 3, "count": 3},
+                {"signal": "finish", "count": 3},
+            ],
+        }
+
+        outcome = replay(SignalCounterUntilFinished, history, [])
+
+        assert len(outcome.commands) == 1
+        assert isinstance(outcome.commands[0], CompleteWorkflow)
+        assert outcome.commands[0].result == expected
+        assert query_state(SignalCounterUntilFinished, history, [], "current") == expected
+
     def test_repeated_wait_after_activity_can_be_satisfied_by_later_signal(self) -> None:
         history = [
             {