From 4d665a5f62757c0bb1d1d256862cb72331486143 Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Mon, 9 Mar 2026 13:40:29 -0700 Subject: [PATCH] =?UTF-8?q?fix:=20SSE=20reconnect=20loop=20=E2=80=94=20rem?= =?UTF-8?q?ove=20=5Fsse=5Fgeneration=20single-consumer=20lock?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The _sse_generation mechanism assumed one SSE consumer per workstream, but the bridge also maintains an SSE connection to each workstream. When a new client connected (browser, proxy, or test), it incremented the generation counter, killing the bridge's connection. The bridge reconnected, killing the new client's connection — creating a mutual-kill cascade that closed every SSE connection after one ping cycle (5s). Fix: remove _sse_generation entirely. sse-starlette handles disconnect detection via its own ASGI task. Also remove the redundant request.is_disconnected() check which raced with sse-starlette's disconnect listener in Starlette 0.52. Root cause confirmed via raw socket test: the server was sending a zero-length chunked terminator (0\r\n\r\n) at exactly 5s, cleanly ending the HTTP response body. --- turnstone/server.py | 14 ++++---------- 1 file changed, 4 insertions(+), 10 deletions(-) diff --git a/turnstone/server.py b/turnstone/server.py index 6a2cc4f8..a739d95f 100644 --- a/turnstone/server.py +++ b/turnstone/server.py @@ -473,11 +473,9 @@ async def events_sse(request: Request) -> Response: if not ws or not ui: return JSONResponse({"error": "Unknown workstream"}, status_code=404) - ui._sse_generation += 1 - my_gen = ui._sse_generation - # Drain stale events. A race with the worker thread is acceptable: - # worst case we discard one fresh event, and the client catches up - # via the history replay above. + # Drain stale events so this client starts fresh. A race with the + # worker thread is acceptable: worst case we discard one fresh event, + # and the client catches up via the history replay below. while not ui._event_queue.empty(): try: ui._event_queue.get_nowait() @@ -509,7 +507,7 @@ async def events_sse(request: Request) -> Response: _metrics.record_sse_connect() try: loop = asyncio.get_running_loop() - while my_gen == ui._sse_generation: + while True: try: event = await loop.run_in_executor( None, functools.partial(ui._event_queue.get, timeout=5) @@ -517,8 +515,6 @@ async def events_sse(request: Request) -> Response: yield {"data": json.dumps(event)} except queue.Empty: pass - if await request.is_disconnected(): - break finally: _metrics.record_sse_disconnect() @@ -545,8 +541,6 @@ async def global_events_sse(request: Request) -> Response: yield {"data": json.dumps(event)} except queue.Empty: pass - if await request.is_disconnected(): - break finally: _metrics.record_sse_disconnect() with listeners_lock: