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..e9df4bc4 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -2205,9 +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 - if timestamp is None and event is not None: - timestamp = event.get("recorded_at") + payload = event.get("payload") if event is not None else None + # 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 @@ -4680,6 +4686,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/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"} +} 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)] 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")