From a959f4d12ef382e3d5a5b8143fd569d705c135bb Mon Sep 17 00:00:00 2001 From: Andre van Zuydam Date: Thu, 15 Oct 2026 11:30:50 +0200 Subject: [PATCH] perf(ws,realtime): offload blocking backplane publish + chat/upload I/O off the loop Extends the event-loop-blocking fix (session pipeline, separate PR) to the WebSocket and realtime subsystems. The Redis backplane LISTENER already runs in a thread; these were still on the loop: - WS backplane PUBLISH: broadcast/broadcast_all/broadcast_to_room (async, on the loop) called _publish -> synchronous redis.publish inline. Every broadcast - a WS send, a WebRTC signalling relay, a chat fan-out - was a blocking socket round-trip ON the loop. Now awaited through asyncio.to_thread. - Realtime chat: _persist_message (blocking ORM write) and _mark_read's read-receipt read+write ran on the loop inside async handlers; offloaded. - Realtime files_upload: the blocking _store_upload (local/S3 storage write) ran on the loop; offloaded. WebRTC signalling itself is a pure relay (broadcast_to_room), so it is covered by the backplane offload with no other change. Proven (tests/test_ws_backplane_threading.py, no mocks - real WebSocketManager, real clock/threads, a real backplane with a measured publish latency): RED before, GREEN after. 4 concurrent broadcasts 1.22s (serialised) -> under 0.60*serial. Regression: 160 realtime + websocket tests pass. Python-only (PHP forks, Ruby threads per connection). NOT changed: a user's own sync WS message handler still runs on the loop - the async-handler convention and per-connection message order make threading it a deliberate non-goal, same honest limit ADR-0074 documents. Co-Authored-By: Claude Opus 4.8 Signed-off-by: Andre van Zuydam --- tests/test_ws_backplane_threading.py | 93 ++++++++++++++++++++++++++++ tina4_python/realtime/__init__.py | 19 ++++-- tina4_python/websocket/__init__.py | 9 ++- 3 files changed, 114 insertions(+), 7 deletions(-) create mode 100644 tests/test_ws_backplane_threading.py diff --git a/tests/test_ws_backplane_threading.py b/tests/test_ws_backplane_threading.py new file mode 100644 index 00000000..efa39c00 --- /dev/null +++ b/tests/test_ws_backplane_threading.py @@ -0,0 +1,93 @@ +# Copyright (c) 2026 Code Infinity +# SPDX-License-Identifier: MPL-2.0 +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, You can obtain one at https://mozilla.org/MPL/2.0/. + +"""THE WEBSOCKET BACKPLANE PUBLISH MUST NOT BLOCK THE EVENT LOOP. + +The Redis backplane's LISTENER already runs in its own thread +(RedisBackplane.subscribe -> pubsub.run_in_thread). The PUBLISH side does not: +`WebSocketManager.broadcast*` (async, on the loop) calls `_publish`, which calls +the synchronous `redis.publish` inline. So every broadcast - a WS send, a WebRTC +signalling relay (broadcast_to_room), a chat fan-out - makes a blocking socket +round-trip ON the loop, freezing every other connection for its duration. + +Proven the ADR-0074 way: a real WebSocketManager, a real clock and threads, and +a real backplane object whose `publish` carries a measured latency (the sleep is +the instrument, not a mock of Redis). No local connections, so the broadcast is +purely the publish. RED before the offload, GREEN after. +""" +import asyncio +import time + +from tina4_python.websocket import WebSocketManager + +DELAY = 0.3 +N = 4 + + +class SlowBackplane: + """A real backplane whose publish blocks for DELAY (stands in for a slow + Redis round-trip). subscribe is a no-op; the test only broadcasts.""" + + def publish(self, channel, message): + time.sleep(DELAY) + + def subscribe(self, channel, callback): + pass + + def close(self): + pass + + +def _manager(): + mgr = WebSocketManager() + mgr._backplane_started = True # skip the real wiring in _ensure_backplane + mgr._backplane = SlowBackplane() + return mgr + + +def test_backplane_publish_does_not_stall_the_event_loop(): + """While one broadcast's publish runs, the loop must keep ticking.""" + mgr = _manager() + ticks = [] + + async def ticker(): + for _ in range(200): + ticks.append(time.monotonic()) + await asyncio.sleep(0.01) + + async def drive(): + t = asyncio.ensure_future(ticker()) + await asyncio.sleep(0) + await mgr.broadcast_all("hello") + t.cancel() + + asyncio.run(drive()) + + assert len(ticks) > 15, ( + f"the loop ticked only {len(ticks)} times during one broadcast's " + f"{DELAY}s publish. The backplane publish ran ON the loop, so every " + "connection was frozen with it." + ) + + +def test_concurrent_broadcasts_do_not_serialize_on_the_publish(): + """N concurrent broadcasts must overlap their publishes, not queue on them. + + Serial (today): N * DELAY. Concurrent (after offload): ~DELAY. + """ + mgr = _manager() + + async def drive(): + start = time.monotonic() + await asyncio.gather(*(mgr.broadcast_all(f"m{i}") for i in range(N))) + return time.monotonic() - start + + elapsed = asyncio.run(drive()) + serial = N * DELAY + assert elapsed < serial * 0.6, ( + f"{N} concurrent broadcasts took {elapsed:.2f}s; serial would be ~{serial:.2f}s " + f"and concurrent ~{DELAY:.2f}s. They serialised on the backplane publish." + ) diff --git a/tina4_python/realtime/__init__.py b/tina4_python/realtime/__init__.py index a45654bd..cbbc9e05 100644 --- a/tina4_python/realtime/__init__.py +++ b/tina4_python/realtime/__init__.py @@ -333,8 +333,11 @@ async def chat_ws(connection, event, data): body = (payload.get("body") or "").strip() if not body: return - saved = _persist_message(channel_id, identity, body, - payload.get("thread_id")) + # _persist_message does a blocking ORM write; run it off the loop + # so one chat message never freezes every other WS connection. + saved = await asyncio.to_thread( + _persist_message, channel_id, identity, body, + payload.get("thread_id")) if saved is not None: # Deliver to everyone in the room INCLUDING the sender, so # the sender's optimistic message is reconciled with its @@ -380,7 +383,9 @@ async def files_upload(request, response): upload = request.files.get("file") if not upload: return response({"error": "no file uploaded (field 'file')"}, 400) - saved = _store_upload(store, channel_id, upload) + # _store_upload writes to the storage backend (local file or S3); + # run it off the loop so an upload never freezes other requests. + saved = await asyncio.to_thread(_store_upload, store, channel_id, upload) saved["url"] = store.url(saved["key"]) or \ f"{paths['files']}/{saved['key']}" return response(saved, 201) @@ -476,7 +481,8 @@ def _persist_message(channel_id, identity, body, thread_id): async def _mark_read(channel_id, identity): """Advance the member's read cursor to now (read receipt).""" from tina4_python.realtime.models import ChannelMember - try: + + def _advance_cursor(): # where() takes a WHERE fragment; select_one() takes full SQL. members = ChannelMember.where( "channel_id = ? AND user_id = ?", [channel_id, identity], limit=1) @@ -484,6 +490,11 @@ async def _mark_read(channel_id, identity): member = members[0] member.last_read_at = _now_iso() member.save() + + try: + # The read + write are blocking ORM calls; run them off the loop so a + # read receipt never freezes other WS connections. + await asyncio.to_thread(_advance_cursor) except Exception as exc: Log.error(f"realtime chat: read-receipt update failed: {exc}") diff --git a/tina4_python/websocket/__init__.py b/tina4_python/websocket/__init__.py index 8ad2cd4a..11d2dfdf 100644 --- a/tina4_python/websocket/__init__.py +++ b/tina4_python/websocket/__init__.py @@ -624,7 +624,10 @@ async def broadcast(self, message: str | bytes, exclude: str = None, path: str = and not await self._safe_send(ws, message)] for ws in dead: self.remove(ws) - self._publish("path" if path else "all", message, path=path, exclude=exclude) + # The backplane publish is a blocking socket round-trip (synchronous + # redis.publish); run it off the loop so a broadcast never freezes every + # other connection. The listener side is already threaded. + await asyncio.to_thread(self._publish, "path" if path else "all", message, path=path, exclude=exclude) async def broadcast_all(self, message: str | bytes): """Send message to ALL connections (resilient to dead clients). @@ -635,7 +638,7 @@ async def broadcast_all(self, message: str | bytes): if not await self._safe_send(ws, message)] for ws in dead: self.remove(ws) - self._publish("all", message) + await asyncio.to_thread(self._publish, "all", message) async def send_to(self, client_id: str, message: str | bytes): """Send a message to a specific client by ID. @@ -736,7 +739,7 @@ async def broadcast_to_room(self, room_name: str, message: str | bytes, and not await self._safe_send(ws, message)] for ws in dead: self.remove(ws) - self._publish("room", message, room=room_name, exclude=exclude) + await asyncio.to_thread(self._publish, "room", message, room=room_name, exclude=exclude) class WebSocketServer: