From b638238107306a4b0f01a21bcd68b1a0324135af Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 6 Nov 2026 07:14:38 +0000 Subject: [PATCH 1/2] feat: qualify business replay time for parallel activity groups --- src/durable_workflow/replay_clock.py | 137 +++++- src/durable_workflow/replay_clock_math.py | 30 ++ src/durable_workflow/workflow.py | 23 +- tests/business_clock_workflows.py | 36 +- .../replay_clock_carbon_reference.json | 93 ++++ .../fixtures/replay_clock_carbon_reference.md | 41 ++ .../business-clock-parallel-activity-all.json | 262 +++++++++++ ...-clock-parallel-failure-selection-gap.json | 135 ++++++ tests/test_business_replay_clock_parallel.py | 417 ++++++++++++++++++ tests/test_replay_regression_corpus.py | 4 + 10 files changed, 1171 insertions(+), 7 deletions(-) create mode 100644 src/durable_workflow/replay_clock_math.py create mode 100644 tests/fixtures/replay_clock_carbon_reference.json create mode 100644 tests/fixtures/replay_clock_carbon_reference.md create mode 100644 tests/fixtures/replay_regressions/business-clock-parallel-activity-all.json create mode 100644 tests/fixtures/replay_regressions/business-clock-parallel-failure-selection-gap.json create mode 100644 tests/test_business_replay_clock_parallel.py diff --git a/src/durable_workflow/replay_clock.py b/src/durable_workflow/replay_clock.py index 20bcad2..7b6ceb1 100644 --- a/src/durable_workflow/replay_clock.py +++ b/src/durable_workflow/replay_clock.py @@ -14,6 +14,7 @@ from typing import Any, ParamSpec, TypeGuard, TypeVar from .errors import ReplayClockUnavailable +from .replay_clock_math import carbon_from_milliseconds, carbon_milliseconds _P = ParamSpec("_P") _T = TypeVar("_T") @@ -27,6 +28,11 @@ _CHILD_EVENTS = frozenset({"ChildWorkflowScheduled", "ChildRunStarted", *_CHILD_TERMINALS}) _CHILD_FAMILY = "child workflow" _REUSE_FIELDS = ("reused_from_run_id", "reused_recorded_at", "reused_activity_execution_id") +_GROUP_FIELDS = ("parallel_group_id", "parallel_group_kind", "parallel_group_base_sequence", + "parallel_group_size", "parallel_group_index") +_SELECTION_FIELDS = ("selection_member_key", "selection_member_index", "selection_member_base_sequence", + "selection_member_size", "selection_member_kind") +_GroupMember = tuple[int | None, int, str, Sequence[Mapping[str, Any]] | None] def _task() -> object | None: @@ -139,8 +145,75 @@ def _declares_redrive(event: Mapping[str, Any]) -> bool: ) +def _all_group_mode(payload: Mapping[str, Any]) -> bool: + return (payload.get("parallel_group_mode") in (None, "all") + and all(payload.get(field) is None for field in _SELECTION_FIELDS)) + + +def _group_entry_matches(actual: Mapping[str, Any], expected: Mapping[str, Any]) -> bool: + return _all_group_mode(actual) and all( + type(actual.get(field)) is type(expected[field]) and actual.get(field) == expected[field] + for field in _GROUP_FIELDS + ) + + +def _group_payload_matches(payload: Mapping[str, Any], expected: Sequence[Mapping[str, Any]]) -> bool: + if (not expected or not _all_group_mode(payload) + or payload.get("execution_mode") not in (None, "remote") or payload.get("local_activity") is True + or payload.get("cancellation_scope_id") not in (None, "root") + or payload.get("selection_group") is not None or payload.get("parallel_group") is not None): + return False + path = payload.get("parallel_group_path") + if path in (None, []): + return len(expected) == 1 and _group_entry_matches(payload, expected[0]) + if not isinstance(path, list) or len(path) != len(expected): + return False + if not all(isinstance(entry, Mapping) and _group_entry_matches(entry, wanted) + for entry, wanted in zip(path, expected, strict=True)): + return False + # Top-level fields are aliases for the innermost path entry. A partial + # alias may not contradict the complete path or coerce its counter types. + return all(field not in payload or ( + type(payload[field]) is type(expected[-1][field]) and payload[field] == expected[-1][field] + ) for field in _GROUP_FIELDS) + + +def _group_event_time(event: Mapping[str, Any]) -> datetime | None: + # Core's Activity-all path uses these model fields, not RedriveReplayClock. + value = event.get("recorded_at") + time = _time(event.get("created_at") if value is None else value) + if "timestamp" in event and _time(event["timestamp"]) != time: + return None + return time + + +def _activity_group_time(events: Sequence[Mapping[str, Any]], failed_member: int | None) -> datetime | None: + candidates: list[tuple[int, datetime, int]] = [] + for offset, event in enumerate(events): + failed = _kind(event) != "ActivityCompleted" + if failed != (failed_member is not None): + if failed: + return None + continue # Successful siblings do not choose Core's failure. + time = _group_event_time(event) + milliseconds = carbon_milliseconds(time) if time is not None else None + if time is None or milliseconds is None: + return None + candidates.append((milliseconds, time, offset)) + if not candidates: + return None + if failed_member is not None: + winner = min(candidates, key=lambda candidate: candidate[0]) + # Preserve the existing SDK exception even when Core chooses another + # failed member. That open compatibility gap has no clock authority. + return carbon_from_milliseconds(winner[0]) if winner[2] == failed_member else None + # max retains the first descriptor on an equal millisecond value. Keep its + # full event precision, never the last row or a maximum over all history. + return max(candidates, key=lambda candidate: candidate[0])[1] + + class BusinessReplayClock: - """Index identities once, then read time only at actual scalar consumption.""" + """Index identities once, then read time only at qualified consumption.""" def __init__( self, events: Sequence[Mapping[str, Any]], workflow_id: str, run_id: str, @@ -149,6 +222,7 @@ def __init__( self.events = events self._by_sequence: dict[int, list[int]] = {} self._child_calls: dict[str, list[int]] = {} + self._activity_executions: dict[str, list[int]] = {} self._source_run: str | None = None self._resume_step: int | None = None self._redrive_inserted: datetime | None = None @@ -161,8 +235,8 @@ def __init__( if (isinstance(payload, Mapping) and self._start_prefix(index, payload) and self._redrive_start(index, payload, workflow_id, run_id)): self.initial = (index, _time(payload.get("replayed_started_at"))) - self._pending = self.initial - self._current = self.initial + self._pending: tuple[int | None, datetime | None] = self.initial + self._current: tuple[int | None, datetime | None] = self.initial for index, event in enumerate(events): self._index_event(index, event, sequence_claim) @@ -179,6 +253,9 @@ def _index_event( call = payload.get("child_call_id") if _kind(event) in _CHILD_EVENTS and _identity(call): self._child_calls.setdefault(call, []).append(index) + execution = payload.get("activity_execution_id") + if _identity(execution): + self._activity_executions.setdefault(execution, []).append(index) def _start_prefix(self, index: int, payload: Mapping[str, Any]) -> bool: if index == 0: @@ -232,6 +309,60 @@ def snapshot(self) -> datetime | None: def unavailable(self) -> None: self._current = self._pending = (None, None) + def preserve_boundary(self) -> None: + """An empty group consumes no new event and retains the proved prefix.""" + self._pending = self._current + + def observe_activity_group( + self, members: Sequence[_GroupMember], failed_member: int | None, *, reconstruction: bool, + ) -> None: + self._pending = (None, None) + # Core's successful query group-send supplies no new time. Body and + # callback reconstruction cannot borrow an executor-only aggregate. + if reconstruction or not members or not members[0][3]: + return + root = members[0][3][0] + if (root.get("parallel_group_base_sequence") != members[0][1] + or root.get("parallel_group_size") != len(members)): + return + events: list[Mapping[str, Any]] = [] + executions: set[str] = set() + for offset, member in enumerate(members): + path = member[3] + expected_root = {**root, "parallel_group_index": offset} + if (not path or not _group_entry_matches(path[0], expected_root) + or member[1] != members[0][1] + offset): + return + event = self._activity_group_boundary(member) + if event is None: + return + execution = event["payload"]["activity_execution_id"] + if execution in executions: + return + executions.add(execution) + events.append(event) + self._pending = (None, _activity_group_time(events, failed_member)) + + def _activity_group_boundary(self, member: _GroupMember) -> Mapping[str, Any] | None: + index, sequence, activity_type, path = member + indices = self._by_sequence.get(sequence, []) + if index is None or not path or len(indices) not in (2, 3) or indices[-1] != index: + return None + kinds = [_kind(self.events[position]) for position in indices] + if (kinds[0] != "ActivityScheduled" or kinds[-1] not in _ACTIVITY_TERMINALS + or len(kinds) == 3 and kinds[1] != "ActivityStarted"): + return None + event = self.events[index] + execution = event["payload"].get("activity_execution_id") + if not _identity(execution) or self._activity_executions.get(execution) != indices: + return None + for position in indices: + body = self.events[position]["payload"] + if (_sequence(body) != sequence or body.get("activity_execution_id") != execution + or body.get("activity_type") != activity_type or not _group_payload_matches(body, path)): + return None + return event + 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}: diff --git a/src/durable_workflow/replay_clock_math.py b/src/durable_workflow/replay_clock_math.py new file mode 100644 index 0000000..e11d1f5 --- /dev/null +++ b/src/durable_workflow/replay_clock_math.py @@ -0,0 +1,30 @@ +"""Carbon 3.15.1's measured millisecond clock domain, without wall time.""" + +from __future__ import annotations + +import math +from datetime import datetime, timedelta, timezone + +_EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc) + + +def carbon_milliseconds(time: datetime) -> int | None: + if time.utcoffset() is None: + return None + delta = time - _EPOCH + micros = (delta.days * 86400 + delta.seconds) * 1_000_000 + delta.microseconds + # Timestamp.php forms a float microsecond value. Outside its exact integer + # range we have no proved quantization, so cannot supply a group clock. + if abs(micros) > 1 << 53: + return None + value = micros / 1000.0 + integral = math.trunc(value) + # PHP 8.4 round(..., 0, PHP_ROUND_HALF_UP): compare with the half edge + # rather than Python's ties-to-even round or a floor/truncation substitute. + if abs(value) >= abs(integral) + 0.5: + integral += -1 if value < 0 else 1 + return integral + + +def carbon_from_milliseconds(value: int) -> datetime: + return _EPOCH + timedelta(milliseconds=value) diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 7fdad3f..2e558f2 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -2862,9 +2862,12 @@ def replay_now(self) -> datetime: """Return the proved UTC instant of the supported consumed boundary. This additive clock supports scalar persisted activities, timers and - children, plus an explicit redrive start. Ordinary entry, inline/local/grouped operations - and unproved callbacks remain unavailable. Repeated reads are stable; - equal/backward recorded times retain their values and precision. Adopt + children, explicit redrive start, and validated lists of remote + activities. Group success retains the selected event's full precision; + matching caught failure uses the engine's millisecond reconstruction. + Ordinary entry, other groups, inline/local operations, unproved times + and group-only query time remain unavailable. Reads are stable, + and proved boundaries may move backwards. Adopt business-time decisions through normal workflow versioning. ``now()`` remains the legacy fixed-start accessor, and timestamps are not order IDs. """ @@ -7589,6 +7592,20 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None vals = [_typed_activity_result(leaf, value) for leaf, value in zip(cmd, vals, strict=True)] except TypedValueError as error: group_typed_failure = error + if not cmd: + ctx._business_clock.preserve_boundary() + elif group_typed_failure is None and all(isinstance(leaf, ScheduleActivity) for leaf in cmd): + selected_failure = next( + (offset for offset, value in enumerate(vals) if failed is not None and value is failed), + None, + ) + ctx._business_clock.observe_activity_group( + [(step.history_index, step.workflow_sequence, leaf.activity_type, leaf._parallel_group_path) + for step, leaf in zip(consumed_steps, cmd, strict=True)], + selected_failure, reconstruction=reconstruction_only, + ) + else: + ctx._business_clock.unavailable() if failed is not None or group_typed_failure is not None: try: _refresh_replay_logging() diff --git a/tests/business_clock_workflows.py b/tests/business_clock_workflows.py index cbc05fb..95da1ff 100644 --- a/tests/business_clock_workflows.py +++ b/tests/business_clock_workflows.py @@ -6,7 +6,7 @@ from typing import Any from durable_workflow import workflow -from durable_workflow.errors import ChildWorkflowFailed, DurableWorkflowError +from durable_workflow.errors import ActivityFailed, ChildWorkflowFailed, DurableWorkflowError from durable_workflow.workflow import WorkflowContext @@ -105,3 +105,37 @@ def clock(self) -> str | None: @workflow.query("explode") def explode(self) -> None: raise ValueError("child-clock-query-error") + + +@workflow.defn(name="tests.replay.business-clock-parallel-activity") +class ParallelActivityBusinessClockWorkflow: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + self.ctx = ctx + observed = [clock_text(ctx)] + legacy = [ctx.now().isoformat(timespec="microseconds")] + values = yield [ctx.schedule_activity("group.0", []), ctx.schedule_activity("group.1", [])] + observed.append(clock_text(ctx)) + legacy.append(ctx.now().isoformat(timespec="microseconds")) + failure = None + try: + yield [ctx.schedule_activity("group.0", []), ctx.schedule_activity("group.1", [])] + except ActivityFailed as error: + failure = {"activity_type": error.activity_type, "message": str(error)} + observed.append(clock_text(ctx)) + legacy.append(ctx.now().isoformat(timespec="microseconds")) + return {"values": values, "clock": observed, "legacy": legacy, "failure": failure} + + @workflow.query("clock") + def clock(self) -> str | None: + return clock_text(self.ctx) + + +@workflow.defn(name="tests.replay.business-clock-parallel-failure-gap") +class ParallelFailureBusinessClockWorkflow: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + try: + yield [ctx.schedule_activity("group.0", []), ctx.schedule_activity("group.1", [])] + except ActivityFailed as error: + return {"activity_type": error.activity_type, "message": str(error), "clock": clock_text(ctx), + "legacy": ctx.now().isoformat(timespec="microseconds")} + raise AssertionError("expected the recorded group failure") diff --git a/tests/fixtures/replay_clock_carbon_reference.json b/tests/fixtures/replay_clock_carbon_reference.json new file mode 100644 index 0000000..328f4d7 --- /dev/null +++ b/tests/fixtures/replay_clock_carbon_reference.json @@ -0,0 +1,93 @@ +{ + "php": "8.4.27", + "carbon": "3.15.1", + "source": "ce0e5636fae7d1e99856eb70cae4c719a284aa08", + "timestamp_source_sha256": "57b2a235656f8edad7c333087a64e45491adaaf1b01ac9c40e53185b73db356d", + "vectors": [ + { + "input": "1970-01-01T00:00:00.000000Z", + "milliseconds": 0, + "reconstructed": "1970-01-01T00:00:00.000000Z" + }, + { + "input": "1970-01-01T00:00:00.000499Z", + "milliseconds": 0, + "reconstructed": "1970-01-01T00:00:00.000000Z" + }, + { + "input": "1970-01-01T00:00:00.000500Z", + "milliseconds": 1, + "reconstructed": "1970-01-01T00:00:00.001000Z" + }, + { + "input": "1970-01-01T00:00:00.000501Z", + "milliseconds": 1, + "reconstructed": "1970-01-01T00:00:00.001000Z" + }, + { + "input": "1969-12-31T23:59:59.999501Z", + "milliseconds": 0, + "reconstructed": "1970-01-01T00:00:00.000000Z" + }, + { + "input": "1969-12-31T23:59:59.999500Z", + "milliseconds": -1, + "reconstructed": "1969-12-31T23:59:59.999000Z" + }, + { + "input": "1969-12-31T23:59:59.999499Z", + "milliseconds": -1, + "reconstructed": "1969-12-31T23:59:59.999000Z" + }, + { + "input": "1969-12-31T23:59:59.998500Z", + "milliseconds": -2, + "reconstructed": "1969-12-31T23:59:59.998000Z" + }, + { + "input": "2026-01-01T00:00:00.123100Z", + "milliseconds": 1767225600123, + "reconstructed": "2026-01-01T00:00:00.123000Z" + }, + { + "input": "2026-01-01T00:00:00.123400Z", + "milliseconds": 1767225600123, + "reconstructed": "2026-01-01T00:00:00.123000Z" + }, + { + "input": "2026-01-01T00:00:00.123499Z", + "milliseconds": 1767225600123, + "reconstructed": "2026-01-01T00:00:00.123000Z" + }, + { + "input": "2026-01-01T00:00:00.123500Z", + "milliseconds": 1767225600124, + "reconstructed": "2026-01-01T00:00:00.124000Z" + }, + { + "input": "2026-01-01T00:00:00.123501Z", + "milliseconds": 1767225600124, + "reconstructed": "2026-01-01T00:00:00.124000Z" + }, + { + "input": "2026-01-01T00:00:00.999500Z", + "milliseconds": 1767225601000, + "reconstructed": "2026-01-01T00:00:01.000000Z" + }, + { + "input": "1900-01-01T00:00:00.123499Z", + "milliseconds": -2208988799877, + "reconstructed": "1900-01-01T00:00:00.123000Z" + }, + { + "input": "1900-01-01T00:00:00.123500Z", + "milliseconds": -2208988799877, + "reconstructed": "1900-01-01T00:00:00.123000Z" + }, + { + "input": "2026-01-01T01:30:00.123500+01:30", + "milliseconds": 1767225600124, + "reconstructed": "2026-01-01T00:00:00.124000Z" + } + ] +} diff --git a/tests/fixtures/replay_clock_carbon_reference.md b/tests/fixtures/replay_clock_carbon_reference.md new file mode 100644 index 0000000..b4d3b40 --- /dev/null +++ b/tests/fixtures/replay_clock_carbon_reference.md @@ -0,0 +1,41 @@ +The adjacent JSON is byte-identical to SDK Rust's retained public reference +packet at commit `3bc747cd05cecbc47c3841193805f1a9490a4a61`, +`tests/fixtures/replay-clock-carbon-reference.json` (Git blob +`328f4d743cb74293ac6e5899fa94e820a317df9d`, 2,663 bytes, SHA-256 +`09a9051249a6dc4c0e3c177d26f058b393d563cf9f42b120c48f5d86371ea38f`). +Its 17 answers were measured with public PHP 8.4.27 and Carbon 3.15.1. +Python independently checks its answers and source bindings; it does not claim +to have rerun that PHP cell or qualified an installed Core/native tuple. + +Carbon tag `3.15.1` resolves to +`ce0e5636fae7d1e99856eb70cae4c719a284aa08`. The independently fetched +`src/Carbon/Traits/Timestamp.php` is Git blob +`43adc5b35a909be96c255b4bc0399ee75dba53e2`, SHA-256 +`57b2a235656f8edad7c333087a64e45491adaaf1b01ac9c40e53185b73db356d`. +It forms a float microsecond value and calls `round(value / 1000)`. +PHP tag `php-8.4.27`, `ext/standard/math.c`, Git blob +`09d98298104333417c233f1f6ba6d78c249b03d7`, SHA-256 +`a6454623cc0f758568724beb47096e8fe6c0d4510c87f1a2243cdcc2b65bd590`, +confirms the zero-place half-away-from-zero comparison. Reconstruction from an +integer millisecond uses the represented instant, including negative epochs. + +The Python group clock qualifies only the exact float-integer microsecond +range `abs(epoch_microseconds) <= 2**53` and represented precision up to six +fractional digits. Other ranges/precision remain unavailable. Success folds +only validated consumed members in authored descriptor order, retains the +first member on equal Carbon milliseconds, and returns its full event time. +Caught failure preserves Python's legacy exception and reconstructs time only +when that member agrees with Core's earliest-millisecond/member-index choice. +The conflicting-winner case remains an OPEN versioned compatibility gap. + +The finite cohort requires a unique schedule, at most one start and one terminal +per execution, complete matching authored Activity-all paths/counters/types and +root remote scope. It reads actual `recorded_at`, or actual `created_at` only +when the former is absent/null, as Core's grouped activity path does. It does +not substitute scalar redrive time, a start time, a sibling, or history maximum. +A conflicting `timestamp` alias or timestamp-only view is not qualified by this +model-field binding. Query body/callback group time remains unavailable because +Core's successful query group-send supplies no new time. Independent later +scalar boundaries can recover their own authority. Empty groups introduce no +new boundary. Mixed/local/select/scoped groups and unsupported lifecycles remain +unavailable; this source cohort does not complete original #215. diff --git a/tests/fixtures/replay_regressions/business-clock-parallel-activity-all.json b/tests/fixtures/replay_regressions/business-clock-parallel-activity-all.json new file mode 100644 index 0000000..e618b52 --- /dev/null +++ b/tests/fixtures/replay_regressions/business-clock-parallel-activity-all.json @@ -0,0 +1,262 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "business-clock-parallel-activity-all", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.business-clock-parallel-activity", + "input": [], + "payload_codec": "avro", + "workflow_id": "business-clock-parallel-activity-all", + "run_id": "business-clock-parallel-activity-all-run" + }, + "history": [ + { + "event_type": "WorkflowStarted", + "recorded_at": "2026-01-01T00:00:00.000000+00:00", + "payload": {}, + "sequence": 1 + }, + { + "event_type": "ActivityScheduled", + "recorded_at": "2026-01-01T00:00:00.000000+00:00", + "payload": { + "sequence": 1, + "activity_execution_id": "group-execution-0", + "activity_type": "group.0", + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0 + } + ] + }, + "sequence": 2 + }, + { + "event_type": "ActivityScheduled", + "recorded_at": "2026-01-01T00:00:00.000000+00:00", + "payload": { + "sequence": 2, + "activity_execution_id": "group-execution-1", + "activity_type": "group.1", + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1 + } + ] + }, + "sequence": 3 + }, + { + "event_type": "ActivityCompleted", + "recorded_at": "2026-01-01T00:00:03.123400+00:00", + "payload": { + "sequence": 2, + "activity_execution_id": "group-execution-1", + "activity_type": "group.1", + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1 + } + ], + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoOdmFsdWUtMQ==" + } + }, + "sequence": 4 + }, + { + "event_type": "ActivityCompleted", + "recorded_at": "2026-01-01T00:00:03.123100+00:00", + "payload": { + "sequence": 1, + "activity_execution_id": "group-execution-0", + "activity_type": "group.0", + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0 + } + ], + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoOdmFsdWUtMA==" + } + }, + "sequence": 5 + }, + { + "event_type": "ActivityScheduled", + "recorded_at": "2026-01-01T00:00:00.000000+00:00", + "payload": { + "sequence": 3, + "activity_execution_id": "group-execution-0-second", + "activity_type": "group.0", + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 0, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 0 + } + ] + }, + "sequence": 6 + }, + { + "event_type": "ActivityScheduled", + "recorded_at": "2026-01-01T00:00:00.000000+00:00", + "payload": { + "sequence": 4, + "activity_execution_id": "group-execution-1-second", + "activity_type": "group.1", + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 1 + } + ] + }, + "sequence": 7 + }, + { + "event_type": "ActivityCompleted", + "recorded_at": "2026-01-01T00:00:10Z", + "payload": { + "sequence": 4, + "activity_execution_id": "group-execution-1-second", + "activity_type": "group.1", + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 1 + } + ], + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoOdmFsdWUtMQ==" + } + }, + "sequence": 8 + }, + { + "event_type": "ActivityFailed", + "recorded_at": "2026-01-01T00:00:01.999500Z", + "payload": { + "sequence": 3, + "activity_execution_id": "group-execution-0-second", + "activity_type": "group.0", + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 0, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 0 + } + ], + "message": "failure-0" + }, + "sequence": 9 + }, + { + "event_type": "WorkflowTaskCompleted", + "recorded_at": "2099-01-01T00:00:00Z", + "payload": {}, + "sequence": 10 + } + ], + "expected": { + "command_sequence": [ + { + "type": "complete_workflow", + "result": { + "values": [ + "value-0", + "value-1" + ], + "clock": [ + null, + "2026-01-01T00:00:03.123100+00:00", + "2026-01-01T00:00:02.000000+00:00" + ], + "legacy": [ + "2026-01-01T00:00:00.000000+00:00", + "2026-01-01T00:00:00.000000+00:00", + "2026-01-01T00:00:00.000000+00:00" + ], + "failure": { + "activity_type": "group.0", + "message": "failure-0" + } + } + } + ] + } +} diff --git a/tests/fixtures/replay_regressions/business-clock-parallel-failure-selection-gap.json b/tests/fixtures/replay_regressions/business-clock-parallel-failure-selection-gap.json new file mode 100644 index 0000000..2c6c85c --- /dev/null +++ b/tests/fixtures/replay_regressions/business-clock-parallel-failure-selection-gap.json @@ -0,0 +1,135 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "business-clock-parallel-failure-selection-gap", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.business-clock-parallel-failure-gap", + "input": [], + "payload_codec": "avro", + "workflow_id": "business-clock-parallel-failure-selection-gap", + "run_id": "business-clock-parallel-failure-selection-gap-run" + }, + "history": [ + { + "event_type": "WorkflowStarted", + "recorded_at": "2026-01-01T00:00:00.000000+00:00", + "payload": {}, + "sequence": 1 + }, + { + "event_type": "ActivityScheduled", + "recorded_at": "2026-01-01T00:00:00.000000+00:00", + "payload": { + "sequence": 1, + "activity_execution_id": "group-execution-0", + "activity_type": "group.0", + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0 + } + ] + }, + "sequence": 2 + }, + { + "event_type": "ActivityScheduled", + "recorded_at": "2026-01-01T00:00:00.000000+00:00", + "payload": { + "sequence": 2, + "activity_execution_id": "group-execution-1", + "activity_type": "group.1", + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1 + } + ] + }, + "sequence": 3 + }, + { + "event_type": "ActivityFailed", + "recorded_at": "2026-01-01T00:00:01Z", + "payload": { + "sequence": 2, + "activity_execution_id": "group-execution-1", + "activity_type": "group.1", + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1 + } + ], + "message": "failure-1" + }, + "sequence": 4 + }, + { + "event_type": "ActivityFailed", + "recorded_at": "2026-01-01T00:00:03.123100+00:00", + "payload": { + "sequence": 1, + "activity_execution_id": "group-execution-0", + "activity_type": "group.0", + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0 + } + ], + "message": "failure-0" + }, + "sequence": 5 + } + ], + "expected": { + "command_sequence": [ + { + "type": "complete_workflow", + "result": { + "activity_type": "group.0", + "message": "failure-0", + "clock": null, + "legacy": "2026-01-01T00:00:00.000000+00:00" + } + } + ] + } +} diff --git a/tests/test_business_replay_clock_parallel.py b/tests/test_business_replay_clock_parallel.py new file mode 100644 index 0000000..04ab988 --- /dev/null +++ b/tests/test_business_replay_clock_parallel.py @@ -0,0 +1,417 @@ +from __future__ import annotations + +import hashlib +import json +from copy import deepcopy +from datetime import datetime, timedelta, timezone +from itertools import permutations +from pathlib import Path +from typing import Any + +import pytest + +from durable_workflow import serializer, workflow +from durable_workflow.errors import ActivityFailed, NonDeterministicReplayError, TypedValueError +from durable_workflow.replay_clock_math import carbon_from_milliseconds, carbon_milliseconds +from durable_workflow.workflow import WorkflowContext, query_state, replay +from tests.business_clock_workflows import clock_text +from tests.test_business_replay_clock import result + +START = "2026-01-01T00:00:00.000000+00:00" +FIRST = "2026-01-01T00:00:03.123100+00:00" +TIED = "2026-01-01T00:00:03.123400+00:00" +REFERENCE = Path(__file__).parent / "fixtures" / "replay_clock_carbon_reference.json" +VECTORS = json.loads(REFERENCE.read_text())["vectors"] + + +def test_retained_carbon_reference_identity_is_exact() -> None: + assert hashlib.sha256(REFERENCE.read_bytes()).hexdigest() == ( + "09a9051249a6dc4c0e3c177d26f058b393d563cf9f42b120c48f5d86371ea38f" + ) + reference = json.loads(REFERENCE.read_text()) + assert (reference["php"], reference["carbon"], reference["source"], reference["timestamp_source_sha256"]) == ( + "8.4.27", "3.15.1", "ce0e5636fae7d1e99856eb70cae4c719a284aa08", + "57b2a235656f8edad7c333087a64e45491adaaf1b01ac9c40e53185b73db356d", + ) + + +@pytest.mark.parametrize("vector", VECTORS, ids=lambda vector: vector["input"]) +def test_actual_carbon_rounding_and_reconstruction_vectors(vector: dict[str, Any]) -> None: + time = datetime.fromisoformat(vector["input"].replace("Z", "+00:00")) + observed = carbon_milliseconds(time) + assert observed == vector["milliseconds"] + assert carbon_from_milliseconds(vector["milliseconds"]).isoformat(timespec="microseconds").replace( + "+00:00", "Z", + ) == vector["reconstructed"] + + +@pytest.mark.parametrize("sign", [-1, 1]) +def test_unproved_float_microsecond_range_stays_unavailable(sign: int) -> None: + time = datetime(1970, 1, 1, tzinfo=timezone.utc) + timedelta(microseconds=sign * ((1 << 53) + 1)) + assert carbon_milliseconds(time) is None + + +def group_path(base: int, size: int, index: int, *, nested: bool = False) -> list[dict[str, Any]]: + def entry(start: int, count: int, position: int) -> dict[str, Any]: + return {"parallel_group_id": f"parallel-activities:{start}:{count}", "parallel_group_kind": "activity", + "parallel_group_base_sequence": start, "parallel_group_size": count, "parallel_group_index": position} + + path = [entry(base, size, index)] + if nested and index: + path.append(entry(base + 1, size - 1, index - 1)) + return path + + +def history( + times: list[Any] | None = None, *, failures: set[int] | None = None, order: tuple[int, ...] | None = None, + nested: bool = False, +) -> list[dict[str, Any]]: + times = [FIRST, TIED] if times is None else times + failures = failures or set() + size = len(times) + rows: list[dict[str, Any]] = [{"event_type": "WorkflowStarted", "recorded_at": START, "payload": {}}] + terminals = [] + for index, time in enumerate(times): + path = group_path(1, size, index, nested=nested) + payload = {"sequence": index + 1, "activity_execution_id": f"group-execution-{index}", + "activity_type": f"group.{index}", **path[-1], "parallel_group_path": path} + rows.append({"event_type": "ActivityScheduled", "recorded_at": START, "payload": deepcopy(payload)}) + if index in failures: + payload["message"] = f"failure-{index}" + else: + payload["result"] = serializer.envelope(f"value-{index}") + terminals.append({"event_type": "ActivityFailed" if index in failures else "ActivityCompleted", + "recorded_at": time, "payload": payload}) + rows.extend(terminals[index] for index in (order if order is not None else range(size))) + return rows + + +@workflow.defn(name="tests.business-clock-parallel-probe") +class ParallelProbe: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + self.ctx = ctx + commands = [ctx.schedule_activity(f"group.{index}", []) for index in range(self.size)] + if self.nested: + commands = [commands[0], commands[1:]] + self.failure = None + try: + self.values = yield commands + except ActivityFailed as error: + self.values = None + self.failure = (error.activity_type, str(error)) + self.time = clock_text(ctx) + return {"values": self.values, "failure": self.failure, "clock": self.time, + "repeated": clock_text(ctx) == self.time, "legacy": ctx.now().isoformat(timespec="microseconds")} + + size = 2 + nested = False + + @workflow.query("clock") + def clock(self) -> str | None: + return clock_text(self.ctx) + + @workflow.query("body-clock") + def body_clock(self) -> str | None: + return self.time + + +@pytest.mark.parametrize("order", [(0, 1), (1, 0)]) +def test_success_uses_authored_tie_order_and_full_precision_not_arrival_or_history_max(order: tuple[int, ...]) -> None: + rows = history(order=order) + rows.append({"event_type": "WorkflowTaskCompleted", "recorded_at": "2099-01-01T00:00:00Z", "payload": {}}) + original = deepcopy(rows) + expected = {"values": ["value-0", "value-1"], "failure": None, "clock": FIRST, + "repeated": True, "legacy": START} + assert result(rows, ParallelProbe) == expected + assert result(rows, ParallelProbe) == expected + assert rows == original + + +@pytest.mark.parametrize("times,expected", [ + ([FIRST, "2026-01-01T00:00:03.123500Z"], "2026-01-01T00:00:03.123500+00:00"), + ([TIED, FIRST], TIED), + (["1969-12-31T23:59:59.999500Z", "1969-12-31T23:59:59.999501Z"], + "1969-12-31T23:59:59.999501+00:00"), +]) +def test_success_uses_carbon_millisecond_comparison_without_replacing_full_precision( + times: list[str], expected: str, +) -> None: + assert result(history(times), ParallelProbe)["clock"] == expected + + +@pytest.mark.parametrize("order", [(0, 1), (1, 0)]) +def test_caught_failure_ignores_successful_sibling_time_and_reconstructs_carbon_milliseconds( + order: tuple[int, ...], +) -> None: + value = result(history(["1969-12-31T23:59:59.999500Z", "2099-01-01T00:00:00Z"], + failures={0}, order=order), ParallelProbe) + assert value == {"values": None, "failure": ("group.0", "failure-0"), + "clock": "1969-12-31T23:59:59.999000+00:00", "repeated": True, "legacy": START} + + +@pytest.mark.parametrize("times,expected", [ + ([FIRST, TIED], "2026-01-01T00:00:03.123000+00:00"), + ([TIED, FIRST], "2026-01-01T00:00:03.123000+00:00"), + ([FIRST, "2026-01-01T00:00:01Z"], None), +]) +def test_multiple_failures_preserve_legacy_winner_and_refuse_a_different_core_winner( + times: list[str], expected: str | None, +) -> None: + value = result(history(times, failures={0, 1}, order=(1, 0)), ParallelProbe) + assert value["failure"] == ("group.0", "failure-0") + assert value["clock"] == expected + + +@pytest.mark.parametrize("order", list(permutations(range(3)))) +def test_nested_all_activity_paths_follow_authored_order(order: tuple[int, ...]) -> None: + class Nested(ParallelProbe): + size = 3 + nested = True + + value = result(history([FIRST, TIED, "2026-01-01T00:00:02Z"], order=order, nested=True), Nested) + assert value["values"] == ["value-0", ["value-1", "value-2"]] + assert value["clock"] == FIRST + + +@pytest.mark.parametrize("bad", [None, False, 17, "", "bad", "2026-02-30T00:00:00Z", "2026-01-01T24:00:00Z", + "2026-01-01T00:00:00.1234567Z", "9999-01-01T00:00:00Z"]) +def test_missing_invalid_or_unproved_required_time_does_not_borrow_a_sibling(bad: Any) -> None: + assert result(history([FIRST, bad]), ParallelProbe)["clock"] is None + + +def test_literal_core_time_precedence_and_actual_created_at_fallback_are_family_specific() -> None: + rows = history() + rows[-2]["payload"]["reused_recorded_at"] = "2099-01-01T00:00:00Z" + rows[-2]["created_at"] = "2098-01-01T00:00:00Z" + assert result(rows, ParallelProbe)["clock"] == FIRST + rows[-2]["recorded_at"] = None + rows[-2]["created_at"] = FIRST + assert result(rows, ParallelProbe)["clock"] == FIRST + rows[-2]["recorded_at"] = "bad" + assert result(rows, ParallelProbe)["clock"] is None + + +@pytest.mark.parametrize("query", ["clock", "body-clock"]) +@pytest.mark.parametrize("failed", [False, True]) +def test_query_body_and_callback_cannot_borrow_the_executor_group_aggregate(query: str, failed: bool) -> None: + rows = history(failures={0} if failed else set()) + assert result(rows, ParallelProbe)["clock"] is not None + assert query_state(ParallelProbe, rows, [], query, payload_codec="avro") is None + + +@pytest.mark.parametrize("field,value", [("activity_execution_id", "foreign"), ("activity_execution_id", None), + ("sequence", "1"), ("sequence", True), ("workflow_sequence", "1"), + ("parallel_group_base_sequence", "1"), ("parallel_group_index", True)]) +def test_unproved_opening_keeps_the_legacy_outcome_without_certifying_time(field: str, value: Any) -> None: + rows = history() + rows[1]["payload"][field] = value + actual = result(rows, ParallelProbe) + assert actual["values"] == ["value-0", "value-1"] + assert actual["clock"] is None + + +def test_duplicate_numeric_claim_and_unindexed_same_execution_cannot_evade_group_qualification() -> None: + for sequence in ("1", "not-a-counter"): + rows = history() + duplicate = deepcopy(rows[1]) + duplicate["payload"]["sequence"] = sequence + rows.insert(1, duplicate) + assert result(rows, ParallelProbe)["clock"] is None + + +def test_absent_lifecycle_and_duplicate_starts_preserve_values_without_proving_time() -> None: + rows = history() + rows.pop(1) + assert result(rows, ParallelProbe)["clock"] is None + rows = history() + started = deepcopy(rows[1]) + started["event_type"] = "ActivityStarted" + rows[2:2] = [started, deepcopy(started)] + assert result(rows, ParallelProbe)["clock"] is None + + +def test_original_activity_type_mismatch_remains_a_replay_error() -> None: + rows = history() + for row in rows[1:]: + if row["payload"]["sequence"] == 1: + row["payload"]["activity_type"] = "different" + with pytest.raises(NonDeterministicReplayError): + result(rows, ParallelProbe) + + +@pytest.mark.parametrize("kind", ["ActivityFailed", "ActivityTimedOut", "ActivityCancelled"]) +def test_each_persisted_failure_kind_keeps_its_exception_and_group_reconstruction(kind: str) -> None: + rows = history(failures={0}) + rows[-2]["event_type"] = kind + value = result(rows, ParallelProbe) + assert value["failure"] == ("group.0", "failure-0") + assert value["clock"] == "2026-01-01T00:00:03.123000+00:00" + + +def test_failed_member_needs_its_time_but_does_not_require_a_successful_sibling_time() -> None: + value = result(history([FIRST, None], failures={0}), ParallelProbe) + assert value["clock"] == "2026-01-01T00:00:03.123000+00:00" + assert result(history([FIRST, None], failures={0, 1}), ParallelProbe)["clock"] is None + + +def test_none_result_remains_a_success_and_does_not_become_a_selected_failure() -> None: + rows = history() + rows[-2]["payload"]["result"] = serializer.envelope(None) + value = result(rows, ParallelProbe) + assert value["values"] == [None, "value-1"] + assert value["failure"] is None + assert value["clock"] == FIRST + + +@pytest.mark.parametrize("empty_path", [False, True]) +def test_complete_flat_metadata_also_binds_without_a_nested_path(empty_path: bool) -> None: + rows = history() + for row in rows[1:]: + if empty_path: + row["payload"]["parallel_group_path"] = [] + else: + row["payload"].pop("parallel_group_path") + assert result(rows, ParallelProbe)["clock"] == FIRST + + +def test_unrepresented_or_conflicting_time_alias_stays_unavailable() -> None: + rows = history() + rows[-1]["timestamp"] = "2099-01-01T00:00:00Z" + assert result(rows, ParallelProbe)["clock"] is None + rows[-1]["timestamp"] = rows[-1].pop("recorded_at") + assert result(rows, ParallelProbe)["clock"] is None # No model-field carrier to qualify. + + +def test_empty_group_preserves_a_proved_boundary_and_a_later_group_can_move_backwards() -> None: + class Multiple: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + yield ctx.schedule_activity("group.0", []) + before = clock_text(ctx) + yield [] + after_empty = clock_text(ctx) + values = yield [ctx.schedule_activity("group.0", []), ctx.schedule_activity("group.1", [])] + return before, after_empty, clock_text(ctx), values + + rows = history(["2026-01-01T00:00:08Z"]) + for row in rows[1:]: + for field in _metadata_fields(): + row["payload"].pop(field) + later = history() + for row in later[1:]: + row["payload"]["sequence"] += 1 + row["payload"]["activity_execution_id"] += "-later" + path = group_path(2, 2, row["payload"]["sequence"] - 2) + row["payload"].update(**path[-1], parallel_group_path=path) + rows.extend(later[1:]) + assert result(rows, Multiple) == ( + "2026-01-01T00:00:08.000000+00:00", "2026-01-01T00:00:08.000000+00:00", FIRST, + ["value-0", "value-1"], + ) + + +def _metadata_fields() -> tuple[str, ...]: + return ("parallel_group_id", "parallel_group_kind", "parallel_group_base_sequence", "parallel_group_size", + "parallel_group_index", "parallel_group_path") + + +def test_pending_and_unmarked_legacy_groups_do_not_receive_a_proved_clock() -> None: + rows = history() + rows.pop() + pending = replay(ParallelProbe, rows, [], payload_codec="avro") + try: + assert pending.commands and all(getattr(command, "activity_type", None) for command in pending.commands) + finally: + pending.close() + rows = history() + for row in rows[1:]: + for field in _metadata_fields(): + row["payload"].pop(field) + assert result(rows, ParallelProbe)["clock"] is None + + +def test_mixed_and_recorded_local_groups_remain_explicitly_unqualified() -> None: + class Mixed: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + values = yield [ctx.schedule_activity("group.0", []), ctx.start_timer(1)] + return values, clock_text(ctx) + + rows = history() + for row in rows[1:]: + payload = row["payload"] + payload["parallel_group_id"] = "parallel-calls:1:2" + payload["parallel_group_kind"] = "mixed" + for entry in payload["parallel_group_path"]: + entry.update(parallel_group_id="parallel-calls:1:2", parallel_group_kind="mixed") + if payload["sequence"] == 2: + row["event_type"] = "TimerScheduled" if row["event_type"] == "ActivityScheduled" else "TimerFired" + for field in ("activity_type", "activity_execution_id", "result"): + payload.pop(field, None) + payload.update(timer_kind="durable_timer", timer_id="group-timer", delay_seconds=1) + assert result(rows, Mixed) == (["value-0", None], None) + + class Local: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + values = yield [ctx.local_activity("prepared.first", []), ctx.local_activity("prepared.second", [])] + return values, clock_text(ctx) + + fixture = json.loads((REFERENCE.parent / "replay_regressions/prepared-local-group-cold-results.json").read_text()) + assert result(fixture["history"], Local) == (["fast-value", "fast-value"], None) + + +def test_typed_adapter_failure_cannot_claim_a_successful_group_clock() -> None: + class Typed: + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + try: + yield [ctx.schedule_activity("group.0", [], result_type=int), ctx.schedule_activity("group.1", [])] + except TypedValueError: + return clock_text(ctx) + raise AssertionError("the original typed-value error was not delivered") + + assert result(history(), Typed) is None + + +def test_a_later_independent_scalar_boundary_recovers_query_authority_after_a_group() -> None: + class Later(ParallelProbe): + def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def] + self.ctx = ctx + yield [ctx.schedule_activity("group.0", []), ctx.schedule_activity("group.1", [])] + group_time = clock_text(ctx) + yield ctx.schedule_activity("after", []) + self.time = clock_text(ctx) + return group_time, self.time + + rows = history() + later = {"sequence": 3, "activity_execution_id": "after-group", "activity_type": "after"} + rows.extend([ + {"event_type": "ActivityScheduled", "payload": dict(later)}, + {"event_type": "ActivityCompleted", "recorded_at": "2026-01-01T00:00:02Z", + "payload": {**later, "result": serializer.envelope("after")}}, + ]) + assert result(rows, Later) == (FIRST, "2026-01-01T00:00:02.000000+00:00") + assert query_state(Later, rows, [], "clock", payload_codec="avro") == "2026-01-01T00:00:02.000000+00:00" + + +def test_optional_started_events_do_not_contribute_their_times() -> None: + rows = history() + starts = [] + for row in rows[1:3]: + started = deepcopy(row) + started.update(event_type="ActivityStarted", recorded_at="2098-01-01T00:00:00Z") + starts.append(started) + rows[3:3] = starts + assert result(rows, ParallelProbe)["clock"] == FIRST + + +@pytest.mark.parametrize("field,value", [("parallel_group_mode", "unknown"), + ("selection_member_key", "other"), ("parallel_group_base_sequence", "1")]) +def test_unqualified_mode_selection_alias_and_raw_path_counter_cannot_be_overwritten_by_a_later_valid_row( + field: str, value: Any, +) -> None: + rows = history() + if field == "parallel_group_base_sequence": + rows[1]["payload"]["parallel_group_path"][0][field] = value + else: + rows[1]["payload"][field] = value + observed = result(rows, ParallelProbe) + assert observed["values"] == ["value-0", "value-1"] + assert observed["clock"] is None diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index e025720..87b50e3 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -28,6 +28,8 @@ from tests.business_clock_workflows import ( ChildBusinessClockWorkflow, NativeRedriveBusinessClockWorkflow, + ParallelActivityBusinessClockWorkflow, + ParallelFailureBusinessClockWorkflow, PrefixedCoreTimerBusinessClockWorkflow, ScalarBusinessClockWorkflow, ) @@ -378,6 +380,8 @@ def run(self, ctx: WorkflowContext, message: str) -> None: WORKFLOWS = [ ChildBusinessClockWorkflow, + ParallelActivityBusinessClockWorkflow, + ParallelFailureBusinessClockWorkflow, NativeRedriveBusinessClockWorkflow, PrefixedCoreTimerBusinessClockWorkflow, ScalarBusinessClockWorkflow, From 4530321ca9f86bbcd5a0311d783d382c9a4561f8 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 6 Nov 2026 07:42:41 +0000 Subject: [PATCH 2/2] fix: use exact canonical-v1 replay clock milliseconds --- src/durable_workflow/replay_clock.py | 14 +- src/durable_workflow/replay_clock_math.py | 25 +- src/durable_workflow/workflow.py | 2 +- .../replay_clock_canonical_v1_reference.json | 255 +++++++++++++++++ .../fixtures/replay_clock_carbon_reference.md | 32 ++- ...s-clock-parallel-integer-milliseconds.json | 262 ++++++++++++++++++ tests/test_business_replay_clock_parallel.py | 61 +++- 7 files changed, 606 insertions(+), 45 deletions(-) create mode 100644 tests/fixtures/replay_clock_canonical_v1_reference.json create mode 100644 tests/fixtures/replay_regressions/business-clock-parallel-integer-milliseconds.json diff --git a/src/durable_workflow/replay_clock.py b/src/durable_workflow/replay_clock.py index 7b6ceb1..171a709 100644 --- a/src/durable_workflow/replay_clock.py +++ b/src/durable_workflow/replay_clock.py @@ -14,7 +14,7 @@ from typing import Any, ParamSpec, TypeGuard, TypeVar from .errors import ReplayClockUnavailable -from .replay_clock_math import carbon_from_milliseconds, carbon_milliseconds +from .replay_clock_math import canonical_milliseconds, from_canonical_milliseconds _P = ParamSpec("_P") _T = TypeVar("_T") @@ -194,9 +194,9 @@ def _activity_group_time(events: Sequence[Mapping[str, Any]], failed_member: int if failed != (failed_member is not None): if failed: return None - continue # Successful siblings do not choose Core's failure. + continue # Successful siblings do not choose the failure clock. time = _group_event_time(event) - milliseconds = carbon_milliseconds(time) if time is not None else None + milliseconds = canonical_milliseconds(time) if time is not None else None if time is None or milliseconds is None: return None candidates.append((milliseconds, time, offset)) @@ -204,10 +204,10 @@ def _activity_group_time(events: Sequence[Mapping[str, Any]], failed_member: int return None if failed_member is not None: winner = min(candidates, key=lambda candidate: candidate[0]) - # Preserve the existing SDK exception even when Core chooses another - # failed member. That open compatibility gap has no clock authority. - return carbon_from_milliseconds(winner[0]) if winner[2] == failed_member else None - # max retains the first descriptor on an equal millisecond value. Keep its + # Preserve the existing SDK exception when the canonical-v1 selector + # chooses another member. That open outcome gap has no clock authority. + return from_canonical_milliseconds(winner[0]) if winner[2] == failed_member else None + # max retains the first descriptor on equal canonical milliseconds. Keep its # full event precision, never the last row or a maximum over all history. return max(candidates, key=lambda candidate: candidate[0])[1] diff --git a/src/durable_workflow/replay_clock_math.py b/src/durable_workflow/replay_clock_math.py index e11d1f5..5a1cd63 100644 --- a/src/durable_workflow/replay_clock_math.py +++ b/src/durable_workflow/replay_clock_math.py @@ -1,30 +1,23 @@ -"""Carbon 3.15.1's measured millisecond clock domain, without wall time.""" +"""Exact canonical-v1 milliseconds for the unreleased business replay clock.""" from __future__ import annotations -import math from datetime import datetime, timedelta, timezone _EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc) -def carbon_milliseconds(time: datetime) -> int | None: +def canonical_milliseconds(time: datetime) -> int | None: if time.utcoffset() is None: return None delta = time - _EPOCH micros = (delta.days * 86400 + delta.seconds) * 1_000_000 + delta.microseconds - # Timestamp.php forms a float microsecond value. Outside its exact integer - # range we have no proved quantization, so cannot supply a group clock. - if abs(micros) > 1 << 53: - return None - value = micros / 1000.0 - integral = math.trunc(value) - # PHP 8.4 round(..., 0, PHP_ROUND_HALF_UP): compare with the half edge - # rather than Python's ties-to-even round or a floor/truncation substitute. - if abs(value) >= abs(integral) + 0.5: - integral += -1 if value < 0 else 1 - return integral + magnitude = (abs(micros) + 500) // 1000 + return -magnitude if micros < 0 else magnitude -def carbon_from_milliseconds(value: int) -> datetime: - return _EPOCH + timedelta(milliseconds=value) +def from_canonical_milliseconds(value: int) -> datetime | None: + try: + return _EPOCH + timedelta(milliseconds=value) + except OverflowError: + return None diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 2e558f2..8999efe 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -2864,7 +2864,7 @@ def replay_now(self) -> datetime: This additive clock supports scalar persisted activities, timers and children, explicit redrive start, and validated lists of remote activities. Group success retains the selected event's full precision; - matching caught failure uses the engine's millisecond reconstruction. + matching caught failure uses canonical-v1 integer milliseconds. Ordinary entry, other groups, inline/local operations, unproved times and group-only query time remain unavailable. Reads are stable, and proved boundaries may move backwards. Adopt diff --git a/tests/fixtures/replay_clock_canonical_v1_reference.json b/tests/fixtures/replay_clock_canonical_v1_reference.json new file mode 100644 index 0000000..9fefeab --- /dev/null +++ b/tests/fixtures/replay_clock_canonical_v1_reference.json @@ -0,0 +1,255 @@ +{ + "numeric_contract": "canonical-v1", + "authority": "https://github.com/durable-workflow/.github/issues/215#issuecomment-6679568700", + "formula": "M = sign(U) * ((abs(U) + 500) div 1000)", + "vectors": [ + { + "input": "1969-12-31T23:59:59.000499Z", + "utc_microseconds": -999501, + "milliseconds": -1000, + "reconstructed": "1969-12-31T23:59:59.000000Z" + }, + { + "input": "1969-12-31T23:59:59.000500Z", + "utc_microseconds": -999500, + "milliseconds": -1000, + "reconstructed": "1969-12-31T23:59:59.000000Z" + }, + { + "input": "1969-12-31T23:59:59.000501Z", + "utc_microseconds": -999499, + "milliseconds": -999, + "reconstructed": "1969-12-31T23:59:59.001000Z" + }, + { + "input": "1969-12-31T23:59:59.998500Z", + "utc_microseconds": -1500, + "milliseconds": -2, + "reconstructed": "1969-12-31T23:59:59.998000Z" + }, + { + "input": "1969-12-31T23:59:59.999499Z", + "utc_microseconds": -501, + "milliseconds": -1, + "reconstructed": "1969-12-31T23:59:59.999000Z" + }, + { + "input": "1969-12-31T23:59:59.999500Z", + "utc_microseconds": -500, + "milliseconds": -1, + "reconstructed": "1969-12-31T23:59:59.999000Z" + }, + { + "input": "1969-12-31T23:59:59.999501Z", + "utc_microseconds": -499, + "milliseconds": 0, + "reconstructed": "1970-01-01T00:00:00.000000Z" + }, + { + "input": "1969-12-31T23:59:59.999999Z", + "utc_microseconds": -1, + "milliseconds": 0, + "reconstructed": "1970-01-01T00:00:00.000000Z" + }, + { + "input": "1970-01-01T00:00:00.000000Z", + "utc_microseconds": 0, + "milliseconds": 0, + "reconstructed": "1970-01-01T00:00:00.000000Z" + }, + { + "input": "1970-01-01T00:00:00.000001Z", + "utc_microseconds": 1, + "milliseconds": 0, + "reconstructed": "1970-01-01T00:00:00.000000Z" + }, + { + "input": "1970-01-01T00:00:00.000499Z", + "utc_microseconds": 499, + "milliseconds": 0, + "reconstructed": "1970-01-01T00:00:00.000000Z" + }, + { + "input": "1970-01-01T00:00:00.000500Z", + "utc_microseconds": 500, + "milliseconds": 1, + "reconstructed": "1970-01-01T00:00:00.001000Z" + }, + { + "input": "1970-01-01T00:00:00.000501Z", + "utc_microseconds": 501, + "milliseconds": 1, + "reconstructed": "1970-01-01T00:00:00.001000Z" + }, + { + "input": "1970-01-01T00:00:00.123400Z", + "utc_microseconds": 123400, + "milliseconds": 123, + "reconstructed": "1970-01-01T00:00:00.123000Z" + }, + { + "input": "1970-01-01T00:00:00.123499Z", + "utc_microseconds": 123499, + "milliseconds": 123, + "reconstructed": "1970-01-01T00:00:00.123000Z" + }, + { + "input": "1970-01-01T00:00:00.123500Z", + "utc_microseconds": 123500, + "milliseconds": 124, + "reconstructed": "1970-01-01T00:00:00.124000Z" + }, + { + "input": "1970-01-01T00:00:00.999500Z", + "utc_microseconds": 999500, + "milliseconds": 1000, + "reconstructed": "1970-01-01T00:00:01.000000Z" + }, + { + "input": "1970-01-01T01:30:00.000500+01:30", + "utc_microseconds": 500, + "milliseconds": 1, + "reconstructed": "1970-01-01T00:00:00.001000Z" + }, + { + "input": "1970-01-01T01:29:59.999500+01:30", + "utc_microseconds": -500, + "milliseconds": -1, + "reconstructed": "1969-12-31T23:59:59.999000Z" + }, + { + "input": "1969-12-31T22:30:00.000500-01:30", + "utc_microseconds": 500, + "milliseconds": 1, + "reconstructed": "1970-01-01T00:00:00.001000Z" + }, + { + "input": "1900-01-01T00:00:00.123500Z", + "utc_microseconds": -2208988799876500, + "milliseconds": -2208988799877, + "reconstructed": "1900-01-01T00:00:00.123000Z" + }, + { + "input": "1600-01-01T00:00:00.123500Z", + "utc_microseconds": -11676095999876500, + "milliseconds": -11676095999877, + "reconstructed": "1600-01-01T00:00:00.123000Z" + }, + { + "input": "1600-01-01T00:00:00.123501Z", + "utc_microseconds": -11676095999876499, + "milliseconds": -11676095999876, + "reconstructed": "1600-01-01T00:00:00.124000Z" + }, + { + "input": "2026-01-01T00:00:00.123499Z", + "utc_microseconds": 1767225600123499, + "milliseconds": 1767225600123, + "reconstructed": "2026-01-01T00:00:00.123000Z" + }, + { + "input": "2026-01-01T00:00:00.123500Z", + "utc_microseconds": 1767225600123500, + "milliseconds": 1767225600124, + "reconstructed": "2026-01-01T00:00:00.124000Z" + }, + { + "input": "2300-01-01T00:00:00.123499Z", + "utc_microseconds": 10413792000123499, + "milliseconds": 10413792000123, + "reconstructed": "2300-01-01T00:00:00.123000Z" + }, + { + "input": "2300-01-01T00:00:00.123500Z", + "utc_microseconds": 10413792000123500, + "milliseconds": 10413792000124, + "reconstructed": "2300-01-01T00:00:00.124000Z" + }, + { + "input": "0001-01-01T00:00:00.123500Z", + "utc_microseconds": -62135596799876500, + "milliseconds": -62135596799877, + "reconstructed": "0001-01-01T00:00:00.123000Z" + }, + { + "input": "9999-01-01T00:00:00.123500Z", + "utc_microseconds": 253370764800123500, + "milliseconds": 253370764800124, + "reconstructed": "9999-01-01T00:00:00.124000Z" + }, + { + "input": "9999-12-31T23:59:59.999499Z", + "utc_microseconds": 253402300799999499, + "milliseconds": 253402300799999, + "reconstructed": "9999-12-31T23:59:59.999000Z" + }, + { + "input": "9999-12-31T23:59:59.999500Z", + "utc_microseconds": 253402300799999500, + "milliseconds": 253402300800000, + "reconstructed": null + } + ], + "groups": [ + { + "id": "authored-tie", + "times": [ + "2026-01-01T00:00:00.123400Z", + "2026-01-01T00:00:00.123499Z" + ], + "success_member": 0, + "failure_member": 0, + "failure_time": "2026-01-01T00:00:00.123000Z" + }, + { + "id": "positive-half", + "times": [ + "2026-01-01T00:00:00.123499Z", + "2026-01-01T00:00:00.123500Z" + ], + "success_member": 1, + "failure_member": 0, + "failure_time": "2026-01-01T00:00:00.123000Z" + }, + { + "id": "negative-half", + "times": [ + "1969-12-31T23:59:59.999501Z", + "1969-12-31T23:59:59.999500Z" + ], + "success_member": 0, + "failure_member": 1, + "failure_time": null + }, + { + "id": "large-tie", + "times": [ + "2300-01-01T00:00:00.123400Z", + "2300-01-01T00:00:00.123499Z" + ], + "success_member": 0, + "failure_member": 0, + "failure_time": "2300-01-01T00:00:00.123000Z" + }, + { + "id": "large-half", + "times": [ + "2300-01-01T00:00:00.123499Z", + "2300-01-01T00:00:00.123500Z" + ], + "success_member": 1, + "failure_member": 0, + "failure_time": "2300-01-01T00:00:00.123000Z" + }, + { + "id": "large-pre-epoch", + "times": [ + "1600-01-01T00:00:00.123500Z", + "1600-01-01T00:00:00.123501Z" + ], + "success_member": 1, + "failure_member": 0, + "failure_time": "1600-01-01T00:00:00.123000Z" + } + ] +} diff --git a/tests/fixtures/replay_clock_carbon_reference.md b/tests/fixtures/replay_clock_carbon_reference.md index b4d3b40..f59b62e 100644 --- a/tests/fixtures/replay_clock_carbon_reference.md +++ b/tests/fixtures/replay_clock_carbon_reference.md @@ -16,17 +16,29 @@ It forms a float microsecond value and calls `round(value / 1000)`. PHP tag `php-8.4.27`, `ext/standard/math.c`, Git blob `09d98298104333417c233f1f6ba6d78c249b03d7`, SHA-256 `a6454623cc0f758568724beb47096e8fe6c0d4510c87f1a2243cdcc2b65bd590`, -confirms the zero-place half-away-from-zero comparison. Reconstruction from an -integer millisecond uses the represented instant, including negative epochs. +binds that particular measured numeric profile. PHP's implementation of +`round()` changed in 8.4; these observations do not establish parity across +all supported PHP runtimes. The retained JSON and its expected answers stay +unchanged. -The Python group clock qualifies only the exact float-integer microsecond -range `abs(epoch_microseconds) <= 2**53` and represented precision up to six -fractional digits. Other ranges/precision remain unavailable. Success folds -only validated consumed members in authored descriptor order, retains the -first member on equal Carbon milliseconds, and returns its full event time. -Caught failure preserves Python's legacy exception and reconstructs time only -when that member agrees with Core's earliest-millisecond/member-index choice. -The conflicting-winner case remains an OPEN versioned compatibility gap. +The separate `replay_clock_canonical_v1_reference.json` defines the unreleased +accessor's canonical-v1 contract selected in +[central #215](https://github.com/durable-workflow/.github/issues/215#issuecomment-6679568700). +For an authoritative signed integer UTC microsecond value `U`, it uses +`M = sign(U) * ((abs(U) + 500) div 1000)`, with nonnegative integer division. +There is no float conversion, runtime rounding or wall-time input. Fixed +answers cover offsets, positive and negative half boundaries, large dates, +pre-epoch dates and reconstruction overflow. Scalar event times are not +quantized. This is a new additive-clock contract, not a claim of universal +legacy Core `Workflow::now()` parity or published conformance. + +Success folds only validated consumed members in authored descriptor order, +retains the first member on equal canonical milliseconds and returns its full +event time. Caught failure preserves Python's legacy exception and reconstructs +the selected canonical millisecond only when that member agrees with the +canonical earliest-millisecond/member-index choice. An unrepresentable +reconstruction stays unavailable. The conflicting-winner case remains an OPEN +versioned compatibility gap; no result or exception is rewritten. The finite cohort requires a unique schedule, at most one start and one terminal per execution, complete matching authored Activity-all paths/counters/types and diff --git a/tests/fixtures/replay_regressions/business-clock-parallel-integer-milliseconds.json b/tests/fixtures/replay_regressions/business-clock-parallel-integer-milliseconds.json new file mode 100644 index 0000000..0f71a4e --- /dev/null +++ b/tests/fixtures/replay_regressions/business-clock-parallel-integer-milliseconds.json @@ -0,0 +1,262 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "business-clock-parallel-integer-milliseconds", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.business-clock-parallel-activity", + "input": [], + "payload_codec": "avro", + "workflow_id": "business-clock-parallel-integer-milliseconds", + "run_id": "business-clock-parallel-integer-milliseconds-run" + }, + "history": [ + { + "event_type": "WorkflowStarted", + "recorded_at": "2026-01-01T00:00:00.000000+00:00", + "payload": {}, + "sequence": 1 + }, + { + "event_type": "ActivityScheduled", + "recorded_at": "2026-01-01T00:00:00.000000+00:00", + "payload": { + "sequence": 1, + "activity_execution_id": "group-execution-0", + "activity_type": "group.0", + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0 + } + ] + }, + "sequence": 2 + }, + { + "event_type": "ActivityScheduled", + "recorded_at": "2026-01-01T00:00:00.000000+00:00", + "payload": { + "sequence": 2, + "activity_execution_id": "group-execution-1", + "activity_type": "group.1", + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1 + } + ] + }, + "sequence": 3 + }, + { + "event_type": "ActivityCompleted", + "recorded_at": "2300-01-01T00:00:00.123500Z", + "payload": { + "sequence": 2, + "activity_execution_id": "group-execution-1", + "activity_type": "group.1", + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 1 + } + ], + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoOdmFsdWUtMQ==" + } + }, + "sequence": 4 + }, + { + "event_type": "ActivityCompleted", + "recorded_at": "2300-01-01T00:00:00.123499Z", + "payload": { + "sequence": 1, + "activity_execution_id": "group-execution-0", + "activity_type": "group.0", + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:1:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 1, + "parallel_group_size": 2, + "parallel_group_index": 0 + } + ], + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoOdmFsdWUtMA==" + } + }, + "sequence": 5 + }, + { + "event_type": "ActivityScheduled", + "recorded_at": "2026-01-01T00:00:00.000000+00:00", + "payload": { + "sequence": 3, + "activity_execution_id": "group-execution-0-second", + "activity_type": "group.0", + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 0, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 0 + } + ] + }, + "sequence": 6 + }, + { + "event_type": "ActivityScheduled", + "recorded_at": "2026-01-01T00:00:00.000000+00:00", + "payload": { + "sequence": 4, + "activity_execution_id": "group-execution-1-second", + "activity_type": "group.1", + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 1 + } + ] + }, + "sequence": 7 + }, + { + "event_type": "ActivityCompleted", + "recorded_at": "2026-01-01T00:00:10Z", + "payload": { + "sequence": 4, + "activity_execution_id": "group-execution-1-second", + "activity_type": "group.1", + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 1, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 1 + } + ], + "result": { + "codec": "avro", + "blob": "wwHioz3/VYAiNwoOdmFsdWUtMQ==" + } + }, + "sequence": 8 + }, + { + "event_type": "ActivityFailed", + "recorded_at": "1600-01-01T00:00:00.123500Z", + "payload": { + "sequence": 3, + "activity_execution_id": "group-execution-0-second", + "activity_type": "group.0", + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 0, + "parallel_group_path": [ + { + "parallel_group_id": "parallel-activities:3:2", + "parallel_group_kind": "activity", + "parallel_group_base_sequence": 3, + "parallel_group_size": 2, + "parallel_group_index": 0 + } + ], + "message": "failure-0" + }, + "sequence": 9 + }, + { + "event_type": "WorkflowTaskCompleted", + "recorded_at": "2099-01-01T00:00:00Z", + "payload": {}, + "sequence": 10 + } + ], + "expected": { + "command_sequence": [ + { + "type": "complete_workflow", + "result": { + "values": [ + "value-0", + "value-1" + ], + "clock": [ + null, + "2300-01-01T00:00:00.123500+00:00", + "1600-01-01T00:00:00.123000+00:00" + ], + "legacy": [ + "2026-01-01T00:00:00.000000+00:00", + "2026-01-01T00:00:00.000000+00:00", + "2026-01-01T00:00:00.000000+00:00" + ], + "failure": { + "activity_type": "group.0", + "message": "failure-0" + } + } + } + ] + } +} diff --git a/tests/test_business_replay_clock_parallel.py b/tests/test_business_replay_clock_parallel.py index 04ab988..842419c 100644 --- a/tests/test_business_replay_clock_parallel.py +++ b/tests/test_business_replay_clock_parallel.py @@ -12,7 +12,7 @@ from durable_workflow import serializer, workflow from durable_workflow.errors import ActivityFailed, NonDeterministicReplayError, TypedValueError -from durable_workflow.replay_clock_math import carbon_from_milliseconds, carbon_milliseconds +from durable_workflow.replay_clock_math import canonical_milliseconds, from_canonical_milliseconds from durable_workflow.workflow import WorkflowContext, query_state, replay from tests.business_clock_workflows import clock_text from tests.test_business_replay_clock import result @@ -22,6 +22,7 @@ TIED = "2026-01-01T00:00:03.123400+00:00" REFERENCE = Path(__file__).parent / "fixtures" / "replay_clock_carbon_reference.json" VECTORS = json.loads(REFERENCE.read_text())["vectors"] +CANONICAL = json.loads(REFERENCE.with_name("replay_clock_canonical_v1_reference.json").read_text()) def test_retained_carbon_reference_identity_is_exact() -> None: @@ -36,19 +37,56 @@ def test_retained_carbon_reference_identity_is_exact() -> None: @pytest.mark.parametrize("vector", VECTORS, ids=lambda vector: vector["input"]) -def test_actual_carbon_rounding_and_reconstruction_vectors(vector: dict[str, Any]) -> None: +def test_canonical_v1_agrees_with_these_specific_retained_php84_observations(vector: dict[str, Any]) -> None: time = datetime.fromisoformat(vector["input"].replace("Z", "+00:00")) - observed = carbon_milliseconds(time) + observed = canonical_milliseconds(time) assert observed == vector["milliseconds"] - assert carbon_from_milliseconds(vector["milliseconds"]).isoformat(timespec="microseconds").replace( + reconstructed = from_canonical_milliseconds(vector["milliseconds"]) + assert reconstructed is not None + assert reconstructed.isoformat(timespec="microseconds").replace( "+00:00", "Z", ) == vector["reconstructed"] @pytest.mark.parametrize("sign", [-1, 1]) -def test_unproved_float_microsecond_range_stays_unavailable(sign: int) -> None: +def test_canonical_v1_qualifies_exact_integer_times_beyond_the_prior_float_profile(sign: int) -> None: time = datetime(1970, 1, 1, tzinfo=timezone.utc) + timedelta(microseconds=sign * ((1 << 53) + 1)) - assert carbon_milliseconds(time) is None + assert canonical_milliseconds(time) == sign * 9007199254741 + + +@pytest.mark.parametrize("vector", CANONICAL["vectors"], ids=lambda vector: vector["input"]) +def test_canonical_v1_fixed_signed_utc_microsecond_and_reconstruction_vectors(vector: dict[str, Any]) -> None: + time = datetime.fromisoformat(vector["input"].replace("Z", "+00:00")) + delta = time - datetime(1970, 1, 1, tzinfo=timezone.utc) + assert (delta.days * 86400 + delta.seconds) * 1_000_000 + delta.microseconds == vector["utc_microseconds"] + assert canonical_milliseconds(time) == vector["milliseconds"] + reconstructed = from_canonical_milliseconds(vector["milliseconds"]) + expected = vector["reconstructed"] + if expected is None: + assert reconstructed is None + else: + assert reconstructed is not None + assert reconstructed == datetime.fromisoformat(expected.replace("Z", "+00:00")) + + +@pytest.mark.parametrize("case", CANONICAL["groups"], ids=lambda case: case["id"]) +def test_canonical_v1_authored_comparison_and_selected_failure_vectors(case: dict[str, Any]) -> None: + value = result(history(case["times"], order=(1, 0)), ParallelProbe) + expected = datetime.fromisoformat(case["times"][case["success_member"]].replace("Z", "+00:00")) + assert value["clock"] == expected.isoformat(timespec="microseconds") + failure = result(history(case["times"], failures={0, 1}, order=(1, 0)), ParallelProbe) + assert failure["failure"] == ("group.0", "failure-0") + time = case["failure_time"] + assert failure["clock"] == (datetime.fromisoformat(time.replace("Z", "+00:00")).isoformat( + timespec="microseconds", + ) if time is not None else None) + + +def test_reconstruction_overflow_refuses_time_without_changing_the_selected_failure() -> None: + rows = history(["9999-12-31T23:59:59.999500Z", FIRST], failures={0}) + value = result(rows, ParallelProbe) + assert value["failure"] == ("group.0", "failure-0") + assert value["clock"] is None def group_path(base: int, size: int, index: int, *, nested: bool = False) -> list[dict[str, Any]]: @@ -133,14 +171,14 @@ def test_success_uses_authored_tie_order_and_full_precision_not_arrival_or_histo (["1969-12-31T23:59:59.999500Z", "1969-12-31T23:59:59.999501Z"], "1969-12-31T23:59:59.999501+00:00"), ]) -def test_success_uses_carbon_millisecond_comparison_without_replacing_full_precision( +def test_success_uses_canonical_millisecond_comparison_without_replacing_full_precision( times: list[str], expected: str, ) -> None: assert result(history(times), ParallelProbe)["clock"] == expected @pytest.mark.parametrize("order", [(0, 1), (1, 0)]) -def test_caught_failure_ignores_successful_sibling_time_and_reconstructs_carbon_milliseconds( +def test_caught_failure_ignores_successful_sibling_time_and_reconstructs_canonical_milliseconds( order: tuple[int, ...], ) -> None: value = result(history(["1969-12-31T23:59:59.999500Z", "2099-01-01T00:00:00Z"], @@ -154,7 +192,7 @@ def test_caught_failure_ignores_successful_sibling_time_and_reconstructs_carbon_ ([TIED, FIRST], "2026-01-01T00:00:03.123000+00:00"), ([FIRST, "2026-01-01T00:00:01Z"], None), ]) -def test_multiple_failures_preserve_legacy_winner_and_refuse_a_different_core_winner( +def test_multiple_failures_preserve_legacy_winner_and_refuse_a_different_canonical_winner( times: list[str], expected: str | None, ) -> None: value = result(history(times, failures={0, 1}, order=(1, 0)), ParallelProbe) @@ -174,7 +212,7 @@ class Nested(ParallelProbe): @pytest.mark.parametrize("bad", [None, False, 17, "", "bad", "2026-02-30T00:00:00Z", "2026-01-01T24:00:00Z", - "2026-01-01T00:00:00.1234567Z", "9999-01-01T00:00:00Z"]) + "2026-01-01T00:00:00.1234567Z"]) def test_missing_invalid_or_unproved_required_time_does_not_borrow_a_sibling(bad: Any) -> None: assert result(history([FIRST, bad]), ParallelProbe)["clock"] is None @@ -319,7 +357,8 @@ def test_pending_and_unmarked_legacy_groups_do_not_receive_a_proved_clock() -> N rows.pop() pending = replay(ParallelProbe, rows, [], payload_codec="avro") try: - assert pending.commands and all(getattr(command, "activity_type", None) for command in pending.commands) + assert pending.commands + assert all(getattr(command, "activity_type", None) for command in pending.commands) finally: pending.close() rows = history()