libsql-server: live checks for the namespace store

libsql-server/scripts/namespace-live-checks/README.md

@@ -0,0 +1,97 @@
+# Namespace live checks
+
+Scripts that run a built `sqld` locally and check how the namespace store
+behaves under load, during shutdown, and across the admin API. They exist
+because the interesting failures here are races: an entry emptied under a
+request, a shutdown while requests are in flight, a destroy racing a lookup.
+Unit tests pin each race deterministically; these scripts show the same
+races against a real server, and let two binaries be compared.
+
+Nothing here touches a remote system. Everything listens on `127.0.0.1`.
+
+## Requirements
+
+`bash` 3.2 or newer, `curl`, `python3`, `lsof`. A debug build is fine and
+widens the race windows, which is what you want.
+
+```
+cargo build -p libsql-server --bin sqld
+cargo build -p libsql --example replica_sync_loop   # for replica-sigterm.sh
+```
+
+## Scripts
+
+| Script | What it does |
+|---|---|
+| `load.py` | Keep-alive clients driving `POST /v2/pipeline`; tallies status codes per phase, with an optional SIGTERM or shell action at a given time. |
+| `scenario.sh` | One focused scenario: `evict` (cache capacity far below the hot set) or `sigterm` (signal under load). Prints the tally, the namespace-store counters and the grouped store log lines. |
+| `regress.sh` | A 65-check matrix across the admin and user APIs, one primary and one primary plus replica, optionally mixed versions. Writes `results.tsv`. |
+| `compare.py` | Side-by-side diff of two `results.tsv` files. |
+| `replica-sigterm.sh` | What embedded replicas (the `libsql` crate, `libsql/examples/replica_sync_loop.rs`) see when their primary is stopped and started again: every distinct `sync()` error with the time it was first seen, and the replicator's own retry log lines. |
+| `window.sh` | How long the shutdown window (503s until the port closes) is for N loaded namespaces, each with a WAL. Compare with your clients' retry budgets. |
+
+## Comparing two binaries
+
+```
+S=libsql-server/scripts/namespace-live-checks
+$S/scenario.sh target/debug/sqld evict            # expect 0 non-200 answers
+$S/scenario.sh target/debug/sqld sigterm          # expect 503s after the signal, no 404s
+
+$S/regress.sh /path/to/old/sqld old /path/to/new/sqld
+$S/regress.sh /path/to/new/sqld new /path/to/old/sqld
+$S/compare.py namespace-live-checks-out/old/results.tsv namespace-live-checks-out/new/results.tsv old new
+
+$S/replica-sigterm.sh target/debug/sqld           # expect no "doesn't exist"; see below
+$S/window.sh target/debug/sqld 2000               # setup takes about 45 s per 1,000 namespaces
+```
+
+`regress.sh` writes server logs next to the results (`<label>/p.log`,
+`r.log`, `d.log`). When a line differs, those logs say why.
+
+## What a healthy run looks like
+
+- `scenario.sh ... evict`: every answer is 200. `libsql_server_namespace_emptied_entry`
+  is well above zero (the race happened and was retried) and
+  `libsql_server_namespace_emptied_entry_gave_up` is absent (never incremented).
+- `scenario.sh ... sigterm`: after the signal only 503 `Namespace store has shutdown`,
+  then connection errors once the port closes. No 404.
+- `regress.sh`: unknown namespaces answer 404 everywhere; deleted namespaces
+  answer 404 and their directory is gone; data survives a restart; the
+  replica catches up, proxies writes, and resumes after a primary restart.
+- `replica-sigterm.sh`: no `sync()` error says `Requested namespace doesn't
+  exist` (a server that reports an existing namespace as missing produces
+  exactly that, at the moment of the signal, and the client does not retry
+  it). What a healthy server produces instead is the replicator's own retry
+  line, `error connecting to primary. retrying. error: status: Unavailable,
+  message: "Namespace store has shutdown"`, and, if the server stays down
+  longer than the client's handshake retries, `Timeout performing handshake
+  with primary`. All replicas sync again once the server is back.
+- `window.sh`: 503 `Namespace store has shutdown` from the first millisecond
+  after the signal until the port closes, no 404. The window grows with the
+  number of loaded namespaces (a debug build on a laptop: about 0.2 to
+  0.3 ms per namespace). It is the time your clients must retry through.
+
+## Reading the output
+
+- A counter that was never incremented is not listed by `/metrics`; absent
+  means zero.
+- `conn_error` in a tally covers refused, reset and timed-out connections.
+  After a SIGTERM it is the expected tail. Elsewhere it usually means the
+  server closed the connection after an error answer.
+- The admin port is gone after a SIGTERM; `scenario.sh` reads `/metrics`
+  only when the server is still up.
+- `window.sh` creates namespaces with a few threads only. More trips sqld's
+  connection-creation throttle (`429 DbCreateTimeout`) during setup; the
+  namespace is still loaded, so the measurement is unaffected.
+- Under eviction churn the server logs `unable to remove .sentinel file`
+  for namespaces reopened while the evicted instance is still shutting
+  down. That is independent of the namespace store and shows on every
+  version.
+
+## Ports
+
+Defaults: user HTTP 18080, admin 19090, gRPC 15001; replica 18081 and
+19091. Override with `PORT`/`ADMIN` (`scenario.sh`) or `P_HTTP`, `P_ADMIN`,
+`P_GRPC`, `R_HTTP`, `R_ADMIN` (`regress.sh`). Both scripts refuse to start
+when a port is already in use, because a stale server answering the health
+check would silently invalidate the run.

libsql-server/scripts/namespace-live-checks/compare.py

@@ -0,0 +1,36 @@
+#!/usr/bin/env python3
+"""Print two regress.sh result files side by side and mark the lines that differ.
+
+Usage: compare.py <a/results.tsv> <b/results.tsv> [label-a] [label-b]
+"""
+import csv
+import sys
+
+
+def load(path):
+    out = {}
+    for row in csv.reader(open(path), delimiter="\t"):
+        if len(row) == 3:
+            out[(row[0], row[1])] = row[2]
+    return out
+
+
+def short(v, n):
+    v = v.replace('{"error":', "").replace('"}', "").replace('"', "")
+    return v[: n - 1] + "…" if len(v) > n else v
+
+
+a, b = load(sys.argv[1]), load(sys.argv[2])
+la = sys.argv[3] if len(sys.argv) > 3 else sys.argv[1]
+lb = sys.argv[4] if len(sys.argv) > 4 else sys.argv[2]
+keys = list(dict.fromkeys(list(a.keys()) + list(b.keys())))
+w = 60
+print(f"{'scenario / check':52s} | {la:{w}s} | {lb:{w}s} |")
+print("-" * 52 + "-+-" + "-" * w + "-+-" + "-" * w + "-+")
+differ = 0
+for k in keys:
+    va, vb = a.get(k, "-"), b.get(k, "-")
+    mark = "" if va == vb else " *"
+    differ += bool(mark)
+    print(f"{(k[0] + ' ' + k[1])[:52]:52s} | {short(va, w):{w}s} | {short(vb, w):{w}s} |{mark}")
+print(f"\n{differ} of {len(keys)} checks differ (marked *). Values with ids or counters are expected to differ.")

libsql-server/scripts/namespace-live-checks/load.py

@@ -0,0 +1,154 @@
+#!/usr/bin/env python3
+"""Drive a running sqld over Hrana HTTP from several keep-alive clients and
+tally the status codes it answers.
+
+Each thread loops over `POST /v2/pipeline` with one statement, selecting the
+namespace with the `x-namespace` header (`ns0`..`nsN-1`, or a fixed `--ns`).
+An optional action splits the tally into a "before" and an "after" phase:
+`--sigterm-after S --pid PID` sends SIGTERM to a server, `--at S --run CMD`
+runs a shell command. The report lists, per phase, how many responses had
+each status, when each status was first seen after the action, and a few
+distinct bodies for non-200 answers.
+
+Examples:
+  load.py --port 18080 --namespaces 6 --threads 24 --duration 20
+  load.py --port 18080 --namespaces 20 --threads 32 --duration 12 \
+          --sigterm-after 3 --pid $(pgrep -n sqld)
+  load.py --port 18080 --ns ns0 --sql "insert into t values (1)" \
+          --threads 8 --duration 4 --at 1.5 --run "curl -X DELETE ..."
+"""
+import argparse
+import collections
+import http.client
+import json
+import os
+import signal
+import subprocess
+import threading
+import time
+
+p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+p.add_argument("--port", type=int, required=True, help="sqld user HTTP port on 127.0.0.1")
+p.add_argument("--namespaces", type=int, default=1, help="cycle over ns0..nsN-1")
+p.add_argument("--ns", type=str, default=None, help="fixed namespace instead of ns0..nsN-1")
+p.add_argument("--sql", type=str, default="select 1")
+p.add_argument("--threads", type=int, default=16)
+p.add_argument("--duration", type=float, default=10)
+p.add_argument("--sigterm-after", type=float, default=None, help="seconds after start to SIGTERM --pid")
+p.add_argument("--pid", type=int, default=None)
+p.add_argument("--at", type=float, default=None, help="seconds after start to run --run")
+p.add_argument("--run", type=str, default=None, help="shell command to run at --at")
+p.add_argument("--quiet", action="store_true", help="one-line summary only")
+args = p.parse_args()
+
+BODY = json.dumps({"requests": [{"type": "execute", "stmt": {"sql": args.sql}}, {"type": "close"}]})
+HEADERS = {"content-type": "application/json"}
+
+lock = threading.Lock()
+tally = collections.defaultdict(collections.Counter)  # phase -> status -> count
+bodies = collections.defaultdict(set)  # status -> distinct bodies (truncated)
+first_seen = {}  # (phase, status) -> seconds after the action
+action_at = [None]
+stop = threading.Event()
+t0 = time.monotonic()
+
+
+def phase():
+    return "after" if action_at[0] is not None and time.monotonic() >= action_at[0] else "before"
+
+
+def since_action():
+    return None if action_at[0] is None else round(time.monotonic() - action_at[0], 3)
+
+
+def record(status, body=None):
+    ph = phase()
+    with lock:
+        tally[ph][status] += 1
+        first_seen.setdefault((ph, status), since_action())
+        if body is not None and len(bodies[status]) < 3:
+            bodies[status].add(body[:160])
+
+
+def worker(i):
+    conn = None
+    consecutive_conn_errors = 0
+    n = 0
+    while not stop.is_set():
+        ns = args.ns or f"ns{(i + n) % args.namespaces}"
+        n += 1
+        try:
+            if conn is None:
+                conn = http.client.HTTPConnection("127.0.0.1", args.port, timeout=5)
+            conn.request("POST", "/v2/pipeline", body=BODY, headers=dict(HEADERS, **{"x-namespace": ns}))
+            resp = conn.getresponse()
+            data = resp.read()
+            consecutive_conn_errors = 0
+            if resp.status == 200:
+                record(200)
+            else:
+                record(resp.status, data.decode("utf-8", "replace"))
+            if resp.getheader("connection", "").lower() == "close":
+                conn.close()
+                conn = None
+        except Exception as e:  # connection refused, reset, timeout
+            record("conn_error", type(e).__name__)
+            consecutive_conn_errors += 1
+            try:
+                if conn:
+                    conn.close()
+            except Exception:
+                pass
+            conn = None
+            # once the server is gone after the action, stop instead of spinning
+            if phase() == "after" and consecutive_conn_errors >= 20:
+                return
+            time.sleep(0.01)
+
+
+threads = [threading.Thread(target=worker, args=(i,), daemon=True) for i in range(args.threads)]
+for t in threads:
+    t.start()
+
+if args.sigterm_after is not None:
+    time.sleep(args.sigterm_after)
+    action_at[0] = time.monotonic()
+    os.kill(args.pid, signal.SIGTERM)
+    if not args.quiet:
+        print(f"sent SIGTERM to {args.pid} at t={round(action_at[0] - t0, 3)}s", flush=True)
+elif args.at is not None:
+    time.sleep(args.at)
+    action_at[0] = time.monotonic()
+    r = subprocess.run(args.run, shell=True, capture_output=True, text=True)
+    if not args.quiet:
+        print(f"ran action at t={round(action_at[0] - t0, 3)}s -> {r.stdout.strip()[:120]}", flush=True)
+
+deadline = t0 + args.duration
+while time.monotonic() < deadline and any(t.is_alive() for t in threads):
+    time.sleep(0.05)
+stop.set()
+for t in threads:
+    t.join(timeout=2)
+
+
+def fmt(counter):
+    return ", ".join(f"{k}:{v}" for k, v in sorted(counter.items(), key=lambda kv: str(kv[0])))
+
+
+if args.quiet:
+    print(" ".join(f"{ph}{{{fmt(tally[ph])}}}" for ph in ("before", "after") if tally[ph]))
+    raise SystemExit(0)
+
+print("\n=== tally (phase -> status: count) ===")
+for ph in ("before", "after"):
+    if tally[ph]:
+        print(f"{ph:7s} {fmt(tally[ph])}")
+if any(v is not None for v in first_seen.values()):
+    print("\n=== first seen (seconds after the action) ===")
+    for (ph, st), t in sorted(first_seen.items(), key=lambda kv: (kv[1] is None, kv[1] or 0)):
+        if ph == "after":
+            print(f"  {st}: +{t}s")
+print("\n=== distinct bodies for non-200 ===")
+for st, bs in bodies.items():
+    for b in bs:
+        print(f"  {st}: {b}")