Accept cancelled activities and later calls in redrive successor evidence

tests/test_redrive_successor_evidence.py

@@ -611,3 +611,180 @@
         _renumber(history)
 
     check(CANCELLATION, changed(CANCELLATION, completed_between))
+
+
+def _cancelled_activity(history: list[dict[str, Any]], before_request: bool = False) -> dict[str, Any]:
+    scheduled = _step_three_activity(history)
+    boundary = _request(history) if before_request else _delivery(history)
+    cancelled = {
+        "event_type": "ActivityCancelled", "recorded_at": boundary["recorded_at"],
+        "payload": {**scheduled["payload"], "cancelled_at": boundary["recorded_at"]},
+    }
+    history.insert(history.index(boundary), cancelled)
+    _renumber(history)
+    return cancelled
+
+
+def test_an_activity_cancelled_after_the_request_is_still_deliverable() -> None:
+    check(CANCELLATION, changed(CANCELLATION, _cancelled_activity))
+
+
+def test_an_activity_cancelled_before_the_request_is_not_deliverable() -> None:
+    with pytest.raises(AssertionError, match="resolved before its request"):
+        check(CANCELLATION, changed(CANCELLATION, lambda history: _cancelled_activity(history, True)))
+
+
+@pytest.mark.parametrize("key", ["sequence", "activity_type", "activity_execution_id"])
+@pytest.mark.parametrize("missing", [False, True], ids=["blank", "missing"])
+def test_activity_cancellation_requires_its_core_identity(key: str, missing: bool) -> None:
+    def mutate(history: list[dict[str, Any]]) -> None:
+        payload = _cancelled_activity(history)["payload"]
+        if missing:
+            del payload[key]
+        else:
+            payload[key] = 0 if key == "sequence" else " "
+
+    with pytest.raises(AssertionError, match=f"ActivityCancelled lacks {key}"):
+        check(CANCELLATION, changed(CANCELLATION, mutate))
+
+
+def _resolve_activity_before_next_call(
+    history: list[dict[str, Any]], resolution: str, before_request: bool,
+) -> None:
+    scheduled = _step_three_activity(history)
+    boundary = _request(history) if before_request else _delivery(history)
+    terminal = {
+        "event_type": resolution, "recorded_at": boundary["recorded_at"],
+        "payload": {**scheduled["payload"], "result": "wwHioz3/VYAiNwoEb2s=", "payload_codec": "avro"},
+    }
+    if resolution == "ActivityCancelled":
+        terminal["payload"].update(cancelled_at=boundary["recorded_at"])
+        terminal["payload"].pop("result")
+        terminal["payload"].pop("payload_codec")
+    history.insert(history.index(boundary), terminal)
+    _delivery(history)["payload"]["sequence"] = 4
+    _renumber(history)
+
+
+@pytest.mark.parametrize("resolution", ["ActivityCompleted", "ActivityCancelled"])
+@pytest.mark.parametrize("before_request", [False, True], ids=["after-request", "before-request"])
+def test_an_unscheduled_next_call_after_a_resolved_activity_is_deliverable(
+    resolution: str, before_request: bool,
+) -> None:
+    check(CANCELLATION, changed(CANCELLATION, lambda history: _resolve_activity_before_next_call(
+        history, resolution, before_request)))
+
+
+@pytest.mark.parametrize("first_resolved", [False, True], ids=["earlier-open", "all-resolved"])
+def test_the_next_call_follows_every_recorded_successor_call(first_resolved: bool) -> None:
+    def mutate(history: list[dict[str, Any]]) -> None:
+        _resolve_activity_before_next_call(history, "ActivityCompleted", True)
+        first = next(item for item in history if item["event_type"] == "ActivityCompleted"
+                     and item["payload"]["sequence"] == 3)
+        fourth = deepcopy(_step_three_activity(history))
+        fourth["recorded_at"] = _request(history)["recorded_at"]
+        fourth["payload"].update(sequence=4, activity_execution_id="fourth-activity")
+        terminal = {**deepcopy(first), "payload": {**first["payload"], **fourth["payload"]}}
+        at = history.index(_request(history))
+        history[at:at] = [fourth, terminal]
+        if not first_resolved:
+            history.remove(first)
+        _delivery(history)["payload"]["sequence"] = 5
+        _renumber(history)
+
+    if first_resolved:
+        check(CANCELLATION, changed(CANCELLATION, mutate))
+    else:
+        with pytest.raises(AssertionError, match="the successor reached"):
+            check(CANCELLATION, changed(CANCELLATION, mutate))
+
+
+def test_the_next_call_follows_an_atomic_memo_step() -> None:
+    def mutate(history: list[dict[str, Any]]) -> None:
+        history.remove(_step_three_activity(history))
+        memo = deepcopy(event(redrive_fixture("redrive-reused-upserts-clock")["history"], "MemoUpserted"))
+        for key in ("reused_from_run_id", "reused_recorded_at"):
+            memo["payload"].pop(key)
+        memo["payload"]["sequence"] = 3
+        memo["recorded_at"] = _request(history)["recorded_at"]
+        history.insert(history.index(_request(history)), memo)
+        _delivery(history)["payload"]["sequence"] = 4
+        _renumber(history)
+
+    check(CANCELLATION, changed(CANCELLATION, mutate))
+
+
+@pytest.mark.parametrize(("opened", "resolved", "identity"), [
+    ("TimerScheduled", "TimerFired", "timer_id"),
+    ("SignalWaitOpened", "SignalApplied", "signal_wait_id"),
+])
+@pytest.mark.parametrize("state", ["resolved", "open", "foreign-resolution"])
+def test_the_next_call_requires_a_resolution_of_the_recorded_timer_or_signal(
+    opened: str, resolved: str, identity: str, state: str,
+) -> None:
+    def mutate(history: list[dict[str, Any]]) -> None:
+        scheduled = _step_three_activity(history)
+        scheduled["event_type"] = opened
+        scheduled["payload"] = {"sequence": 3, identity: "successor-wait"}
+        if opened == "SignalWaitOpened":
+            scheduled["payload"]["signal_name"] = "approve"
+        terminal = {"event_type": resolved, "recorded_at": _request(history)["recorded_at"],
+                    "payload": dict(scheduled["payload"])}
+        if opened == "SignalWaitOpened":
+            terminal["payload"].update(workflow_command_id="signal-command", signal_id="successor-signal")
+            received = {"event_type": "SignalReceived", "recorded_at": terminal["recorded_at"],
+                        "payload": {**terminal["payload"],
+                                    "workflow_instance_id": _request(history)["payload"]["workflow_instance_id"],
+                                    "workflow_run_id": _request(history)["payload"]["workflow_run_id"]}}
+            received["payload"].pop("sequence")
+            history.insert(history.index(_request(history)), received)
+        if state != "open":
+            if state == "foreign-resolution":
+                terminal["payload"][identity] = "another-wait"
+            history.insert(history.index(_request(history)), terminal)
+        _delivery(history)["payload"]["sequence"] = 4
+        _renumber(history)
+
+    if state == "resolved":
+        check(CANCELLATION, changed(CANCELLATION, mutate))
+    else:
+        with pytest.raises(AssertionError, match="the successor reached|no opened wait"):
+            check(CANCELLATION, changed(CANCELLATION, mutate))
+
+
+@pytest.mark.parametrize("mutation", ["open", "resolved-after-delivery", "foreign-resolution", "skipped-step"])
+def test_an_unscheduled_delivery_cannot_skip_an_unreached_call(mutation: str) -> None:
+    def mutate(history: list[dict[str, Any]]) -> None:
+        _delivery(history)["payload"]["sequence"] = 4
+        if mutation == "open":
+            return
+        _resolve_activity_before_next_call(history, "ActivityCompleted", True)
+        terminal = next(
+            item for item in history if item["event_type"] == "ActivityCompleted"
+            and item["payload"]["sequence"] == 3)
+        if mutation == "resolved-after-delivery":
+            history.remove(terminal)
+            terminal["recorded_at"] = _delivery(history)["recorded_at"]
+            history.append(terminal)
+        elif mutation == "foreign-resolution":
+            terminal["payload"]["activity_execution_id"] = "other-activity"
+        else:
+            _delivery(history)["payload"]["sequence"] = 5
+        _renumber(history)
+
+    with pytest.raises(AssertionError, match="the successor reached"):
+        check(CANCELLATION, changed(CANCELLATION, mutate))
+
+
+def test_selection_handle_delivery_remains_outside_the_successor_model() -> None:
+    def mutate(history: list[dict[str, Any]]) -> None:
+        _delivery(history)["payload"].update(call_kind="selection_handle", operation_sequence=3)
+
+    with pytest.raises(AssertionError, match="single call kind"):
+        check(CANCELLATION, changed(CANCELLATION, mutate))
+
+
+@pytest.mark.parametrize("field", ["sequence_span", "operation_sequence", "operation_sequence_span"])
+def test_group_and_selection_operands_remain_outside_the_successor_model(field: str) -> None:
+    with pytest.raises(AssertionError, match="one single call"):
+        check(CANCELLATION, changed(CANCELLATION, lambda history: _delivery(history)["payload"].__setitem__(field, 1)))

