Healthy relay connections: heartbeat, per-browser queues, send backpressure, hint for several tabs

README.md

@@ -143,6 +143,7 @@ Notes:
 - **Web sockets must be on**, otherwise the board stays offline. **Always On** prevents the app from idling and dropping connections.
 - Prefer a pinned version (`k3mpaxl/factgraph:0.5.3`) over `latest` during an incident, so a restart never changes the version.
 - App Service **HTTP logging** is off by default. If you enable it, it records request URLs, which contain board IDs.
+- **WebSocket limits per plan:** every open FactGraph tab holds one connection. App Service allows 5 per instance on **F1 (Free)**, 35 on Shared and 350 from **B1**; above the limit, browsers get HTTP 429 and stay offline. Use B1 or higher. Browsers send a heartbeat, and the server closes connections that stay silent (closed laptops, changed networks), so slots are freed without restarting the app.
 - **App Service Authentication** (Entra ID sign-in) can be put in front of the UI. Agents and scripts then also need an Entra token, and if the incident involves your own tenant, an identity provider you cannot trust is no protection. IP access restrictions are often the better choice there.
 - When the incident is closed: export the boards as JSON, then `az group delete -n $RG`. Nothing remains on the server side.
 
@@ -430,6 +431,9 @@ curl -s -X POST "http://127.0.0.1:8080/api/boards/$BOARD/export" \
 A **Syncing … changes** indicator in the header shows larger syncs (reconnects, imports, colleagues' batches) and a short *Synced* when done. Messages above 1 MB go to the relay gzip-compressed in a binary frame, so the 16 MiB WebSocket limit applies to the compressed size (up to 192 MiB unpacked); large imports are additionally kept in source parts.
 
 - Every browser sorts all actions the same way (logical clock, actor, ID). Any arrival order therefore produces the same graph, and concurrent changes to the same field are resolved identically everywhere.
+- Large syncs do not break connections: the relay queues messages per browser and sends them at that browser's pace (a slow or stalled browser holds back neither the sender nor the others; one more than 64 MiB behind is disconnected and fetches the rest when it reconnects), and a browser puts the next message on the wire only when its backlog is below 1 MB, so its pong and heartbeat never wait behind megabytes (a tab in the background sends at once, because browsers slow its timers down). The image allows 120 s for a WebSocket pong (`UVICORN_WS_PING_TIMEOUT`).
+- Every tab loads, stores and syncs the whole board, so a large board open in several tabs makes them all slower. A tab says so when the board is open in other tabs of the same browser.
+- Every 25 s a browser sends a heartbeat that the server answers. The server closes a browser silent for 2.5 minutes, and a browser that hears nothing from the server for 2.5 minutes (e.g. after sleep) connects again; after failed attempts it waits longer (2 s up to 30 s).
 - When joining and reconnecting, a browser sends a short summary of what it holds (per author: last clock, count, digest) to one peer, which answers with only what is missing; then the others are asked for what only they have. Changes made offline stay local and are delivered when a peer asks back. Large batches travel gzip-compressed; the relay forwards them without unpacking.
 - A duplicated browser tab gets its own identity before it connects. If two connections still claim the same identity, the older tab stops reconnecting and offers to continue as a new session.
 - All times are stored in one format, ISO 8601 in UTC with milliseconds. A time without a zone (from an export, an agent or a dialog) is read as UTC, never as the browser's local time. Exports whose time columns are marked *[Local]* are refused, so they cannot shift silently.

app/main.py

@@ -15,7 +15,7 @@
 import re
 import secrets
 import zlib
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
 from urllib.parse import urlsplit
 from uuid import UUID
@@ -68,6 +68,11 @@ class Peer:
     websocket: WebSocket
     name: str
     session_token: str | None = None
+    # Messages for this browser, sent by its own writer at its pace (see deliver).
+    outbox: asyncio.Queue = field(default_factory=asyncio.Queue)
+    queued: int = 0
+    writer: asyncio.Task | None = None
+    dropped: bool = False
 
 
 # Presence only. No board data or action history is retained on the server.
@@ -207,27 +212,66 @@ def valid_uuid(value: str) -> bool:
         return False
 
 
+# A browser may fall this far behind (a large sync to a stalled or very slow browser) before it is disconnected; it
+# fetches what it lacks when it reconnects (sync by summary).
+MAX_OUTBOX_BYTES = 64 * 1024 * 1024
+background: set[asyncio.Task] = set()
+
+
+def deliver(peer: Peer, message: str | bytes) -> None:
+    """Queue a message for one browser. Its own writer sends it at that browser's pace, so a slow browser holds back
+    neither the sender nor the others: waiting for it inside the sender's loop stopped reading the sender (uvicorn reads one
+    message ahead), the sender's pong stayed unread and uvicorn closed the healthy sender ("keepalive ping timeout"); on
+    large boards the sync then started over and over."""
+    if peer.dropped:
+        return
+    if peer.queued + len(message) > MAX_OUTBOX_BYTES:
+        peer.dropped = True
+        task = asyncio.create_task(refuse(peer.websocket, 1013))  # try again later
+        background.add(task)
+        task.add_done_callback(background.discard)
+        return
+    peer.queued += len(message)
+    peer.outbox.put_nowait(message)
+
+
+async def write(peer: Peer) -> None:
+    """Send what is queued for one browser, in order; ends with its connection."""
+    try:
+        while True:
+            message = await peer.outbox.get()
+            try:
+                if isinstance(message, bytes):
+                    await peer.websocket.send_bytes(message)
+                else:
+                    await peer.websocket.send_text(message)
+            finally:
+                peer.queued -= len(message)
+    except (RuntimeError, OSError, WebSocketDisconnect):
+        pass
+
+
+def start_peer(peer: Peer) -> Peer:
+    peer.writer = asyncio.create_task(write(peer))
+    return peer
+
+
 async def send_to_room(board_id: str, message: dict, *, exclude: str | None = None,
                        target: str | None = None) -> None:
+    text = json.dumps(message)
     async with rooms_lock:
-        recipients = [peer.websocket for actor, peer in rooms.get(board_id, {}).items()
+        recipients = [peer for actor, peer in rooms.get(board_id, {}).items()
                       if actor != exclude and (target is None or actor == target)]
-    for websocket in recipients:
-        try:
-            await websocket.send_json(message)
-        except (RuntimeError, WebSocketDisconnect):
-            pass
+    for peer in recipients:
+        deliver(peer, text)
 
 
 async def send_bytes_to_room(board_id: str, data: bytes, *, exclude: str | None = None, target: str | None = None) -> None:
     async with rooms_lock:
-        recipients = [peer.websocket for actor, peer in rooms.get(board_id, {}).items()
+        recipients = [peer for actor, peer in rooms.get(board_id, {}).items()
                       if actor != exclude and (target is None or actor == target)]
-    for websocket in recipients:
-        try:
-            await websocket.send_bytes(data)
-        except (RuntimeError, WebSocketDisconnect):
-            pass
+    for peer in recipients:
+        deliver(peer, data)
 
 
 async def browser_command(board_id: str, operation: str, **values) -> dict:
@@ -254,8 +298,8 @@ async def browser_command(board_id: str, operation: str, **values) -> dict:
         future: asyncio.Future[dict] = loop.create_future()
         pending[request_id] = (peer.websocket, future)
     try:
-        await peer.websocket.send_json({"type": "api-command", "requestId": request_id,
-                                        "boardId": board_id, "operation": operation, **values})
+        deliver(peer, json.dumps({"type": "api-command", "requestId": request_id,
+                                  "boardId": board_id, "operation": operation, **values}))
         result = await asyncio.wait_for(future, timeout=30)
         if not result.get("ok"):
             error = str(result.get("error") or "Browser rejected the command")
@@ -844,6 +888,18 @@ async def receive_message(websocket: WebSocket):
         return None
 
 
+def idle_timeout(heartbeat) -> float | None:
+    """How long a browser may stay silent: six of the heartbeat intervals it announced in hello (25 s → 150 s). Browsers
+    without a heartbeat (older versions) are only closed by WebSocket pings, as before.
+
+    Hosting platforms count WebSocket connections (Azure App Service: 5 on the free plan, 350 from Basic) and their front
+    end may keep a connection alive whose browser vanished (closed laptop, changed network): the slot stays taken until a
+    restart. Application data cannot be answered by a proxy, so a silent browser is detected and its slot freed."""
+    if isinstance(heartbeat, bool) or not isinstance(heartbeat, (int, float)) or heartbeat <= 0:
+        return None
+    return min(max(float(heartbeat), 0.5), 60.0) * 6
+
+
 async def refuse(websocket: WebSocket, code: int = 1008) -> None:
     """Close a connection whose browser may already be gone: a reload or a closed tab right after connecting, many at once
     after a server restart. Closing it again is no error worth a traceback."""
@@ -873,30 +929,39 @@ async def board_socket(websocket: WebSocket, board_id: str):
         return
     name = str(hello.get("name", "Guest")).strip()[:40] or "Guest"
     session_token = str(hello.get("token", "")).strip()[:200] or None
+    idle = idle_timeout(hello.get("heartbeat"))
 
     board_id = str(UUID(board_id))
     async with rooms_lock:
         room = rooms.setdefault(board_id, {})
         previous = room.get(actor)
         peers = [{"id": peer_id, "name": peer.name} for peer_id, peer in room.items()
                  if peer_id != actor]
-        room[actor] = Peer(websocket=websocket, name=name, session_token=session_token)
+        own = room[actor] = start_peer(Peer(websocket=websocket, name=name, session_token=session_token))
     if previous:
         # 4001: replaced by a connection with the same identity. The browser does not reconnect on it (two tabs with
         # one actor would otherwise replace each other every few seconds) and offers a new identity instead.
+        previous.dropped = True
+        if previous.writer:
+            previous.writer.cancel()
         try:
             await previous.websocket.close(code=4001, reason="replaced")
         except (RuntimeError, OSError, WebSocketDisconnect):  # the replaced tab may be gone already
             pass
         for request_id, (writer, future) in list(pending.items()):
             if writer is previous.websocket and not future.done():
                 future.set_result({"ok": False, "error": "Browser connection replaced"})
-    await websocket.send_json({"type": "welcome", "peers": peers})
-    await send_to_room(board_id, {"type": "peer-joined", "peer": {"id": actor, "name": name}}, exclude=actor)
-
     try:
+        # Inside the try: a browser that leaves right now is still removed from the room below.
+        deliver(own, json.dumps({"type": "welcome", "peers": peers}))
+        await send_to_room(board_id, {"type": "peer-joined", "peer": {"id": actor, "name": name}}, exclude=actor)
         while True:
-            message = await receive_message(websocket)
+            try:
+                message = await (asyncio.wait_for(receive_message(websocket), timeout=idle) if idle else receive_message(websocket))
+            except asyncio.TimeoutError:
+                # Silent for six heartbeats: the browser is gone, even if a proxy still answers WebSocket pings for it.
+                await refuse(websocket, 4000)
+                break
             if isinstance(message, tuple):
                 header, data = message
                 target = header.get("target")
@@ -906,7 +971,10 @@ async def board_socket(websocket: WebSocket, board_id: str):
             if not isinstance(message, dict):
                 continue
             kind = message.get("type")
-            if kind == "actions":
+            if kind == "heartbeat":
+                # Answered, so the browser knows the whole path is alive (a laptop waking up finds a dead connection).
+                deliver(own, '{"type":"heartbeat"}')
+            elif kind == "actions":
                 actions = message.get("actions")
                 if isinstance(actions, list) and len(actions) <= 1000:
                     target = message.get("target")
@@ -954,6 +1022,9 @@ async def board_socket(websocket: WebSocket, board_id: str):
     except (WebSocketDisconnect, RuntimeError):
         pass
     finally:
+        own.dropped = True
+        if own.writer:
+            own.writer.cancel()
         async with rooms_lock:
             room = rooms.get(board_id)
             if room and room.get(actor, None) and room[actor].websocket is websocket:

deploy/Dockerfile

@@ -12,9 +12,12 @@ WORKDIR /app
 # Logs show the client's real IP from X-Forwarded-For, trusted only from loopback and link-local proxies (169.254.0.0/16:
 # Azure App Service front ends; not reachable as a client, so nobody can fake an IP in the log). Behind another reverse
 # proxy, set FORWARDED_ALLOW_IPS to its address.
+# WebSocket keepalive: a pong may wait behind megabytes of a large sync (and a hosting front end's buffers); 20 s closed
+# healthy browsers mid-sync. The application heartbeat (2.5 min) detects browsers that are really gone.
 ENV PYTHONDONTWRITEBYTECODE=1 \
     PYTHONUNBUFFERED=1 \
-    FORWARDED_ALLOW_IPS=127.0.0.1,::1,169.254.0.0/16
+    FORWARDED_ALLOW_IPS=127.0.0.1,::1,169.254.0.0/16 \
+    UVICORN_WS_PING_TIMEOUT=120
 # Apply Debian security updates at build time (e.g. libpcre2-8-0), even when the base image lags behind.
 RUN apt-get update \
     && apt-get -y upgrade --no-install-recommends \

tests/test_relay.py

@@ -321,6 +321,28 @@ def call(method, path, body=None, content_type="application/json"):
             self.assertEqual(caught.exception.code, 422)
             self.assertIn("caller_oid", json.load(caught.exception)["detail"])
 
+    async def test_heartbeat_keeps_browsers_and_silence_frees_their_connection(self):
+        """Hosting front ends may keep a vanished browser's connection open (and answer pings for it); a browser that
+        announced a heartbeat and stays silent for six of them is closed, which frees the platform's connection slot."""
+        url = f"ws://127.0.0.1:{self.port}/ws/boards/{uuid4()}"
+        async with websockets.connect(url) as websocket:
+            await websocket.send(json.dumps({"type": "hello", "actor": str(uuid4()), "name": "Beat", "heartbeat": 0.5}))
+            self.assertEqual((await receive(websocket))["type"], "welcome")
+            # Heartbeats are answered and keep the connection beyond the 3 s silence limit.
+            for _ in range(5):
+                await websocket.send(json.dumps({"type": "heartbeat"}))
+                self.assertEqual((await receive(websocket))["type"], "heartbeat")
+                await asyncio.sleep(1)
+            # Then silence: closed with 4000.
+            with self.assertRaises(websockets.ConnectionClosed) as closed:
+                await asyncio.wait_for(websocket.recv(), timeout=6)
+            self.assertEqual(closed.exception.rcvd.code, 4000)
+        # Browsers without a heartbeat (older versions) are not closed for silence.
+        async with join(url, str(uuid4()), "Old") as websocket:
+            await receive(websocket)
+            with self.assertRaises(asyncio.TimeoutError):
+                await asyncio.wait_for(websocket.recv(), timeout=4)
+
     async def test_large_replies_arrive_in_parts_and_lists_query_the_browser(self):
         board, actor, token = str(uuid4()), str(uuid4()), "parts-token"
 

tests/test_relay_flow.py

@@ -0,0 +1,93 @@
+"""A large sync to a slow browser must not cost the sending browser its connection.
+
+The relay forwards a browser's messages to the others. When it waited for a slow receiver inside the sender's loop, it
+stopped reading the sender (uvicorn reads one message ahead), the sender's pong stayed unread, and uvicorn closed the
+healthy sender with "keepalive ping timeout": both reconnected and the sync of a large board started over and over.
+"""
+import asyncio
+import json
+import os
+import socket
+import subprocess
+import sys
+import time
+import unittest
+from pathlib import Path
+from urllib.request import urlopen
+from uuid import uuid4
+
+import websockets
+
+FRAME = 512 * 1024
+FRAMES = 40
+
+
+class RelayFlowTest(unittest.IsolatedAsyncioTestCase):
+    @classmethod
+    def setUpClass(cls):
+        with socket.socket() as sock:
+            sock.bind(("127.0.0.1", 0))
+            cls.port = sock.getsockname()[1]
+        # Fast keepalive pings, so a stalled reader is noticed within seconds.
+        cls.process = subprocess.Popen([sys.executable, "-m", "uvicorn", "app.main:app", "--host", "127.0.0.1", "--port", str(cls.port),
+                                        "--log-level", "error", "--ws-ping-interval", "1", "--ws-ping-timeout", "4"],
+                                       cwd=Path(__file__).resolve().parents[1], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+                                       env={**os.environ})
+        for _ in range(100):
+            try:
+                with urlopen(f"http://127.0.0.1:{cls.port}/api/health", timeout=0.1):
+                    return
+            except Exception:
+                time.sleep(0.05)
+        cls.process.terminate()
+        raise RuntimeError("Test server could not be started")
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.process.terminate()
+        cls.process.wait(5)
+
+    async def hello(self, websocket, name):
+        await websocket.send(json.dumps({"type": "hello", "actor": str(uuid4()), "name": name}))
+        return json.loads(await asyncio.wait_for(websocket.recv(), timeout=3))
+
+    async def test_a_slow_browser_holds_back_neither_the_sender_nor_the_others(self):
+        url = f"ws://127.0.0.1:{self.port}/ws/boards/{uuid4()}"
+        header = json.dumps({"type": "actions"}).encode()
+        frame = b"FGZ1" + len(header).to_bytes(4, "big") + header + os.urandom(FRAME)
+        # The stalled browser stops reading after one message (a closed laptop, a very slow link).
+        async with websockets.connect(url, max_queue=1) as stalled, websockets.connect(url) as fast, websockets.connect(url) as sender:
+            await self.hello(stalled, "Stalled")
+            await self.hello(fast, "Fast")
+            await self.hello(sender, "Sender")
+            received = 0
+
+            async def read_fast():
+                nonlocal received
+                while received < FRAMES:
+                    if isinstance(await fast.recv(), bytes):
+                        received += 1
+
+            async def drain(websocket):
+                async for _ in websocket:
+                    pass
+            reader, drainer = asyncio.create_task(read_fast()), asyncio.create_task(drain(sender))
+            started = time.monotonic()
+            try:
+                for _ in range(FRAMES):
+                    await sender.send(frame)
+                await asyncio.wait_for(reader, timeout=60)
+                took = time.monotonic() - started
+                # The sender stays connected and answers a ping.
+                await asyncio.wait_for(await sender.ping(), timeout=5)
+            finally:
+                reader.cancel()
+                drainer.cancel()
+            self.assertEqual(received, FRAMES)
+            # 20 MB over loopback take well under a second; waiting for the stalled browser would take until its keepalive
+            # timeout (about 5 s here, 120 s in production).
+            self.assertLess(took, 3, f"the fast browser waited {took:.1f} s for the stalled one")
+
+
+if __name__ == "__main__":
+    unittest.main()