diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py index 744a311..00f73d6 100644 --- a/src/durable_workflow/workflow.py +++ b/src/durable_workflow/workflow.py @@ -6751,6 +6751,11 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None continue ctx.logger._set_replaying(False) _assert_pending_step_matches(cmd) + if _next_unconsumed_recorded_step() is not None: + # Admission already has its durable sequence. A later task + # waits for that call's terminal history instead of starting + # the same operation again at a new sequence. + return _state(pending) pending.append(cmd) return _state(pending) if isinstance(cmd, RecordSideEffect): diff --git a/tests/fixtures/replay_regressions/nexus-pending-cancellation-no-readmission.json b/tests/fixtures/replay_regressions/nexus-pending-cancellation-no-readmission.json new file mode 100644 index 0000000..3f5c997 --- /dev/null +++ b/tests/fixtures/replay_regressions/nexus-pending-cancellation-no-readmission.json @@ -0,0 +1,20 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "nexus-pending-cancellation-no-readmission", + "protocol_version": "1.20", + "bindings": ["python"], + "workflow": {"type": "tests.replay.nexus-terminal-delivery", "input": [], "payload_codec": "avro"}, + "history": [ + {"event_type": "ServiceCallStarted", "payload": { + "sequence": 1, "service_call_id": "original-service-call", "endpoint_name": "greeter", + "service_name": "shared", "operation_name": "greet", "operation_mode": "sync", "wait_for": "completed", + "status": "started", "service_call": {"service_call_id": "original-service-call", "target_namespace": "shared", "status": "started"} + }}, + {"event_type": "CooperativeCancellationRequested", "recorded_at": "2026-10-12T07:00:00Z", "payload": { + "workflow_command_id": "original-cancellation", "workflow_run_id": "caller-run", + "cleanup_deadline_at": "2026-10-12T07:00:30Z" + }} + ], + "expected": {"command_sequence": []} +} diff --git a/tests/test_nexus_durable_wait.py b/tests/test_nexus_durable_wait.py index 5500687..6a3975e 100644 --- a/tests/test_nexus_durable_wait.py +++ b/tests/test_nexus_durable_wait.py @@ -84,8 +84,7 @@ def test_new_nexus_call_serializes_the_leased_service_operation() -> None: def test_started_call_stays_pending_and_cold_replay_receives_terminal_result() -> None: started = _service_event("ServiceCallStarted", "started") pending = replay(DurableNexusCaller, [started], [], workflow_id="caller", run_id="caller-run") - assert len(pending.commands) == 1 - assert isinstance(pending.commands[0], NexusServiceCall) + assert pending.commands == [] expected = {"greeting": "hello world"} completed = _service_event("ServiceCallCompleted", "completed", expected) @@ -95,6 +94,26 @@ def test_started_call_stays_pending_and_cold_replay_receives_terminal_result() - assert resumed.commands[0].result == expected +def test_cancellation_request_and_cold_replays_do_not_readmit_pending_call() -> None: + started = _service_event("ServiceCallStarted", "started") + requested = { + "event_type": "CooperativeCancellationRequested", + "recorded_at": "2026-10-12T07:00:00Z", + "payload": { + "workflow_command_id": "original-cancellation", + "workflow_run_id": "caller-run", + "cleanup_deadline_at": "2026-10-12T07:00:30Z", + }, + } + for history in ([started], [started, requested]): + for _ in range(3): + # A fresh workflow instance represents both another task and a + # replacement worker. Neither may allocate another admission. + outcome = replay(DurableNexusCaller, history, [], workflow_id="caller", run_id="caller-run") + assert outcome.commands == [] + assert outcome.cancellation_delivery is None + + @workflow.defn(name="tests.nexus.catch-failure") class CatchingNexusCaller: def run(self, ctx: WorkflowContext): # type: ignore[no-untyped-def]