tests/test_replay_clock.py

@@ -497,6 +497,7 @@ def test_reused_signal_wait_events_do_not_move_the_cancellation_budget() -> None
     # followed by the reused TimerFired.
     "ActivityScheduled": ("sequence", "activity_type", "activity_execution_id"),
     "ActivityCompleted": ("sequence", "activity_type", "activity_execution_id"),
+    "ActivityCancelled": ("sequence", "activity_type", "activity_execution_id"),
     "TimerScheduled": ("sequence", "timer_id"),
     "TimerFired": ("sequence", "timer_id"),
     "SideEffectRecorded": ("sequence",),
@@ -512,8 +513,8 @@ def test_reused_signal_wait_events_do_not_move_the_cancellation_budget() -> None
 # 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",
+    "ActivityCompleted", "ActivityCancelled", "TimerScheduled", "TimerFired", "SideEffectRecorded",
+    "VersionMarkerRecorded", "CooperativeCancellationDelivered",
 })
 
 
@@ -583,7 +584,7 @@ def _recorded_call_kind(event: dict[str, Any]) -> str:
     "activity", "local_activity", "timer", "condition", "signal", "child", "service_operation",
 })
 # Core CooperativeCancellationDelivery::RESOLUTION_TYPES this check can see.
-REDRIVE_RESOLUTION_EVENTS = frozenset({"ActivityCompleted", "TimerFired", "SignalApplied"})
+REDRIVE_RESOLUTION_EVENTS = frozenset({"ActivityCompleted", "ActivityCancelled", "TimerFired", "SignalApplied"})
 
 
 def check_redrive_successor(history: list[dict[str, Any]], workflow: dict[str, Any]) -> None:
