From 356e56bc4ad261297e22f5569a52537babd5b5d4 Mon Sep 17 00:00:00 2001 From: DABH Date: Thu, 10 Sep 2026 00:46:29 -0500 Subject: [PATCH 1/2] Cancel activities Core stopped tracking at worker shutdown When a workflow run is evicted while one of its local activities is executing, Core queues a cancel for the activity and then, once the eviction activation completes, invalidates the run, which removes the activity from Core's outstanding set. If the cancel is still queued at that point Core discards it as no longer tracked, while the Python worker keeps running the activity. Core then reports activity polling as finished and wait_all_completed waits for that task forever, so worker shutdown hangs. Once Core's activity poll has shut down it tracks no activity, so anything still executing can only be finished from here. Cancel it with worker_shutdown cancellation details, log a warning, and let its completion be ignored by Core as untracked. This is the shutdown hang behind the frequent macOS CI timeouts of test_workflow_cancel_activity[True], whose captured logs show a workflow task eviction followed by two local activity starts for one run and one fewer cancel. The regression test forces the same race by delaying the activity poll that follows a local activity start and terminating the workflow so the workflow task heartbeat fails. --- CHANGELOG.md | 3 ++ temporalio/worker/_activity.py | 13 +++++++ tests/worker/test_workflow.py | 63 ++++++++++++++++++++++++++++++++++ 3 files changed, 79 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 40e806916..5faaea339 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -52,6 +52,9 @@ to include examples, links to docs, or any other relevant information. ### Fixed +- Worker shutdown no longer waits forever for an activity that Core has stopped tracking, such as a + local activity whose cancellation was lost when its workflow run was evicted. Once activity polling + has shut down, any activity still executing is cancelled with `worker_shutdown` cancellation details. - **Experimental**: External storage metrics now report the wall-clock time storage was in flight. Previously each batch's duration was summed, over-reporting the time whenever storage operations ran concurrently. diff --git a/temporalio/worker/_activity.py b/temporalio/worker/_activity.py index ded3047fc..fa36cd3de 100644 --- a/temporalio/worker/_activity.py +++ b/temporalio/worker/_activity.py @@ -195,6 +195,19 @@ async def drain_poll_queue(self) -> None: # Only call this after run()/drain_poll_queue() have returned. This will not # raise an exception. async def wait_all_completed(self) -> None: + # Core tracks no activities once polling has shut down, so cancel stragglers + for task_token, activity in self._running_activities.items(): + if not activity.done: + logger.warning( + "Cancelling activity %s still running after worker shutdown", + task_token, + ) + activity.cancellation_details.details = ( + temporalio.activity.ActivityCancellationDetails( + worker_shutdown=True + ) + ) + activity.cancel(cancelled_by_request=True) running_tasks = [v.task for v in self._running_activities.values() if v.task] if running_tasks: await asyncio.gather(*running_tasks, return_exceptions=False) diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py index 42ba0b69c..f677046cb 100644 --- a/tests/worker/test_workflow.py +++ b/tests/worker/test_workflow.py @@ -27,6 +27,7 @@ NoReturn, cast, ) +from unittest.mock import patch from urllib.request import urlopen import pydantic @@ -36,6 +37,7 @@ import temporalio.activity import temporalio.api.sdk.v1 +import temporalio.bridge.worker import temporalio.client import temporalio.converter import temporalio.converter._extstore @@ -1135,6 +1137,67 @@ async def activity_started() -> bool: ) +@workflow.defn +class OrphanedLocalActivityWorkflow: + @workflow.run + async def run(self) -> None: + await workflow.execute_local_activity( + "wait_forever_local_activity", + start_to_close_timeout=timedelta(minutes=5), + ) + + +async def test_worker_shutdown_cancels_local_activity_untracked_by_core( + client: Client, +): + started = asyncio.Event() + details: list[temporalio.activity.ActivityCancellationDetails | None] = [] + + @activity.defn(name="wait_forever_local_activity") + async def wait_forever_local_activity() -> None: + started.set() + try: + await asyncio.sleep(1000) + except asyncio.CancelledError: + details.append(activity.cancellation_details()) + + # Delay the next poll so run invalidation drops the queued cancel before dispatch + orig_poll = temporalio.bridge.worker.Worker.poll_activity_task + delay_next_poll = False + + async def slow_poll(self: temporalio.bridge.worker.Worker): + nonlocal delay_next_poll + if delay_next_poll: + delay_next_poll = False + await asyncio.sleep(3) + task = await orig_poll(self) + if task.HasField("start") and task.start.is_local: + delay_next_poll = True + return task + + with patch.object(temporalio.bridge.worker.Worker, "poll_activity_task", slow_poll): + worker = new_worker( + client, + OrphanedLocalActivityWorkflow, + activities=[wait_forever_local_activity], + ) + run_task = asyncio.create_task(worker.run()) + handle = await client.start_workflow( + OrphanedLocalActivityWorkflow.run, + id=f"workflow-{uuid.uuid4()}", + task_queue=worker.task_queue, + task_timeout=timedelta(seconds=1), + ) + await started.wait() + # Terminating fails the workflow task heartbeat, which evicts the run + await handle.terminate() + await asyncio.sleep(4) + await asyncio.wait_for(worker.shutdown(), 20) + await run_task + assert len(details) == 1 and details[0] + assert details[0].worker_shutdown or details[0].cancel_requested + + @workflow.defn class SimpleChildWorkflow: @workflow.run From 2c5b8217f35cc6af9272408d029f4b344c676eb8 Mon Sep 17 00:00:00 2001 From: DABH Date: Thu, 10 Sep 2026 03:15:22 -0500 Subject: [PATCH 2/2] Make the untracked local activity shutdown test deterministic The regression test forced the lost-cancel race with wall-clock timing: a 3s delay on the activity poll that follows a local activity start and a 4s wait after terminating the workflow. On the slow fips CI job shutdown still hung for its 20s limit twice, and the fallback cancel never ran, so the worker was still waiting for its pollers to exit when the test gave up. Gate the test on events instead. The activity poll after the local activity start is held until the workflow poller has returned shutdown: Core only ends the workflow stream once it has processed the eviction completion that invalidates the run, and it marks workflows as shut down for the local activity manager before returning, so the cancel queued at eviction can never reach the worker and shutdown has to cancel the activity itself. The assertion now requires worker_shutdown details. A second test covers the other ordering: the cancel is delivered while Core still tracks the activity, the activity ignores it, and eviction is held until then so the run is invalidated afterwards. Shutdown must still end the activity without replacing its cancel_requested details, which are set once, so the fallback now only sets worker_shutdown details when none were recorded and the changelog entry says so. The workflow task timeout is 3s so that on a loaded runner the task cannot time out before the terminate lands, which replays the workflow and starts a second local activity. --- CHANGELOG.md | 3 +- temporalio/worker/_activity.py | 10 +-- tests/worker/test_workflow.py | 109 ++++++++++++++++++++++++++++----- 3 files changed, 100 insertions(+), 22 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5faaea339..a86c865a6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -54,7 +54,8 @@ to include examples, links to docs, or any other relevant information. - Worker shutdown no longer waits forever for an activity that Core has stopped tracking, such as a local activity whose cancellation was lost when its workflow run was evicted. Once activity polling - has shut down, any activity still executing is cancelled with `worker_shutdown` cancellation details. + has shut down, any activity still executing is cancelled, with `worker_shutdown` cancellation + details if it has none yet. - **Experimental**: External storage metrics now report the wall-clock time storage was in flight. Previously each batch's duration was summed, over-reporting the time whenever storage operations ran concurrently. diff --git a/temporalio/worker/_activity.py b/temporalio/worker/_activity.py index fa36cd3de..f88dd24f1 100644 --- a/temporalio/worker/_activity.py +++ b/temporalio/worker/_activity.py @@ -202,11 +202,13 @@ async def wait_all_completed(self) -> None: "Cancelling activity %s still running after worker shutdown", task_token, ) - activity.cancellation_details.details = ( - temporalio.activity.ActivityCancellationDetails( - worker_shutdown=True + # Cancellation details are set once, so keep any already received + if not activity.cancellation_details.details: + activity.cancellation_details.details = ( + temporalio.activity.ActivityCancellationDetails( + worker_shutdown=True + ) ) - ) activity.cancel(cancelled_by_request=True) running_tasks = [v.task for v in self._running_activities.values() if v.task] if running_tasks: diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py index f677046cb..91657a3f1 100644 --- a/tests/worker/test_workflow.py +++ b/tests/worker/test_workflow.py @@ -43,6 +43,7 @@ import temporalio.converter._extstore import temporalio.worker import temporalio.worker._command_aware_visitor +import temporalio.worker._workflow import temporalio.worker._workflow_instance import temporalio.workflow from temporalio import activity, workflow @@ -55,7 +56,10 @@ PollWorkflowExecutionUpdateResponse, ResetStickyTaskQueueRequest, ) -from temporalio.bridge.proto.workflow_activation import WorkflowActivation +from temporalio.bridge.proto.workflow_activation import ( + RemoveFromCache, + WorkflowActivation, +) from temporalio.bridge.proto.workflow_completion import WorkflowActivationCompletion from temporalio.client import ( AsyncActivityCancelledError, @@ -1151,6 +1155,7 @@ async def test_worker_shutdown_cancels_local_activity_untracked_by_core( client: Client, ): started = asyncio.Event() + workflow_poll_shut_down = asyncio.Event() details: list[temporalio.activity.ActivityCancellationDetails | None] = [] @activity.defn(name="wait_forever_local_activity") @@ -1161,21 +1166,35 @@ async def wait_forever_local_activity() -> None: except asyncio.CancelledError: details.append(activity.cancellation_details()) - # Delay the next poll so run invalidation drops the queued cancel before dispatch - orig_poll = temporalio.bridge.worker.Worker.poll_activity_task - delay_next_poll = False - - async def slow_poll(self: temporalio.bridge.worker.Worker): - nonlocal delay_next_poll - if delay_next_poll: - delay_next_poll = False - await asyncio.sleep(3) - task = await orig_poll(self) + # Hold the next poll until Core has evicted the run, so its queued cancel is never delivered + bridge_worker = temporalio.bridge.worker.Worker + orig_poll_activity = bridge_worker.poll_activity_task + orig_poll_workflow = bridge_worker.poll_workflow_activation + hold_next_poll = False + + async def poll_activity_task(self: temporalio.bridge.worker.Worker): + nonlocal hold_next_poll + if hold_next_poll: + hold_next_poll = False + await workflow_poll_shut_down.wait() + task = await orig_poll_activity(self) if task.HasField("start") and task.start.is_local: - delay_next_poll = True + hold_next_poll = True return task - with patch.object(temporalio.bridge.worker.Worker, "poll_activity_task", slow_poll): + async def poll_workflow_activation(self: temporalio.bridge.worker.Worker): + try: + return await orig_poll_workflow(self) + except temporalio.bridge.worker.PollShutdownError: # type: ignore[reportPrivateLocalImportUsage] + workflow_poll_shut_down.set() + raise + + with ( + patch.object(bridge_worker, "poll_activity_task", poll_activity_task), + patch.object( + bridge_worker, "poll_workflow_activation", poll_workflow_activation + ), + ): worker = new_worker( client, OrphanedLocalActivityWorkflow, @@ -1186,16 +1205,72 @@ async def slow_poll(self: temporalio.bridge.worker.Worker): OrphanedLocalActivityWorkflow.run, id=f"workflow-{uuid.uuid4()}", task_queue=worker.task_queue, - task_timeout=timedelta(seconds=1), + task_timeout=timedelta(seconds=3), ) await started.wait() # Terminating fails the workflow task heartbeat, which evicts the run await handle.terminate() - await asyncio.sleep(4) await asyncio.wait_for(worker.shutdown(), 20) await run_task - assert len(details) == 1 and details[0] - assert details[0].worker_shutdown or details[0].cancel_requested + assert details == [ + temporalio.activity.ActivityCancellationDetails(worker_shutdown=True) + ] + + +async def test_worker_shutdown_keeps_details_of_local_activity_ignoring_cancel( + client: Client, +): + started = asyncio.Event() + cancel_seen = asyncio.Event() + details: list[temporalio.activity.ActivityCancellationDetails | None] = [] + + @activity.defn(name="wait_forever_local_activity") + async def wait_forever_local_activity() -> None: + started.set() + while True: + try: + await asyncio.sleep(1000) + except asyncio.CancelledError: + details.append(activity.cancellation_details()) + cancel_seen.set() + if activity.is_worker_shutdown(): + raise + + # Finish evicting only after the cancel reached the activity, so Core drops it afterwards + workflow_worker = temporalio.worker._workflow._WorkflowWorker + orig_evict = workflow_worker._handle_cache_eviction + + async def handle_cache_eviction( + self: temporalio.worker._workflow._WorkflowWorker, + act: WorkflowActivation, + job: RemoveFromCache, + ): + await cancel_seen.wait() + await orig_evict(self, act, job) + + with patch.object(workflow_worker, "_handle_cache_eviction", handle_cache_eviction): + worker = new_worker( + client, + OrphanedLocalActivityWorkflow, + activities=[wait_forever_local_activity], + ) + run_task = asyncio.create_task(worker.run()) + handle = await client.start_workflow( + OrphanedLocalActivityWorkflow.run, + id=f"workflow-{uuid.uuid4()}", + task_queue=worker.task_queue, + task_timeout=timedelta(seconds=3), + ) + await started.wait() + # Terminating fails the workflow task heartbeat, which evicts the run + await handle.terminate() + await asyncio.wait_for(cancel_seen.wait(), 20) + await asyncio.wait_for(worker.shutdown(), 20) + await run_task + assert ( + details + == [temporalio.activity.ActivityCancellationDetails(cancel_requested=True)] * 2 + ) @workflow.defn