mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
fix: SSE reconnect loop — remove _sse_generation single-consumer lock
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.
This commit is contained in:
+4
-10
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user