@@ -773,9 +774,28 @@ def text(value: Any) -> bool:
         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")
+        # Core nextDurableCommandSequence is one past the highest typed step
+        # already recorded. This single-call model also requires earlier calls
+        # to have resolved before delivery; it cannot infer parallel progress.
+        before_delivery = [event for event in events if event["sequence"] < delivery["sequence"]
+                           and event["event_type"] in REDRIVE_SUCCESSOR_STEP_EVENTS]
+        next_step = max((event["payload"]["sequence"] for event in before_delivery), default=resume - 1) + 1
+        require(step == next_step and not at_step, "the delivery targets a successor step the successor reached")
+        resolutions = {
+            "ActivityScheduled": ("activity_execution_id", {"ActivityCompleted", "ActivityCancelled"}),
+            "TimerScheduled": ("timer_id", {"TimerFired"}),
+            "SignalWaitOpened": ("signal_wait_id", {"SignalApplied"}),
+        }
+        for earlier in before_delivery:
+            if earlier["event_type"] not in resolutions:
+                continue
+            identity, kinds = resolutions[earlier["event_type"]]
+            require(any(
+                terminal["sequence"] > earlier["sequence"] and terminal["event_type"] in kinds
+                and terminal["payload"]["sequence"] == earlier["payload"]["sequence"]
+                and terminal["payload"][identity] == earlier["payload"][identity]
+                for terminal in before_delivery
+            ), "the delivery targets a successor step the successor reached: an earlier call is unresolved")
 
 
 REDRIVE_SUCCESSOR_FIXTURES = VALID_SUCCESSORS