fix: drain dispatcher queue via single pump task, not per-connection

Each connection used to run its own handler_send loop consuming from
the shared dispatcher queue. With zero clients connected (e.g. an
iPad that silently drops on sleep), nothing drained the queue, so
state updates piled up unbounded and got replayed as a burst on
reconnect. Now one pump task drains the queue continuously regardless
of connection count, and reconnecting clients resync from
global_state instead of a backlog.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TvgC7oy9MxaA4ZQxXqkNdS
This commit is contained in:
2026-07-01 08:46:25 +02:00
co-authored by Claude Sonnet 5
parent f7d3001ca5
commit 0736f76ca3
2 changed files with 16 additions and 15 deletions
+1 -9
View File
@@ -22,11 +22,7 @@ class WsServer:
async def handler(self, websocket, path): async def handler(self, websocket, path):
await self.register(websocket) await self.register(websocket)
try: try:
consumer_task = asyncio.ensure_future(self.handler_recv(websocket, path)) await self.handler_recv(websocket, path)
producer_task = asyncio.ensure_future(self.handler_send(websocket, path))
done, pending = await asyncio.wait([consumer_task, producer_task], return_when=asyncio.FIRST_COMPLETED,)
for task in pending:
task.cancel()
finally: finally:
await self.unregister(websocket) await self.unregister(websocket)
@@ -34,7 +30,3 @@ class WsServer:
async def handler_recv(self, websocket, path): async def handler_recv(self, websocket, path):
pass pass
@abc.abstractmethod
async def handler_send(self, websocket, path):
pass
+15 -6
View File
@@ -11,6 +11,21 @@ class WsServerMultiUser(WsServer):
self.listener = listener self.listener = listener
self.USERS = UserSet() self.USERS = UserSet()
self.global_state = {} self.global_state = {}
self.pump_task = None
async def listen(self, host, port):
# A single task drains the dispatcher's queue regardless of how many
# clients are connected, so state updates never pile up in the queue
# while nobody is listening. A reconnecting client resyncs from
# global_state instead of replaying a backlog (see handler_recv).
self.pump_task = asyncio.ensure_future(self.state_pump())
return await super().listen(host, port)
async def state_pump(self):
while True:
data = await self.listener.on_send()
self.global_state = update(self.global_state, data)
await self.notify_state(data)
async def notify_state(self, data): async def notify_state(self, data):
if self.USERS: # asyncio.wait doesn't accept an empty list if self.USERS: # asyncio.wait doesn't accept an empty list
@@ -43,9 +58,3 @@ class WsServerMultiUser(WsServer):
await self.listener.on_recv(data) await self.listener.on_recv(data)
except websockets.exceptions.ConnectionClosed: except websockets.exceptions.ConnectionClosed:
pass pass
async def handler_send(self, websocket, path):
while True:
data = await self.listener.on_send()
await self.notify_state(data)
self.global_state = update(self.global_state, data)