From 32f751e1150a5ca3c8f5c6c6703d14c5bb7927c0 Mon Sep 17 00:00:00 2001 From: stroblme Date: Sun, 6 Sep 2026 17:36:49 +0200 Subject: [PATCH] End the live socket when its client goes, instead of spinning on it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `receive_text` raises `WebSocketDisconnect` once the peer has gone and `RuntimeError` on every call after that. The loop swallowed both — the exception was checked only to decide whether the frame held a media subscription — and immediately made another read, which raised again. A closed tab therefore left a coroutine reading a dead socket at full tilt: around 15,000 turns a second, one orphaned `Queue.get()` per turn parked on the bus queue's waiter list. The event loop stalled behind it, `/health` answered 503 on its loop-lag check, and Docker's probe plus Traefik took the API out of the pool mid-request — which is what `make verify` was dying on, with `TypeError: Failed to fetch` from the browser. The spin ended only when an event finally reached the send path and the socket raised there, so it lasted from milliseconds to minutes depending on what the bus happened to be carrying. Four abandoned sockets logged 2,518 `Task was destroyed but it is pending!` lines and up to 2.6s of loop lag; twenty now log none and cost 1.0ms. A failed read is the disconnect, so it ends the stream. The getter created beside it is cancelled rather than abandoned: nothing reads it after the turn that made it, and left behind it would also take an event the live reader wanted. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01CL9zvnnvcp1mvA8o7impxk --- backend/fluksio/api/routes/flows.py | 19 +++++++--- backend/tests/api/routes/test_flows.py | 50 ++++++++++++++++++++++++++ 2 files changed, 65 insertions(+), 4 deletions(-) diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py index f538b25..9c4a9ae 100644 --- a/backend/fluksio/api/routes/flows.py +++ b/backend/fluksio/api/routes/flows.py @@ -1054,14 +1054,25 @@ async def flow_events(websocket: WebSocket, token: str = "") -> None: {sender, receiver}, return_when=asyncio.FIRST_COMPLETED ) if receiver in done: + if sender not in done: + # Nobody will ever read this one: the next turn makes + # its own. Left behind it stays parked on the queue's + # waiter list, takes an event the live reader wanted, + # and is reported as a task destroyed while pending. + sender.cancel() # A frame from the client. Only a disconnect ends the # stream — a keepalive, or anything else it decides to # say, used to be read as the client going away and cost # it every live update from then on. - if receiver.exception() is None: - asked = wanted_names(receiver.result(), only) - if asked is not None: - wanted = asked + failure = receiver.exception() + if failure is not None: + # The client is gone, and `receive_text` raises for + # good from here on. Reading it again would spin this + # loop at full tilt against a dead socket. + raise failure + asked = wanted_names(receiver.result(), only) + if asked is not None: + wanted = asked receiver = asyncio.create_task(websocket.receive_text()) if sender not in done: continue diff --git a/backend/tests/api/routes/test_flows.py b/backend/tests/api/routes/test_flows.py index 548c273..c548aa5 100644 --- a/backend/tests/api/routes/test_flows.py +++ b/backend/tests/api/routes/test_flows.py @@ -1,10 +1,13 @@ +import asyncio from datetime import UTC, datetime import pytest +from fastapi import WebSocketDisconnect from fastapi.testclient import TestClient from sqlalchemy import func from sqlmodel import Session, select +from fluksio.api.routes import flows as flows_route from fluksio.core.config import settings from fluksio.flow.events import event_bus from fluksio.models import Run, RunArtifact, RunMetric, RunNode @@ -493,3 +496,50 @@ def test_renaming_a_flow_leaves_nothing_under_the_old_name( assert [key for key in state.keys() if "movable." in key] == [] client.delete(f"{PREFIX}/moved", headers=superuser_token_headers) + + +@pytest.mark.anyio +async def test_the_live_socket_ends_when_its_client_goes( + client: TestClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """A departing client ends the stream instead of restarting the read. + + `receive_text` raises for good once the peer has gone, so reading it again + spun this loop at full tilt: the event loop stalled, `/health` answered + 503 and one queue getter was orphaned per turn. + """ + monkeypatch.setattr(flows_route, "user_from_token", lambda session, token: object()) + + class Peer: + """One frame, then gone — the way a closed tab leaves.""" + + def __init__(self) -> None: + self.app = client.app + self.frames = ['{"type": "media", "names": []}'] + self.sent: list[str] = [] + + async def accept(self) -> None: + return None + + async def send_text(self, text: str) -> None: + self.sent.append(text) + + async def receive_text(self) -> str: + if self.frames: + return self.frames.pop(0) + raise WebSocketDisconnect(1006) + + peer = Peer() + subscribed = len(event_bus._subscribers) + + await asyncio.wait_for(flows_route.flow_events(peer, token="not-a-jwt"), timeout=5) + + # The snapshot went out, and the frame in between did not end the stream: + # reaching the disconnect at all means the second read happened. + assert peer.sent + assert peer.frames == [] + assert len(event_bus._subscribers) == subscribed + # A turn of the loop for the cancellations the handler asked for. + await asyncio.sleep(0.05) + parked = [t for t in asyncio.all_tasks() if "Queue.get" in repr(t.get_coro())] + assert parked == []