From bd8a825ea79e58fa3070e533f686d97ef317e0ee Mon Sep 17 00:00:00 2001 From: RachelWanggg Date: Thu, 15 Oct 2026 15:16:53 -0700 Subject: [PATCH 1/2] Fix cancelling the task that awaits an external workflow cancel --- changelog/fixed/impatient-pangolin.md | 1 + temporalio/worker/_workflow_instance.py | 4 + tests/worker/test_workflow.py | 116 ++++++++++++++++++++++++ 3 files changed, 121 insertions(+) create mode 100644 changelog/fixed/impatient-pangolin.md diff --git a/changelog/fixed/impatient-pangolin.md b/changelog/fixed/impatient-pangolin.md new file mode 100644 index 0000000000..551e000962 --- /dev/null +++ b/changelog/fixed/impatient-pangolin.md @@ -0,0 +1 @@ +Cancelling a workflow, or the task awaiting `ExternalWorkflowHandle.cancel()`, before the external cancel request resolves no longer fails every following workflow task with `InvalidStateError`. diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index 31b655b90c..d12db11361 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -1108,6 +1108,10 @@ def _apply_resolve_request_cancel_external_workflow( f"Failed finding pending external cancel for sequence {job.seq}" ) fut, external_workflow_id = pending + # Cancelling the task that awaits the request also cancels this future, + # but the request was already sent and Core still resolves it + if fut.cancelled(): + return # We intentionally let this error if future is already done if job.HasField("failure"): workflow_context = temporalio.converter.WorkflowSerializationContext( diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py index 7b61ca16a1..f48f51fe48 100644 --- a/tests/worker/test_workflow.py +++ b/tests/worker/test_workflow.py @@ -2046,6 +2046,122 @@ async def test_workflow_cancel_external(client: Client): assert isinstance(err.value.cause, CancelledError) +@workflow.defn +class CancelExternalStopWaitingWorkflow: + def __init__(self) -> None: + self._finish = False + + @workflow.run + async def run(self, external_workflow_id: str) -> None: + cancel_task = asyncio.create_task( + workflow.get_external_workflow_handle(external_workflow_id).cancel() + ) + # Let the task send the request, then stop waiting on it the way + # asyncio.timeout or a failing TaskGroup sibling would + await asyncio.sleep(0) + cancel_task.cancel() + try: + await cancel_task + except asyncio.CancelledError: + pass + await workflow.wait_condition(lambda: self._finish) + + @workflow.signal + def finish(self) -> None: + self._finish = True + + +@workflow.defn +class CancelExternalThenCleanUpWorkflow: + def __init__(self) -> None: + self._finish = False + + @workflow.run + async def run(self, external_workflow_id: str) -> None: + try: + await workflow.get_external_workflow_handle(external_workflow_id).cancel() + except asyncio.CancelledError: + await workflow.wait_condition(lambda: self._finish) + raise + + @workflow.signal + def finish(self) -> None: + self._finish = True + + +@pytest.mark.parametrize( + "scenario", ["stop_waiting", "stop_waiting_target_missing", "workflow_cancelled"] +) +async def test_workflow_cancel_external_wait_cancelled( + client: Client, env: WorkflowEnvironment, scenario: str +): + if scenario == "stop_waiting_target_missing" and env.supports_time_skipping_v1: + pytest.skip( + "Java test server does not resolve a cancel request for a missing workflow" + ) + task_queue = str(uuid.uuid4()) + target: WorkflowHandle[Any, str] | None = None + if scenario == "stop_waiting_target_missing": + target_id = f"workflow-{uuid.uuid4()}" + else: + # No timer, so time-skipping V2 cannot finish it before the cancel + target = await client.start_workflow( + TrapCancelWorkflow.run, + id=f"workflow-{uuid.uuid4()}", + task_queue=task_queue, + ) + target_id = target.id + handle: WorkflowHandle[Any, None] + if scenario == "workflow_cancelled": + handle = await client.start_workflow( + CancelExternalThenCleanUpWorkflow.run, + target_id, + id=f"workflow-{uuid.uuid4()}", + task_queue=task_queue, + ) + # Cancelling before the first workflow task means the cancellation + # arrives in the same activation that sends the external cancel + await handle.cancel() + else: + handle = await client.start_workflow( + CancelExternalStopWaitingWorkflow.run, + target_id, + id=f"workflow-{uuid.uuid4()}", + task_queue=task_queue, + ) + async with new_worker( + client, + CancelExternalStopWaitingWorkflow, + CancelExternalThenCleanUpWorkflow, + TrapCancelWorkflow, + task_queue=task_queue, + ): + + async def cancel_request_resolved() -> None: + async for e in handle.fetch_history_events(): + if e.event_type in ( + EventType.EVENT_TYPE_EXTERNAL_WORKFLOW_EXECUTION_CANCEL_REQUESTED, + EventType.EVENT_TYPE_REQUEST_CANCEL_EXTERNAL_WORKFLOW_EXECUTION_FAILED, + ): + return + assert False, "External cancel request not resolved" + + # Finishing any earlier could complete the workflow before Core + # delivers the resolution + await assert_eventually(cancel_request_resolved) + await handle.signal("finish") + if scenario == "workflow_cancelled": + with pytest.raises(WorkflowFailureError) as err: + await handle.result() + assert isinstance(err.value.cause, CancelledError) + else: + await handle.result() + event_types = [e.event_type async for e in handle.fetch_history_events()] + assert EventType.EVENT_TYPE_WORKFLOW_TASK_FAILED not in event_types + if target: + assert "cancelled" == await target.result() + + @dataclass class SignalExternalWorkflowArgs: external_workflow_id: str From 8d09f71c9b2272be157b564373ae9c95aee91c6a Mon Sep 17 00:00:00 2001 From: RachelWanggg Date: Fri, 16 Oct 2026 09:43:23 -0700 Subject: [PATCH 2/2] Split external cancel tests into separate tests --- tests/worker/test_workflow.py | 119 ++++++++++++++++++++-------------- 1 file changed, 69 insertions(+), 50 deletions(-) diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py index f48f51fe48..cfc0190314 100644 --- a/tests/worker/test_workflow.py +++ b/tests/worker/test_workflow.py @@ -2089,77 +2089,96 @@ def finish(self) -> None: self._finish = True -@pytest.mark.parametrize( - "scenario", ["stop_waiting", "stop_waiting_target_missing", "workflow_cancelled"] -) -async def test_workflow_cancel_external_wait_cancelled( - client: Client, env: WorkflowEnvironment, scenario: str -): - if scenario == "stop_waiting_target_missing" and env.supports_time_skipping_v1: - pytest.skip( - "Java test server does not resolve a cancel request for a missing workflow" - ) - task_queue = str(uuid.uuid4()) - target: WorkflowHandle[Any, str] | None = None - if scenario == "stop_waiting_target_missing": - target_id = f"workflow-{uuid.uuid4()}" - else: +async def wait_for_external_cancel_resolution(handle: WorkflowHandle) -> None: + async def resolved() -> None: + async for e in handle.fetch_history_events(): + if e.event_type in ( + EventType.EVENT_TYPE_EXTERNAL_WORKFLOW_EXECUTION_CANCEL_REQUESTED, + EventType.EVENT_TYPE_REQUEST_CANCEL_EXTERNAL_WORKFLOW_EXECUTION_FAILED, + ): + return + assert False, "External cancel request not resolved" + + # Letting the workflow finish any earlier could complete it before Core + # delivers the resolution + await assert_eventually(resolved) + + +async def test_workflow_cancel_external_stop_waiting(client: Client): + async with new_worker( + client, CancelExternalStopWaitingWorkflow, TrapCancelWorkflow + ) as worker: # No timer, so time-skipping V2 cannot finish it before the cancel target = await client.start_workflow( TrapCancelWorkflow.run, id=f"workflow-{uuid.uuid4()}", - task_queue=task_queue, + task_queue=worker.task_queue, ) - target_id = target.id - handle: WorkflowHandle[Any, None] - if scenario == "workflow_cancelled": handle = await client.start_workflow( - CancelExternalThenCleanUpWorkflow.run, - target_id, + CancelExternalStopWaitingWorkflow.run, + target.id, id=f"workflow-{uuid.uuid4()}", - task_queue=task_queue, + task_queue=worker.task_queue, ) - # Cancelling before the first workflow task means the cancellation - # arrives in the same activation that sends the external cancel - await handle.cancel() - else: + await wait_for_external_cancel_resolution(handle) + await handle.signal(CancelExternalStopWaitingWorkflow.finish) + await handle.result() + event_types = [e.event_type async for e in handle.fetch_history_events()] + assert EventType.EVENT_TYPE_WORKFLOW_TASK_FAILED not in event_types + assert "cancelled" == await target.result() + + +async def test_workflow_cancel_external_stop_waiting_target_missing( + client: Client, env: WorkflowEnvironment +): + if env.supports_time_skipping_v1: + pytest.skip( + "Java test server does not resolve a cancel request for a missing workflow" + ) + async with new_worker(client, CancelExternalStopWaitingWorkflow) as worker: handle = await client.start_workflow( CancelExternalStopWaitingWorkflow.run, - target_id, + f"workflow-{uuid.uuid4()}", id=f"workflow-{uuid.uuid4()}", - task_queue=task_queue, + task_queue=worker.task_queue, ) + await wait_for_external_cancel_resolution(handle) + await handle.signal(CancelExternalStopWaitingWorkflow.finish) + await handle.result() + event_types = [e.event_type async for e in handle.fetch_history_events()] + assert EventType.EVENT_TYPE_WORKFLOW_TASK_FAILED not in event_types + + +async def test_workflow_cancel_external_workflow_cancelled(client: Client): + task_queue = str(uuid.uuid4()) + target = await client.start_workflow( + TrapCancelWorkflow.run, + id=f"workflow-{uuid.uuid4()}", + task_queue=task_queue, + ) + handle = await client.start_workflow( + CancelExternalThenCleanUpWorkflow.run, + target.id, + id=f"workflow-{uuid.uuid4()}", + task_queue=task_queue, + ) + # Cancelling before the first workflow task means the cancellation arrives + # in the same activation that sends the external cancel + await handle.cancel() async with new_worker( client, - CancelExternalStopWaitingWorkflow, CancelExternalThenCleanUpWorkflow, TrapCancelWorkflow, task_queue=task_queue, ): - - async def cancel_request_resolved() -> None: - async for e in handle.fetch_history_events(): - if e.event_type in ( - EventType.EVENT_TYPE_EXTERNAL_WORKFLOW_EXECUTION_CANCEL_REQUESTED, - EventType.EVENT_TYPE_REQUEST_CANCEL_EXTERNAL_WORKFLOW_EXECUTION_FAILED, - ): - return - assert False, "External cancel request not resolved" - - # Finishing any earlier could complete the workflow before Core - # delivers the resolution - await assert_eventually(cancel_request_resolved) - await handle.signal("finish") - if scenario == "workflow_cancelled": - with pytest.raises(WorkflowFailureError) as err: - await handle.result() - assert isinstance(err.value.cause, CancelledError) - else: + await wait_for_external_cancel_resolution(handle) + await handle.signal(CancelExternalThenCleanUpWorkflow.finish) + with pytest.raises(WorkflowFailureError) as err: await handle.result() + assert isinstance(err.value.cause, CancelledError) event_types = [e.event_type async for e in handle.fetch_history_events()] assert EventType.EVENT_TYPE_WORKFLOW_TASK_FAILED not in event_types - if target: - assert "cancelled" == await target.result() + assert "cancelled" == await target.result() @dataclass