From 32ae30a52805a391b71906ba958710dc364f7df8 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 13 Oct 2026 04:53:58 +0000 Subject: [PATCH 1/2] Seed replay clocks from persisted workflow start time --- src/durable_workflow/workflow.py | 33 +++++-- .../workflow-start-recorded-clock.json | 10 ++ tests/test_replay_clock.py | 93 +++++++++++++++++++ tests/test_replay_regression_corpus.py | 2 + 4 files changed, 132 insertions(+), 6 deletions(-) create mode 100644 tests/fixtures/replay_regressions/workflow-start-recorded-clock.json create mode 100644 tests/test_replay_clock.py diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 8b157aa..9d62e30 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -1937,10 +1937,14 @@ def __init__( workflow_command_id: str | None = None, cancel_requested: bool = False, allow_cancellation_scope_authoring: bool = False, + _allow_wall_clock: bool = True, ) -> None: self._workflow_id = workflow_id self._run_id = run_id - self._current_time = current_time or datetime.now(timezone.utc) + self._current_time = current_time + 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._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 @@ -2538,6 +2542,12 @@ def continue_as_new( return ContinueAsNew(workflow_type=workflow_type, arguments=list(args), task_queue=task_queue) def now(self) -> datetime: + """Return the persisted workflow-start time, fixed during replay.""" + if self._current_time is None: + raise NonDeterministicReplayError( + 0, "persisted workflow-start clock", [], + detail=self._replay_clock_error or "Workflow replay clock unavailable: missing persisted start time", + ) return self._current_time def random(self) -> random.Random: @@ -2548,7 +2558,7 @@ def uuid4(self) -> uuid.UUID: return uuid.UUID(bytes=rand_bytes, version=4) def uuid7(self) -> uuid.UUID: - timestamp_ms = int(self._current_time.timestamp() * 1000) + timestamp_ms = int(self.now().timestamp() * 1000) if timestamp_ms < 0 or timestamp_ms >= (1 << 48): raise ValueError("ctx.uuid7() requires ctx.now() within the UUIDv7 timestamp range") @@ -4276,13 +4286,22 @@ def _replay_state( resolved_sequences.add(sequence) workflow_start_time: datetime | None = None + replay_clock_error = "Workflow replay clock unavailable: missing WorkflowStarted persisted start time" for ev in events: etype = _history_event_type(ev) if etype == "WorkflowStarted": - ts = (ev.get("payload") or {}).get("timestamp") - if ts: - with contextlib.suppress(ValueError, TypeError): - workflow_start_time = datetime.fromisoformat(str(ts).replace("Z", "+00:00")) + start_payload = ev.get("payload") + if not isinstance(start_payload, Mapping): + start_payload = {} + 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" + replay_clock_error = f"Workflow replay clock unavailable: invalid or missing WorkflowStarted.{clock_source}" + if isinstance(ts, str): + with contextlib.suppress(ValueError): + parsed_time = datetime.fromisoformat(ts.replace("Z", "+00:00")) + if parsed_time.tzinfo is not None and parsed_time.utcoffset() is not None: + workflow_start_time = parsed_time break for sequence, membership in scope_history.memberships.items(): @@ -4298,7 +4317,9 @@ def _replay_state( workflow_command_id=workflow_command_id, cancel_requested=cancel_requested, allow_cancellation_scope_authoring=allow_cancellation_scope_authoring, + _allow_wall_clock=not events, ) + ctx._replay_clock_error = replay_clock_error if workflow_start_time is None else None # Continue-as-new cursor checkpoints describe instance state that predates # this run. Seed them before the workflow opens its first stream wait so # both the durable wait key and worker acknowledgement use the global diff --git a/tests/fixtures/replay_regressions/workflow-start-recorded-clock.json b/tests/fixtures/replay_regressions/workflow-start-recorded-clock.json new file mode 100644 index 0000000..59cf5da --- /dev/null +++ b/tests/fixtures/replay_regressions/workflow-start-recorded-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": "workflow-start-recorded-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-01T12:00:00.123456Z", "payload": {}}], + "expected": {"type": "complete_workflow", "result": "2026-06-01T12:00:00.123456+00:00"} +} diff --git a/tests/test_replay_clock.py b/tests/test_replay_clock.py new file mode 100644 index 0000000..affc94b --- /dev/null +++ b/tests/test_replay_clock.py @@ -0,0 +1,93 @@ +from __future__ import annotations + +from copy import deepcopy +from datetime import datetime +from typing import Any +from uuid import UUID + +import pytest + +from durable_workflow import workflow +from durable_workflow.errors import NonDeterministicReplayError +from durable_workflow.workflow import CompleteWorkflow, WorkflowContext, replay + + +@workflow.defn(name="tests.replay.persisted-clock") +class PersistedClockWorkflow: + def run(self, ctx: WorkflowContext) -> str: + return ctx.now().isoformat(timespec="microseconds") + + +@workflow.defn(name="tests.replay.persisted-uuid7") +class PersistedUUID7Workflow: + def run(self, ctx: WorkflowContext) -> str: + return str(ctx.uuid7()) + + +@workflow.defn(name="tests.replay.marker-clock") +class MarkerClockWorkflow: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + before = ctx.now().isoformat(timespec="microseconds") + version = yield ctx.get_version("clock-case", 0, 7) + after = ctx.now().isoformat(timespec="microseconds") + return [before, version, after, str(ctx.uuid7())] + + +def result(workflow_cls: type, history: list[dict[str, Any]]) -> Any: + outcome = replay(workflow_cls, history, [], run_id="original-clock-run", payload_codec="avro") + assert len(outcome.commands) == 1 + command = outcome.commands[0] + assert isinstance(command, CompleteWorkflow) + return command.result + + +def test_recorded_start_time_survives_repeated_cold_reads_and_later_marker() -> None: + history = [ + {"event_type": "WorkflowStarted", "recorded_at": "2026-06-01T12:00:00.123456Z", "payload": {}}, + {"event_type": "VersionMarkerRecorded", "recorded_at": "2026-06-01T12:00:03.987654Z", + "payload": {"sequence": 1, "change_id": "clock-case", "version": 7, + "min_supported": 0, "max_supported": 7}}, + ] + original = deepcopy(history) + first = result(MarkerClockWorkflow, history) + second = result(MarkerClockWorkflow, history) + assert first == second + assert first[:3] == ["2026-06-01T12:00:00.123456+00:00", 7, "2026-06-01T12:00:00.123456+00:00"] + assert UUID(first[3]).int >> 80 == int(datetime.fromisoformat(first[0]).timestamp() * 1000) + assert history == original + + +def test_explicit_payload_time_retains_precedence() -> None: + history = [{"event_type": "WorkflowStarted", "recorded_at": "2026-06-01T12:00:00.123456Z", + "payload": {"timestamp": "2026-06-01T14:01:02.654321+02:00"}}] + assert result(PersistedClockWorkflow, history) == "2026-06-01T14:01:02.654321+02:00" + + +@pytest.mark.parametrize("invalid", [None, False, 123, "", "not-a-time", "2026-06-01T12:00:00"]) +@pytest.mark.parametrize("explicit", [False, True]) +@pytest.mark.parametrize("workflow_cls", [PersistedClockWorkflow, PersistedUUID7Workflow]) +def test_clock_helpers_refuse_invalid_time_without_fallback( + invalid: Any, explicit: bool, workflow_cls: type, +) -> None: + event: dict[str, Any] = {"event_type": "WorkflowStarted", "payload": {}} + if explicit: + event["recorded_at"] = "2026-06-01T12:00:00.123456Z" + event["payload"]["timestamp"] = invalid + else: + event["recorded_at"] = invalid + with pytest.raises(NonDeterministicReplayError, match="Workflow replay clock unavailable"): + result(workflow_cls, [event]) + + +@pytest.mark.parametrize("history", [ + [{"event_type": "WorkflowStarted", "payload": {}}], + [{"event_type": "SignalReceived", "payload": {"signal_name": "unhandled"}}], +]) +def test_missing_replay_time_is_explicit(history: list[dict[str, Any]]) -> None: + with pytest.raises(NonDeterministicReplayError, match="Workflow replay clock unavailable"): + result(PersistedClockWorkflow, history) + + +def test_empty_local_authoring_remains_available() -> None: + assert datetime.fromisoformat(result(PersistedClockWorkflow, [])).tzinfo is not None + assert UUID(result(PersistedUUID7Workflow, [])).version == 7 diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index 4a75623..144b6da 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -34,6 +34,7 @@ RepeatedPatchWorkflow, ) from tests.test_pending_operation_replay import PendingOperation +from tests.test_replay_clock import PersistedClockWorkflow from tests.test_update_signal_condition_replay import ( PostConditionReceiversWorkflow, UpdateSignalConditionTimerWorkflow, @@ -278,6 +279,7 @@ def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] WORKFLOWS = [ + PersistedClockWorkflow, NexusTargetWorkflow, UnsupportedAsyncWorkflow, ChildPolicyAuthorWorkflow, From 2467639c7a456438f1d8da197922538365dfab91 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Tue, 13 Oct 2026 04:56:59 +0000 Subject: [PATCH 2/2] Release stable persisted-clock correction as 2.5.13 --- CHANGELOG.md | 7 +++++++ pyproject.toml | 6 +++--- 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a346fc6..3fde81b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [2.5.13] - 2026-10-13 + +### Fixed +- Seed deterministic workflow-start time from recorded history when the explicit + payload timestamp is absent. Cold `now()` and UUIDv7 reads use persisted time; + missing or invalid replay time produces an explicit diagnostic. + ## [2.5.12] - 2026-10-13 ### Fixed diff --git a/pyproject.toml b/pyproject.toml index 85f6afa..0158f0c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "durable-workflow" -version = "2.5.12" +version = "2.5.13" description = "Python client and worker SDK for Durable Workflow Cloud and self-hosted Server" readme = "README.md" requires-python = ">=3.10" @@ -71,8 +71,8 @@ durable-workflow-replay-conformance = "durable_workflow.replay_conformance:main" durable-workflow-workflow-updates-conformance = "durable_workflow.workflow_updates_conformance:main" [tool.durable-workflow] -product-train = "2.5.12" -registry-version = "2.5.12" +product-train = "2.5.13" +registry-version = "2.5.13" supported-server-versions = "2.5.16" worker-protocol-version = "1.20" control-plane-version = "2"