Finish fenced remote claims without delaying worker shutdown on observer cancellation

Author: durable-workflow-ops

src/durable_workflow/worker.py

@@ -2318,22 +2318,17 @@ async def observe_ownership() -> None:
             return result
         finally:
             abandoned.set()
-            shutdown.cancel()
-            observation.cancel()
-            with contextlib.suppress(asyncio.CancelledError, Exception):
-                await shutdown
-            with contextlib.suppress(asyncio.CancelledError, Exception):
-                await observation
-            if not invocation.done():
-                invocation.cancel()
 
-                def discard_late_result(future: asyncio.Task[Any]) -> None:
-                    if not future.cancelled():
-                        future.exception()
+            def discard_late_result(future: asyncio.Task[Any]) -> None:
+                if not future.cancelled():
+                    future.exception()
 
-                # A running thread or cancellation-resistant callable can outlive
-                # the await. Its heartbeat and eventual publication stay fenced.
-                invocation.add_done_callback(discard_late_result)
+            # The abandoned claim must finish without waiting for callback or
+            # observer cancellation. All late progress/publication stays fenced.
+            for background in (invocation, observation, shutdown):
+                if not background.done():
+                    background.cancel()
+                background.add_done_callback(discard_late_result)
 
     async def _run_activity_task(self, task: dict[str, Any]) -> str:
         self._track_worker_session_from_task(task)

tests/test_cooperative_remote_worker.py

@@ -121,6 +121,32 @@ def callback() -> object:
         release.set()
 
 
+async def test_shutdown_expiry_of_tracked_remote_attempt_fences_a_cancellation_resistant_result(owner) -> None:
+    worker, client = owner
+    worker._shutdown_timeout = 0.01
+    entered, late_fenced = asyncio.Event(), asyncio.Event()
+
+    async def callback() -> object:
+        entered.set()
+        try:
+            await asyncio.Event().wait()
+        except asyncio.CancelledError:
+            with pytest.raises(_RemoteActivityExecutionAborted):
+                await activity.context().heartbeat()
+            late_fenced.set()
+            return object()
+
+    worker.activities["remote"] = callback
+    execution = worker._track(worker._run_activity_task(task()))
+    await asyncio.wait_for(entered.wait(), timeout=2)
+    await asyncio.wait_for(worker.stop(), timeout=2)
+    await asyncio.wait_for(late_fenced.wait(), timeout=2)
+    assert execution.cancelled() or execution.result() == "claim_aborted"
+    client.heartbeat_activity_task.assert_not_awaited()
+    client.complete_activity_task.assert_not_awaited()
+    client.fail_activity_task.assert_not_awaited()
+
+
 @pytest.mark.parametrize("synchronous", [False, True])
 async def test_authored_heartbeat_stays_on_owner_loop_and_preserves_typed_result(owner, synchronous: bool) -> None:
     worker, client = owner