From 0d4eb53be2c58514dd2ba6b7fd62b0f51aff2079 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 11:14:05 -0600 Subject: [PATCH 1/3] Replay a redriven run from the failed run's start time A redrive successor's WorkflowStarted is recorded at redrive time. Core replays it from payload.replayed_started_at when recovery_kind is redrive; the SDK used recorded_at, so ctx.now() and ctx.uuid7() before the failed step differed from the failed run. Use the replayed start, refuse an invalid one, and let the cancellation clock prefer reused_recorded_at. Fixes #232 Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 4 +++ src/durable_workflow/workflow.py | 8 ++++++ tests/test_history_event_contract.py | 27 +++++++++++++++++- tests/test_replay_clock.py | 42 +++++++++++++++++++++++++++- 4 files changed, 79 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 19ecddbd..2d3f4e27 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- A run created by redrive now replays from the failed run's start time. `ctx.now()` and `ctx.uuid7()` read `WorkflowStarted.replayed_started_at` when `recovery_kind` is `redrive`, as Core does, so workflow code before the failed step sees the same clock as the failed run. An invalid value is refused rather than ignored. The cancellation clock prefers a reused step's `reused_recorded_at`. (#232) + ## [2.13.0] - 2026-11-03 ### Added diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 5dbc80c2..3963e894 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -2206,6 +2206,10 @@ def cancellation_context(self) -> CancellationContext | ScopedCancellationContex def _observe_cancellation_replay_time(self, event: Mapping[str, Any] | None) -> None: timestamp = event.get("timestamp") if event is not None else None + payload = event.get("payload") if event is not None else None + # A step reused by redrive keeps the time it was first recorded. + if timestamp is None and isinstance(payload, Mapping) and isinstance(payload.get("reused_recorded_at"), str): + timestamp = payload["reused_recorded_at"] if timestamp is None and event is not None: timestamp = event.get("recorded_at") self._cancellation_replay_time_available = False @@ -4680,6 +4684,10 @@ def _replay_state( explicit_time = "timestamp" in start_payload ts = start_payload.get("timestamp") if explicit_time else ev.get("recorded_at") clock_source = "payload.timestamp" if explicit_time else "recorded_at" + # A redriven run replays from the failed run's start time, as Core does. + if start_payload.get("recovery_kind") == "redrive" and "replayed_started_at" in start_payload: + ts = start_payload.get("replayed_started_at") + clock_source = "payload.replayed_started_at" replay_clock_error = f"Workflow replay clock unavailable: invalid or missing WorkflowStarted.{clock_source}" if isinstance(ts, str): with contextlib.suppress(ValueError): diff --git a/tests/test_history_event_contract.py b/tests/test_history_event_contract.py index 466ea162..6311c1b8 100644 --- a/tests/test_history_event_contract.py +++ b/tests/test_history_event_contract.py @@ -674,6 +674,22 @@ } +# Keys Core adds only to runs created by redrive (workflow#864). The SDK reads +# replayed_started_at for the start clock and reused_recorded_at for step time. +REDRIVE_HISTORY_EVENT_PAYLOAD_KEYS = { + "WorkflowStarted": ("resume_step_sequence", "recovery_kind", "replayed_started_at"), + "ActivityCompleted": ("reused_from_run_id", "reused_activity_execution_id", "reused_recorded_at"), + "TimerFired": ("reused_from_run_id", "reused_recorded_at"), + "SideEffectRecorded": ("reused_from_run_id", "reused_recorded_at"), + "VersionMarkerRecorded": ("reused_from_run_id", "reused_recorded_at"), +} + +for _event_type, _keys in REDRIVE_HISTORY_EVENT_PAYLOAD_KEYS.items(): + DOCUMENTED_HISTORY_EVENT_PAYLOAD_KEYS[_event_type] = ( + *DOCUMENTED_HISTORY_EVENT_PAYLOAD_KEYS[_event_type], + *(key for key in _keys if key not in DOCUMENTED_HISTORY_EVENT_PAYLOAD_KEYS[_event_type]), + ) + SDK_INTERPRETED_HISTORY_EVENTS = frozenset( { "WorkflowStarted", @@ -732,9 +748,18 @@ def test_schedule_history_events_are_preserved_only() -> None: def test_php_history_fixture_key_expectations_match_documented_contract() -> None: for event_type, fixture_keys in PHP_HISTORY_EVENT_EXPECTED_KEYS.items(): - documented_keys = set(DOCUMENTED_HISTORY_EVENT_PAYLOAD_KEYS[event_type]) + # Published PHP fixtures come from ordinary runs, which never carry redrive keys. + documented_keys = set(DOCUMENTED_HISTORY_EVENT_PAYLOAD_KEYS[event_type]) - set( + REDRIVE_HISTORY_EVENT_PAYLOAD_KEYS.get(event_type, ()) + ) expected_keys = set(fixture_keys) tolerated_extra_keys = set(SDK_TOLERATED_FIXTURE_EXTRA_KEYS.get(event_type, frozenset())) assert expected_keys - documented_keys == tolerated_extra_keys, event_type assert documented_keys - expected_keys == set(), event_type + + +def test_redrive_payload_keys_are_documented_on_the_interpreted_events() -> None: + for event_type, keys in REDRIVE_HISTORY_EVENT_PAYLOAD_KEYS.items(): + assert event_type in SDK_INTERPRETED_HISTORY_EVENTS, event_type + assert set(keys) <= set(DOCUMENTED_HISTORY_EVENT_PAYLOAD_KEYS[event_type]), event_type diff --git a/tests/test_replay_clock.py b/tests/test_replay_clock.py index affc94b7..f2e2cfcc 100644 --- a/tests/test_replay_clock.py +++ b/tests/test_replay_clock.py @@ -7,7 +7,7 @@ import pytest -from durable_workflow import workflow +from durable_workflow import serializer, workflow from durable_workflow.errors import NonDeterministicReplayError from durable_workflow.workflow import CompleteWorkflow, WorkflowContext, replay @@ -91,3 +91,43 @@ def test_missing_replay_time_is_explicit(history: list[dict[str, Any]]) -> None: def test_empty_local_authoring_remains_available() -> None: assert datetime.fromisoformat(result(PersistedClockWorkflow, [])).tzinfo is not None assert UUID(result(PersistedUUID7Workflow, [])).version == 7 + + +@workflow.defn(name="tests.replay.redrive-clock") +class RedriveClockWorkflow: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + before = ctx.now().isoformat(timespec="microseconds") + token = str(ctx.uuid7()) + recorded = yield ctx.schedule_activity("tests.first", [before]) + return [before, token, recorded] + + +def redriven_history(**start_payload: Any) -> list[dict[str, Any]]: + return [ + {"event_type": "WorkflowStarted", "recorded_at": "2026-06-01T10:05:00.000000Z", "payload": start_payload}, + {"event_type": "ActivityCompleted", "recorded_at": "2026-06-01T10:05:00.500000Z", + "payload": {"sequence": 1, "activity_type": "tests.first", + "result": serializer.encode("recorded", codec="avro"), "payload_codec": "avro", + "reused_from_run_id": "failed-run", + "reused_recorded_at": "2026-06-01T10:00:01.000000Z"}}, + ] + + +def test_redriven_run_replays_from_the_failed_run_start_time() -> None: + history = redriven_history(recovery_kind="redrive", replayed_started_at="2026-06-01T10:00:00.000000Z") + before, token, recorded = result(RedriveClockWorkflow, history) + assert before == "2026-06-01T10:00:00.000000+00:00" + assert UUID(token).int >> 80 == int(datetime.fromisoformat(before).timestamp() * 1000) + assert recorded == "recorded" + + +def test_replayed_start_time_requires_a_redrive_start() -> None: + history = redriven_history(replayed_started_at="2026-06-01T10:00:00.000000Z") + assert result(RedriveClockWorkflow, history)[0] == "2026-06-01T10:05:00.000000+00:00" + + +@pytest.mark.parametrize("invalid", [None, 123, "", "not-a-time", "2026-06-01T10:00:00"]) +def test_invalid_replayed_start_time_is_refused_without_fallback(invalid: Any) -> None: + history = redriven_history(recovery_kind="redrive", replayed_started_at=invalid) + with pytest.raises(NonDeterministicReplayError, match="replayed_started_at"): + replay(RedriveClockWorkflow, history, [], run_id="redriven-run", payload_codec="avro") From d5fbb9308dcf358a28b9328c97525caea10e3d95 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 11:16:45 -0600 Subject: [PATCH 2/3] Pin the redriven start clock in the replay corpus Co-Authored-By: Claude Opus 5.5 --- .../redrive-start-replayed-clock.json | 10 ++++++++++ 1 file changed, 10 insertions(+) create mode 100644 tests/fixtures/replay_regressions/redrive-start-replayed-clock.json diff --git a/tests/fixtures/replay_regressions/redrive-start-replayed-clock.json b/tests/fixtures/replay_regressions/redrive-start-replayed-clock.json new file mode 100644 index 00000000..c8c537a9 --- /dev/null +++ b/tests/fixtures/replay_regressions/redrive-start-replayed-clock.json @@ -0,0 +1,10 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "redrive-start-replayed-clock", + "protocol_version": "1.20", + "bindings": ["python"], + "workflow": {"type": "tests.replay.persisted-clock", "input": [], "payload_codec": "avro"}, + "history": [{"event_type": "WorkflowStarted", "recorded_at": "2026-06-01T10:05:00.000000Z", "payload": {"recovery_kind": "redrive", "replayed_started_at": "2026-06-01T10:00:00.000000Z"}}], + "expected": {"type": "complete_workflow", "result": "2026-06-01T10:00:00.000000+00:00"} +} From 8db78ed1f8658343e7363175cb04cc706657554f Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 11:32:38 -0600 Subject: [PATCH 3/3] Let a reused step's original time override its event time The cancellation clock read the event's top-level timestamp before reused_recorded_at, so a step reused by redrive still advanced cleanup time to the redrive. Prefer reused_recorded_at, as Core's RedriveReplayClock::recordedAt does, and pin the remaining budget. Co-Authored-By: Claude Opus 5.5 --- src/durable_workflow/workflow.py | 14 ++++++++------ tests/test_cancellation_remaining_time.py | 19 +++++++++++++++++++ 2 files changed, 27 insertions(+), 6 deletions(-) diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 3963e894..e9df4bc4 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -2205,13 +2205,15 @@ def cancellation_context(self) -> CancellationContext | ScopedCancellationContex return self._cancellation_context def _observe_cancellation_replay_time(self, event: Mapping[str, Any] | None) -> None: - timestamp = event.get("timestamp") if event is not None else None payload = event.get("payload") if event is not None else None - # A step reused by redrive keeps the time it was first recorded. - if timestamp is None and isinstance(payload, Mapping) and isinstance(payload.get("reused_recorded_at"), str): - timestamp = payload["reused_recorded_at"] - if timestamp is None and event is not None: - timestamp = event.get("recorded_at") + # A step reused by redrive keeps the time it was first recorded, as in + # Core's RedriveReplayClock::recordedAt; it overrides the event's own time. + if isinstance(payload, Mapping) and isinstance(payload.get("reused_recorded_at"), str): + timestamp: Any = payload["reused_recorded_at"] + else: + timestamp = event.get("timestamp") if event is not None else None + if timestamp is None and event is not None: + timestamp = event.get("recorded_at") self._cancellation_replay_time_available = False if not isinstance(timestamp, str): return diff --git a/tests/test_cancellation_remaining_time.py b/tests/test_cancellation_remaining_time.py index db7d5c1e..cc32e1e0 100644 --- a/tests/test_cancellation_remaining_time.py +++ b/tests/test_cancellation_remaining_time.py @@ -263,3 +263,22 @@ def execute(command: RecordLocalActivity) -> str: ] assert observations == [22.123456, 22.123456] assert calls == ["once"] + + +def test_a_reused_step_keeps_its_original_time_for_the_remaining_budget() -> None: + class Cleanup: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + try: + yield ctx.start_timer(60) + except WorkflowCancelled as error: + assert error.context is not None + with ctx.cancellation_shield(): + yield ctx.schedule_activity("cleanup", []) + return error.context.remaining() + + reused = completed_activity(2, "cleanup", "cleaned") + reused["payload"] = {**reused["payload"], "reused_from_run_id": "failed-run", + "reused_recorded_at": "2026-10-01T00:00:12Z"} + events = history() + [{**reused, "timestamp": "2026-10-01T00:00:29Z"}] + # Redrive recorded the step again at 00:29; its workflow time stays 00:12. + assert replay(Cleanup, events, [], run_id="run-1").commands == [CompleteWorkflow(18.123456)]