mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
fix: per-workstream SSE fan-out — multiple consumers no longer steal … (#38)
* fix: per-workstream SSE fan-out — multiple consumers no longer steal each other's tokens
After 4d665a5 removed the single-consumer SSE lock, the shared
_event_queue let concurrent consumers (browser, bridge, console proxy)
race on Queue.get(), each receiving ~1/N of content tokens and producing
garbled streaming text.
Replace the single queue with per-client fan-out: each SSE connection
registers its own bounded queue (maxsize=500) on WebUI._listeners, and
_enqueue() copies every event to all registered queues. On eviction or
close, a ws_closed sentinel is injected so SSE generators exit promptly.
* fix: address CI failures and Copilot review feedback
- Handle ws_closed sentinel in events_sse generator (break on close)
- Guarantee sentinel delivery by evicting one item when queue is full
- Clear listeners list after injecting sentinels on cleanup
- Fix test_slow_consumer to fill only slow queue directly
- Fix ruff SIM117 (nested with), unused import, mypy unused-ignore
This commit is contained in:
@@ -460,13 +460,13 @@ The server sends an SSE comment every 5 seconds when no events are pending:
|
||||
This prevents proxies and browsers from closing the connection due to
|
||||
inactivity.
|
||||
|
||||
#### Generation mechanism
|
||||
#### Multi-consumer fan-out
|
||||
|
||||
Each new SSE connection to a workstream increments an internal
|
||||
`_sse_generation` counter. The previous SSE handler detects the generation
|
||||
mismatch and exits its event loop, ensuring only one active SSE connection per
|
||||
workstream at a time. The event queue is drained of stale events before the new
|
||||
connection begins streaming.
|
||||
Each SSE connection to a workstream receives its own delivery queue. Events
|
||||
produced by the worker thread are fanned out to all registered listener queues,
|
||||
so multiple consumers (browser, bridge, console proxy, SDK) can connect
|
||||
simultaneously and each receives every event. On reconnect the client receives
|
||||
a full history replay, so no catch-up mechanism is needed.
|
||||
|
||||
---
|
||||
|
||||
|
||||
Reference in New Issue
Block a user