From 0736f76ca3285cad78be734a0f0491ef0fe15c59 Mon Sep 17 00:00:00 2001 From: Jens Ahrensfeld Date: Wed, 1 Jul 2026 08:46:25 +0200 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_01TvgC7oy9MxaA4ZQxXqkNdS --- ws/server/ws_server.py | 10 +--------- ws/server/ws_server_multi_user.py | 21 +++++++++++++++------ 2 files changed, 16 insertions(+), 15 deletions(-) diff --git a/ws/server/ws_server.py b/ws/server/ws_server.py index 3af5b91..59b584a 100644 --- a/ws/server/ws_server.py +++ b/ws/server/ws_server.py @@ -22,11 +22,7 @@ class WsServer: async def handler(self, websocket, path): await self.register(websocket) try: - consumer_task = asyncio.ensure_future(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() + await self.handler_recv(websocket, path) finally: await self.unregister(websocket) @@ -34,7 +30,3 @@ class WsServer: async def handler_recv(self, websocket, path): pass - @abc.abstractmethod - async def handler_send(self, websocket, path): - pass - diff --git a/ws/server/ws_server_multi_user.py b/ws/server/ws_server_multi_user.py index 272772d..64b8627 100644 --- a/ws/server/ws_server_multi_user.py +++ b/ws/server/ws_server_multi_user.py @@ -11,6 +11,21 @@ class WsServerMultiUser(WsServer): self.listener = listener self.USERS = UserSet() 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): if self.USERS: # asyncio.wait doesn't accept an empty list @@ -43,9 +58,3 @@ class WsServerMultiUser(WsServer): await self.listener.on_recv(data) except websockets.exceptions.ConnectionClosed: 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)