diff --git a/CHANGELOG.md b/CHANGELOG.md index 91b0c55d..2936b414 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Replay now enforces the recorded search attributes (Workflow protocol 1.20 replay identity). A workflow whose `upsert_search_attributes` differs from its `SearchAttributesUpserted` history fails replay with a `NonDeterministicReplayError` whose detail starts `search_attribute_value_mismatch`, as memo replay already does for memos. Values are compared as Core's `UpsertSearchAttributesCall` normalizes them (trimmed strings and list items, a blank string as null) and with their types (`1`, `1.0`, `True` and `"1"` differ). An authored tuple is checked as the JSON list it is sent as. Values outside that portable domain, on either side, are refused rather than coerced: non-string or non-portable keys, objects, sets, bytes, maps, non-string list items, non-finite floats and out-of-range integers. A `SearchAttributesUpserted` event without `attributes` is refused rather than skipped, with a detail starting `search_attribute_update_missing`. One without a workflow sequence is refused with `durable_command_sequence_missing` (or `durable_command_sequence_invalid` for zero, negative, string, float or bool) rather than given one, and a second upsert at the same sequence with `duplicate_search_attribute_update`. An upsert that shares its workflow sequence with any other durable command record is refused with `durable_command_sequence_collision`. Every lifecycle row that carries a command's sequence claims it, including activity heartbeat, retry and cancellation-acknowledged rows, child cancellation rows, service operation scheduling, cancellation scope openings, continue-as-new, and cooperative and scope cancellation deliveries, which claim every sequence in their `sequence_span` (absent or null means 1, as in Core; any other value that is not a positive integer is refused with `durable_command_sequence_invalid`); events that only point at a step, such as a signal receipt, an update or a cancellation request, claim no sequence. When the history records `attribute_types`, it must type exactly the update's non-null keys, and each type must be a known type that can hold the value; anything else, including a malformed type value, fails with `search_attribute_type_mismatch`. The SDK's upsert declares no types and the Server records a key's namespace-registered type ahead of the inferred one, so a string may be recorded as `keyword`, `string` or `datetime` and an integer as `int` or `float`; search-attribute type identity is therefore not pinned for Python. History without `attribute_types` keeps value-only identity. Replay diagnostics name reasons, keys, JSON types, sequences and event types, never a recorded or authored value. (#235) +### Corrected + +- **2.13.1 claim withdrawn (#237).** The 2.13.1 entry said "The cancellation clock prefers a reused step's `reused_recorded_at`." As an observable result, that is incorrect. The test behind it was not a history Workflow 2.19 can write: it had no redrive `WorkflowStarted` and delivered the cancellation before the reused step. Redrive writes the successor's start and its reused prefix together, so a successor cancellation always follows the prefix and `remaining()` is computed from that delivery. A reused event's `reused_recorded_at` therefore cannot reach the budget, and the preference changes no observable result. Replay behavior is unchanged. The 2.13.1 start-clock change (`ctx.now()` from `replayed_started_at`) is unaffected. Valid successor fixtures replace the evidence, and the 2.13.1 fixture `redrive-start-replayed-clock`, which has only a partial redrive envelope, is registered as non-qualifying. + ### Added - Replay of Workflow 2.19 redrive reuse is qualified for `MemoUpserted` and for `SearchAttributesUpserted` value identity. Neither event is a replay clock source: `ctx.now()` stays the persisted start time, which on a redriven run is the failed run's start, and the cancellation budget does not move on an upsert. Search-attribute type identity is not qualified, because the Server records namespace-registered types the SDK cannot see. Signal waits (`SignalWaitOpened`, `SignalReceived`, `SignalApplied`) are not qualified for Python: the SDK has no signal-wait primitive. Replaying such reused events through a signal handler applies the signal once and takes no time from them, but that does not qualify signal waits. (#235) diff --git a/tests/fixtures/replay_regressions/redrive-successor-reused-cancellation-clock.json b/tests/fixtures/replay_regressions/redrive-successor-reused-cancellation-clock.json new file mode 100644 index 00000000..919cee2e --- /dev/null +++ b/tests/fixtures/replay_regressions/redrive-successor-reused-cancellation-clock.json @@ -0,0 +1,162 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "redrive-successor-reused-cancellation-clock", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.redrive-successor-cancellation-clock", + "input": [ + "Taylor", + 5 + ], + "payload_codec": "avro", + "workflow_id": "regression-corpus-redrive-successor-cancellation", + "run_id": "01JBREDRIVESUCCESSORRUN0000" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "workflow_type": "tests.replay.redrive-successor-cancellation-clock", + "workflow_instance_id": "regression-corpus-redrive-successor-cancellation", + "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000", + "workflow_command_id": "01JBREDRIVECOMMAND00000000", + "continued_from_run_id": "01JBREDRIVESOURCERUN000000", + "resume_step_sequence": 3, + "recovery_kind": "redrive", + "execution_deadline_at": null, + "run_deadline_at": null, + "replayed_started_at": "2026-11-03T00:00:00+00:00", + "workflow_link_id": "01JBREDRIVELINK00000000000" + } + }, + { + "sequence": 2, + "event_type": "ActivityCompleted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "activity_execution_id": "01JBREDRIVESUCCESSORACT000", + "activity_type": "tests.redrive.first", + "sequence": 1, + "result": "wwHioz3/VYAiNwocSGVsbG8sIFRheWxvciE=", + "payload_codec": "avro", + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_activity_execution_id": "01JBREDRIVESOURCEACTIVITY0", + "reused_recorded_at": "2026-11-03T00:00:00+00:00" + } + }, + { + "sequence": 3, + "event_type": "TimerScheduled", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "timer_id": "01JBREDRIVESUCCESSORTIMER0", + "sequence": 2, + "delay_seconds": 5, + "fire_at": "2026-11-03T00:00:05.000000Z" + } + }, + { + "sequence": 4, + "event_type": "TimerFired", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "timer_id": "01JBREDRIVESUCCESSORTIMER0", + "sequence": 2, + "delay_seconds": 5, + "fire_at": "2026-11-03T00:00:05.000000Z", + "fired_at": "2026-11-03T00:00:06.000000Z", + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:06+00:00" + } + }, + { + "sequence": 5, + "event_type": "ActivityScheduled", + "recorded_at": "2026-11-03T00:05:00.500000+00:00", + "payload": { + "activity_execution_id": "01JBREDRIVESUCCESSORACT001", + "activity_type": "tests.redrive.second", + "sequence": 3 + } + }, + { + "sequence": 6, + "event_type": "CooperativeCancellationRequested", + "recorded_at": "2026-11-03T00:05:01+00:00", + "payload": { + "workflow_command_id": "successor-request", + "workflow_instance_id": "regression-corpus-redrive-successor-cancellation", + "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000", + "command_type": "request_cancellation", + "cleanup_deadline_at": "2026-11-03T00:05:31.000000Z", + "cancellation": { + "schema": "durable-workflow.cancellation-context/v1", + "request_id": "successor-request", + "root_request_id": "successor-request", + "root_workflow_instance_id": "regression-corpus-redrive-successor-cancellation", + "root_workflow_run_id": "01JBREDRIVESUCCESSORRUN0000", + "parent_request_id": null, + "reason": null, + "requester": { + "type": "operator", + "id": "fixture-operator" + }, + "source": "control_plane", + "requested_at": "2026-11-03T00:05:01.000000Z", + "cleanup_deadline_at": "2026-11-03T00:05:31.000000Z", + "lineage": [ + { + "request_id": "successor-request", + "workflow_instance_id": "regression-corpus-redrive-successor-cancellation", + "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000" + } + ] + } + } + }, + { + "sequence": 7, + "event_type": "CooperativeCancellationDelivered", + "recorded_at": "2026-11-03T00:05:02+00:00", + "payload": { + "workflow_command_id": "successor-request", + "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000", + "sequence": 3, + "call_kind": "activity", + "cancellation": { + "schema": "durable-workflow.cancellation-context/v1", + "request_id": "successor-request", + "root_request_id": "successor-request", + "root_workflow_instance_id": "regression-corpus-redrive-successor-cancellation", + "root_workflow_run_id": "01JBREDRIVESUCCESSORRUN0000", + "parent_request_id": null, + "reason": null, + "requester": { + "type": "operator", + "id": "fixture-operator" + }, + "source": "control_plane", + "requested_at": "2026-11-03T00:05:01.000000Z", + "cleanup_deadline_at": "2026-11-03T00:05:31.000000Z", + "lineage": [ + { + "request_id": "successor-request", + "workflow_instance_id": "regression-corpus-redrive-successor-cancellation", + "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000" + } + ] + } + } + } + ], + "expected": { + "type": "complete_workflow", + "result": 29.0 + } +} diff --git a/tests/fixtures/replay_regressions/redrive-successor-reused-markers.json b/tests/fixtures/replay_regressions/redrive-successor-reused-markers.json new file mode 100644 index 00000000..751a07ca --- /dev/null +++ b/tests/fixtures/replay_regressions/redrive-successor-reused-markers.json @@ -0,0 +1,90 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "redrive-successor-reused-markers", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.redrive-successor-markers", + "input": [ + "Taylor" + ], + "payload_codec": "avro", + "workflow_id": "regression-corpus-redrive-successor-markers", + "run_id": "01JBREDRIVESUCCESSORRUN0000" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "workflow_type": "tests.replay.redrive-successor-markers", + "workflow_instance_id": "regression-corpus-redrive-successor-markers", + "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000", + "workflow_command_id": "01JBREDRIVECOMMAND00000000", + "continued_from_run_id": "01JBREDRIVESOURCERUN000000", + "resume_step_sequence": 4, + "recovery_kind": "redrive", + "execution_deadline_at": null, + "run_deadline_at": null, + "replayed_started_at": "2026-11-03T00:00:00+00:00", + "workflow_link_id": "01JBREDRIVELINK00000000000" + } + }, + { + "sequence": 2, + "event_type": "SideEffectRecorded", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "sequence": 1, + "result": "wwHioz3/VYAiNwogMDEyMzQ1Njc4OWFiY2RlZg==", + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:00+00:00" + } + }, + { + "sequence": 3, + "event_type": "VersionMarkerRecorded", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "sequence": 2, + "change_id": "redrive-markers", + "version": 1, + "min_supported": -1, + "max_supported": 1, + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:00+00:00" + } + }, + { + "sequence": 4, + "event_type": "ActivityCompleted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "activity_execution_id": "01JBREDRIVESUCCESSORACT000", + "activity_type": "tests.redrive.first", + "sequence": 3, + "result": "wwHioz3/VYAiNwocSGVsbG8sIFRheWxvciE=", + "payload_codec": "avro", + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_activity_execution_id": "01JBREDRIVESOURCEACTIVITY0", + "reused_recorded_at": "2026-11-03T00:00:01+00:00" + } + } + ], + "expected": { + "command_sequence": [ + { + "type": "schedule_activity", + "activity_type": "tests.redrive.second", + "arguments": [ + "Hello, Taylor!", + "2026-11-03T00:00:00+00:00" + ] + } + ] + } +} diff --git a/tests/fixtures/replay_regressions/redrive-successor-reused-timer.json b/tests/fixtures/replay_regressions/redrive-successor-reused-timer.json new file mode 100644 index 00000000..97a918f9 --- /dev/null +++ b/tests/fixtures/replay_regressions/redrive-successor-reused-timer.json @@ -0,0 +1,90 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "redrive-successor-reused-timer", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.redrive-successor-timer", + "input": [ + "Taylor", + 5 + ], + "payload_codec": "avro", + "workflow_id": "regression-corpus-redrive-successor-timer", + "run_id": "01JBREDRIVESUCCESSORRUN0000" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "workflow_type": "tests.replay.redrive-successor-timer", + "workflow_instance_id": "regression-corpus-redrive-successor-timer", + "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000", + "workflow_command_id": "01JBREDRIVECOMMAND00000000", + "continued_from_run_id": "01JBREDRIVESOURCERUN000000", + "resume_step_sequence": 3, + "recovery_kind": "redrive", + "execution_deadline_at": null, + "run_deadline_at": null, + "replayed_started_at": "2026-11-03T00:00:00+00:00", + "workflow_link_id": "01JBREDRIVELINK00000000000" + } + }, + { + "sequence": 2, + "event_type": "ActivityCompleted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "activity_execution_id": "01JBREDRIVESUCCESSORACT000", + "activity_type": "tests.redrive.first", + "sequence": 1, + "result": "wwHioz3/VYAiNwocSGVsbG8sIFRheWxvciE=", + "payload_codec": "avro", + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_activity_execution_id": "01JBREDRIVESOURCEACTIVITY0", + "reused_recorded_at": "2026-11-03T00:00:00+00:00" + } + }, + { + "sequence": 3, + "event_type": "TimerScheduled", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "timer_id": "01JBREDRIVESUCCESSORTIMER0", + "sequence": 2, + "delay_seconds": 5, + "fire_at": "2026-11-03T00:00:05.000000Z" + } + }, + { + "sequence": 4, + "event_type": "TimerFired", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "timer_id": "01JBREDRIVESUCCESSORTIMER0", + "sequence": 2, + "delay_seconds": 5, + "fire_at": "2026-11-03T00:00:05.000000Z", + "fired_at": "2026-11-03T00:00:06.000000Z", + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:06+00:00" + } + } + ], + "expected": { + "command_sequence": [ + { + "type": "schedule_activity", + "activity_type": "tests.redrive.second", + "arguments": [ + "Hello, Taylor!" + ] + } + ] + } +} diff --git a/tests/redrive_fixture_registry.py b/tests/redrive_fixture_registry.py new file mode 100644 index 00000000..36b4141d --- /dev/null +++ b/tests/redrive_fixture_registry.py @@ -0,0 +1,64 @@ +"""Every redrive replay fixture, and whether it is redrive-successor evidence. + +The replay corpus is immutable, so a fixture that is not a history Workflow +can write stays byte-identical and keeps replaying through the official +runner. This registry records that it is not qualification evidence: its +SHA-256 is pinned, it must keep failing ``check_redrive_successor`` with the +named error, and its named replacement must pass. The gate in +``tests/test_redrive_successor_evidence.py`` requires every fixture that +looks like a redrive to be registered here. +""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class NonQualifying: + sha256: str + error: str + replacement: str + release: str + reason: str + + +# Valid redrive successors and their resume_step_sequence. +VALID_SUCCESSORS: dict[str, int] = { + # Added by #236 (unreleased at the time of #237). + "redrive-reused-upserts-clock": 3, + "redrive-reused-signal-wait-clock": 2, + "redrive-reused-upserts-cancellation-clock": 3, + "redrive-reused-search-attribute-value-mismatch": 3, + "redrive-reused-search-attribute-type-mismatch": 3, + "redrive-reused-search-attribute-update-missing": 3, + # Added by #237, built to Workflow 28893f97 FailedRunRedrivePlan::successorEvents. + "redrive-successor-reused-markers": 4, + "redrive-successor-reused-timer": 3, + "redrive-successor-reused-cancellation-clock": 3, +} + +NON_QUALIFYING: dict[str, NonQualifying] = { + "redrive-start-replayed-clock": NonQualifying( + sha256="70653171faf5ddba5ed264d0a7ce8bb1af97112dbd6e3bb0e39c6f155fee0fd2", + error="history order", + replacement="redrive-successor-reused-markers", + release="2.13.1", + reason=( + "Partial redrive envelope (only recovery_kind and replayed_started_at, no history sequence, " + "identity, link or resume step) and no reused prefix. It asserts only ctx.now(), which is " + "correct; it is ambiguous rather than a clock defect. The replacement asserts the same " + "start clock on a valid successor." + ), + ), +} + +# Fixtures for a run that redrive starts from. None in this corpus. +SOURCE_RUNS: frozenset[str] = frozenset() + +# The 2.13.1 unit test test_a_reused_step_keeps_its_original_time_for_the_remaining_budget +# (tests/test_cancellation_remaining_time.py, #233) was not a corpus fixture. +# It delivered the cancellation before a "reused" step whose reused_recorded_at +# followed the delivery. #237 replaced it with the non-leakage evidence on +# redrive-successor-reused-cancellation-clock. +REPLACED_UNIT_TEST = "test_a_reused_step_keeps_its_original_time_for_the_remaining_budget" diff --git a/tests/test_cancellation_remaining_time.py b/tests/test_cancellation_remaining_time.py index 24d85410..fddaa13e 100644 --- a/tests/test_cancellation_remaining_time.py +++ b/tests/test_cancellation_remaining_time.py @@ -265,24 +265,10 @@ def execute(command: RecordLocalActivity) -> str: 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)] - +# test_a_reused_step_keeps_its_original_time_for_the_remaining_budget (2.13.1, +# #233) was removed by #237. It delivered the cancellation before a "reused" +# step whose reused_recorded_at followed the delivery, a history redrive cannot +# write. Valid successor evidence is in tests/test_redrive_successor_evidence.py. @pytest.mark.parametrize("later", ["2026-10-01T00:00:12Z", "2026-10-01T00:00:29Z"]) diff --git a/tests/test_redrive_successor_evidence.py b/tests/test_redrive_successor_evidence.py new file mode 100644 index 00000000..731597cd --- /dev/null +++ b/tests/test_redrive_successor_evidence.py @@ -0,0 +1,462 @@ +"""Valid redrive successor evidence (#237). + +The 2.13.1 claim that the cancellation clock prefers a reused step's +reused_recorded_at rested on a history Workflow cannot write: the +cancellation was delivered before a "reused" step. Redrive writes the +successor's WorkflowStarted and its reused prefix in one write, so the +successor's own cancellation always follows the prefix, and remaining() is +computed from that delivery. Reused times are therefore not observable +through remaining(). These tests prove that non-leakage on successors built +to Workflow 28893f97 FailedRunRedrivePlan::successorEvents, and gate every +redrive fixture through the registry. +""" + +from __future__ import annotations + +import hashlib +import json +from copy import deepcopy +from datetime import datetime +from typing import Any + +import pytest + +from durable_workflow import workflow +from durable_workflow.errors import WorkflowCancelled +from durable_workflow.workflow import CompleteWorkflow, ScheduleActivity, WorkflowContext, query_state, replay +from tests.redrive_fixture_registry import NON_QUALIFYING, SOURCE_RUNS, VALID_SUCCESSORS +from tests.test_replay_clock import REDRIVE_FIXTURES, check_redrive_successor, redrive_fixture, redrive_time + +SIDE_EFFECT_CALLS: list[str] = [] + + +@workflow.defn(name="tests.replay.redrive-successor-markers") +class RedriveSuccessorMarkersWorkflow: + """Core TestRedriveMarkersWorkflow, plus the start clock it observes.""" + + def __init__(self) -> None: + self.markers: dict[str, Any] = {} + + def run(self, ctx: WorkflowContext, name: str): # type: ignore[no-untyped-def] + started = ctx.now().isoformat() + self.markers["token"] = yield ctx.side_effect(lambda: SIDE_EFFECT_CALLS.append("ran") or "fresh") + self.markers["version"] = yield ctx.get_version("redrive-markers", -1, 1) + greeting = yield ctx.schedule_activity("tests.redrive.first", [name]) + return (yield ctx.schedule_activity("tests.redrive.second", [greeting, started])) + + @workflow.query("markers") + def query_markers(self) -> dict[str, Any]: + return self.markers + + +@workflow.defn(name="tests.replay.redrive-successor-timer") +class RedriveSuccessorTimerWorkflow: + """Core TestRedriveTimerWorkflow.""" + + def run(self, ctx: WorkflowContext, name: str, seconds: int): # type: ignore[no-untyped-def] + greeting = yield ctx.schedule_activity("tests.redrive.first", [name]) + yield ctx.start_timer(seconds) + return (yield ctx.schedule_activity("tests.redrive.second", [greeting])) + + +@workflow.defn(name="tests.replay.redrive-successor-cancellation-clock") +class RedriveSuccessorCancellationClockWorkflow: + """The timer successor; its own cancellation reaches the second activity.""" + + def run(self, ctx: WorkflowContext, name: str, seconds: int): # type: ignore[no-untyped-def] + greeting = yield ctx.schedule_activity("tests.redrive.first", [name]) + yield ctx.start_timer(seconds) + try: + return (yield ctx.schedule_activity("tests.redrive.second", [greeting])) + except WorkflowCancelled as error: + assert error.context is not None + return error.context.remaining() + + +SUCCESSOR_WORKFLOWS = { + "redrive-successor-reused-markers": RedriveSuccessorMarkersWorkflow, + "redrive-successor-reused-timer": RedriveSuccessorTimerWorkflow, + "redrive-successor-reused-cancellation-clock": RedriveSuccessorCancellationClockWorkflow, +} +CANCELLATION = "redrive-successor-reused-cancellation-clock" + + +def commands(name: str, history: list[dict[str, Any]] | None = None) -> list[Any]: + fixture = redrive_fixture(name) + task = fixture["workflow"] + outcome = replay(SUCCESSOR_WORKFLOWS[name], fixture["history"] if history is None else history, task["input"], + run_id=task["run_id"], workflow_id=task["workflow_id"], payload_codec="avro") + return list(outcome.commands) + + +def check(name: str, history: list[dict[str, Any]]) -> None: + check_redrive_successor(history, redrive_fixture(name)["workflow"]) + + +def event(history: list[dict[str, Any]], event_type: str) -> dict[str, Any]: + matches = [item for item in history if item["event_type"] == event_type] + assert len(matches) == 1, event_type + return matches[0] + + +def changed(name: str, change: Any) -> list[dict[str, Any]]: + history = deepcopy(redrive_fixture(name)["history"]) + change(history) + return history + + +# Registry gate. +def _looks_like_redrive(path: Any) -> bool: + text = path.read_text(encoding="utf-8") + if "redrive" in path.stem or "reused_from_run_id" in text or "reused_recorded_at" in text: + return True + history = json.loads(text).get("history") or [] + return any(item.get("event_type") == "WorkflowStarted" and (item.get("payload") or {}).get("recovery_kind") + == "redrive" for item in history if isinstance(item, dict)) + + +def _verdict(name: str) -> str | None: + fixture = redrive_fixture(name) + try: + check_redrive_successor(fixture["history"], fixture["workflow"]) + except AssertionError as error: + return str(error) + return None + + +def test_every_redrive_fixture_is_registered() -> None: + selected = {path.stem for path in REDRIVE_FIXTURES.glob("*.json") if _looks_like_redrive(path)} + assert selected == set(VALID_SUCCESSORS) | set(NON_QUALIFYING) | SOURCE_RUNS + + +def test_exactly_the_registered_successors_pass() -> None: + passing = {path.stem for path in REDRIVE_FIXTURES.glob("*.json") if _looks_like_redrive(path) + and _verdict(path.stem) is None} + assert passing == set(VALID_SUCCESSORS) + + +@pytest.mark.parametrize("name", sorted(VALID_SUCCESSORS)) +def test_registered_successor_resumes_after_its_reused_prefix(name: str) -> None: + assert redrive_fixture(name)["history"][0]["payload"]["resume_step_sequence"] == VALID_SUCCESSORS[name] + assert _verdict(name) is None + + +@pytest.mark.parametrize("name", sorted(NON_QUALIFYING)) +def test_non_qualifying_fixture_is_pinned_and_replaced(name: str) -> None: + entry = NON_QUALIFYING[name] + raw = (REDRIVE_FIXTURES / f"{name}.json").read_bytes() + assert hashlib.sha256(raw).hexdigest() == entry.sha256 + assert _verdict(name) == entry.error + assert entry.replacement in VALID_SUCCESSORS and _verdict(entry.replacement) is None + + +# The replacements replay as Core would: nothing re-executes. +def test_markers_successor_reuses_its_markers_and_the_start_clock() -> None: + SIDE_EFFECT_CALLS.clear() + assert commands("redrive-successor-reused-markers") == [ + ScheduleActivity("tests.redrive.second", ["Hello, Taylor!", "2026-11-03T00:00:00+00:00"]), + ] + fixture = redrive_fixture("redrive-successor-reused-markers") + task = fixture["workflow"] + markers = query_state(RedriveSuccessorMarkersWorkflow, fixture["history"], task["input"], "markers", + run_id=task["run_id"], workflow_id=task["workflow_id"], payload_codec="avro") + assert markers == {"token": "0123456789abcdef", "version": 1} + assert SIDE_EFFECT_CALLS == [] + + +def test_timer_successor_does_not_start_its_timer_again() -> None: + assert commands("redrive-successor-reused-timer") == [ScheduleActivity("tests.redrive.second", ["Hello, Taylor!"])] + + +def test_cancellation_successor_budget_is_computed_from_its_delivery() -> None: + # Deadline 00:05:31 less the delivery at 00:05:02. + assert commands(CANCELLATION) == [CompleteWorkflow(29.0)] + + +# Non-leakage. +REUSED_TIMES = ["2026-11-03T00:00:00+00:00", "2026-11-03T00:04:59.999999+00:00"] + + +@pytest.mark.parametrize("name", sorted(SUCCESSOR_WORKFLOWS)) +@pytest.mark.parametrize("reused_at", REUSED_TIMES) +def test_valid_retiming_of_reused_events_changes_nothing(name: str, reused_at: str) -> None: + def retime(history: list[dict[str, Any]]) -> None: + for item in history: + if "reused_recorded_at" in item["payload"]: + item["payload"]["reused_recorded_at"] = reused_at + + history = changed(name, retime) + check(name, history) + assert commands(name, history) == commands(name) + + +def _time(item: dict[str, Any]) -> datetime: + payload = item["payload"] + return redrive_time(payload.get("reused_recorded_at") or item["recorded_at"]) + + +def leaking_clock_remaining(history: list[dict[str, Any]]) -> float: + """A clock that read every event up to the delivery, preferring reuse times.""" + delivered = next(index for index, item in enumerate(history) + if item["event_type"] == "CooperativeCancellationDelivered") + request = event(history, "CooperativeCancellationRequested") + deadline = redrive_time(request["payload"]["cleanup_deadline_at"]) + return (deadline - max(_time(item) for item in history[:delivered + 1])).total_seconds() + + +LATE = "2026-11-03T00:05:20+00:00" + + +def _late(event_type: str) -> Any: + def change(history: list[dict[str, Any]]) -> None: + item = event(history, event_type) + item["recorded_at"] = LATE + if "reused_recorded_at" in item["payload"]: + item["payload"]["reused_recorded_at"] = LATE + + return change + + +@pytest.mark.parametrize("event_type", ["ActivityCompleted", "TimerFired"]) +def test_a_reused_time_after_the_delivery_does_not_reach_the_budget(event_type: str) -> None: + original = redrive_fixture(CANCELLATION)["history"] + retimed = changed(CANCELLATION, _late(event_type)) + # The probe would see the reused event: 00:05:31 - 00:05:20. + assert leaking_clock_remaining(original) == 29.0 + assert leaking_clock_remaining(retimed) == 11.0 + # The SDK does not: the budget is the delivery's. + assert commands(CANCELLATION, retimed) == commands(CANCELLATION) == [CompleteWorkflow(29.0)] + # Redrive cannot write this probe history. + with pytest.raises(AssertionError): + check(CANCELLATION, retimed) + + +def test_positive_control_the_delivery_time_reaches_the_budget() -> None: + retimed = changed(CANCELLATION, _late("CooperativeCancellationDelivered")) + assert commands(CANCELLATION, retimed) == [CompleteWorkflow(11.0)] + + +def test_the_2_13_1_order_is_not_a_successor() -> None: + # Move the successor's cancellation ahead of the reused prefix. + def early(history: list[dict[str, Any]]) -> None: + request = event(history, "CooperativeCancellationRequested") + delivery = event(history, "CooperativeCancellationDelivered") + history.remove(request) + history.remove(delivery) + history[1:1] = [request, delivery] + for number, item in enumerate(history, start=1): + item["sequence"] = number + + with pytest.raises(AssertionError, match="reused events must directly follow WorkflowStarted"): + check(CANCELLATION, changed(CANCELLATION, early)) + + +@pytest.mark.parametrize("name", sorted(SUCCESSOR_WORKFLOWS)) +@pytest.mark.parametrize("resume", [lambda value: value - 1, lambda value: value + 1, lambda value: None, str]) +def test_successor_resume_must_follow_the_reused_prefix(name: str, resume: Any) -> None: + def change(history: list[dict[str, Any]]) -> None: + history[0]["payload"]["resume_step_sequence"] = resume(history[0]["payload"]["resume_step_sequence"]) + + with pytest.raises(AssertionError, match="resume_step_sequence"): + check(name, changed(name, change)) + + +# Core's copied timer schedule (FailedRunRedrivePlan.php:437-455). +def _schedule(history: list[dict[str, Any]]) -> dict[str, Any]: + return event(history, "TimerScheduled") + + +def _renumber(history: list[dict[str, Any]]) -> None: + for number, item in enumerate(history, start=1): + item["sequence"] = number + + +def _move_schedule_after_fired(history: list[dict[str, Any]]) -> None: + schedule = _schedule(history) + history.remove(schedule) + history.insert(history.index(event(history, "TimerFired")) + 1, schedule) + _renumber(history) + + +def _drop(event_type: str) -> Any: + def change(history: list[dict[str, Any]]) -> None: + history.remove(event(history, event_type)) + _renumber(history) + + return change + + +@pytest.mark.parametrize("name", ["redrive-successor-reused-timer", CANCELLATION]) +@pytest.mark.parametrize("change", [ + lambda history: _schedule(history)["payload"].__setitem__("reused_from_run_id", "01JBREDRIVESOURCERUN000000"), + lambda history: _schedule(history)["payload"].__setitem__("reused_recorded_at", "2026-11-03T00:00:05+00:00"), + lambda history: _schedule(history)["payload"].__setitem__("timer_id", "another-timer"), + lambda history: _schedule(history)["payload"].__setitem__("timer_id", " "), + lambda history: _schedule(history)["payload"].__setitem__("sequence", 1), + lambda history: _schedule(history).__setitem__("recorded_at", "2026-11-03T00:05:00.100000+00:00"), + _move_schedule_after_fired, + _drop("TimerScheduled"), + _drop("TimerFired"), + lambda history: _schedule(history)["payload"].__setitem__("workflow_run_id", "bogus-run"), + lambda history: _schedule(history)["payload"].__setitem__("workflow_run_id", "01JBREDRIVESUCCESSORRUN0000"), + lambda history: _schedule(history)["payload"].__setitem__("timer_kind", "condition_timeout"), + lambda history: _schedule(history)["payload"].__setitem__("cancellation_scope_id", "scope-one"), + lambda history: _schedule(history)["payload"].__setitem__("cancellation_cleanup", {"scope_id": "scope-one"}), + lambda history: _schedule(history)["payload"].__setitem__("parallel_group_id", "group-one"), + lambda history: _schedule(history)["payload"].__setitem__("condition_wait_id", "wait-one"), + lambda history: _schedule(history)["payload"].__setitem__("signal_wait_id", "wait-one"), +], ids=["schedule-reused-from", "schedule-reused-at", "other-timer-id", "blank-timer-id", "other-step", + "schedule-after-start", "schedule-after-fired", "no-schedule", "no-fired-timer", "bogus-run", + "successor-run", "timer-kind", "cancellation-scope", "cancellation-cleanup", "parallel-group", + "condition-wait", "signal-wait"]) +def test_only_core_shaped_copied_timer_schedules_are_accepted(name: str, change: Any) -> None: + with pytest.raises(AssertionError): + check(name, changed(name, change)) + + +def _delivery(history: list[dict[str, Any]]) -> dict[str, Any]: + return event(history, "CooperativeCancellationDelivered") + + +def _request(history: list[dict[str, Any]]) -> dict[str, Any]: + return event(history, "CooperativeCancellationRequested") + + +SOURCE_RUN = "01JBREDRIVESOURCERUN000000" + + +def _both_contexts(history: list[dict[str, Any]], change: Any) -> None: + change(_request(history)["payload"]["cancellation"]) + change(_delivery(history)["payload"]["cancellation"]) + + +@pytest.mark.parametrize(("change", "message"), [ + (lambda history: _delivery(history)["payload"].__setitem__("sequence", 2), "reuses a reused step"), + (lambda history: _delivery(history)["payload"].__setitem__("sequence", 1), "reuses a reused step"), + (lambda history: _request(history).__setitem__("recorded_at", "2026-11-03T00:04:59+00:00"), "predates"), + (lambda history: _request(history)["payload"].__setitem__("workflow_run_id", SOURCE_RUN), "another run"), + (lambda history: _delivery(history)["payload"].__setitem__("workflow_run_id", SOURCE_RUN), "another run"), + (lambda history: _request(history)["payload"].__setitem__("workflow_instance_id", "another-instance"), + "another workflow instance"), + (lambda history: _both_contexts(history, lambda context: context.__setitem__("root_workflow_run_id", SOURCE_RUN)), + "rooted at the successor run"), + (lambda history: _both_contexts(history, lambda context: context["lineage"][-1].__setitem__( + "workflow_run_id", SOURCE_RUN)), "rooted at the successor run"), + (lambda history: _request(history)["payload"].__setitem__("reused_from_run_id", SOURCE_RUN), + "reused events must directly follow"), + (lambda history: event(history, "ActivityScheduled")["payload"].__setitem__("sequence", 2), + "reuses a reused step"), +]) +def test_the_successor_cancellation_is_its_own(change: Any, message: str) -> None: + with pytest.raises(AssertionError, match=message): + check(CANCELLATION, changed(CANCELLATION, change)) + + +# The successor cancellation is exactly one request, then exactly one delivery +# of it, carrying one context bound to the successor, delivered to a call the +# successor reached. Each mutation below was accepted by the bd41be60 check. +CANCELLED_SUCCESSORS = [CANCELLATION, "redrive-reused-upserts-cancellation-clock"] + + +def _duplicate(event_type: str) -> Any: + def change(history: list[dict[str, Any]]) -> None: + original = event(history, event_type) + history.insert(history.index(original) + 1, deepcopy(original)) + _renumber(history) + + return change + + +def _without(event_type: str) -> Any: + def change(history: list[dict[str, Any]]) -> None: + history.remove(event(history, event_type)) + _renumber(history) + + return change + + +def _context(target: str, change: Any) -> Any: + def apply(history: list[dict[str, Any]]) -> None: + for item in ([_request(history)] if target in {"request", "both"} else []) + ( + [_delivery(history)] if target in {"delivery", "both"} else [] + ): + change(item["payload"]["cancellation"]) + + return apply + + +def _drop_context(target: str) -> Any: + def apply(history: list[dict[str, Any]]) -> None: + for item in ([_request(history)] if target in {"request", "both"} else []) + ( + [_delivery(history)] if target in {"delivery", "both"} else [] + ): + del item["payload"]["cancellation"] + + return apply + + +CANCELLATION_MUTATIONS = { + "context-deleted-from-both": (_drop_context("both"), "carries its cancellation context"), + "context-deleted-from-request": (_drop_context("request"), "carries its cancellation context"), + "context-deleted-from-delivery": (_drop_context("delivery"), "carries its request's context"), + "duplicate-request": (_duplicate("CooperativeCancellationRequested"), "exactly one request and one delivery"), + "duplicate-delivery": (_duplicate("CooperativeCancellationDelivered"), "exactly one request and one delivery"), + "request-deleted": (_without("CooperativeCancellationRequested"), "exactly one request and one delivery"), + "delivery-deleted": (_without("CooperativeCancellationDelivered"), "exactly one request and one delivery"), + "mismatched-context": (_context("delivery", lambda context: context["requester"].__setitem__("id", "other")), + "carries its request's context"), + "delivery-before-request-time": ( + lambda history: _delivery(history).__setitem__("recorded_at", "2026-11-03T00:05:00.750000+00:00"), + "not earlier than its request"), + "context-request-id": (_context("both", lambda context: context.__setitem__("request_id", "other-request")), + "the context is this request"), + "context-root-request": (_context("both", lambda context: context.__setitem__("root_request_id", "other")), + "its own root"), + "context-root-instance": (_context("both", lambda context: context.__setitem__( + "root_workflow_instance_id", "another-instance")), "rooted at the successor run"), + "lineage-request": (_context("both", lambda context: context["lineage"][-1].__setitem__("request_id", "other")), + "rooted at the successor run"), + "context-deadline": (_context("both", lambda context: context.__setitem__( + "cleanup_deadline_at", "2026-11-03T00:05:40.000000Z")), "cleanup deadline"), + "delivery-to-unrecorded-step-4": (lambda history: _delivery(history)["payload"].__setitem__("sequence", 4), + "a successor step the successor reached"), +} + + +@pytest.mark.parametrize("name", CANCELLED_SUCCESSORS) +@pytest.mark.parametrize("mutation", sorted(CANCELLATION_MUTATIONS)) +def test_the_successor_cancellation_is_bound_exactly(name: str, mutation: str) -> None: + change, message = CANCELLATION_MUTATIONS[mutation] + if mutation == "context-root-request": + # A root request: its context names itself as its root. + assert redrive_fixture(name)["history"][-1]["payload"]["cancellation"]["parent_request_id"] is None + with pytest.raises(AssertionError, match=message): + check(name, changed(name, change)) + + +def test_a_delivery_to_a_later_step_the_successor_recorded_is_accepted() -> None: + # The successor finishes step 3 and schedules step 4 before the cancel. + def later(history: list[dict[str, Any]]) -> None: + scheduled = event(history, "ActivityScheduled") + completed = {"event_type": "ActivityCompleted", "recorded_at": "2026-11-03T00:05:00.600000+00:00", + "payload": {**scheduled["payload"], "result": "wwHioz3/VYAiNwoEb2s=", "payload_codec": "avro"}} + fourth = deepcopy(scheduled) + fourth["recorded_at"] = "2026-11-03T00:05:00.700000+00:00" + fourth["payload"].update(sequence=4, activity_execution_id="01JBREDRIVESUCCESSORACT002") + at = history.index(scheduled) + 1 + history[at:at] = [completed, fourth] + _delivery(history)["payload"]["sequence"] = 4 + _renumber(history) + + check(CANCELLATION, changed(CANCELLATION, later)) + + +@pytest.mark.parametrize("name", ["redrive-successor-reused-timer", CANCELLATION]) +@pytest.mark.parametrize("change", [ + lambda history: _schedule(history)["payload"].__setitem__("workflow_run_id", "01JBREDRIVESOURCERUN000000"), + lambda history: _schedule(history)["payload"].__setitem__("timer_kind", None), + lambda history: _schedule(history)["payload"].__setitem__("cancellation_scope_id", None), +], ids=["source-run", "null-timer-kind", "null-scope"]) +def test_a_plain_copied_schedule_may_name_the_source_run(name: str, change: Any) -> None: + # Core copies the source run's schedule payload; a null key is plain. + check(name, changed(name, change)) diff --git a/tests/test_replay_clock.py b/tests/test_replay_clock.py index 7a4ee13e..a991e606 100644 --- a/tests/test_replay_clock.py +++ b/tests/test_replay_clock.py @@ -12,6 +12,7 @@ from durable_workflow import serializer, workflow from durable_workflow.errors import NonDeterministicReplayError, WorkflowCancelled from durable_workflow.workflow import CompleteWorkflow, WorkflowContext, query_state, replay +from tests.redrive_fixture_registry import VALID_SUCCESSORS @workflow.defn(name="tests.replay.persisted-clock") @@ -468,10 +469,83 @@ def test_reused_signal_wait_events_do_not_move_the_cancellation_budget() -> None "SignalApplied": ("workflow_command_id", "signal_id", "signal_name", "signal_wait_id", "sequence"), "CooperativeCancellationRequested": ("workflow_command_id", "workflow_instance_id", "workflow_run_id"), "CooperativeCancellationDelivered": ("workflow_command_id", "workflow_run_id", "sequence"), + # Core FailedRunRedrivePlan::successorEvents (workflow 28893f97) writes a + # reused activity as one ActivityCompleted with a new execution id, and a + # reused timer as a copied TimerScheduled (new timer_id, no reuse keys) + # followed by the reused TimerFired. + "ActivityScheduled": ("sequence", "activity_type", "activity_execution_id"), + "ActivityCompleted": ("sequence", "activity_type", "activity_execution_id"), + "TimerScheduled": ("sequence", "timer_id"), + "TimerFired": ("sequence", "timer_id"), + "SideEffectRecorded": ("sequence",), + "VersionMarkerRecorded": ("sequence", "change_id"), } # The events one reused workflow step writes, in order. SignalReceived belongs -# to its wait and has no step of its own. -REDRIVE_STEP_SHAPES = {("MemoUpserted",), ("SearchAttributesUpserted",), ("SignalWaitOpened", "SignalApplied")} +# to its wait and has no step of its own, and a copied TimerScheduled is not a +# reused event. +REDRIVE_STEP_SHAPES = { + ("MemoUpserted",), ("SearchAttributesUpserted",), ("SignalWaitOpened", "SignalApplied"), + ("ActivityCompleted",), ("TimerFired",), ("SideEffectRecorded",), ("VersionMarkerRecorded",), +} +# Successor events that occupy a workflow step of their own. +REDRIVE_SUCCESSOR_STEP_EVENTS = frozenset({ + "MemoUpserted", "SearchAttributesUpserted", "SignalWaitOpened", "SignalApplied", "ActivityScheduled", + "ActivityCompleted", "TimerScheduled", "TimerFired", "SideEffectRecorded", "VersionMarkerRecorded", + "CooperativeCancellationDelivered", +}) + + +# Core FailedRunRedrivePlan::isPlainTimer: redrive copies only plain timers. +_NON_PLAIN_TIMER_KEYS = ( + "timer_kind", "cancellation_cleanup", "cancellation_scope_id", "parallel_group_id", "condition_wait_id", + "signal_wait_id", +) + + +def _copied_timer_schedule(history: list[dict[str, Any]], index: int, started: dict[str, Any]) -> bool: + """Core's copy of a reused timer's schedule (FailedRunRedrivePlan.php:437-455). + + Accepted in the reused prefix only when it carries neither reuse key, is + recorded at the successor start, is a plain timer (Core isPlainTimer, + FailedRunRedrivePlan.php:497), keeps the source run if it names one, and + is directly followed by the reused TimerFired for the same non-blank + timer_id and the same step. + """ + event = history[index] + payload = event.get("payload") or {} + source = (started.get("payload") or {}).get("continued_from_run_id") + if event.get("event_type") != "TimerScheduled" or "reused_from_run_id" in payload or ( + "reused_recorded_at" in payload + ) or event.get("recorded_at") != started.get("recorded_at") or index + 1 >= len(history): + return False + # The copy is the source run's schedule payload: any run it names is the + # source, never the successor or another run. + if "workflow_run_id" in payload and payload["workflow_run_id"] != source: + return False + if any(payload.get(key) is not None for key in _NON_PLAIN_TIMER_KEYS): + return False + fired = history[index + 1] + fired_payload = fired.get("payload") or {} + timer_id = payload.get("timer_id") + return (fired.get("event_type") == "TimerFired" and "reused_from_run_id" in fired_payload + and isinstance(timer_id, str) and bool(timer_id.strip()) + and fired_payload.get("timer_id") == timer_id + and fired_payload.get("sequence") == payload.get("sequence")) + + +def redrive_time(value: Any) -> datetime: + """Parse a history time on every supported Python; 3.10 rejects a trailing Z.""" + if not isinstance(value, str): + raise AssertionError("history time must be a string") + parsed = datetime.fromisoformat(value[:-1] + "+00:00" if value.endswith("Z") else value) + if parsed.tzinfo is None: + raise AssertionError("history time must carry an offset") + return parsed + + +# Events a successor cancellation can be delivered to: a call the successor +# recorded itself and that waits. +REDRIVE_CANCELLABLE_EVENTS = frozenset({"ActivityScheduled", "TimerScheduled", "SignalWaitOpened"}) def check_redrive_successor(history: list[dict[str, Any]], workflow: dict[str, Any]) -> None: @@ -498,20 +572,23 @@ def present(payload: dict[str, Any], key: str) -> bool: for key in REDRIVE_START_DEADLINES: deadline = start[key] if deadline is not None: - require(isinstance(deadline, str) and datetime.fromisoformat(deadline).tzinfo is not None, key) + require(isinstance(deadline, str) and redrive_time(deadline) is not None, key) # The start binds the fixture's workflow type, instance and successor run. require(start["workflow_type"] == workflow.get("type"), "workflow_type must be the fixture workflow") require(start["workflow_instance_id"] == workflow.get("workflow_id"), "workflow_instance_id must be the fixture's") require(start["workflow_run_id"] == workflow.get("run_id"), "workflow_run_id must be the fixture's") instance, run, source = start["workflow_instance_id"], start["workflow_run_id"], start["continued_from_run_id"] require(run != source, "the successor is a new run") - started_at = datetime.fromisoformat(started["recorded_at"]) - require(datetime.fromisoformat(start["replayed_started_at"]) <= started_at, "replayed_started_at") + started_at = redrive_time(started["recorded_at"]) + require(redrive_time(start["replayed_started_at"]) <= started_at, "replayed_started_at") prefix = True steps: list[tuple[int, list[str]]] = [] signals: dict[str, dict[str, Any]] = {} - cancellations: dict[str, dict[str, Any]] = {} + requests: list[tuple[dict[str, Any], set[int]]] = [] + deliveries: list[tuple[dict[str, Any], set[int]]] = [] + cancellable_steps: set[int] = set() + successor_steps: list[int] = [] for index, event in enumerate(history[1:], start=2): require(event.get("sequence") == index, "history order") kind = event.get("event_type") @@ -524,6 +601,8 @@ def present(payload: dict[str, Any], key: str) -> bool: if "workflow_instance_id" in payload: require(payload["workflow_instance_id"] == instance, "event belongs to another workflow instance") reused = "reused_from_run_id" in payload or "reused_recorded_at" in payload + if prefix and not reused and _copied_timer_schedule(history, index - 1, started): + continue if reused: require(prefix, "reused events must directly follow WorkflowStarted") for key in ("reused_from_run_id", "reused_recorded_at"): @@ -533,7 +612,11 @@ def present(payload: dict[str, Any], key: str) -> bool: if "workflow_run_id" in payload: require(payload["workflow_run_id"] == source, "reused event belongs to another run") require(event["recorded_at"] == started["recorded_at"], "reused events are recorded with the start") - require(datetime.fromisoformat(payload["reused_recorded_at"]) <= started_at, "reused_recorded_at") + require(redrive_time(payload["reused_recorded_at"]) <= started_at, "reused_recorded_at") + if kind == "TimerFired": + # Core always writes the copied schedule directly before it. + require(_copied_timer_schedule(history, index - 2, started), + "reused TimerFired lacks its copied TimerScheduled") if kind != "SignalReceived": if steps and steps[-1][0] == payload["sequence"]: steps[-1][1].append(kind) @@ -543,8 +626,16 @@ def present(payload: dict[str, Any], key: str) -> bool: prefix = False if "workflow_run_id" in payload: require(payload["workflow_run_id"] == run, "successor event belongs to another run") - require(datetime.fromisoformat(event["recorded_at"]) >= started_at, + require(redrive_time(event["recorded_at"]) >= started_at, "successor event predates the successor start") + if kind in REDRIVE_SUCCESSOR_STEP_EVENTS: + successor_steps.append(payload["sequence"]) + if kind in REDRIVE_CANCELLABLE_EVENTS: + cancellable_steps.add(payload["sequence"]) + if kind == "CooperativeCancellationRequested": + requests.append((event, set(cancellable_steps))) + if kind == "CooperativeCancellationDelivered": + deliveries.append((event, set(cancellable_steps))) if kind.startswith("Signal"): if kind != "SignalWaitOpened": require(payload["signal_wait_id"] in signals, "signal event has no opened wait") @@ -552,11 +643,6 @@ def present(payload: dict[str, Any], key: str) -> bool: for key in ("signal_name", "workflow_command_id", "signal_id"): if key in payload: require(wait.setdefault(key, payload[key]) == payload[key], f"signal events disagree on {key}") - if kind.startswith("CooperativeCancellation"): - command = cancellations.setdefault(kind, payload["workflow_command_id"]) - require(command == payload["workflow_command_id"], "duplicate cancellation event") - if cancellations: - require(len(set(cancellations.values())) == 1, "cancellation events disagree on workflow_command_id") # The reused steps are the contiguous prefix 1..resume-1, in order, one # whole step each. require(bool(steps), "no reused workflow step") @@ -567,24 +653,58 @@ def present(payload: dict[str, Any], key: str) -> bool: resume = start.get("resume_step_sequence") require(type(resume) is int and resume == len(steps) + 1, "resume_step_sequence must follow the last reused step") + # The successor never records, or delivers to, a step it reused. + require(all(step >= resume for step in successor_steps), "successor event reuses a reused step") + if requests or deliveries: + _check_successor_cancellation(requests, deliveries, instance, run, resume) -REDRIVE_SUCCESSOR_FIXTURES = { - "redrive-reused-search-attribute-update-missing": 3, - "redrive-reused-upserts-clock": 3, - "redrive-reused-signal-wait-clock": 2, - "redrive-reused-upserts-cancellation-clock": 3, - "redrive-reused-search-attribute-value-mismatch": 3, - "redrive-reused-search-attribute-type-mismatch": 3, -} +def _check_successor_cancellation( + requests: list[tuple[dict[str, Any], set[int]]], + deliveries: list[tuple[dict[str, Any], set[int]]], + instance: str, run: str, resume: Any, +) -> None: + """The successor's own cancellation: one request, then one delivery of it.""" + def require(condition: bool, message: str) -> None: + if not condition: + raise AssertionError(message) -def test_every_reused_fixture_is_a_checked_redrive_successor() -> None: - reused = sorted( - path.stem for path in REDRIVE_FIXTURES.glob("*.json") - if "reused_from_run_id" in path.read_text(encoding="utf-8") - ) - assert reused == sorted(REDRIVE_SUCCESSOR_FIXTURES) + require(len(requests) == 1 and len(deliveries) == 1, + "a successor cancellation is exactly one request and one delivery") + (request, _), (delivery, recorded_steps) = requests[0], deliveries[0] + require(request["sequence"] < delivery["sequence"], "the delivery follows its request") + require(redrive_time(delivery["recorded_at"]) >= redrive_time(request["recorded_at"]), + "the delivery is not earlier than its request") + asked, delivered = request["payload"], delivery["payload"] + command = asked["workflow_command_id"] + require(delivered["workflow_command_id"] == command, "the delivery names its request") + context = asked.get("cancellation") + require(isinstance(context, dict) and bool(context), "the request carries its cancellation context") + require(delivered.get("cancellation") == context, "the delivery carries its request's context") + assert isinstance(context, dict) + lineage = context.get("lineage") + require(context.get("request_id") == command, "the context is this request") + require(context.get("root_workflow_run_id") == run and context.get("root_workflow_instance_id") == instance, + "successor cancellation must be rooted at the successor run") + require(isinstance(lineage, list) and bool(lineage) and isinstance(lineage[-1], dict) + and lineage[-1].get("request_id") == command + and lineage[-1].get("workflow_run_id") == run + and lineage[-1].get("workflow_instance_id") == instance, + "successor cancellation must be rooted at the successor run") + if context.get("parent_request_id") is None: + require(context.get("root_request_id") == command, "a root request is its own root") + require(asked.get("cleanup_deadline_at") == context.get("cleanup_deadline_at") + and redrive_time(asked["cleanup_deadline_at"]) >= redrive_time(request["recorded_at"]), + "the request's cleanup deadline is its context's and follows it") + # The interrupted call is the first unreused step (a call not yet + # scheduled) or a cancellable call the successor recorded before it. + step = delivered["sequence"] + require(step == resume or step in recorded_steps, + "the delivery targets a successor step the successor reached") + + +REDRIVE_SUCCESSOR_FIXTURES = VALID_SUCCESSORS @pytest.mark.parametrize(("fixture", "resume"), REDRIVE_SUCCESSOR_FIXTURES.items()) @@ -651,7 +771,7 @@ def _last_reused(history: list[dict[str, Any]]) -> dict[str, Any]: "recorded with the start"), (lambda history: [history[1]["payload"].pop(key) for key in ("reused_from_run_id", "reused_recorded_at")], "directly follow"), - (lambda history: history.append({"sequence": len(history) + 1, "event_type": "ActivityScheduled", + (lambda history: history.append({"sequence": len(history) + 1, "event_type": "UpdateApplied", "recorded_at": "2026-11-03T00:05:01+00:00", "payload": {"sequence": 3}}), "unexpected event"), (lambda history: history[1].__setitem__("sequence", 9), "history order"), @@ -694,7 +814,7 @@ def _event(history: list[dict[str, Any]], event_type: str) -> dict[str, Any]: ("redrive-reused-upserts-cancellation-clock", lambda history: _event(history, "CooperativeCancellationDelivered")["payload"].__setitem__( "workflow_command_id", "other-cancel"), - "disagree on workflow_command_id"), + "the delivery names its request"), ("redrive-reused-signal-wait-clock", lambda history: _event(history, "SignalApplied")["payload"].__setitem__("signal_wait_id", "other-wait"), "no opened wait"), diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index c455ab69..d51a1e75 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -43,6 +43,11 @@ ) from tests.test_pending_operation_replay import PendingOperation from tests.test_poll_failure_backoff import RecoveredWorkflow +from tests.test_redrive_successor_evidence import ( + RedriveSuccessorCancellationClockWorkflow, + RedriveSuccessorMarkersWorkflow, + RedriveSuccessorTimerWorkflow, +) from tests.test_replay_clock import ( PersistedClockWorkflow, RedriveChangedStageWorkflow, @@ -365,6 +370,9 @@ def run(self, ctx: WorkflowContext, message: str) -> None: PersistedClockWorkflow, RedriveChangedStageWorkflow, RedriveSignalWorkflow, + RedriveSuccessorCancellationClockWorkflow, + RedriveSuccessorMarkersWorkflow, + RedriveSuccessorTimerWorkflow, RedriveVisibilityCleanupWorkflow, RedriveVisibilityWorkflow, NexusCooperativeDeliveryWorkflow,