Service node ends its call when the loop cancels every task; test order and timing leaks

_serve_once treated a cancelled call future as finished, so an event loop
cancelling all tasks at once (shutdown) never sent the worker a cancel and the
call's thread kept asyncio.run waiting for SILENCE_S. The cameras audio test
now removes its voice flow and device, the unpaired-frame test waits for its
first frame before the unpair, and test_connector no longer inherits the
controller an app test bound to demand.

Claude-Session: https://claude.ai/code/session_01SSi5AvZ7pFmySK8zy3NPMh

backend/fluksio/flow/nodes/service.py

@@ -190,7 +190,10 @@ class ServiceNode(Node):
             # The worker went away or went silent, or the url was refused.
             return healthy, f"{type(exc).__name__}: {exc}"
         finally:
-            if not call.done():
+            # A cancelled future is done while its thread still waits for the
+            # worker, which is how an event loop cancelling every task at once
+            # (a shutdown) reaches here: only an answer means it has ended.
+            if not call.done() or call.cancelled():
                 # The flow stopped, or this side failed. Ending the call kills
                 # the server's process group on the worker, and its answer says
                 # that is done, so the next start finds the port free.

backend/tests/api/routes/test_cameras.py

@@ -102,6 +102,31 @@ def camera(client: TestClient, superuser_token_headers: dict[str, str]):
     client.delete(f"{PREFIX}/cam1", headers=superuser_token_headers)
 
 
+@pytest.fixture
+def microphone(client: TestClient, superuser_token_headers: dict[str, str]):
+    """A paired device with a microphone, wired in flow ``voice``. Yields its header.
+
+    Both go again afterwards: a published flow outlives the module in the
+    flow store, and the next module's app would load it without its device.
+    """
+    _panels(client, superuser_token_headers, {"panels": [{"id": "mic1"}]})
+    device = _pair(client, superuser_token_headers, "mic1")
+    port = {"port": "mic", "direction": "in", "dtype": "audio"}
+    reported = client.put(
+        f"{PREFIX}/mic1/report", headers=device, json={**REPORT, "ports": [port]}
+    )
+    assert reported.status_code == 200, reported.text
+    node = _device_node(
+        "mic1",
+        node="panel",
+        provides=[{"name": "mic", "port": "mic", "dtype": "audio"}],
+    )
+    _publish_flow(client, superuser_token_headers, "voice", [node])
+    yield device
+    client.delete(f"{FLOWS}/voice", headers=superuser_token_headers)
+    client.delete(f"{PREFIX}/mic1", headers=superuser_token_headers)
+
+
 def _ticket(client: TestClient, headers: dict[str, str]) -> str:
     answer = client.post(f"{settings.API_V1_STR}/flows/ws-ticket", headers=headers)
     return f"{SOCKET}?ticket={answer.json()['ticket']}"
