docs(relay): state the enforced limits and the legacy log-pruning procedure

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.

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())