From 6a1426d59e6b96d31f04e4a09e6e044737cc4dc1 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 6 Nov 2026 05:56:35 +0000 Subject: [PATCH 1/2] fix: reject ambiguous legacy sequence claims for replay clocks --- src/durable_workflow/replay_clock.py | 18 ++- src/durable_workflow/workflow.py | 8 +- ...clock-legacy-sequence-claim-ambiguity.json | 107 ++++++++++++++++++ ...t_business_replay_clock_sequence_claims.py | 104 +++++++++++++++++ 4 files changed, 231 insertions(+), 6 deletions(-) create mode 100644 tests/fixtures/replay_regressions/business-clock-legacy-sequence-claim-ambiguity.json create mode 100644 tests/test_business_replay_clock_sequence_claims.py diff --git a/src/durable_workflow/replay_clock.py b/src/durable_workflow/replay_clock.py index 7084eb9..20bcad2 100644 --- a/src/durable_workflow/replay_clock.py +++ b/src/durable_workflow/replay_clock.py @@ -142,7 +142,10 @@ def _declares_redrive(event: Mapping[str, Any]) -> bool: class BusinessReplayClock: """Index identities once, then read time only at actual scalar consumption.""" - def __init__(self, events: Sequence[Mapping[str, Any]], workflow_id: str, run_id: str) -> None: + def __init__( + self, events: Sequence[Mapping[str, Any]], workflow_id: str, run_id: str, + *, sequence_claim: Callable[[Mapping[str, Any]], int | None], + ) -> None: self.events = events self._by_sequence: dict[int, list[int]] = {} self._child_calls: dict[str, list[int]] = {} @@ -161,13 +164,17 @@ def __init__(self, events: Sequence[Mapping[str, Any]], workflow_id: str, run_id self._pending = self.initial self._current = self.initial for index, event in enumerate(events): - self._index_event(index, event) + self._index_event(index, event, sequence_claim) - def _index_event(self, index: int, event: Mapping[str, Any]) -> None: + def _index_event( + self, index: int, event: Mapping[str, Any], sequence_claim: Callable[[Mapping[str, Any]], int | None], + ) -> None: payload = event.get("payload") if not isinstance(payload, Mapping): return - if (sequence := _sequence(payload)) is not None: + # Visibility follows the resolver's actual claims, including legacy + # counters. Authority still requires strict integers on every row. + if (sequence := sequence_claim(payload)) is not None: self._by_sequence.setdefault(sequence, []).append(index) call = payload.get("child_call_id") if _kind(event) in _CHILD_EVENTS and _identity(call): @@ -229,6 +236,9 @@ def observe(self, index: int | None, family: str, sequence: int) -> None: self._pending = (None, None) if index is None or not 0 <= index < len(self.events) or family not in {"activity", "timer", _CHILD_FAMILY}: return + if any(_sequence(self.events[position]["payload"]) != sequence + for position in self._by_sequence.get(sequence, [])): + return boundary = (self._child_boundary(index, sequence) if family == _CHILD_FAMILY else self._scalar_boundary(index, family, sequence)) if not boundary: diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 2f3f0f9..7fdad3f 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -2202,7 +2202,9 @@ def __init__( if self._current_time is None and _allow_wall_clock: self._current_time = datetime.now(timezone.utc) self._replay_clock_error: str | None = None - self._business_clock = replay_clock.BusinessReplayClock([], workflow_id, run_id) + self._business_clock = replay_clock.BusinessReplayClock( + [], workflow_id, run_id, sequence_claim=_workflow_sequence, + ) self._workflow_command_id = workflow_command_id or run_id or workflow_id self._cancel_requested = bool(cancel_requested) self._cancellation_request_id: str | None = None @@ -5139,7 +5141,9 @@ def _replay_state( _allow_wall_clock=not events, ) ctx._replay_clock_error = replay_clock_error if workflow_start_time is None else None - ctx._business_clock = replay_clock.BusinessReplayClock(events, workflow_id or "", run_id) + ctx._business_clock = replay_clock.BusinessReplayClock( + events, workflow_id or "", run_id, sequence_claim=_workflow_sequence, + ) lifetime = _ReplayLifetime(ctx.logger) def _set_replaying(replaying: bool) -> None: diff --git a/tests/fixtures/replay_regressions/business-clock-legacy-sequence-claim-ambiguity.json b/tests/fixtures/replay_regressions/business-clock-legacy-sequence-claim-ambiguity.json new file mode 100644 index 0000000..70abcf0 --- /dev/null +++ b/tests/fixtures/replay_regressions/business-clock-legacy-sequence-claim-ambiguity.json @@ -0,0 +1,107 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "business-clock-legacy-sequence-claim-ambiguity", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.business-clock", + "input": [], + "payload_codec": "avro", + "workflow_id": "business-clock-instance", + "run_id": "business-clock-run" + }, + "history": [ + { + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-01T00:00:00.123456Z", + "payload": { + "workflow_type": "tests.replay.business-clock", + "workflow_instance_id": "business-clock-instance", + "workflow_run_id": "business-clock-run" + }, + "sequence": 1 + }, + { + "event_type": "ActivityScheduled", + "payload": { + "sequence": "1", + "activity_execution_id": "clock-activity", + "activity_type": "clock.first" + }, + "sequence": 2 + }, + { + "event_type": "ActivityScheduled", + "payload": { + "sequence": 1, + "activity_execution_id": "clock-activity", + "activity_type": "clock.first" + }, + "sequence": 3 + }, + { + "event_type": "ActivityCompleted", + "recorded_at": "2026-11-01T00:00:03.123456Z", + "payload": { + "sequence": 1, + "activity_execution_id": "clock-activity", + "activity_type": "clock.first", + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoKZmlyc3Q=" + } + }, + "sequence": 4 + }, + { + "event_type": "TimerScheduled", + "payload": { + "sequence": 2, + "timer_id": "clock-timer", + "timer_kind": "durable_timer", + "delay_seconds": 2 + }, + "sequence": 5 + }, + { + "event_type": "TimerFired", + "recorded_at": "2026-11-01T00:00:05.654321Z", + "payload": { + "sequence": 2, + "timer_id": "clock-timer", + "timer_kind": "durable_timer", + "delay_seconds": 2 + }, + "sequence": 6 + }, + { + "event_type": "WorkflowTaskCompleted", + "recorded_at": "2026-12-01T00:00:00Z", + "payload": {}, + "sequence": 7 + } + ], + "expected": { + "command_sequence": [ + { + "type": "complete_workflow", + "result": { + "clock": [ + null, + null, + "2026-11-01T00:00:05.654321+00:00" + ], + "legacy": [ + "2026-11-01T00:00:00.123456+00:00", + "2026-11-01T00:00:00.123456+00:00", + "2026-11-01T00:00:00.123456+00:00" + ], + "elapsed_seconds": null + } + } + ] + } +} diff --git a/tests/test_business_replay_clock_sequence_claims.py b/tests/test_business_replay_clock_sequence_claims.py new file mode 100644 index 0000000..86d6442 --- /dev/null +++ b/tests/test_business_replay_clock_sequence_claims.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +from copy import deepcopy +from typing import Any + +import pytest + +from durable_workflow.errors import NonDeterministicReplayError +from tests.test_business_replay_clock import history, result +from tests.test_business_replay_clock_child import execute +from tests.test_business_replay_clock_child import history as child_history + + +def malformed_claim(payload: dict[str, Any], shape: str) -> None: + sequence = payload["sequence"] + if shape == "digits": + payload["sequence"] = str(sequence) + elif shape == "leading-zero": + payload["sequence"] = f"0{sequence}" + elif shape == "unicode-digits": + payload["sequence"] = "١" if sequence == 1 else "٢" + elif shape == "alias-only": + payload["workflow_sequence"] = payload.pop("sequence") + elif shape == "alias-conflict": + payload["workflow_sequence"] = sequence + 1 + elif shape == "alias-string": + payload["workflow_sequence"] = str(sequence) + elif shape == "alias-fallback": + payload.update(sequence=0, workflow_sequence=sequence) + else: + raise AssertionError(shape) + + +CLAIMS = ["digits", "leading-zero", "unicode-digits", "alias-only", "alias-conflict", "alias-string", "alias-fallback"] + + +@pytest.mark.parametrize("family,position,observation", [("activity", 1, 1), ("timer", 3, 2)]) +@pytest.mark.parametrize("shape", CLAIMS) +def test_legacy_claimed_duplicate_opening_is_visible_but_cannot_prove_time( + family: str, position: int, observation: int, shape: str, +) -> None: + events = history() + expected = result(events) + duplicate = deepcopy(events[position]) + malformed_claim(duplicate["payload"], shape) + events.insert(position, duplicate) + expected["clock"][observation] = None + expected["elapsed_seconds"] = None + original = deepcopy(events) + assert result(events) == expected, family + assert result(events) == expected + assert events == original + + +@pytest.mark.parametrize("shape", CLAIMS) +@pytest.mark.parametrize("foreign_call", [False, True]) +def test_child_duplicate_claim_cannot_hide_behind_an_unindexed_counter_or_another_call( + shape: str, foreign_call: bool, +) -> None: + events = child_history() + expected = execute(events) + duplicate = deepcopy(events[1]) + malformed_claim(duplicate["payload"], shape) + if foreign_call: + duplicate["payload"]["child_call_id"] = "foreign-call" + events.insert(1, duplicate) + expected["clock"][1] = None + assert execute(events) == expected + assert execute(events) == expected + + +@pytest.mark.parametrize("value", [True, "1", "01", "١"]) +def test_boolean_and_string_claims_do_not_weaken_the_required_actual_integer(value: Any) -> None: + events = history() + duplicate = deepcopy(events[1]) + duplicate["payload"]["sequence"] = value + events.insert(1, duplicate) + assert result(events)["clock"] == [None, None, "2026-11-01T00:00:05.654321+00:00"] + + +@pytest.mark.parametrize("position", [2, 4]) +def test_duplicate_terminal_preserves_the_original_replay_error(position: int) -> None: + events = history() + duplicate = deepcopy(events[position]) + duplicate["payload"]["sequence"] = str(duplicate["payload"]["sequence"]) + events.insert(position, duplicate) + with pytest.raises(NonDeterministicReplayError, match="Keep yielded workflow steps stable"): + result(events) + + +def test_nonparseable_digit_counter_preserves_the_original_parser_error() -> None: + events = history() + events[1]["payload"]["sequence"] = "²" + with pytest.raises(ValueError, match="invalid literal for int"): + result(events) + + +def test_unclaimed_opening_does_not_poison_an_unrelated_proved_sequence() -> None: + events = history() + expected = result(events) + duplicate = deepcopy(events[1]) + duplicate["payload"]["sequence"] = "not-a-counter" + events.insert(1, duplicate) + assert result(events) == expected From eb233bfa9da8918eaa2f12ad2aed6c03c35277a5 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 6 Nov 2026 06:03:48 +0000 Subject: [PATCH 2/2] test: keep the unclaimed-row control on a distinct resource --- tests/test_business_replay_clock_sequence_claims.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/test_business_replay_clock_sequence_claims.py b/tests/test_business_replay_clock_sequence_claims.py index 86d6442..5f49a59 100644 --- a/tests/test_business_replay_clock_sequence_claims.py +++ b/tests/test_business_replay_clock_sequence_claims.py @@ -95,10 +95,11 @@ def test_nonparseable_digit_counter_preserves_the_original_parser_error() -> Non result(events) -def test_unclaimed_opening_does_not_poison_an_unrelated_proved_sequence() -> None: +def test_unclaimed_foreign_opening_does_not_poison_a_proved_sequence() -> None: events = history() expected = result(events) duplicate = deepcopy(events[1]) duplicate["payload"]["sequence"] = "not-a-counter" + duplicate["payload"]["activity_execution_id"] = "unrelated-activity" events.insert(1, duplicate) assert result(events) == expected