Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions src/durable_workflow/workflow.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
Original file line number Diff line number Diff line change
@@ -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": []}
}
23 changes: 21 additions & 2 deletions tests/test_nexus_durable_wait.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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]
Expand Down
Loading