RemoteWorker.cancel makes no coroutine once the loop is closed

run_coroutine_threadsafe refused it and it was never awaited, a RuntimeWarning
under pytest 9. The cancel still logs that it could not reach the worker.

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

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
 #