/ws/bus subscriber thread leakEvery client that connects to /ws/bus/{session_id} spawns a thread that never exits after disconnect.
_subscribe() in gateway.py:148 runs in a thread-pool thread via run_in_executor with an unbounded while True loop. The WebSocketDisconnect handler at line 169 exits the coroutine but has no way to signal the thread. The thread keeps running — and holding a ZMQ subscriber socket — until the process restarts.
client connects
→ coroutine starts
→ run_in_executor(_subscribe) ← thread T1 starts, loops forever
→ coroutine pumps queue
client disconnects
→ WebSocketDisconnect caught
→ coroutine exits
→ T1 still running ✗ ZMQ socket still open ✗
client connects
→ stop = threading.Event()
→ run_in_executor(_subscribe) ← thread T1 starts, checks stop
→ coroutine pumps queue
client disconnects
→ WebSocketDisconnect caught
→ finally: stop.set()
→ T1 exits within ≤200 ms (next receive_with_timeout wake-up) ✓
threading.Event stop flag. Thread checks it on every loop iteration. Coroutine sets it in a finally block, guaranteeing cleanup on both normal close and error.
Single file: src/local/api/gateway.py, function ws_bus.
loop = asyncio.get_event_loop()
queue: asyncio.Queue = asyncio.Queue()
def _subscribe() -> None:
sub = ZmqSubscriber(PROXY_BACKEND_ADDR, subscriptions=[""], bind=False)
try:
while True:
msg = sub.receive_with_timeout(200)
if msg is not None:
asyncio.run_coroutine_threadsafe(queue.put(msg), loop)
finally:
sub.close()
loop.run_in_executor(None, _subscribe)
try:
while True:
env: MessageEnvelope = await queue.get()
await websocket.send_json({...})
except WebSocketDisconnect:
pass
import threading
loop = asyncio.get_event_loop()
queue: asyncio.Queue = asyncio.Queue()
stop = threading.Event()
def _subscribe() -> None:
sub = ZmqSubscriber(PROXY_BACKEND_ADDR, subscriptions=[""], bind=False)
try:
while not stop.is_set():
msg = sub.receive_with_timeout(200)
if msg is not None:
asyncio.run_coroutine_threadsafe(queue.put(msg), loop)
finally:
sub.close()
loop.run_in_executor(None, _subscribe)
try:
while True:
env: MessageEnvelope = await queue.get()
await websocket.send_json({...})
except WebSocketDisconnect:
pass
finally:
stop.set()
threading is already in the stdlib — no new dependency. receive_with_timeout(200) already exists, so the thread wakes every 200 ms regardless. The finally block fires on both clean close and any exception, so no code path leaks.
| File | Change |
|---|---|
src/local/api/gateway.py | Add stop = threading.Event(); change while True → while not stop.is_set(); add finally: stop.set() to coroutine |
/ws/bus/{id}, then disconnect — confirm no orphaned threads in threading.enumerate()/ws/chat operation unaffected