From b92101628557a34f838c97ea1d12ed1d56743adb Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 20:23:27 -0600 Subject: [PATCH 1/2] Bind the redrive successor cancellation to Core's full shape The successor check now requires a root v1 cancellation context whose lineage is exactly the request's own address, a null parent_request_id, requested_at and schema, the request's command_type and the delivery's single-call call_kind. A delivery to a call resolved before its request, or to an unscheduled step that is recorded later, is refused (Workflow 28893f97 CancellationContext::fromArray, WorkflowStub::cancel and CooperativeCancellationDelivery). The unreleased #236 fixture redrive-reused-upserts-cancellation-clock has no command_type. It stays byte-identical, is registered as non-qualifying, and redrive-successor-reused-upserts-cancellation-clock replaces it. Fixes #239. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 1 + ...sor-reused-upserts-cancellation-clock.json | 155 ++++++++++++++++++ tests/redrive_fixture_registry.py | 15 +- tests/test_redrive_successor_evidence.py | 121 +++++++++++++- tests/test_replay_clock.py | 115 ++++++++----- .../test_search_attribute_replay_identity.py | 2 +- 6 files changed, 368 insertions(+), 41 deletions(-) create mode 100644 tests/fixtures/replay_regressions/redrive-successor-reused-upserts-cancellation-clock.json diff --git a/CHANGELOG.md b/CHANGELOG.md index 2936b414..0f628137 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### 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. +- **Redrive successor evidence is checked against Core's full cancellation shape (#239).** The test check for a successor's own cancellation now requires what Workflow 28893f97 writes: a root `durable-workflow.cancellation-context/v1` context whose lineage is exactly the request's own address, a null `parent_request_id`, `requested_at` no later than the request, `command_type` `request_cancellation` on the request, and a single-call `call_kind` on the delivery that matches the recorded call. A delivery to a call resolved before the request is refused, and so is any event recorded at an unscheduled step the delivery consumed. The unreleased #236 fixture `redrive-reused-upserts-cancellation-clock` has no `command_type`, so it is registered as non-qualifying and `redrive-successor-reused-upserts-cancellation-clock` replaces it. Tests and fixtures only; replay behavior is unchanged. ### Added diff --git a/tests/fixtures/replay_regressions/redrive-successor-reused-upserts-cancellation-clock.json b/tests/fixtures/replay_regressions/redrive-successor-reused-upserts-cancellation-clock.json new file mode 100644 index 00000000..c848c908 --- /dev/null +++ b/tests/fixtures/replay_regressions/redrive-successor-reused-upserts-cancellation-clock.json @@ -0,0 +1,155 @@ +{ + "$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-upserts-cancellation-clock", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.redrive-visibility-cleanup", + "input": [ + "Taylor" + ], + "payload_codec": "avro", + "workflow_id": "regression-corpus-redrive-visibility", + "run_id": "01JBREDRIVESUCCESSORRUN0000" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "workflow_type": "tests.replay.redrive-visibility-cleanup", + "workflow_instance_id": "regression-corpus-redrive-visibility", + "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": "MemoUpserted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "sequence": 1, + "entries": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + }, + "merged": { + "codec": "avro", + "blob": "wwHioz3/VYAiNw4EEGN1c3RvbWVyCgxUYXlsb3IKc3RhZ2UKEGdyZWV0aW5nAA==" + }, + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:00+00:00" + } + }, + { + "sequence": 3, + "event_type": "SearchAttributesUpserted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "sequence": 2, + "attributes": { + "Stage": "greeting" + }, + "attribute_types": { + "Stage": "keyword" + }, + "merged": { + "Stage": "greeting" + }, + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_recorded_at": "2026-11-03T00:00:00+00:00" + } + }, + { + "sequence": 4, + "event_type": "CooperativeCancellationRequested", + "recorded_at": "2026-11-03T00:05:02+00:00", + "payload": { + "workflow_command_id": "redrive-cancel-1", + "workflow_instance_id": "regression-corpus-redrive-visibility", + "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000", + "command_type": "request_cancellation", + "reason": "maintenance", + "cleanup_deadline_at": "2026-11-03T00:05:30.000000Z", + "cancellation": { + "schema": "durable-workflow.cancellation-context/v1", + "request_id": "redrive-cancel-1", + "root_request_id": "redrive-cancel-1", + "root_workflow_instance_id": "regression-corpus-redrive-visibility", + "root_workflow_run_id": "01JBREDRIVESUCCESSORRUN0000", + "parent_request_id": null, + "reason": "maintenance", + "requester": { + "type": "operator", + "id": "operator-1", + "label": "Maintainer" + }, + "source": "control_plane", + "requested_at": "2026-11-03T00:05:02.000000Z", + "cleanup_deadline_at": "2026-11-03T00:05:30.000000Z", + "lineage": [ + { + "request_id": "redrive-cancel-1", + "workflow_instance_id": "regression-corpus-redrive-visibility", + "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000" + } + ] + } + } + }, + { + "sequence": 5, + "event_type": "CooperativeCancellationDelivered", + "recorded_at": "2026-11-03T00:05:04+00:00", + "payload": { + "workflow_command_id": "redrive-cancel-1", + "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000", + "sequence": 3, + "call_kind": "activity", + "cancellation": { + "schema": "durable-workflow.cancellation-context/v1", + "request_id": "redrive-cancel-1", + "root_request_id": "redrive-cancel-1", + "root_workflow_instance_id": "regression-corpus-redrive-visibility", + "root_workflow_run_id": "01JBREDRIVESUCCESSORRUN0000", + "parent_request_id": null, + "reason": "maintenance", + "requester": { + "type": "operator", + "id": "operator-1", + "label": "Maintainer" + }, + "source": "control_plane", + "requested_at": "2026-11-03T00:05:02.000000Z", + "cleanup_deadline_at": "2026-11-03T00:05:30.000000Z", + "lineage": [ + { + "request_id": "redrive-cancel-1", + "workflow_instance_id": "regression-corpus-redrive-visibility", + "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000" + } + ] + } + } + } + ], + "expected": { + "type": "complete_workflow", + "result": { + "after_memo": "2026-11-03T00:00:00+00:00", + "after_search_attributes": "2026-11-03T00:00:00+00:00", + "remaining": 26.0 + } + } +} diff --git a/tests/redrive_fixture_registry.py b/tests/redrive_fixture_registry.py index 36b4141d..b98d1841 100644 --- a/tests/redrive_fixture_registry.py +++ b/tests/redrive_fixture_registry.py @@ -28,7 +28,6 @@ class NonQualifying: # 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, @@ -36,6 +35,9 @@ class NonQualifying: "redrive-successor-reused-markers": 4, "redrive-successor-reused-timer": 3, "redrive-successor-reused-cancellation-clock": 3, + # Added by #239: redrive-reused-upserts-cancellation-clock with the + # request's command_type, which Core WorkflowStub::cancel always writes. + "redrive-successor-reused-upserts-cancellation-clock": 3, } NON_QUALIFYING: dict[str, NonQualifying] = { @@ -51,6 +53,17 @@ class NonQualifying: "start clock on a valid successor." ), ), + "redrive-reused-upserts-cancellation-clock": NonQualifying( + sha256="f8c40d18fa71a0656089290fd5be44e7df318fafd12721d7a6680e1113ec4856", + error="the request's command_type is request_cancellation", + replacement="redrive-successor-reused-upserts-cancellation-clock", + release="unreleased (#236, after 2.13.1)", + reason=( + "Its CooperativeCancellationRequested has no command_type. Core WorkflowStub::cancel " + "(workflow 28893f97) always writes command_type request_cancellation, so Workflow cannot " + "write this request. Its replay result is correct; the replacement adds only that field." + ), + ), } # Fixtures for a run that redrive starts from. None in this corpus. diff --git a/tests/test_redrive_successor_evidence.py b/tests/test_redrive_successor_evidence.py index 731597cd..bf01bd59 100644 --- a/tests/test_redrive_successor_evidence.py +++ b/tests/test_redrive_successor_evidence.py @@ -355,7 +355,7 @@ def test_the_successor_cancellation_is_its_own(change: Any, message: str) -> Non # 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"] +CANCELLED_SUCCESSORS = [CANCELLATION, "redrive-successor-reused-upserts-cancellation-clock"] def _duplicate(event_type: str) -> Any: @@ -460,3 +460,122 @@ def later(history: list[dict[str, Any]]) -> None: 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)) + + +# #239: the successor cancellation as Core 28893f97 writes it. Each mutation +# below was accepted by the 262eb6ef check and is refused here. +def _step_three_activity(history: list[dict[str, Any]]) -> dict[str, Any]: + """The successor's step-3 ActivityScheduled, inserted before the request if absent.""" + scheduled = [item for item in history if item["event_type"] == "ActivityScheduled"] + if scheduled: + return scheduled[0] + item = {"event_type": "ActivityScheduled", "recorded_at": _request(history)["recorded_at"], + "payload": {"activity_execution_id": "01JBREDRIVESUCCESSORACT001", + "activity_type": "tests.redrive.second", "sequence": 3}} + history.insert(history.index(_request(history)), item) + _renumber(history) + return item + + +def _completed_before_request(history: list[dict[str, Any]]) -> None: + scheduled = _step_three_activity(history) + completed = {"event_type": "ActivityCompleted", "recorded_at": _request(history)["recorded_at"], + "payload": {**scheduled["payload"], "result": "wwHioz3/VYAiNwoEb2s=", "payload_codec": "avro"}} + history.insert(history.index(scheduled) + 1, completed) + _renumber(history) + + +def _scheduled_after_delivery(history: list[dict[str, Any]]) -> None: + scheduled = _step_three_activity(history) + history.remove(scheduled) + scheduled["recorded_at"] = _delivery(history)["recorded_at"] + history.insert(history.index(_delivery(history)) + 1, scheduled) + _renumber(history) + + +def _source_run_prepended(context: dict[str, Any]) -> None: + context["lineage"].insert(0, {"request_id": "source-request", "workflow_instance_id": + context["root_workflow_instance_id"], "workflow_run_id": SOURCE_RUN}) + + +CORE_CANCELLATION_MUTATIONS = { + # 1. Root binding: a parent no longer skips the root check. + "parent-with-another-root": (_context("both", lambda context: context.update( + parent_request_id="parent-request", root_request_id="parent-request")), "its own root"), + "parent-request-id-missing": (_context("both", lambda context: context.pop("parent_request_id")), "its own root"), + # 2. Lineage: the whole lineage is the request's own address. + "lineage-source-run-prepended": (_context("both", _source_run_prepended), "rooted at the successor run"), + "lineage-entry-repeated": (_context("both", lambda context: context["lineage"].append( + dict(context["lineage"][0]))), "rooted at the successor run"), + # 3. Delivery step. + "delivery-to-a-call-completed-before-the-request": (_completed_before_request, "resolved before its request"), + "schedule-after-a-delivery-to-the-resume-step": (_scheduled_after_delivery, "the successor reached"), + # 4. Fields Core always writes. + "request-command-type-missing": (lambda history: _request(history)["payload"].pop("command_type"), + "command_type"), + "request-command-type-other": (lambda history: _request(history)["payload"].__setitem__( + "command_type", "request_termination"), "command_type"), + "delivery-call-kind-missing": (lambda history: _delivery(history)["payload"].pop("call_kind"), "call_kind"), + "delivery-call-kind-unknown": (lambda history: _delivery(history)["payload"].__setitem__( + "call_kind", "workflow"), "call_kind"), + "delivery-call-kind-parallel": (lambda history: _delivery(history)["payload"].__setitem__( + "call_kind", "parallel"), "call_kind"), + "delivery-sequence-span": (lambda history: _delivery(history)["payload"].__setitem__("sequence_span", 2), + "one single call"), + "context-schema-missing": (_context("both", lambda context: context.pop("schema")), "schema"), + "context-schema-unknown": (_context("both", lambda context: context.__setitem__( + "schema", "durable-workflow.cancellation-context/v3")), "schema"), + "context-schema-scoped": (_context("both", lambda context: context.__setitem__( + "schema", "durable-workflow.cancellation-context/v2")), "schema"), + "context-scope-origin": (_context("both", lambda context: context.__setitem__("scope_origin", {})), "schema"), + "context-requested-at-missing": (_context("both", lambda context: context.pop("requested_at")), + "history time must be a string"), + "context-requested-after-the-request": (_context("both", lambda context: context.__setitem__( + "requested_at", "2026-11-03T00:05:05.000000Z")), "requested_at"), + "context-requested-before-the-start": (_context("both", lambda context: context.__setitem__( + "requested_at", "2026-11-03T00:04:59.000000Z")), "requested_at"), + "context-deadline-at-the-request": (lambda history: [ + target.__setitem__("cleanup_deadline_at", _request(history)["payload"]["cancellation"]["requested_at"]) + for target in (_request(history)["payload"], _request(history)["payload"]["cancellation"], + _delivery(history)["payload"]["cancellation"])], "cleanup deadline"), + "context-requester-empty": (_context("both", lambda context: context.__setitem__("requester", {})), + "requester"), + "context-requester-unknown-key": (_context("both", lambda context: context["requester"].__setitem__( + "email", "operator@example.test")), "requester"), + "context-source-blank": (_context("both", lambda context: context.__setitem__("source", " ")), "source"), + "request-reason-differs": (lambda history: _request(history)["payload"].__setitem__("reason", "other"), + "reason"), +} + + +@pytest.mark.parametrize("name", CANCELLED_SUCCESSORS) +@pytest.mark.parametrize("mutation", sorted(CORE_CANCELLATION_MUTATIONS)) +def test_the_successor_cancellation_has_core_shape(name: str, mutation: str) -> None: + change, message = CORE_CANCELLATION_MUTATIONS[mutation] + with pytest.raises(AssertionError, match=message): + check(name, changed(name, change)) + + +def test_a_delivery_to_a_recorded_call_names_its_kind() -> None: + with pytest.raises(AssertionError, match="the recorded call's kind"): + check(CANCELLATION, changed(CANCELLATION, lambda history: _delivery(history)["payload"].__setitem__( + "call_kind", "timer"))) + + +@pytest.mark.parametrize("call_kind", ["activity", "timer", "child", "service_operation"]) +def test_a_delivery_to_the_unscheduled_resume_step_may_name_any_single_call(call_kind: str) -> None: + # Core lets a delivery consume a call not yet scheduled; its shape is open. + name = "redrive-successor-reused-upserts-cancellation-clock" + check(name, changed(name, lambda history: _delivery(history)["payload"].__setitem__("call_kind", call_kind))) + + +def test_a_call_completed_after_the_request_is_still_deliverable() -> None: + # Core resolvedBeforeRequest: only a resolution before the request refuses it. + def completed_between(history: list[dict[str, Any]]) -> None: + scheduled = _step_three_activity(history) + completed = {"event_type": "ActivityCompleted", "recorded_at": _delivery(history)["recorded_at"], + "payload": {**scheduled["payload"], "result": "wwHioz3/VYAiNwoEb2s=", "payload_codec": "avro"}} + history.insert(history.index(_delivery(history)), completed) + _renumber(history) + + check(CANCELLATION, changed(CANCELLATION, completed_between)) diff --git a/tests/test_replay_clock.py b/tests/test_replay_clock.py index a991e606..9327142b 100644 --- a/tests/test_replay_clock.py +++ b/tests/test_replay_clock.py @@ -401,7 +401,7 @@ def replay_cleanup(workflow_cls: type, history: list[dict[str, Any]], fixture: d def test_reused_upserts_do_not_move_the_cancellation_budget() -> None: - fixture = redrive_fixture("redrive-reused-upserts-cancellation-clock") + fixture = redrive_fixture("redrive-successor-reused-upserts-cancellation-clock") history = fixture["history"] kinds = [event["event_type"] for event in history] assert kinds.index("CooperativeCancellationRequested") > max(kinds.index(kind) for kind in REUSED_UPSERT_EVENTS) @@ -414,7 +414,7 @@ def test_reused_upserts_do_not_move_the_cancellation_budget() -> None: def _signal_successor_with_cancellation() -> tuple[list[dict[str, Any]], dict[str, Any]]: - upserts = redrive_fixture("redrive-reused-upserts-cancellation-clock") + upserts = redrive_fixture("redrive-successor-reused-upserts-cancellation-clock") # The same post-prefix request and delivery, bound to the signal workflow. instance = upserts["workflow"]["workflow_id"] tail = json.loads(json.dumps(upserts["history"][-2:]).replace(instance, "regression-corpus-redrive-signal")) @@ -543,9 +543,15 @@ def redrive_time(value: Any) -> datetime: 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"}) +# Events a successor cancellation can be delivered to (a call the successor +# recorded itself and that waits), with the call_kind a delivery to it names. +REDRIVE_CANCELLABLE_EVENTS = {"ActivityScheduled": "activity", "TimerScheduled": "timer", "SignalWaitOpened": "signal"} +# Core CooperativeCancellationDelivery::CALL_SHAPES: single-call delivery kinds. +REDRIVE_DELIVERY_CALL_KINDS = frozenset({ + "activity", "local_activity", "timer", "condition", "signal", "child", "service_operation", +}) +# Core CooperativeCancellationDelivery::RESOLUTION_TYPES this check can see. +REDRIVE_RESOLUTION_EVENTS = frozenset({"ActivityCompleted", "TimerFired", "SignalApplied"}) def check_redrive_successor(history: list[dict[str, Any]], workflow: dict[str, Any]) -> None: @@ -585,9 +591,7 @@ def present(payload: dict[str, Any], key: str) -> bool: prefix = True steps: list[tuple[int, list[str]]] = [] signals: 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_events: list[dict[str, Any]] = [] successor_steps: list[int] = [] for index, event in enumerate(history[1:], start=2): require(event.get("sequence") == index, "history order") @@ -630,12 +634,7 @@ def present(payload: dict[str, Any], key: str) -> bool: "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))) + successor_events.append(event) if kind.startswith("Signal"): if kind != "SignalWaitOpened": require(payload["signal_wait_id"] in signals, "signal event has no opened wait") @@ -655,53 +654,93 @@ 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) + if any(event["event_type"].startswith("CooperativeCancellation") for event in successor_events): + _check_successor_cancellation(successor_events, instance, run, resume, started_at) 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, + events: list[dict[str, Any]], instance: str, run: str, resume: Any, started_at: datetime, ) -> None: - """The successor's own cancellation: one request, then one delivery of it.""" + """The successor's own cancellation: one request, then one delivery of it. + + Shapes follow Workflow 28893f97: WorkflowStub::cancel writes the request + and a root CancellationContext (CancellationContext::fromArray), and + CooperativeCancellationDelivery::record and ::validateCallBoundary write + and gate the delivery. + """ def require(condition: bool, message: str) -> None: if not condition: raise AssertionError(message) + def text(value: Any) -> bool: + return isinstance(value, str) and bool(value.strip()) + + requests = [event for event in events if event["event_type"] == "CooperativeCancellationRequested"] + deliveries = [event for event in events if event["event_type"] == "CooperativeCancellationDelivered"] 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] + request, delivery = 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(asked.get("command_type") == "request_cancellation", "the request's command_type is request_cancellation") 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") + # A run's own cancellation is a root v1 context: it names itself as its + # root, has no parent, and its lineage is exactly its own address. A + # propagated (parent or scope) context is rooted at another run and is + # not a successor's own cancellation. + require(context.get("schema") == "durable-workflow.cancellation-context/v1" + and "scope_origin" not in context and "scope_authority_deadline_at" not in context, + "the context schema is a root durable-workflow.cancellation-context/v1") 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(context.get("lineage") == [ + {"request_id": command, "workflow_instance_id": instance, "workflow_run_id": run}, + ], "successor cancellation must be rooted at the successor run") + require("parent_request_id" in context and context["parent_request_id"] is None + and context.get("root_request_id") == command, "a root request is its own root") + requester = context.get("requester") + require(isinstance(requester, dict) and bool(requester) + and all(key in {"type", "id", "label"} and text(value) for key, value in requester.items()), + "the context requester identifies its caller") + require(text(context.get("source")), "the context names its source") + require(context.get("reason") is None or isinstance(context.get("reason"), str), "the context reason") + require(asked.get("reason") == context.get("reason"), "the request's reason is its context's") + requested_at = redrive_time(context.get("requested_at")) + require(started_at <= requested_at <= redrive_time(request["recorded_at"]), + "the context requested_at is the request's time") require(asked.get("cleanup_deadline_at") == context.get("cleanup_deadline_at") - and redrive_time(asked["cleanup_deadline_at"]) >= redrive_time(request["recorded_at"]), + and redrive_time(asked["cleanup_deadline_at"]) > requested_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. + # One single call: span and selection operands are other delivery shapes. + call_kind = delivered.get("call_kind") + require(call_kind in REDRIVE_DELIVERY_CALL_KINDS, "the delivery's call_kind is a single call kind") + require(not any(key in delivered for key in ("sequence_span", "operation_sequence", "operation_sequence_span")), + "the delivery is one single call") step = delivered["sequence"] - require(step == resume or step in recorded_steps, - "the delivery targets a successor step the successor reached") + at_step = [event for event in events if event is not delivery and event["payload"].get("sequence") == step + and event["event_type"] in REDRIVE_SUCCESSOR_STEP_EVENTS] + opened = [event for event in at_step + if event["event_type"] in REDRIVE_CANCELLABLE_EVENTS and event["sequence"] < delivery["sequence"]] + if opened: + # A cancellable call the successor recorded, still open at the request + # (Core resolvedBeforeRequest), of the kind the delivery names. + require(call_kind == REDRIVE_CANCELLABLE_EVENTS[opened[0]["event_type"]], + "the delivery's call_kind is the recorded call's kind") + require(not any(event["event_type"] in REDRIVE_RESOLUTION_EVENTS and event["sequence"] < request["sequence"] + for event in at_step), "the delivery targets a call resolved before its request") + else: + # The first unreused step, a call not yet scheduled: the delivery + # consumes it, so no successor event occupies it before or after. + require(step == resume and not at_step, "the delivery targets a successor step the successor reached") REDRIVE_SUCCESSOR_FIXTURES = VALID_SUCCESSORS @@ -799,19 +838,19 @@ def _event(history: list[dict[str, Any]], event_type: str) -> dict[str, Any]: ("redrive-reused-signal-wait-clock", lambda history: _event(history, "SignalWaitOpened")["payload"].pop("reused_recorded_at"), "lacks reused_recorded_at"), - ("redrive-reused-upserts-cancellation-clock", + ("redrive-successor-reused-upserts-cancellation-clock", lambda history: _event(history, "CooperativeCancellationRequested")["payload"].__setitem__( "workflow_run_id", REDRIVE_SOURCE_RUN), "successor event belongs to another run"), - ("redrive-reused-upserts-cancellation-clock", + ("redrive-successor-reused-upserts-cancellation-clock", lambda history: _event(history, "CooperativeCancellationRequested")["payload"].__setitem__( "workflow_instance_id", "other"), "another workflow instance"), - ("redrive-reused-upserts-cancellation-clock", + ("redrive-successor-reused-upserts-cancellation-clock", lambda history: _event(history, "CooperativeCancellationRequested").__setitem__( "recorded_at", "2026-11-03T00:04:59+00:00"), "predates"), - ("redrive-reused-upserts-cancellation-clock", + ("redrive-successor-reused-upserts-cancellation-clock", lambda history: _event(history, "CooperativeCancellationDelivered")["payload"].__setitem__( "workflow_command_id", "other-cancel"), "the delivery names its request"), diff --git a/tests/test_search_attribute_replay_identity.py b/tests/test_search_attribute_replay_identity.py index 46ce7ab3..bdf88336 100644 --- a/tests/test_search_attribute_replay_identity.py +++ b/tests/test_search_attribute_replay_identity.py @@ -445,7 +445,7 @@ def test_pointer_records_claim_no_sequence(event_type: str) -> None: # Cancellation deliveries are durable command positions (Core # WorkflowStepHistory::nextSequenceFromEvents). A delivery claims its sequence # and, with sequence_span, every sequence through sequence + span - 1. -DELIVERY_FIXTURE = "redrive-reused-upserts-cancellation-clock" +DELIVERY_FIXTURE = "redrive-successor-reused-upserts-cancellation-clock" def _cooperative(fixture: str, sequence: int, span: int, order: str, drop_memo: bool = False) -> list[dict[str, Any]]: From 7cf1a63a95de69e6deae8f78eee01017c17e0b2c Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 20:45:03 -0600 Subject: [PATCH 2/2] Derive the delivered call kind as Core callKindAt does An ActivityScheduled with local_activity true is a local_activity call, and a TimerScheduled's timer_kind maps condition and signal timeouts, so a valid local-activity delivery is accepted and an activity delivery to it is refused. The requester check refuses only an empty string, as CancellationContext::fromArray does. Citations now name WorkflowStub::requestCancellation / recordCancellationRequest, and the parallel and selection_handle refusal is documented as deliberate. Refs #239. Co-Authored-By: Claude Opus 5.5 --- tests/redrive_fixture_registry.py | 4 +-- tests/test_redrive_successor_evidence.py | 32 ++++++++++++++++++++++++ tests/test_replay_clock.py | 29 +++++++++++++++------ 3 files changed, 55 insertions(+), 10 deletions(-) diff --git a/tests/redrive_fixture_registry.py b/tests/redrive_fixture_registry.py index b98d1841..fda7570e 100644 --- a/tests/redrive_fixture_registry.py +++ b/tests/redrive_fixture_registry.py @@ -36,7 +36,7 @@ class NonQualifying: "redrive-successor-reused-timer": 3, "redrive-successor-reused-cancellation-clock": 3, # Added by #239: redrive-reused-upserts-cancellation-clock with the - # request's command_type, which Core WorkflowStub::cancel always writes. + # request's command_type, which Core WorkflowStub::recordCancellationRequest always writes. "redrive-successor-reused-upserts-cancellation-clock": 3, } @@ -59,7 +59,7 @@ class NonQualifying: replacement="redrive-successor-reused-upserts-cancellation-clock", release="unreleased (#236, after 2.13.1)", reason=( - "Its CooperativeCancellationRequested has no command_type. Core WorkflowStub::cancel " + "Its CooperativeCancellationRequested has no command_type. Core WorkflowStub::recordCancellationRequest " "(workflow 28893f97) always writes command_type request_cancellation, so Workflow cannot " "write this request. Its replay result is correct; the replacement adds only that field." ), diff --git a/tests/test_redrive_successor_evidence.py b/tests/test_redrive_successor_evidence.py index bf01bd59..aeae84ee 100644 --- a/tests/test_redrive_successor_evidence.py +++ b/tests/test_redrive_successor_evidence.py @@ -562,6 +562,38 @@ def test_a_delivery_to_a_recorded_call_names_its_kind() -> None: "call_kind", "timer"))) +def _local_activity(call_kind: str) -> Any: + def change(history: list[dict[str, Any]]) -> None: + event(history, "ActivityScheduled")["payload"]["local_activity"] = True + _delivery(history)["payload"]["call_kind"] = call_kind + + return change + + +def test_a_delivery_to_a_recorded_local_activity_names_local_activity() -> None: + # Core callKindAt: an ActivityScheduled with local_activity true is a + # local_activity call. b9210162 refused this valid delivery. + check(CANCELLATION, changed(CANCELLATION, _local_activity("local_activity"))) + + +def test_an_activity_delivery_to_a_recorded_local_activity_is_refused() -> None: + # b9210162 accepted this delivery, which Core refuses. + with pytest.raises(AssertionError, match="the recorded call's kind"): + check(CANCELLATION, changed(CANCELLATION, _local_activity("activity"))) + + +def test_a_local_activity_delivery_to_a_recorded_activity_is_refused() -> None: + with pytest.raises(AssertionError, match="the recorded call's kind"): + check(CANCELLATION, changed(CANCELLATION, lambda history: _delivery(history)["payload"].__setitem__( + "call_kind", "local_activity"))) + + +@pytest.mark.parametrize("name", CANCELLED_SUCCESSORS) +def test_a_whitespace_requester_value_is_accepted_as_core_does(name: str) -> None: + # Core CancellationContext::fromArray refuses only an empty string. + check(name, changed(name, _context("both", lambda context: context["requester"].__setitem__("label", " ")))) + + @pytest.mark.parametrize("call_kind", ["activity", "timer", "child", "service_operation"]) def test_a_delivery_to_the_unscheduled_resume_step_may_name_any_single_call(call_kind: str) -> None: # Core lets a delivery consume a call not yet scheduled; its shape is open. diff --git a/tests/test_replay_clock.py b/tests/test_replay_clock.py index 9327142b..4eb5858c 100644 --- a/tests/test_replay_clock.py +++ b/tests/test_replay_clock.py @@ -543,9 +543,19 @@ def redrive_time(value: Any) -> datetime: return parsed -# Events a successor cancellation can be delivered to (a call the successor -# recorded itself and that waits), with the call_kind a delivery to it names. -REDRIVE_CANCELLABLE_EVENTS = {"ActivityScheduled": "activity", "TimerScheduled": "timer", "SignalWaitOpened": "signal"} +# 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 _recorded_call_kind(event: dict[str, Any]) -> str: + """The call_kind Core CooperativeCancellationDelivery::callKindAt derives.""" + payload = event["payload"] + if event["event_type"] == "ActivityScheduled": + return "local_activity" if payload.get("local_activity") is True else "activity" + if event["event_type"] == "TimerScheduled": + return {"condition_timeout": "condition", "signal_timeout": "signal"}.get(payload.get("timer_kind"), "timer") + return "signal" # Core CooperativeCancellationDelivery::CALL_SHAPES: single-call delivery kinds. REDRIVE_DELIVERY_CALL_KINDS = frozenset({ "activity", "local_activity", "timer", "condition", "signal", "child", "service_operation", @@ -663,10 +673,12 @@ def _check_successor_cancellation( ) -> None: """The successor's own cancellation: one request, then one delivery of it. - Shapes follow Workflow 28893f97: WorkflowStub::cancel writes the request - and a root CancellationContext (CancellationContext::fromArray), and + Shapes follow Workflow 28893f97: WorkflowStub::requestCancellation, + through recordCancellationRequest, writes the request and a root + CancellationContext (CancellationContext::fromArray), and CooperativeCancellationDelivery::record and ::validateCallBoundary write - and gate the delivery. + and gate the delivery. Parallel and selection_handle deliveries are + deliberately refused: no successor fixture uses those shapes. """ def require(condition: bool, message: str) -> None: @@ -709,7 +721,8 @@ def text(value: Any) -> bool: and context.get("root_request_id") == command, "a root request is its own root") requester = context.get("requester") require(isinstance(requester, dict) and bool(requester) - and all(key in {"type", "id", "label"} and text(value) for key, value in requester.items()), + and all(key in {"type", "id", "label"} and isinstance(value, str) and value != "" + for key, value in requester.items()), "the context requester identifies its caller") require(text(context.get("source")), "the context names its source") require(context.get("reason") is None or isinstance(context.get("reason"), str), "the context reason") @@ -733,7 +746,7 @@ def text(value: Any) -> bool: if opened: # A cancellable call the successor recorded, still open at the request # (Core resolvedBeforeRequest), of the kind the delivery names. - require(call_kind == REDRIVE_CANCELLABLE_EVENTS[opened[0]["event_type"]], + require(call_kind == _recorded_call_kind(opened[0]), "the delivery's call_kind is the recorded call's kind") require(not any(event["event_type"] in REDRIVE_RESOLUTION_EVENTS and event["sequence"] < request["sequence"] for event in at_step), "the delivery targets a call resolved before its request")