From bd41be60990791141923bcc264f9ff82c84ed902 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 16:57:29 -0600 Subject: [PATCH 1/3] Replace the invalid 2.13.1 reused cancellation-clock evidence The 2.13.1 test test_a_reused_step_keeps_its_original_time_for_the_ remaining_budget (#233) used a history Workflow cannot write: no redrive WorkflowStarted, the cancellation delivered before a "reused" step, and reused_recorded_at after the delivery. Its 18.123456 depends on that order (the valid order gives 22.123456). It backed the 2.13.1 clause "The cancellation clock prefers a reused step's reused_recorded_at". Remove that unit test (it is not a corpus fixture) and add three valid successor fixtures built to Workflow 28893f97 FailedRunRedrivePlan::successorEvents, each with the full redrive envelope, a Core-identical reused prefix and resume = highest + 1: - redrive-successor-reused-markers: side effect 1, version marker 2, activity 3, resume 4. The side effect does not run again, and the next activity receives ctx.now() = replayed_started_at, which replaces the start-clock claim of redrive-start-replayed-clock. - redrive-successor-reused-timer: activity 1, Core's copied TimerScheduled and the reused TimerFired 2, resume 3. - redrive-successor-reused-cancellation-clock: the timer successor, its own ActivityScheduled 3, and its own cancellation after the prefix: remaining() is 29.0 from the delivery. The claim becomes non-leakage: valid retiming of reused times changes nothing; a leaking-clock probe reads 11.0 where the SDK keeps 29.0; and a positive control on the delivery gives 11.0. check_redrive_successor accepts Core's activity, timer, side-effect and version-marker shapes, and Core's copied TimerScheduled only when it has no reuse keys, is recorded at the successor start and is directly followed by the reused TimerFired with the same timer_id and step. A reused TimerFired requires that copy, and no successor step may reuse a reused step. A sweep over every corpus fixture changes exactly the three new verdicts. tests/redrive_fixture_registry.py registers every redrive fixture. The 2.13.1 redrive-start-replayed-clock stays byte-identical, pinned by SHA-256, failing the check with "history order", and replaced. The gate requires the registry to match the corpus and exactly the registered successors to pass. The CHANGELOG corrects the 2.13.1 claim under Unreleased. Fixes #237 Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 4 + ...e-successor-reused-cancellation-clock.json | 162 +++++++++ .../redrive-successor-reused-markers.json | 90 +++++ .../redrive-successor-reused-timer.json | 90 +++++ tests/redrive_fixture_registry.py | 64 ++++ tests/test_cancellation_remaining_time.py | 22 +- tests/test_redrive_successor_evidence.py | 337 ++++++++++++++++++ tests/test_replay_clock.py | 86 ++++- tests/test_replay_regression_corpus.py | 8 + 9 files changed, 826 insertions(+), 37 deletions(-) create mode 100644 tests/fixtures/replay_regressions/redrive-successor-reused-cancellation-clock.json create mode 100644 tests/fixtures/replay_regressions/redrive-successor-reused-markers.json create mode 100644 tests/fixtures/replay_regressions/redrive-successor-reused-timer.json create mode 100644 tests/redrive_fixture_registry.py create mode 100644 tests/test_redrive_successor_evidence.py 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..ff039c1b --- /dev/null +++ b/tests/test_redrive_successor_evidence.py @@ -0,0 +1,337 @@ +"""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 + +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 datetime.fromisoformat(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 = datetime.fromisoformat(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"), +], 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"]) +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" + + +@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: _request(history)["payload"]["cancellation"].__setitem__("root_workflow_run_id", SOURCE_RUN), + "rooted at the successor run"), + (lambda history: _delivery(history)["payload"]["cancellation"]["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)) diff --git a/tests/test_replay_clock.py b/tests/test_replay_clock.py index 7a4ee13e..2306a6dc 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,52 @@ 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", +}) + + +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, 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 {} + 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 + 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 check_redrive_successor(history: list[dict[str, Any]], workflow: dict[str, Any]) -> None: @@ -512,6 +555,7 @@ def present(payload: dict[str, Any], key: str) -> bool: steps: list[tuple[int, list[str]]] = [] signals: dict[str, dict[str, Any]] = {} cancellations: dict[str, dict[str, Any]] = {} + 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 +568,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"): @@ -534,6 +580,10 @@ def present(payload: dict[str, Any], key: str) -> bool: 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") + 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) @@ -545,6 +595,17 @@ def present(payload: dict[str, Any], key: str) -> bool: require(payload["workflow_run_id"] == run, "successor event belongs to another run") require(datetime.fromisoformat(event["recorded_at"]) >= started_at, "successor event predates the successor start") + if kind in REDRIVE_SUCCESSOR_STEP_EVENTS: + successor_steps.append(payload["sequence"]) + for key in ("cancellation",): + context = payload.get(key) + if isinstance(context, dict): + lineage = context.get("lineage") + require(context.get("root_workflow_run_id") == run + and isinstance(lineage, list) and bool(lineage) + 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 kind.startswith("Signal"): if kind != "SignalWaitOpened": require(payload["signal_wait_id"] in signals, "signal event has no opened wait") @@ -567,24 +628,11 @@ 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") -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 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) +REDRIVE_SUCCESSOR_FIXTURES = VALID_SUCCESSORS @pytest.mark.parametrize(("fixture", "resume"), REDRIVE_SUCCESSOR_FIXTURES.items()) @@ -651,7 +699,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"), 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, From 52940fb7522b632e9d303755ae4e5621ec8e81a3 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 17:08:50 -0600 Subject: [PATCH 2/3] Bind the successor cancellation exactly and parse times on Python 3.10 Security's hold on bd41be60: - Python 3.10 rejects a trailing Z in datetime.fromisoformat, so the non-leakage tests failed on Core's 00:05:31.000000Z deadline. The validator and the evidence tests now parse history times through redrive_time, which maps a trailing Z to +00:00 and requires an offset. - The successor cancellation is now exactly one request followed by exactly one delivery. Both carry the same non-blank context, the delivery names its request, and it is not earlier than the request. The context is that request, is rooted at the successor run and instance, and its lineage ends at them; a root request is its own root; the request's cleanup deadline is the context's and follows the request. The delivery targets the first unreused step, a call not yet scheduled, or a cancellable call (activity, timer or signal wait) the successor recorded before it, never an unrecorded later step. Fifteen mutations (context deleted from either or both events, duplicate or missing request or delivery, mismatched context, delivery earlier than its request, context request, root or lineage identity, deadline, and delivery to unrecorded step 4) run on both cancellation successors; the bd41be60 check accepted all 30 and this one refuses all 30. A delivery to step 4 after the successor recorded step 4 is accepted. The fixture verdict sweep still changes exactly the three new successors. Refs #237 Co-Authored-By: Claude Opus 5.5 --- tests/test_redrive_successor_evidence.py | 116 +++++++++++++++++++++-- tests/test_replay_clock.py | 98 +++++++++++++++---- 2 files changed, 187 insertions(+), 27 deletions(-) diff --git a/tests/test_redrive_successor_evidence.py b/tests/test_redrive_successor_evidence.py index ff039c1b..cdac389a 100644 --- a/tests/test_redrive_successor_evidence.py +++ b/tests/test_redrive_successor_evidence.py @@ -25,7 +25,7 @@ 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 +from tests.test_replay_clock import REDRIVE_FIXTURES, check_redrive_successor, redrive_fixture, redrive_time SIDE_EFFECT_CALLS: list[str] = [] @@ -192,7 +192,7 @@ def retime(history: list[dict[str, Any]]) -> None: def _time(item: dict[str, Any]) -> datetime: payload = item["payload"] - return datetime.fromisoformat(payload.get("reused_recorded_at") or item["recorded_at"]) + return redrive_time(payload.get("reused_recorded_at") or item["recorded_at"]) def leaking_clock_remaining(history: list[dict[str, Any]]) -> float: @@ -200,7 +200,7 @@ def leaking_clock_remaining(history: list[dict[str, Any]]) -> float: delivered = next(index for index, item in enumerate(history) if item["event_type"] == "CooperativeCancellationDelivered") request = event(history, "CooperativeCancellationRequested") - deadline = datetime.fromisoformat(request["payload"]["cleanup_deadline_at"]) + deadline = redrive_time(request["payload"]["cleanup_deadline_at"]) return (deadline - max(_time(item) for item in history[:delivered + 1])).total_seconds() @@ -315,6 +315,11 @@ def _request(history: list[dict[str, Any]]) -> dict[str, Any]: 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"), @@ -323,10 +328,10 @@ def _request(history: list[dict[str, Any]]) -> dict[str, Any]: (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: _request(history)["payload"]["cancellation"].__setitem__("root_workflow_run_id", SOURCE_RUN), + (lambda history: _both_contexts(history, lambda context: context.__setitem__("root_workflow_run_id", SOURCE_RUN)), "rooted at the successor run"), - (lambda history: _delivery(history)["payload"]["cancellation"]["lineage"][-1].__setitem__( - "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), @@ -335,3 +340,102 @@ def _request(history: list[dict[str, Any]]) -> dict[str, Any]: 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)) diff --git a/tests/test_replay_clock.py b/tests/test_replay_clock.py index 2306a6dc..72579fe3 100644 --- a/tests/test_replay_clock.py +++ b/tests/test_replay_clock.py @@ -517,6 +517,21 @@ def _copied_timer_schedule(history: list[dict[str, Any]], index: int, started: d 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: """Require a successor Workflow 2.19 can write for this fixture's workflow.""" @@ -541,20 +556,22 @@ 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") @@ -579,7 +596,7 @@ 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), @@ -593,19 +610,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"]) - for key in ("cancellation",): - context = payload.get(key) - if isinstance(context, dict): - lineage = context.get("lineage") - require(context.get("root_workflow_run_id") == run - and isinstance(lineage, list) and bool(lineage) - 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 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") @@ -613,11 +627,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") @@ -630,6 +639,53 @@ def present(payload: dict[str, Any], key: str) -> bool: "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) + + +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) + + 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 @@ -742,7 +798,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"), From 262eb6ef1352d1ff4eb494bcd744ec4bb05ce517 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 17:28:19 -0600 Subject: [PATCH 3/3] Bind the copied timer schedule to the source run and plain timers Interim review of 52940fb7 (F1): a copied TimerScheduled in the reused prefix skipped the run checks, so one naming any run was accepted. The copy is the source run's schedule payload (Core 28893f97 FailedRunRedrivePlan.php:437-455), so a workflow_run_id must be absent or the continued_from_run_id. Redrive copies only plain timers (isPlainTimer, :497), so a copy carrying a non-null timer_kind, cancellation_cleanup, cancellation_scope_id, parallel_group_id, condition_wait_id or signal_wait_id is refused. Counterfactuals on both timer successors: a bogus run, the successor run, and each non-plain key fail here and passed at 52940fb7 (16 cases). The source run and null keys are accepted. The fixture verdict sweep still changes exactly the three new successors. Refs #237 Co-Authored-By: Claude Opus 5.5 --- tests/test_redrive_successor_evidence.py | 23 ++++++++++++++++++++++- tests/test_replay_clock.py | 20 ++++++++++++++++++-- 2 files changed, 40 insertions(+), 3 deletions(-) diff --git a/tests/test_redrive_successor_evidence.py b/tests/test_redrive_successor_evidence.py index cdac389a..731597cd 100644 --- a/tests/test_redrive_successor_evidence.py +++ b/tests/test_redrive_successor_evidence.py @@ -297,8 +297,18 @@ def change(history: list[dict[str, Any]]) -> None: _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"]) + "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)) @@ -439,3 +449,14 @@ def later(history: list[dict[str, Any]]) -> None: _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 72579fe3..a991e606 100644 --- a/tests/test_replay_clock.py +++ b/tests/test_replay_clock.py @@ -495,19 +495,35 @@ def test_reused_signal_wait_events_do_not_move_the_cancellation_budget() -> None }) +# 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, and is directly followed by the reused - TimerFired for the same non-blank timer_id and the same step. + 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")