From f52d9c3077d9076d0229e380b49667477d51fdb2 Mon Sep 17 00:00:00 2001 From: stroblme Date: Tue, 27 Oct 2026 21:23:40 +0100 Subject: [PATCH] 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. --- backend/fluksio/flow/nodes/service.py | 5 ++- backend/tests/api/routes/test_cameras.py | 48 ++++++++++++++++-------- backend/tests/flow/test_connector.py | 4 +- backend/tests/flow/test_service_node.py | 45 ++++++++++++++++++++++ 4 files changed, 84 insertions(+), 18 deletions(-) diff --git a/backend/fluksio/flow/nodes/service.py b/backend/fluksio/flow/nodes/service.py index 9b2263c..ba4eff5 100644 --- a/backend/fluksio/flow/nodes/service.py +++ b/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. diff --git a/backend/tests/api/routes/test_cameras.py b/backend/tests/api/routes/test_cameras.py index e01a41d..ed30583 100644 --- a/backend/tests/api/routes/test_cameras.py +++ b/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" diff --git a/backend/tests/flow/test_connector.py b/backend/tests/flow/test_connector.py index c843eec..e13f165 100644 --- a/backend/tests/flow/test_connector.py +++ b/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() diff --git a/backend/tests/flow/test_service_node.py b/backend/tests/flow/test_service_node.py index e6685e6..fdea4d3 100644 --- a/backend/tests/flow/test_service_node.py +++ b/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,))