diff --git a/backend/fluksio/flow/remote.py b/backend/fluksio/flow/remote.py index 30678ca..fcf638d 100644 --- a/backend/fluksio/flow/remote.py +++ b/backend/fluksio/flow/remote.py @@ -249,6 +249,11 @@ class RemoteWorker: def cancel(self, call_id: str) -> None: """Ask the worker to kill what it is running for this call.""" + # A closed loop refuses the coroutine and leaves it never awaited, so it + # is not made at all. + if self._loop.is_closed(): + logger.warning("Could not cancel '%s' on '%s'", call_id, self.name) + return try: asyncio.run_coroutine_threadsafe( self._send({"op": "cancel", "call_id": call_id}), self._loop diff --git a/backend/tests/flow/test_remote.py b/backend/tests/flow/test_remote.py index 16b6177..e872e5f 100644 --- a/backend/tests/flow/test_remote.py +++ b/backend/tests/flow/test_remote.py @@ -8,8 +8,10 @@ what happened rather than a wait that never ends. """ import asyncio +import gc import threading import time +import warnings from collections.abc import Iterator import pytest @@ -241,6 +243,21 @@ def test_cancelling_a_run_reaches_only_that_run(loop): assert [frame["call_id"] for frame in cancels] == ["run-a:flow.node"] +def test_a_cancel_after_the_loop_closed_leaves_no_coroutine_behind(): + closed = asyncio.new_event_loop() + closed.close() + worker = RemoteWorker( + name="gpu1", labels=[], send=FakeSocket().send_json, loop=closed + ) + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + worker.cancel("run-a:flow.node") + gc.collect() + + assert not [w for w in caught if "never awaited" in str(w.message)] + + # ----------------------------------------------------------------------------- # Liveness, and what a heartbeat is evidence of #