diff --git a/CHANGELOG.md b/CHANGELOG.md index 9aed71b..190e8ae 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 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) + ## [2.13.2] - 2026-11-04 ### Fixed diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index a017aec..475c67b 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -7020,6 +7020,30 @@ def _consume_cancellation(command: Any, boundary: CancellationDelivery) -> _Repl except StopIteration as stop: return _terminal_state(stop.value, include_pending=True) + def _assert_reopen_within_first_timeout(first_index: int, reopen_index: int) -> None: + """A physical reopen keeps the first open's timeout or undercuts it. + + Earlier SDKs reopened with the full timeout and later ones with the time + remaining, so a reopen may equal or undercut the first open. Anything + else, including a reopen that adds or drops the timeout, is not a + history this workflow code writes. This is the sdk-php#271 rule. + """ + first = wait_opened[first_index].get("timeout_seconds") + reopen = wait_opened[reopen_index].get("timeout_seconds") + if first is None and reopen is None: + return + if (isinstance(first, int) and not isinstance(first, bool) and isinstance(reopen, int) + and not isinstance(reopen, bool) and reopen <= first): + return + sequence = wait_opened[reopen_index].get("sequence") + raise NonDeterministicReplayError( + sequence if isinstance(sequence, int) else current_call_sequence, + "condition wait reopen within its first open's timeout", + ["ConditionWaitOpened"], + detail="condition_wait_reopen_timeout_exceeded: a reopen records a longer timeout than the wait's " + "first open, or adds or drops it", + ) + def _reopened_with_remaining_timeout(command: WaitCondition, first_index: int) -> WaitCondition: """Reopen a logical wait with the time left before its first deadline. @@ -7699,7 +7723,6 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None and next_wait_index < len(wait_opened) and _same_logical_condition_wait(wait_opened[next_wait_index], cmd) ) - if resolution == "timed_out": terminal_condition_reopen_cmd = None next_value = False @@ -7714,6 +7737,10 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None )]) 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. + _assert_reopen_within_first_timeout(first_open_index, next_wait_index) consumed_reopen_for_current_wait = True wait_yield_count = next_wait_index continue @@ -7741,6 +7768,7 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None # with the same key/fingerprint before declaring the # logical wait still pending. if has_reopened_same_wait: + _assert_reopen_within_first_timeout(first_open_index, next_wait_index) consumed_reopen_for_current_wait = True wait_yield_count = next_wait_index continue diff --git a/tests/fixtures/replay_regressions/condition-wait-reopen-longer-timeout-refused.json b/tests/fixtures/replay_regressions/condition-wait-reopen-longer-timeout-refused.json new file mode 100644 index 0000000..11fa1b9 --- /dev/null +++ b/tests/fixtures/replay_regressions/condition-wait-reopen-longer-timeout-refused.json @@ -0,0 +1,106 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "condition-wait-reopen-longer-timeout-refused", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.condition-remaining-timeout", + "input": [], + "payload_codec": "avro" + }, + "history": [ + { + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": { + "workflow_type": "tests.replay.condition-remaining-timeout" + }, + "sequence": 1 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": { + "condition_wait_id": "w1", + "condition_key": "approval", + "sequence": 1, + "timeout_seconds": 300 + }, + "sequence": 2 + }, + { + "event_type": "TimerScheduled", + "recorded_at": "2026-11-03T00:00:00+00:00", + "payload": { + "timer_kind": "condition_timeout", + "condition_wait_id": "w1", + "condition_key": "approval", + "sequence": 1, + "delay_seconds": 300 + }, + "sequence": 3 + }, + { + "event_type": "SignalReceived", + "payload": { + "signal_name": "note", + "value": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwA" + }, + "payload_codec": "avro", + "workflow_sequence": 1 + }, + "recorded_at": "2026-11-03T00:02:00+00:00", + "sequence": 4 + }, + { + "event_type": "ConditionWaitOpened", + "recorded_at": "2026-11-03T00:02:01+00:00", + "payload": { + "condition_wait_id": "w2", + "condition_key": "approval", + "sequence": 2, + "timeout_seconds": 400 + }, + "sequence": 5 + }, + { + "event_type": "TimerScheduled", + "recorded_at": "2026-11-03T00:02:01+00:00", + "payload": { + "timer_kind": "condition_timeout", + "condition_wait_id": "w2", + "condition_key": "approval", + "sequence": 2, + "delay_seconds": 400 + }, + "sequence": 6 + }, + { + "event_type": "SignalReceived", + "payload": { + "signal_name": "note", + "value": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwwA" + }, + "payload_codec": "avro", + "workflow_sequence": 2 + }, + "recorded_at": "2026-11-03T00:03:00+00:00", + "sequence": 7 + } + ], + "expected": { + "command_sequence": [] + }, + "expected_replay_error": { + "type": "NonDeterministicReplayError", + "workflow_sequence": 2, + "message_contains": "condition_wait_reopen_timeout_exceeded" + } +} diff --git a/tests/test_wait_condition.py b/tests/test_wait_condition.py index 609ad0c..c2c69e8 100644 --- a/tests/test_wait_condition.py +++ b/tests/test_wait_condition.py @@ -625,17 +625,6 @@ def test_signal_received_before_reopened_wait_uses_next_wait_when_sequence_absen "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, }, }, ] @@ -674,17 +663,6 @@ def test_signal_received_before_reopened_wait_uses_next_wait_when_sequence_is_st "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, }, }, ] @@ -1377,6 +1355,18 @@ def test_an_equal_timeout_grace_wait_is_still_a_new_wait(self) -> None: assert isinstance(command, WaitCondition) and command.timeout_seconds == 271 +@workflow.defn(name="tests.replay.condition-longer-grace-after-timeout") +class LongerGraceAfterTimeout(GraceAfterTimeout): + grace_seconds = 400 + + +def test_a_longer_grace_wait_after_a_timeout_is_a_separate_call() -> None: + # The 300 s wait timed out, so a 400 s wait with the same key is not a + # reopen and the reopen bound (#248) does not apply to it. + (command,) = replay(LongerGraceAfterTimeout, _timed_out_then_grace(400, "note"), []).commands + assert isinstance(command, WaitCondition) and command.timeout_seconds == 371 + + def test_a_selection_condition_member_is_not_reopened_by_a_wrong_signal() -> None: # Ordinary parallel lists refuse WaitCondition. A condition inside # ctx.select is never re-sent while open, so its original timer keeps its @@ -1394,3 +1384,103 @@ 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) == [] + + +# #248: a physical reopen never records more time than the wait's first open +# (parity with sdk-php#271). Earlier SDKs reopened with the full timeout, later +# ones with the remainder, so equal and shorter reopens replay. +def _reopen_history(first: int | None, reopen: int | None) -> list[dict]: + def opened(wait_id: str, sequence: int, at: str, timeout: int | None) -> list[dict]: + events = _timed_open(wait_id, sequence, at, timeout=timeout or 0) + if timeout is None: + del events[0]["payload"]["timeout_seconds"] + events = events[:1] + return events + + return [START, *opened("w1", 1, "2026-11-03T00:00:00+00:00", first), + _timed_signal("note", 1, "2026-11-03T00:02:00+00:00"), + *opened("w2", 2, "2026-11-03T00:02:01+00:00", reopen), + _timed_signal("note", 2, "2026-11-03T00:03:00+00:00")] + + +@workflow.defn(name="tests.replay.condition-untimed-reopen") +class UntimedApproval(RemainingTimeoutApproval): + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + return (yield ctx.wait_condition(lambda: self.approved, key="approval")) + + +class TestReopenTimeoutBound: + @pytest.mark.parametrize("reopen", [300, 179, 0]) + def test_an_equal_or_shorter_reopen_replays(self, reopen: int) -> None: + assert _reopened_timeout(_reopen_history(300, reopen)) == 120 + + @pytest.mark.parametrize(("first", "reopen"), [(300, 301), (300, 600)]) + def test_a_longer_reopen_is_refused(self, first: int, reopen: int) -> None: + with pytest.raises(NonDeterministicReplayError, match="condition_wait_reopen_timeout_exceeded") as raised: + replay(RemainingTimeoutApproval, _reopen_history(first, reopen), []) + assert raised.value.workflow_sequence == 2 + assert str(reopen) not in str(raised.value) + + def test_untimed_reopens_of_an_untimed_wait_replay(self) -> None: + (command,) = replay(UntimedApproval, _reopen_history(None, None), []).commands + assert isinstance(command, WaitCondition) and command.timeout_seconds is None + + @pytest.mark.parametrize(("first", "reopen"), [(None, 60), (300, None)]) + def test_a_reopen_that_adds_or_drops_the_timeout_is_refused(self, first: int | None, reopen: int | None) -> None: + # An untimed wait never records a timeout and a timed one never drops + # it, so either history comes from changed code (sdk-php#271). + workflow_cls = UntimedApproval if first is None else RemainingTimeoutApproval + with pytest.raises(NonDeterministicReplayError, match="condition_wait_reopen_timeout_exceeded"): + replay(workflow_cls, _reopen_history(first, reopen), []) + + +# A same-key wait that follows a satisfied one is a separate call, so the +# reopen bound (#248) does not apply to it, whatever its timeout. +@workflow.defn(name="tests.replay.condition-two-approvals") +class TwoApprovals: + first_timeout: int | None = 300 + second_timeout: int | None = 400 + + def __init__(self) -> None: + self.approvals = 0 + + @workflow.signal("approve") + def approve(self) -> None: + self.approvals += 1 + + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + cls = type(self) + yield ctx.wait_condition(lambda: self.approvals >= 1, key="approval", timeout=cls.first_timeout) + yield ctx.wait_condition(lambda: self.approvals >= 2, key="approval", timeout=cls.second_timeout) + return self.approvals + + +def _two_approvals_history(first: int | None, second: int | None, satisfied_recorded: bool) -> list[dict]: + def open_wait(wait_id: str, sequence: int, at: str, timeout: int | None) -> dict: + payload: dict = {"condition_wait_id": wait_id, "condition_key": "approval", "sequence": sequence} + if timeout is not None: + payload["timeout_seconds"] = timeout + return {"event_type": "ConditionWaitOpened", "recorded_at": at, "payload": payload} + + history = [START, open_wait("w1", 1, "2026-11-03T00:00:00+00:00", first), + _timed_signal("approve", 1, "2026-11-03T00:00:10+00:00")] + if satisfied_recorded: + history.append({"event_type": "ConditionWaitSatisfied", "recorded_at": "2026-11-03T00:00:10+00:00", + "payload": {"condition_wait_id": "w1", "condition_key": "approval", "sequence": 1}}) + history.append(open_wait("w2", 2, "2026-11-03T00:00:11+00:00", second)) + return history + + +@pytest.mark.parametrize("satisfied_recorded", [True, False]) +@pytest.mark.parametrize(("first", "second"), [(300, 400), (None, 60), (300, None)], + ids=["longer", "added", "dropped"]) +def test_a_separate_wait_after_a_satisfied_one_keeps_its_own_timeout( + first: int | None, second: int | None, satisfied_recorded: bool, +) -> None: + TwoApprovals.first_timeout, TwoApprovals.second_timeout = first, second + try: + (command,) = replay(TwoApprovals, _two_approvals_history(first, second, satisfied_recorded), []).commands + finally: + TwoApprovals.first_timeout, TwoApprovals.second_timeout = 300, 400 + assert isinstance(command, WaitCondition) and command.condition_key == "approval" + assert command.timeout_seconds == second