From cece594d9e6262fb2c95dfac5fffb5cb34b6e5cc Mon Sep 17 00:00:00 2001 From: k3mpaxl <> Date: Wed, 7 Oct 2026 13:04:15 +0200 Subject: [PATCH 1/2] Keep relay connections healthy: heartbeat, per-browser queues, send backpressure Users who were disconnected could not reconnect until App Service restarted, and large boards reconnected every few seconds during sync. - Heartbeat: every 25 s a browser sends a message the server answers. The server closes a browser silent for six of them (2.5 min), even if the hosting front end still answers WebSocket pings for it, so connections of vanished browsers (closed laptops, changed networks) free their slot in the platform's WebSocket limit instead of piling up until a restart. A browser that hears nothing for 2.5 min connects again; failed attempts wait longer (2 s up to 30 s). Browsers without a heartbeat (older versions) are only closed by pings, as before. - Per-browser queues: the relay queues messages per browser and a writer sends them at that browser's pace. Forwarding inside the sender's loop stopped reading the sender while a slow browser drained; the sender's pong stayed unread and uvicorn closed the healthy sender ("keepalive ping timeout"), so the sync of a large board started over and over. A browser more than 64 MiB behind is disconnected (1013) and fetches the rest when it reconnects. - Send backpressure: a browser puts the next message on the wire only when its backlog is below 1 MB (also for reply parts), so its own pong and heartbeat never wait behind megabytes of upload. - The image allows 120 s for a WebSocket pong (UVICORN_WS_PING_TIMEOUT). - A browser that leaves between hello and welcome is removed from the room. tests/test_relay_flow.py reproduces the closed sender with a stalled browser (fails on the old relay); test_relay covers heartbeat and silence; README documents the App Service WebSocket limits per plan. Co-Authored-By: Claude Opus 5.5 --- README.md | 3 ++ app/main.py | 113 +++++++++++++++++++++++++++++++-------- deploy/Dockerfile | 5 +- tests/test_relay.py | 22 ++++++++ tests/test_relay_flow.py | 93 ++++++++++++++++++++++++++++++++ web/src/query.ts | 12 +++++ web/src/useBoard.ts | 65 ++++++++++++++++++---- web/tests/query.cjs | 16 ++++++ 8 files changed, 298 insertions(+), 31 deletions(-) create mode 100644 tests/test_relay_flow.py diff --git a/README.md b/README.md index ac537bd..19d8df0 100644 --- a/README.md +++ b/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,8 @@ 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. The image allows 120 s for a WebSocket pong (`UVICORN_WS_PING_TIMEOUT`). +- 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. diff --git a/app/main.py b/app/main.py index 22a14bf..cede776 100644 --- a/app/main.py +++ b/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,6 +929,7 @@ 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: @@ -880,10 +937,13 @@ async def board_socket(websocket: WebSocket, board_id: str): 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 @@ -891,12 +951,17 @@ async def board_socket(websocket: WebSocket, board_id: str): 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: diff --git a/deploy/Dockerfile b/deploy/Dockerfile index 9ce4c2e..e0d978d 100644 --- a/deploy/Dockerfile +++ b/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 \ diff --git a/tests/test_relay.py b/tests/test_relay.py index 1b64a7f..84c38eb 100644 --- a/tests/test_relay.py +++ b/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" diff --git a/tests/test_relay_flow.py b/tests/test_relay_flow.py new file mode 100644 index 0000000..fa26460 --- /dev/null +++ b/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() diff --git a/web/src/query.ts b/web/src/query.ts index 252bd33..4a060c5 100644 --- a/web/src/query.ts +++ b/web/src/query.ts @@ -73,6 +73,18 @@ export const COMPRESSED_RAW_LIMIT = 60 * 1024 * 1024 /** Messages above this are sent gzip-compressed in a binary frame (the relay inflates them). */ export const COMPRESS_ABOVE = 1_000_000 +/** Keep at most this much queued on a socket before sending more (see drained). */ +export const SEND_BACKLOG = 1_000_000 + +/** + * Wait until the browser has passed most of what is queued on a socket to the network. Queuing a whole large sync at + * once put the browser's own pong and heartbeat behind megabytes of upload: the server took the healthy browser for gone + * and closed it, and the sync of a large board started over and over. + */ +export async function drained(socket: WebSocket, limit = SEND_BACKLOG) { + while (socket.readyState === WebSocket.OPEN && socket.bufferedAmount > limit) await new Promise(resolve => setTimeout(resolve, 50)) +} + /** gzip in the browser (CompressionStream); null where it is not available. */ export async function gzipText(text: string): Promise { if (typeof CompressionStream === 'undefined') return null diff --git a/web/src/useBoard.ts b/web/src/useBoard.ts index 2539cb9..5e4ae93 100644 --- a/web/src/useBoard.ts +++ b/web/src/useBoard.ts @@ -4,7 +4,7 @@ import { useCallback, useEffect, useMemo, useRef, useState } from 'react' import { type ActionDraft, type BoardAction, type UndoScope, isAction, isDraft, participantKey, project, redoTarget, sortActions, undoTargets } from './board' import { listBoards, loadActions, saveActions, touchBoard, type BoardMeta } from './store' import { placeNew } from './layout' -import { actionChunks, messageParts, queryRecords, COMPRESS_ABOVE, COMPRESSED_RAW_LIMIT, RELAY_LIMIT, gzipText, packFrame, unpackFrame } from './query' +import { actionChunks, drained, messageParts, queryRecords, COMPRESS_ABOVE, COMPRESSED_RAW_LIMIT, RELAY_LIMIT, gzipText, packFrame, unpackFrame } from './query' import { uuid } from './uuid' import { trustedDrafts } from './importBatches' import { onBehalf } from './provenance' @@ -17,6 +17,14 @@ import { browserMeasure, buildGraphSvg, svgToPng } from './exportGraph' import { iconMarkup, prepareIconMarkup } from './KindIcon' const MAX_EXPORT_CHARS = 12_000_000 +/** + * Keepalive on the relay. Every HEARTBEAT_MS the browser sends a small message the server answers. The server closes a + * browser that stays silent for six of them (frees the connection slot hosting platforms count, also when their front + * end still answers WebSocket pings for a vanished browser), and the browser reconnects when the server stays silent for + * SILENCE_MS (a laptop waking up finds a connection that only looks open). Timers in hidden tabs may run once a minute. + */ +const HEARTBEAT_MS = 25_000 +const SILENCE_MS = 150_000 /** Render the board for REST/MCP with a saved perspective (or everything) and the stored group state. */ async function renderExport({ data, name }: ReturnType, options: Record) { @@ -117,17 +125,21 @@ export function useBoard(boardId: string, listener?: ChangeListener) { if (count) { syncCounts.current.outgoing += count; reportSync() } const done = () => { if (count) { syncCounts.current.outgoing -= count; reportSync() } } sendQueue.current = sendQueue.current.then(async () => { - if (socketRef.current?.readyState !== WebSocket.OPEN) return + const socket = socketRef.current + if (socket?.readyState !== WebSocket.OPEN) return + // One message at a time into the network: pong and heartbeat must not wait behind a whole large sync. + await drained(socket) + if (socketRef.current !== socket || socket.readyState !== WebSocket.OPEN) return if (text.length > COMPRESS_ABOVE) { const packed = await gzipText(text) if (packed) { if (packed.byteLength > RELAY_LIMIT) { onTooLarge?.(); return } const { type, target, deferRender } = message as Record - if (socketRef.current?.readyState === WebSocket.OPEN) socketRef.current.send(packFrame({ type, ...(target ? { target } : {}), deferRender: deferRender === true }, packed)) + if (socket.readyState === WebSocket.OPEN) socket.send(packFrame({ type, ...(target ? { target } : {}), deferRender: deferRender === true }, packed)) return } } - socketRef.current.send(text) + socket.send(text) }).catch(() => {}).finally(done) }, [reportSync]) @@ -145,9 +157,14 @@ export function useBoard(boardId: string, listener?: ChangeListener) { }, [send, actor]) // Replies to REST/MCP commands; large ones go in parts that the server reassembles. const reply = useCallback((requestId: string, payload: Record) => { - const socket = socketRef.current - if (socket?.readyState !== WebSocket.OPEN) return - for (const part of messageParts(requestId, payload)) socket.send(part) + void (async () => { + for (const part of messageParts(requestId, payload)) { + const socket = socketRef.current + if (socket?.readyState !== WebSocket.OPEN) return + await drained(socket) + socket.send(part) + } + })() }, []) const merge = useCallback((incoming: unknown[], deferRender = false): Promise => { @@ -257,6 +274,15 @@ export function useBoard(boardId: string, listener?: ChangeListener) { settleJoin(300) requestSync() } + // Failed attempts in a row: the next one waits longer (2 s, 4 s … 30 s), so a refusing server is not hammered. + let failures = 0 + let heartbeat: number | undefined + const reconnect = () => { + const delay = Math.min(30_000, 2000 * 2 ** failures) * (0.8 + Math.random() * 0.4) + failures++ + window.clearTimeout(retry) + retry = window.setTimeout(connect, delay) + } const connect = () => { if (cancelled) return const scheme = window.location.protocol === 'https:' ? 'wss:' : 'ws:' @@ -264,14 +290,32 @@ export function useBoard(boardId: string, listener?: ChangeListener) { const socket = new WebSocket(`${scheme}//${window.location.host}/ws/boards/${boardId}`) websocket = socket socketRef.current = socket + let lastHeard = Date.now() socket.onopen = () => { if (cancelled) return - socket.send(JSON.stringify({ type: 'hello', actor, name: nameRef.current, token: sessionToken })) + socket.send(JSON.stringify({ type: 'hello', actor, name: nameRef.current, token: sessionToken, heartbeat: HEARTBEAT_MS / 1000 })) setConnected(true) + lastHeard = Date.now() + window.clearInterval(heartbeat) + heartbeat = window.setInterval(() => { + if (socketRef.current !== socket || socket.readyState !== WebSocket.OPEN) return + if (Date.now() - lastHeard > SILENCE_MS) { + // The connection only looks open: give it up and connect again (its late close event is ignored). + window.clearInterval(heartbeat) + socketRef.current = null + try { socket.close(4000, 'silent') } catch { /* already closing */ } + setConnected(false) + setPeers([]) + reconnect() + return + } + socket.send('{"type":"heartbeat"}') + }, HEARTBEAT_MS) // Syncing starts with the welcome (who is online), with summaries instead of the whole log. } socket.binaryType = 'arraybuffer' socket.onmessage = event => { + lastHeard = Date.now() // Large batches arrive as compressed binary frames (forwarded unchanged by the relay). if (event.data instanceof ArrayBuffer) { void unpackFrame(event.data).then(message => { if (message && socketRef.current === socket) handle(message) }); return } let message: Record @@ -280,6 +324,7 @@ export function useBoard(boardId: string, listener?: ChangeListener) { } const handle = (message: Record) => { if (message.type === 'welcome' && Array.isArray(message.peers)) { + failures = 0 const others = message.peers.filter((peer): peer is Peer => !!peer && typeof peer === 'object' && typeof (peer as Peer).id === 'string' && (peer as Peer).id !== actor) if (!actionsRef.current.length && others.length) { setJoining(true); settleJoin(15000) } // First one peer (it answers with what is missing here), then the others for what only they have. @@ -367,11 +412,12 @@ export function useBoard(boardId: string, listener?: ChangeListener) { } socket.onclose = event => { if (cancelled || socketRef.current !== socket) return + window.clearInterval(heartbeat) setConnected(false) setPeers([]) // Another tab connected with this identity: reconnecting would push it out again, and so on every 2 s. if (event.code === 4001) { setReplaced(true); return } - retry = window.setTimeout(connect, 2000) + reconnect() } } @@ -399,6 +445,7 @@ export function useBoard(boardId: string, listener?: ChangeListener) { cancelled = true releaseActor(actor) if (retry) window.clearTimeout(retry) + window.clearInterval(heartbeat) if (flushTimer.current) window.clearTimeout(flushTimer.current) websocket?.close() socketRef.current = null diff --git a/web/tests/query.cjs b/web/tests/query.cjs index b13b56f..b304b9f 100644 --- a/web/tests/query.cjs +++ b/web/tests/query.cjs @@ -66,3 +66,19 @@ console.log('Targeted queries, shortened lists, reply parts and size-bounded act assert.ok(isSummary(summarize(mine)) && !isSummary({ a: [1, 2] }) && !isSummary([1])) console.log('Sync by summary: deltas, gaps, unknown actors, older peers passed') } + +// Send backpressure: wait until the socket's backlog is below the limit before sending more, so pong and heartbeat are +// not stuck behind a whole large sync. +;(async () => { + const { drained } = require(require('node:path').resolve(process.argv[2]) + '/query.js') + globalThis.WebSocket ??= { OPEN: 1 } + const socket = { readyState: 1, bufferedAmount: 3_000_000 } + const shrink = setInterval(() => { socket.bufferedAmount = Math.max(0, socket.bufferedAmount - 1_000_000) }, 60) + const started = Date.now() + await drained(socket) + clearInterval(shrink) + assert.ok(socket.bufferedAmount <= 1_000_000 && Date.now() - started >= 100, 'waited until the backlog was small') + const closed = { readyState: 3, bufferedAmount: 9_000_000 } + await drained(closed) + console.log('Send backpressure passed') +})() From 1fcc0be94d1529fc73f8663ab223f7642e966079 Mon Sep 17 00:00:00 2001 From: k3mpaxl <> Date: Thu, 8 Oct 2026 08:13:18 +0200 Subject: [PATCH 2/2] Say when a board is open in several tabs; background tabs send without waiting - A tab shows a dismissible hint when the same board is open in other tabs of this browser: every tab loads, stores and syncs the whole board, so a large board open several times makes them all slow. Tabs count each other over the existing identity BroadcastChannel; they count again when a tab opens or closes the board, when a peer leaves (a killed tab never says goodbye) and once a minute. - Send backpressure no longer waits in a hidden tab. Browsers run its timers once a second or once a minute, so a background tab answering a sync request would have sent about 1 MB per minute. Co-Authored-By: Claude Opus 5.5 --- README.md | 3 ++- web/e2e/workflows.spec.ts | 8 ++++++ web/src/App.tsx | 7 +++++- web/src/identity.ts | 52 ++++++++++++++++++++++++++++++++++++++- web/src/query.ts | 6 +++-- web/src/styles.css | 9 ++++--- web/src/useBoard.ts | 23 +++++++++++++++-- web/tests/query.cjs | 6 +++++ 8 files changed, 104 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index 19d8df0..21e04bf 100644 --- a/README.md +++ b/README.md @@ -431,7 +431,8 @@ 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. The image allows 120 s for a WebSocket pong (`UVICORN_WS_PING_TIMEOUT`). +- 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. diff --git a/web/e2e/workflows.spec.ts b/web/e2e/workflows.spec.ts index ef704ea..ef05cb4 100644 --- a/web/e2e/workflows.spec.ts +++ b/web/e2e/workflows.spec.ts @@ -464,7 +464,15 @@ test('identity: a duplicated tab gets its own identity; a tab replaced by its ow await page.waitForTimeout(4500) await expect(page.getByText('2 online',{exact:true})).toBeVisible() await expect(dup.getByText('2 online',{exact:true})).toBeVisible() + // Both tabs say that the board is open twice in this browser; the hint can be dismissed. + const twice=(tab:typeof page)=>tab.getByRole('status').filter({hasText:'also open in another tab'}) + await expect(twice(page)).toBeVisible() + await expect(twice(dup)).toBeVisible() + await dup.getByRole('button',{name:'Got it'}).click() + await expect(twice(dup)).toHaveCount(0) await dup.close() + // The other tab is gone, and so is the hint. + await expect(twice(page)).toHaveCount(0) // Something connects with this tab's identity anyway: the tab says so and does not push it out again. const socket=new WebSocket(`${base.replace('http','ws')}/ws/boards/${id}`) await new Promise(resolve=>socket.addEventListener('open',resolve)) diff --git a/web/src/App.tsx b/web/src/App.tsx index 7e22ee6..d1ae5ba 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -1,7 +1,7 @@ import { useCallback, useEffect, useMemo, useRef, useState, type CSSProperties, type ReactNode, Fragment } from 'react' import { Activity, ArrowDownToLine, ArrowLeft, ArrowRight, ChevronDown, ChevronRight, Clock, Command, Copy, CornerDownLeft, - FileText, GitBranch, Keyboard, Link2, HardDrive, Server, Users, AlertTriangle, Lock, CircleHelp, ListTree, Merge, Moon, Pause, Pencil, Play, Plus, Plug, Search, ShieldCheck, + FileText, GitBranch, Keyboard, Link2, HardDrive, Info, Server, Users, AlertTriangle, Lock, CircleHelp, ListTree, Merge, Moon, Pause, Pencil, Play, Plus, Plug, Search, ShieldCheck, SkipBack, SkipForward, Sparkles, Sun, Trash2, Undo2, Redo2, Upload, X, Crosshair, CheckCircle2, CircleDashed, Boxes, Zap, Layers, Ungroup, ImageDown, RotateCcw, ShieldAlert, Braces, Bot, FileCheck2, } from 'lucide-react' @@ -961,6 +961,9 @@ export default function App() { const changes: ChangeListener = useRef(null) const board = useBoard(boardId, changes) const data = board.ready ? board.data : null + // The hint that this board is open in other tabs too: once dismissed, not again in this tab. + const [tabsHintSeen, setTabsHintSeen] = useState(() => { try { return sessionStorage.getItem(`factgraph:tabsHint:${boardId}`) === '1' } catch { return false } }) + const dismissTabsHint = () => { setTabsHintSeen(true); try { sessionStorage.setItem(`factgraph:tabsHint:${boardId}`, '1') } catch { /* private mode */ } } // Filters, layers, time window, selection and zoom are this analyst's own: per board in this browser, never synced. const saved = useMemo(() => loadView(boardId), [boardId]) const [selection, setSelectionState] = useState(saved.selection) @@ -1615,6 +1618,8 @@ export default function App() { { const file = event.target.files?.[0]; event.target.value = ''; void importFile(file) }} /> {board.replaced &&
This board was opened with this tab's identity in another tab (for example a duplicated tab), so this tab is no longer connected. Your changes here are kept.
} + {board.otherTabs > 0 && !tabsHintSeen && !board.replaced &&
This board is also open in {board.otherTabs === 1 ? 'another tab' : `${board.otherTabs} other tabs`} of this browser. Every tab loads and syncs the whole board, so a large board stays faster with one tab. +
}
{board.storageError &&
{board.storageError}
} {error && !dialog &&
setError('')}>{error}
} diff --git a/web/src/identity.ts b/web/src/identity.ts index 78efd9b..8169c43 100644 --- a/web/src/identity.ts +++ b/web/src/identity.ts @@ -7,16 +7,29 @@ import { uuid } from './uuid' */ const instance = uuid() const held = new Set() +/** Boards open in this tab, so other tabs can tell how often a board is open in this browser (see otherTabs). */ +const boards = new Map() +const boardListeners = new Set<(boardId: string) => void>() +let leaving = false let channel: BroadcastChannel | null | undefined function open(): BroadcastChannel | null { if (channel !== undefined) return channel channel = typeof BroadcastChannel === 'undefined' ? null : new BroadcastChannel('factgraph-identity') channel?.addEventListener('message', event => { - const message = event.data as { type?: string; actor?: string; instance?: string } + const message = event.data as { type?: string; actor?: string; instance?: string; board?: string } if (message?.type === 'probe' && message.actor && held.has(message.actor) && message.instance !== instance) channel!.postMessage({ type: 'taken', actor: message.actor, to: message.instance }) + if (message?.type === 'board-probe' && message.board && boards.has(message.board) && !leaving && message.instance !== instance) + channel!.postMessage({ type: 'board-here', board: message.board, to: message.instance, from: instance }) + if (message?.type === 'board-changed' && message.board) boardListeners.forEach(listener => listener(message.board!)) }) + // A closed tab never unmounts its board: it stops answering and lets the others count again. + if (channel && typeof window !== 'undefined') { + const changed = () => boards.forEach((_, board) => channel!.postMessage({ type: 'board-changed', board })) + window.addEventListener('pagehide', () => { leaving = true; changed() }) + window.addEventListener('pageshow', event => { if (event.persisted) { leaving = false; changed() } }) + } return channel } @@ -40,6 +53,43 @@ export function actorTaken(actor: string, wait = 150): Promise { export function holdActor(actor: string) { open(); held.add(actor) } export function releaseActor(actor: string) { held.delete(actor) } +/** This tab shows the board: other tabs with it count this one. */ +export function holdBoard(boardId: string) { + boards.set(boardId, (boards.get(boardId) ?? 0) + 1) + open()?.postMessage({ type: 'board-changed', board: boardId }) +} +export function releaseBoard(boardId: string) { + const count = (boards.get(boardId) ?? 1) - 1 + if (count > 0) boards.set(boardId, count); else boards.delete(boardId) + open()?.postMessage({ type: 'board-changed', board: boardId }) +} + +/** + * How many other tabs of this browser show the board (they answer within `wait` ms). Each tab loads, stores and syncs + * the whole board, which with a large board makes them all slow. + */ +export function otherTabs(boardId: string, wait = 200): Promise { + const bus = open() + if (!bus) return Promise.resolve(0) + return new Promise(resolve => { + const seen = new Set() + const listener = (event: MessageEvent) => { + const message = event.data as { type?: string; board?: string; to?: string; from?: string } + if (message?.type === 'board-here' && message.board === boardId && message.to === instance && message.from) seen.add(message.from) + } + bus.addEventListener('message', listener) + bus.postMessage({ type: 'board-probe', board: boardId, instance }) + window.setTimeout(() => { bus.removeEventListener('message', listener); resolve(seen.size) }, wait) + }) +} + +/** Calls the listener with a board's ID when a tab opens or closes it. */ +export function onBoardTabs(listener: (boardId: string) => void) { + open() + boardListeners.add(listener) + return () => { boardListeners.delete(listener) } +} + const actorKey = (boardId: string) => `factgraph:actor:${boardId}` const tokenKey = (boardId: string) => `factgraph:sessionToken:${boardId}` const newToken = () => `${uuid()}${uuid()}`.replaceAll('-', '') diff --git a/web/src/query.ts b/web/src/query.ts index 4a060c5..dab744b 100644 --- a/web/src/query.ts +++ b/web/src/query.ts @@ -79,11 +79,13 @@ export const SEND_BACKLOG = 1_000_000 /** * Wait until the browser has passed most of what is queued on a socket to the network. Queuing a whole large sync at * once put the browser's own pong and heartbeat behind megabytes of upload: the server took the healthy browser for gone - * and closed it, and the sync of a large board started over and over. + * and closed it, and the sync of a large board started over and over. A tab in the background does not wait: browsers run + * its timers once a second or even once a minute, which would leave another tab waiting for its sync answer for minutes. */ export async function drained(socket: WebSocket, limit = SEND_BACKLOG) { - while (socket.readyState === WebSocket.OPEN && socket.bufferedAmount > limit) await new Promise(resolve => setTimeout(resolve, 50)) + while (socket.readyState === WebSocket.OPEN && socket.bufferedAmount > limit && !hidden()) await new Promise(resolve => setTimeout(resolve, 50)) } +const hidden = () => typeof document !== 'undefined' && document.hidden /** gzip in the browser (CompressionStream); null where it is not available. */ export async function gzipText(text: string): Promise { diff --git a/web/src/styles.css b/web/src/styles.css index 1712c56..ca53818 100644 --- a/web/src/styles.css +++ b/web/src/styles.css @@ -1161,10 +1161,13 @@ button.impact-stat { text-align: left; font: inherit; color: inherit; cursor: po .agent-errors dl { display: grid; grid-template-columns: max-content 1fr; gap: 3px 10px; margin: 6px 0 0 } .agent-errors dt { font-family: var(--mono); color: var(--text) } .agent-errors dd { margin: 0 } .export-preview { display: flex; flex-direction: column; gap: 4px } .export-preview img { width: 100%; max-height: 180px; object-fit: contain; border: 1px solid var(--border); border-radius: var(--radius-sm); background: var(--canvas) } .export-preview small { font-size: 11px; color: var(--text-3) } -.replaced-banner { position: fixed; left: 50%; top: 52px; transform: translateX(-50%); z-index: 90; display: flex; align-items: center; gap: 10px; max-width: min(720px, calc(100% - 32px)); padding: 10px 14px; border-radius: var(--radius-lg); background: var(--surface); border: 1px solid var(--warn); box-shadow: var(--shadow-lg); font-size: 12.5px; color: var(--text-2) } +.replaced-banner, .tabs-banner { position: fixed; left: 50%; top: 52px; transform: translateX(-50%); z-index: 90; display: flex; align-items: center; gap: 10px; max-width: min(720px, calc(100% - 32px)); padding: 10px 14px; border-radius: var(--radius-lg); background: var(--surface); border: 1px solid var(--warn); box-shadow: var(--shadow-lg); font-size: 12.5px; color: var(--text-2) } .replaced-banner svg { color: var(--warn); flex: none } -.replaced-banner b { color: var(--text) } -.replaced-banner button { flex: none } +.replaced-banner b, .tabs-banner b { color: var(--text) } +.replaced-banner button, .tabs-banner button { flex: none } +/* The hint that a board is open in other tabs of this browser too. */ +.tabs-banner { border-color: var(--accent) } +.tabs-banner svg { color: var(--accent); flex: none } /* Import dialog: files, column mapping, saved formats */ .modal.import-modal { width: min(1000px, 100%) } diff --git a/web/src/useBoard.ts b/web/src/useBoard.ts index 5e4ae93..feca923 100644 --- a/web/src/useBoard.ts +++ b/web/src/useBoard.ts @@ -8,7 +8,7 @@ import { actionChunks, drained, messageParts, queryRecords, COMPRESS_ABOVE, COMP import { uuid } from './uuid' import { trustedDrafts } from './importBatches' import { onBehalf } from './provenance' -import { actorTaken, freshIdentity, holdActor, releaseActor, storedIdentity } from './identity' +import { actorTaken, freshIdentity, holdActor, holdBoard, onBoardTabs, otherTabs as countOtherTabs, releaseActor, releaseBoard, storedIdentity } from './identity' import { isSummary, summarize, syncPlan } from './sync' import { validateDrafts } from './validation' import { loadFormats } from './importFormats' @@ -75,6 +75,8 @@ export function useBoard(boardId: string, listener?: ChangeListener) { const [boards, setBoards] = useState([]) const [peers, setPeers] = useState([]) const [ready, setReady] = useState(false) + /** Other tabs of this browser that show this board. */ + const [otherTabs, setOtherTabs] = useState(0) // A fresh browser joining analysts who are online: the board is on its way, not empty. const [joining, setJoining] = useState(false) const joinTimer = useRef | undefined>(undefined) @@ -262,6 +264,7 @@ export function useBoard(boardId: string, listener?: ChangeListener) { useEffect(() => { let cancelled = false + let holding = false let retry: number | undefined let websocket: WebSocket | null = null @@ -427,6 +430,8 @@ export function useBoard(boardId: string, listener?: ChangeListener) { if (cancelled) return if (taken) { setIdentity(freshIdentity(boardId)); return } holdActor(actor) + holdBoard(boardId) + holding = true const stored = sortActions(loaded) actionsRef.current = stored actionIdsRef.current = new Set(stored.map(item => item.id)) @@ -444,6 +449,7 @@ export function useBoard(boardId: string, listener?: ChangeListener) { return () => { cancelled = true releaseActor(actor) + if (holding) releaseBoard(boardId) if (retry) window.clearTimeout(retry) window.clearInterval(heartbeat) if (flushTimer.current) window.clearTimeout(flushTimer.current) @@ -452,6 +458,19 @@ export function useBoard(boardId: string, listener?: ChangeListener) { } }, [actor, boardId, emitMany, merge, projectCached, reply, send, sendActions, sessionToken, undo, redo]) - return { actor, name, setName, actions, boards, peers, ready, joining, replaced, reidentify, connected, storageError, unsynced, sync, + // Count again when a tab opens or closes the board or someone leaves the board (a tab killed without closing, which + // never says so), and now and then. + const peerCount = peers.length + useEffect(() => { + if (!ready) return + let alive = true + const count = () => void countOtherTabs(boardId).then(tabs => { if (alive) setOtherTabs(tabs) }) + count() + const stop = onBoardTabs(board => { if (board === boardId) count() }) + const timer = window.setInterval(count, 60_000) + return () => { alive = false; stop(); window.clearInterval(timer) } + }, [boardId, ready, peerCount]) + + return { actor, name, setName, actions, boards, peers, ready, joining, replaced, reidentify, connected, storageError, unsynced, sync, otherTabs, boardName: projection.name, data: projection.data, emit, emitMany, undo, redo, importActions, sessionToken } } diff --git a/web/tests/query.cjs b/web/tests/query.cjs index b304b9f..b885562 100644 --- a/web/tests/query.cjs +++ b/web/tests/query.cjs @@ -80,5 +80,11 @@ console.log('Targeted queries, shortened lists, reply parts and size-bounded act assert.ok(socket.bufferedAmount <= 1_000_000 && Date.now() - started >= 100, 'waited until the backlog was small') const closed = { readyState: 3, bufferedAmount: 9_000_000 } await drained(closed) + // A tab in the background sends at once (its timers are throttled). + globalThis.document = { hidden: true } + const quick = Date.now() + await drained({ readyState: 1, bufferedAmount: 9_000_000 }) + assert.ok(Date.now() - quick < 50, 'a hidden tab does not wait') + delete globalThis.document console.log('Send backpressure passed') })()