diff --git a/deploy/relay-v1/README.md b/deploy/relay-v1/README.md index ba38e2def4..7b93149d52 100644 --- a/deploy/relay-v1/README.md +++ b/deploy/relay-v1/README.md @@ -104,3 +104,148 @@ Never retire a prefix that current clients still use: an unconfigured relay is never retired, so an omitted or mistyped variable fails safe. An operator turnover hazard is the reverse: a prefix listed but no longer proxied stays retired only for as long as the edge announces it. + +## Keeping a legacy 1.0.x deployment alive + +A 1.0.x Relay persisted every realtime session message in `realtime_messages` +and never pruned it: `log_bytes` was tracked per account but no code enforced a +ceiling, so the file grew with traffic forever. One production deployment +reached **38.6 GiB in six days** with 97 accounts (~6.4 GB/day, ~2.8 M rows, +~14.5 KB per encrypted row), while every other table stayed under a megabyte. +1.0.2 stores no session content and drops those tables at startup (see +[Forwarding only](../../src/apps/relay-server/README.md#forwarding-only)), but a +legacy version cannot be patched after the fact, so its log has to be bounded by +hand until that version is retired. + +`prune-legacy-message-log.py` in this directory does that from the database +side, without a service window. The rules below are the contract it implements; +do not weaken them without re-reading the legacy read path. + +### What a legacy database may lose without affecting users + +- Messages are forwarded after being committed, so a client cursor is only ever + valid against the rows that still exist. Reads tolerate gaps: catch-up asks + for `seq > cursor` and history pages ask for `seq < cursor`, and neither + reports a missing range. +- Delete by age (default: keep 24 hours), keep the newest `--min-keep` messages + of every session regardless of age, and never delete anything newer than + `--safety-minutes` (default: 60 minutes). A device that was offline longer + than the window cannot backfill that gap; nothing else changes for it. +- Never delete or rewrite `realtime_sessions`, `realtime_account_sequence` or + `realtime_sessions.seq`. They carry the session identity and the cursor + counters; removing them would break `open_session` and the client cursors + instead of just shortening history. The tool leaves them untouched. +- The `local_id` uniqueness that makes a send retry idempotent only covers the + rows still retained. A retry of a message older than the window inserts a new + row rather than returning the old one, which is acceptable because clients + retry within seconds. + +### Prune online + +```bash +# Analyse first: prints how many rows each policy would delete. +python3 prune-legacy-message-log.py --report +python3 prune-legacy-message-log.py --dry-run --keep-hours 24 + +# Delete for real, gently. +nice -n 10 ionice -c2 -n6 python3 prune-legacy-message-log.py --apply \ + --keep-hours 24 --safety-minutes 60 --max-msgs 5000 --min-keep 200 \ + --batch-rows 2000 --sleep-ms 20 +``` + +The Relay runs SQLite in WAL mode with `busy_timeout = 5s`, so users only stay +unaffected while every statement holds the single write lock for far less than +five seconds: + +- Delete primary-key ranges in small committed batches with a pause between + them; never delete a large range in one transaction. A single oversized + `DELETE`, an online `VACUUM`, or a checkpoint stall makes every account wait + for the write lock and then fail with `database is locked`. +- Count rows that were really removed (`total_changes`), never the planned range + length. A planned range is recomputed on every run, so re-planning already + pruned sessions reports phantom deletions and issues empty statements that + still take the write lock. +- Verify with the Relay's own log: `slow statement` warnings (sqlx logs anything + over one second) mean the batches are too large. + +### Reclaim the space (one short window) + +Pruning alone does not shrink the file: an `auto_vacuum = 0` database keeps the +freed pages in its freelist and reuses them, which stops the growth but leaves +the file at its high-water mark. Either reclaim in a short window: + +```bash +docker stop openbitfun-relay-v1.0.1 +# Keep the previous file (plus its -wal/-shm) until the new one is verified. +sqlite3 /srv/openbitfun-relay-v1.0.1/data/relay.db \ + "PRAGMA auto_vacuum=INCREMENTAL; PRAGMA journal_mode=WAL; VACUUM;" +docker start openbitfun-relay-v1.0.1 +``` + +or compact online first and swap inside the window, which is what turned a +38.6 GiB file into 7.15 GiB here: run +`VACUUM INTO '/srv/.../data/relay-compact.db'` while the service keeps serving +(a consistent snapshot; `VACUUM INTO` output is `journal_mode=delete` and keeps +`auto_vacuum` of the source, so the window still has to set both), then stop the +container, move the live `relay.db` and its `-wal`/`-shm` aside, move the +compact copy into place, `chown` it to the container's UID (10001 in the Compose +file), and start. Budget the window around the in-place `VACUUM`, not the copy: +on a 7 GiB file it took 138 of the 151 seconds here. + +Enabling `auto_vacuum = INCREMENTAL` in that same window is what makes later +prunes return space without stopping the service: the tool then calls +`PRAGMA incremental_vacuum(...)` after each run, which moves free pages to the +end of the file and truncates them (verified: 6,843 free pages returned and the +file shrank by exactly 26.7 MiB while the service stayed up). + +After the swap, verify before deleting anything: `PRAGMA quick_check`, +row counts for `users`/`devices`/`auth_tokens`/`realtime_*`, `auto_vacuum = 2`, +`journal_mode = wal`, `/health`, and that devices reconnect. A swap installs a +snapshot taken when the copy started, so writes made during the copy are +missing; that is acceptable for the message log, but expect a few devices to +sign in again because their credential row moved. + +### Run it on a timer + +Install the tool on the deployment host and prune hourly, so the database stops +growing without anyone watching it: + +```bash +install -m 0755 prune-legacy-message-log.py /root/ops/ +``` + +```ini +# /etc/systemd/system/openbitfun-relay-1.0.1-prune.service +[Unit] +Description=Prune the legacy OpenBitFun Relay 1.0.1 realtime message log +After=network-online.target docker.service + +[Service] +Type=oneshot +Nice=10 +IOSchedulingClass=best-effort +IOSchedulingPriority=6 +ExecStart=/usr/bin/python3 /root/ops/prune-legacy-message-log.py --apply \ + --keep-hours 24 --safety-minutes 60 --max-msgs 5000 --min-keep 200 \ + --batch-rows 500 --sleep-ms 25 --incremental-vacuum-pages 2000 +StandardOutput=append:/var/log/openbitfun-relay-1.0.1-prune.log +StandardError=append:/var/log/openbitfun-relay-1.0.1-prune.log + +# /etc/systemd/system/openbitfun-relay-1.0.1-prune.timer +[Unit] +Description=Hourly prune of the legacy OpenBitFun Relay 1.0.1 message log + +[Timer] +OnCalendar=hourly +RandomizedDelaySec=300 +Persistent=true +Unit=openbitfun-relay-1.0.1-prune.service + +[Install] +WantedBy=timers.target +``` + +`systemctl enable --now openbitfun-relay-1.0.1-prune.timer`. The steady-state run +is cheap: hourly traffic moves past the window in seconds, versus 455 seconds +for the first pass over 2.17 M rows. Pruning and retiring are separate steps — +pruning only buys time until the version reaches the retirement switch above. diff --git a/deploy/relay-v1/prune-legacy-message-log.py b/deploy/relay-v1/prune-legacy-message-log.py new file mode 100644 index 0000000000..51e3d5d54e --- /dev/null +++ b/deploy/relay-v1/prune-legacy-message-log.py @@ -0,0 +1,212 @@ +#!/usr/bin/env python3 +"""Online pruner for the pre-1.0.2 relay realtime message log. + +A 1.0.x Relay persisted an unbounded, never-pruned message log (the retired +`realtime/store.rs` tracked `log_bytes` but enforced no ceiling), so its +`relay.db` grew until the disk filled. 1.0.2 stores no session content and drops +those tables at startup; until a legacy version is retired, this tool keeps its +log bounded by hand. See README.md in this directory for the operational +procedure and for what a legacy database may lose without affecting users. + +How it stays invisible to users: + - WAL plus batched, committed deletes of primary-key ranges: each statement + holds the single write lock for milliseconds, far below the Relay's + `busy_timeout` of 5 seconds. + - nothing newer than --safety-minutes is deleted; + - every session always keeps its newest --min-keep messages; + - older rows go only beyond --keep-hours or --max-msgs; + - `realtime_sessions`, both sequence counters and every other table stay + untouched, so existing client cursors and session ids remain valid. + +Usage: + prune-legacy-message-log.py # dry run (the default) + prune-legacy-message-log.py --apply # delete + prune-legacy-message-log.py --report # current size and statistics +""" +import argparse +import fcntl +import os +import sqlite3 +import sys +import time + +DEFAULT_DB = "/srv/openbitfun-relay-v1.0.1/data/relay.db" +LOCK_FILE = "/var/lock/openbitfun-relay-1.0.1-prune.lock" + + +def human(n): + n = float(n) + for unit in ("B", "KiB", "MiB", "GiB", "TiB"): + if abs(n) < 1024 or unit == "TiB": + return f"{int(n)} B" if unit == "B" else f"{n:.2f} {unit}" + n /= 1024 + + +def connect(db, readonly=False): + if readonly: + con = sqlite3.connect(f"file:{db}?mode=ro", uri=True, timeout=10.0) + con.execute("PRAGMA query_only=ON") + else: + con = sqlite3.connect(db, timeout=10.0, isolation_level=None) + con.execute("PRAGMA busy_timeout=8000") + return con + + +def boundary_seq(cur, account, sid, lo, hi, cutoff_ms): + """Smallest seq in [lo,hi] whose created_at >= cutoff_ms (hi+1 if none). + + created_at is monotonic in seq because the log is append-only, so a binary + search over the primary key needs ~log2(n) single-row probes instead of a + full walk over the session. + """ + probe = cur.connection.cursor() # own cursor: never reset the caller's iteration + lo, hi = lo, hi + 1 + while lo < hi: + mid = (lo + hi) // 2 + row = probe.execute( + "SELECT created_at FROM realtime_messages WHERE account_id=? AND session_id=? AND seq=?", + (account, sid, mid), + ).fetchone() + if row is None: + # Missing seq (already pruned): treat as older, keep searching right. + lo = mid + 1 + elif row[0] >= cutoff_ms: + hi = mid + else: + lo = mid + 1 + return lo + + +def plan(cur, now_ms, keep_hours, safety_minutes, max_msgs, min_keep): + cutoff_ms = now_ms - int(keep_hours * 3600 * 1000) + safety_ms = now_ms - int(safety_minutes * 60 * 1000) + plan_rows = [] + sessions = cur.execute( + "SELECT account_id, id, seq, updated_at FROM realtime_sessions WHERE seq>0" + ).fetchall() + for account, sid, seq, updated in sessions: + if seq <= min_keep: + continue # everything sits inside the always-keep tail + # Deletion ceiling: never cross the always-keep tail, the retention + # window, or the safety line. + keep_seq = min( + seq - min_keep + 1, + boundary_seq(cur, account, sid, 1, seq, cutoff_ms), + boundary_seq(cur, account, sid, 1, seq, safety_ms), + ) + # Size ceiling: the newest max_msgs always survive, even inside the window. + keep_seq = max(keep_seq, seq - max_msgs + 1, 1) + if keep_seq > 1: + plan_rows.append((account, sid, keep_seq, keep_seq - 1)) + return plan_rows + + +def report(args): + con = connect(args.db, readonly=True) + cur = con.cursor() + print(f"file {human(os.path.getsize(args.db))}") + print(f"messages {cur.execute('SELECT count(*) FROM realtime_messages').fetchone()[0]:,} rows") + print(f"sessions {cur.execute('SELECT count(*) FROM realtime_sessions').fetchone()[0]:,}") + freelist = cur.execute("PRAGMA freelist_count").fetchone()[0] + print(f"freelist {freelist:,} pages ({human(freelist * 4096)})") + print(f"auto_vacuum {cur.execute('PRAGMA auto_vacuum').fetchone()[0]} (0=none, 2=incremental)") + con.close() + + +def run(args): + con = connect(args.db, readonly=not args.apply) + cur = con.cursor() + now_ms = int(time.time() * 1000) + started = time.time() + plan_rows = plan(cur, now_ms, args.keep_hours, args.safety_minutes, args.max_msgs, args.min_keep) + total = sum(p[3] for p in plan_rows) + print(f"plan: {total:,} rows in {len(plan_rows):,} sessions " + f"(keep_hours={args.keep_hours}, safety={args.safety_minutes}min, " + f"max_msgs={args.max_msgs}, min_keep={args.min_keep}) " + f"[{time.time()-started:.0f}s]") + if not args.apply: + print("\nlargest deletions:") + for account, sid, keep_seq, count in sorted(plan_rows, key=lambda p: -p[3])[:8]: + print(f" {account:>12} {sid[:16]:>16} delete {count:>7,} rows (keep seq >= {keep_seq})") + pages = cur.execute("PRAGMA page_count").fetchone()[0] + print(f"\nDRY RUN: nothing changed. file={human(os.path.getsize(args.db))} pages={pages:,}") + con.close() + return 0 + deleted = 0 + for account, sid, keep_seq, _count in plan_rows: + # A session whose lowest surviving seq already sits at the ceiling has + # nothing left in the delete range: skip it instead of issuing empty + # range deletes that would still take the write lock. + lowest = cur.execute( + "SELECT min(seq) FROM realtime_messages WHERE account_id=? AND session_id=?", + (account, sid), + ).fetchone()[0] + if lowest is None or lowest >= keep_seq: + continue + before = con.total_changes + start = lowest + while start < keep_seq: + end = min(start + args.batch_rows - 1, keep_seq - 1) + cur.execute( + "DELETE FROM realtime_messages WHERE account_id=? AND session_id=? " + "AND seq>=? AND seq<=?", + (account, sid, start, end), + ) + start = end + 1 + if args.sleep_ms: + time.sleep(args.sleep_ms / 1000.0) + # Report rows that were really removed, never the planned range length. + real = con.total_changes - before + deleted += real + if real: + print(f" {account}/{sid[:12]}: -{real:,} rows (keep seq >= {keep_seq}) " + f"[{time.time()-started:.0f}s]") + sys.stdout.flush() + cur.execute("PRAGMA wal_checkpoint(PASSIVE)") + auto = cur.execute("PRAGMA auto_vacuum").fetchone()[0] + if auto == 2 and args.incremental_vacuum_pages: + # Only an incremental-auto-vacuum database returns freed pages online; + # without it the file keeps its size and reuses free pages instead. + cur.execute(f"PRAGMA incremental_vacuum({args.incremental_vacuum_pages})") + con.close() + print(f"deleted {deleted:,} rows (real) in {time.time()-started:.0f}s") + print(f"file now {human(os.path.getsize(args.db))}, " + f"wal {human(os.path.getsize(args.db + '-wal'))}, auto_vacuum={auto}") + return 0 + + +def main(): + ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + ap.add_argument("--db", default=DEFAULT_DB, help="legacy relay.db to prune") + ap.add_argument("--keep-hours", type=float, default=24.0, + help="delete only messages older than this window") + ap.add_argument("--safety-minutes", type=float, default=60.0, + help="never delete anything newer than this") + ap.add_argument("--max-msgs", type=int, default=5000, + help="newest messages retained per session") + ap.add_argument("--min-keep", type=int, default=200, + help="newest messages kept per session regardless of age") + ap.add_argument("--batch-rows", type=int, default=2000, + help="rows per committed delete statement") + ap.add_argument("--sleep-ms", type=int, default=20, + help="pause between statements so the relay keeps the write lock") + ap.add_argument("--incremental-vacuum-pages", type=int, default=2000, + help="pages returned to the filesystem per run when auto_vacuum=2") + ap.add_argument("--dry-run", action="store_true", help="analyse only (the default)") + ap.add_argument("--apply", action="store_true", help="really delete") + ap.add_argument("--report", action="store_true", help="print current size and statistics") + args = ap.parse_args() + if args.report: + return report(args) + if args.apply: + lock = open(LOCK_FILE, "w") + try: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + print("another prune is already running; exiting") + return 0 + return run(args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/apps/relay-server/README.md b/src/apps/relay-server/README.md index 67b86418f0..231e9146ab 100644 --- a/src/apps/relay-server/README.md +++ b/src/apps/relay-server/README.md @@ -110,25 +110,47 @@ They are implemented in the shared Relay service, not in the agent loop. | Authentication request body | 16 KiB; oversized bodies return 413 | | Buffered HTTP request bodies | 512 MiB total reserved before buffering; overload returns 503 | | Concurrent HTTP API requests | 2,048; overload returns 503 | -| Body read / device RPC handler | 15 seconds / 130 seconds | -| HTTP request rate | 6,000/minute per source IP; device APIs also per account; overload returns 429 | +| Body read / API handler | 15 seconds / 130 seconds | +| HTTP request rate | 6,000/minute per source IP, and per account on the device API; overload returns 429 | | GitHub authorization start / poll | 10 / 120 per minute per IP | -| Identity exchanges | 10/minute per IP; 64 concurrent outbound identity requests | -| WebSocket upgrades | 120/minute per IP; 4,096 active sockets globally | -| WebSocket authentication | Must complete within 10 seconds | -| WebSocket ingress | 16 KiB per message/frame; 4 KiB read buffer per connection | -| WebSocket messages | 12,000/minute per connection | -| WebSocket outgoing queue | 128 messages per socket, 256 MiB total queued/writing bytes | -| Slow WebSocket writes | Close after a 15-second write timeout | -| RPC response memory | 256 MiB covering queued payloads and serialized replies, retained until read or disconnect | -| Pending device RPCs | 2,048 globally; 64 per account; cancellation releases capacity | +| Identity verification | 10 attempts/minute per IP; 3-second connect and 5-second response timeout; 64 concurrent outbound requests | +| WebSocket connections | 4,096 active sockets per Relay process; a further handshake is rejected with `connection capacity exceeded` | +| WebSocket connect / heartbeat | Namespace connect must finish within 15 seconds; ping every 15 seconds with a 45-second pong timeout; 30-second acknowledgement lifetime | +| WebSocket frames | 256 KiB per message; larger encrypted bodies use the short-lived HTTP payload lane | +| WebSocket outgoing queue | 128 queued messages per socket, plus 256 MiB of outbound message memory server-wide; a frame that cannot be queued within 2 seconds is dropped | +| RPC memory budgets | 64 MiB server-wide and 16 MiB per account, reserved against the estimated serialized size of in-flight calls; an exhausted budget answers `server RPC memory budget busy` or `account RPC memory budget busy` instead of queueing | +| Pending device RPCs | 2,048 globally, 64 per account, with a 256 MiB response budget: legacy HTTP-to-WebSocket bridge limits, kept in the shared crate but not driven by any current server route | | Registered devices / active credentials | 64 / 256 per account; database-atomic admission | | Device RPC ciphertext | 48 MiB, with JSON envelope allowance | +| Published Pages | 100 MiB per page, 10 MiB per file, 4,096 files per page; 1 GiB content-addressed asset store and 256 MiB in-memory asset volume by default | +| Page data | 4 MiB per blob, 2,048 blobs and 64 MiB mutable bytes per page, 10,000 blobs and 256 MiB mutable bytes per account; 20 MiB per page database, 1,000 rows and 2 MiB per query | +| Page functions | 64 concurrent workers globally, 16 per user, 8 per page; 3,000 requests/minute per user and 600 per page; 1 MiB request body | Existing devices can reconnect at the registration limit. Idempotent token replays remain valid at the credential limit. Limits never delete a user's session, device, workspace, or other product data. +Every value in this table is a compile-time constant of the shared service. +Only deployment-level knobs are configurable: the `RELAY_*` settings below, the +container's CPU, memory, PID and file-descriptor limits, and the reverse proxy's +`limit_req`, `limit_conn`, body size and timeouts. A deployment that quotes a +larger limit than its container or proxy allows will fail at the smaller one. + +### Scaling + +One Relay process owns its sockets, rooms, presence, payload store, asset store +and SQLite database in-process, and the Socket.IO build has no external adapter. +A second instance cannot route to the first instance's sockets, so horizontal +scaling needs per-account stickiness at the proxy; the 4,096-socket ceiling is +per process, not per deployment. + +Two paths cost work per connection rather than per message. Connect and +disconnect pass one global gate that is held across the device row's SQLite +round trips, and every open socket re-validates its credential every five +seconds. Both are linear in concurrent sockets, and they matter when a whole +fleet reconnects at once — a Relay restart is the event that produces that +load, so size and test against a mass reconnect, not only steady traffic. + Bearer authentication precedes body buffering on device APIs. Device discovery, public-key lookup, message routing, and RPC correlation all enforce account ownership. Delegated controller credentials cannot register sockets, mint more @@ -325,7 +347,10 @@ Compatibility on the same connection: - On start-up the service drops the retired `realtime_sessions`, `realtime_messages` and `realtime_account_sequence` tables from an existing database and runs `VACUUM`, so previously stored ciphertext is removed from - disk without an operator step. + disk without an operator step. A version that is still serving clients cannot + be patched after the fact: bound its growing log by hand until it reaches the + retirement switch, as described in the + [v1 deployment guide](../../../deploy/relay-v1/README.md#keeping-a-legacy-10x-deployment-alive). The old `/ws`, HTTP device `rpc` and `messages` routes are retired. Deploy the new client and server together under a separate versioned relay prefix.