@@ -269,24 +294,11 @@ def test_an_unpaired_device_is_closed_and_its_frames_stop(
 
 
 def test_a_device_streams_audio_chunks_on_its_socket(
-    client: TestClient, superuser_token_headers: dict[str, str]
+    client: TestClient, microphone
 ) -> None:
     """A microphone's chunks take the camera's path, in their own family."""
-    _panels(client, superuser_token_headers, {"panels": [{"id": "mic1"}]})
-    device = _pair(client, superuser_token_headers, "mic1")
-    port = {"port": "mic", "direction": "in", "dtype": "audio"}
-    reported = client.put(
-        f"{PREFIX}/mic1/report", headers=device, json={**REPORT, "ports": [port]}
-    )
-    assert reported.status_code == 200, reported.text
-    node = _device_node(
-        "mic1",
-        node="panel",
-        provides=[{"name": "mic", "port": "mic", "dtype": "audio"}],
-    )
-    _publish_flow(client, superuser_token_headers, "voice", [node])
     controller = client.app.state.flow_controller
-    with client.websocket_connect(SOCKET, headers=device) as ws:
+    with client.websocket_connect(SOCKET, headers=microphone) as ws:
         ws.receive_json()
         ws.send_bytes(frame("voice.mic", b"\x00\x01"))  # an image on an audio port
         ws.send_bytes(frame("voice.mic", b"\x00\x02", "audio/L16;rate=16000"))
@@ -329,9 +341,13 @@ def test_an_unpaired_device_is_closed_at_its_next_frame(
     """Without the event an unpair sends, which a full queue may have lost."""
     from starlette.websockets import WebSocketDisconnect
 
+    controller = client.app.state.flow_controller
     with client.websocket_connect(SOCKET, headers=camera) as device:
         device.receive_json()
         device.send_bytes(frame("cams.camera", b"\xff\xd8 one"))
+        # The credential is asked as each frame is read, so a bump before the
+        # server has read the first one would refuse that one too.
+        wait_for(lambda: (controller.values().get("cams.camera") or {}).get("value"))
         client.app.state.device_store.bump("cam1")
         device.send_bytes(frame("cams.camera", b"\xff\xd8 two"))
         with pytest.raises(WebSocketDisconnect) as closed:
@@ -339,7 +355,7 @@ def test_an_unpaired_device_is_closed_at_its_next_frame(
                 device.receive_json()
         assert closed.value.code == 1008
     store = client.app.state.artifact_store
-    current = client.app.state.flow_controller.values()["cams.camera"]["value"]
+    current = controller.values()["cams.camera"]["value"]
     assert store.volatile.get(current["digest"]) == b"\xff\xd8 one"
 
 

backend/tests/flow/test_connector.py

@@ -313,7 +313,9 @@ def test_a_port_is_wanted_while_watched_or_read(monkeypatch):
     from fluksio.flow.demand import demand
 
     node = a_sensor([])
-    # A worker process binds no controller and cannot see who watches.
+    # A worker process binds no controller and cannot see who watches. The
+    # one an app test bound stays on the process-wide `demand` after its stop.
+    monkeypatch.setattr(demand, "_controller", None)
     assert node.wanted("reading")
 
     reads: set[str] = set()

backend/tests/flow/test_service_node.py

@@ -1,8 +1,10 @@
 """The service node: a server on a worker, restarted when it ends, its url following."""
 
 import asyncio
+import time
 from collections.abc import Callable
 
+from fluksio.flow import remote
 from fluksio.flow.messages import DType, MessageSpec
 from fluksio.flow.nodes import ServiceNode, service
 from fluksio.flow.pipeline import Pipeline
@@ -99,6 +101,49 @@ def test_a_service_comes_back_and_its_url_follows(monkeypatch):
     asyncio.run(scenario())
 
 
+def test_a_service_cancelled_with_its_loop_ends_its_call(monkeypatch):
+    """A loop that cancels every task at once must not strand the call's thread.
+
+    The call's own future is cancelled with the rest, so it reads done while the
+    thread still waits for the worker, which then keeps the loop's shutdown
+    waiting until the silence bound.
+    """
+    monkeypatch.setattr(remote, "SILENCE_S", 3.0)
+
+    async def healthy(client, url):
+        return None
+
+    monkeypatch.setattr(ServiceNode, "_probe", staticmethod(healthy))
+    agent = Agent()
+
+    async def scenario() -> None:
+        agent.worker = RemoteWorker(
+            name="gpu-worker",
+            labels=["gpu"],
+            send=agent.send_json,
+            loop=asyncio.get_running_loop(),
+            host="192.0.2.5",
+        )
+        hub = RemoteWorkerHub()
+        hub.attach(agent.worker)
+        node = ServiceNode(
+            provides=[MessageSpec(name="url", dtype=DType.STR)],
+            params={"command": "serve", "worker": "gpu"},
+        )
+        node.assign_flow("voice", "parrot")
+        pipeline = Pipeline(nodes=[node])
+        node.hub = hub
+        await node.start()
+        await until(lambda: bool(agent.serves()))
+        await until(lambda: pipeline.state.get("voice.url") == "http://192.0.2.5:8100")
+        # No stop: asyncio.run cancels the node's task and the call together.
+
+    started = time.monotonic()
+    asyncio.run(scenario())
+    assert time.monotonic() - started < 2
+    assert [frame["op"] for frame in agent.sent] == ["serve", "cancel"]
+
+
 def test_a_service_that_never_gets_healthy_reads_down(monkeypatch):
     monkeypatch.setattr(service, "BACKOFF", (0.01,))