From 31702240831aa1f644bd47ffe23c3be074a0574f Mon Sep 17 00:00:00 2001 From: DABH Date: Thu, 10 Sep 2026 00:46:29 -0500 Subject: [PATCH 1/3] 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 5a8995696..589fec7ce 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -164,6 +164,9 @@ to include examples, links to docs, or any other relevant information. repeated identical call (the conservative direction). - `contrib.deepagents`: summarization middleware configured with a model name string now routes its LLM calls through Activities instead of running them in the Workflow. +- 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 f72df9cf4..b5951a00e 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 d751fdb8b..459bac76a 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 @@ -1136,6 +1138,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 436eb309634972fb95f5f717b7465624d02d13d4 Mon Sep 17 00:00:00 2001 From: DABH Date: Thu, 10 Sep 2026 03:15:22 -0500 Subject: [PATCH 2/3] 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 589fec7ce..1bcb8ae39 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -166,7 +166,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 b5951a00e..0c971db4b 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 459bac76a..d82d0978e 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, @@ -1152,6 +1156,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") @@ -1162,21 +1167,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, @@ -1187,16 +1206,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 From 2fffae55bbef17003ff387aea26f96baa65ac93c Mon Sep 17 00:00:00 2001 From: DABH Date: Wed, 30 Sep 2026 14:03:03 -0500 Subject: [PATCH 3/3] Keep shutdown going when the straggler cancel interrupts failure encoding The new cancel in wait_all_completed can land while an orphaned activity is still encoding its failure through a codec that yields, which ends the activity task with CancelledError. gather() re-raised it, Worker._run skipped finalize_shutdown, and shutdown() hung, the very symptom this branch fixes. Collect task exceptions instead, logging anything other than cancellation, so wait_all_completed keeps its no-raise contract. Add a regression test with a slow payload codec, share the poll-hold setup between the orphan tests, and move the changelog entry from the released 1.33.0 section to Unreleased. --- CHANGELOG.md | 8 +-- temporalio/worker/_activity.py | 11 +++- tests/worker/test_workflow.py | 95 ++++++++++++++++++++++++++++------ 3 files changed, 93 insertions(+), 21 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1bcb8ae39..b3aaf0595 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,6 +28,10 @@ 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 if it has none yet. ### Security ## [1.34.0] - 2026-09-30 @@ -164,10 +168,6 @@ to include examples, links to docs, or any other relevant information. repeated identical call (the conservative direction). - `contrib.deepagents`: summarization middleware configured with a model name string now routes its LLM calls through Activities instead of running them in the Workflow. -- 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 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 0c971db4b..cbb504857 100644 --- a/temporalio/worker/_activity.py +++ b/temporalio/worker/_activity.py @@ -212,7 +212,16 @@ async def wait_all_completed(self) -> None: 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) + # A cancel above can land while an activity is still encoding its + # result, ending its task with CancelledError, so never let a task + # exception escape and stall shutdown + for result in await asyncio.gather(*running_tasks, return_exceptions=True): + if isinstance(result, BaseException) and not isinstance( + result, asyncio.CancelledError + ): + logger.warning( + "Activity task raised during worker shutdown", exc_info=result + ) def _handle_cancel_activity_task( self, diff --git a/tests/worker/test_workflow.py b/tests/worker/test_workflow.py index d82d0978e..299636624 100644 --- a/tests/worker/test_workflow.py +++ b/tests/worker/test_workflow.py @@ -16,7 +16,8 @@ import typing import uuid from abc import ABC, abstractmethod -from collections.abc import Awaitable, Mapping, Sequence +from collections.abc import Awaitable, Iterator, Mapping, Sequence +from contextlib import contextmanager from dataclasses import dataclass from datetime import datetime, timedelta, timezone from enum import IntEnum @@ -1152,22 +1153,15 @@ async def run(self) -> None: ) -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] = [] +@contextmanager +def _orphan_local_activity_cancel() -> Iterator[None]: + """Make Core drop a running local activity's cancel on eviction. - @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()) - - # Hold the next poll until Core has evicted the run, so its queued cancel is never delivered + Holds the activity poll that follows a local activity start until workflow + polling has shut down, so the cancel Core queues on eviction is never + delivered and the activity keeps running untracked. + """ + workflow_poll_shut_down = asyncio.Event() bridge_worker = temporalio.bridge.worker.Worker orig_poll_activity = bridge_worker.poll_activity_task orig_poll_workflow = bridge_worker.poll_workflow_activation @@ -1196,6 +1190,24 @@ async def poll_workflow_activation(self: temporalio.bridge.worker.Worker): bridge_worker, "poll_workflow_activation", poll_workflow_activation ), ): + yield + + +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()) + + with _orphan_local_activity_cancel(): worker = new_worker( client, OrphanedLocalActivityWorkflow, @@ -1274,6 +1286,57 @@ async def handle_cache_eviction( ) +class _SlowPayloadCodec(PayloadCodec): + """Yields while encoding, like a codec that calls a remote KMS.""" + + async def encode(self, payloads: Sequence[Payload]) -> list[Payload]: + if payloads: + await asyncio.sleep(5) + return list(payloads) + + async def decode(self, payloads: Sequence[Payload]) -> list[Payload]: + return list(payloads) + + +async def test_worker_shutdown_completes_while_untracked_activity_encodes_failure( + client: Client, +): + # The shutdown cancel lands while the orphaned activity is still encoding its + # failure through a slow codec; shutdown must still complete + started = asyncio.Event() + + @activity.defn(name="wait_forever_local_activity") + async def raise_on_shutdown_local_activity() -> None: + started.set() + await activity.wait_for_worker_shutdown() + # Details send the failure through the codec + raise ApplicationError("failing at shutdown", {"detail": "x"}) + + config = client.config() + config["data_converter"] = dataclasses.replace( + DataConverter.default, payload_codec=_SlowPayloadCodec() + ) + codec_client = Client(**config) + with _orphan_local_activity_cancel(): + worker = new_worker( + codec_client, + OrphanedLocalActivityWorkflow, + activities=[raise_on_shutdown_local_activity], + ) + run_task = asyncio.create_task(worker.run()) + handle = await codec_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(worker.shutdown(), 20) + await run_task + + @workflow.defn class SimpleChildWorkflow: @workflow.run