perf(ws,realtime): offload blocking backplane publish + chat/upload I/O off the event loop

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."
+    )

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,14 +481,20 @@ 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)
         if members:
             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}")
 

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: