test(reliability): add connect/disconnect soak load test (Week 4)

load-test/connect-disconnect-soak.js

@@ -0,0 +1,261 @@
+/**
+ * connect-disconnect-soak.js
+ * -----------------------------------------------------------------------------
+ * Week 4 (State & reliability) — connect / disconnect soak test.
+ *
+ * Churns many Socket.IO  connect -> (optional joinRoom) -> disconnect  cycles to
+ * validate that the server releases resources on disconnect (no leak) and that
+ * clients keep connecting cleanly under sustained churn.
+ *
+ * This script drives the churn and reports CLIENT-SIDE metrics (connect / join
+ * latency, success / failure, throughput). Leak detection itself is SERVER-SIDE:
+ * capture the backend process RSS (and mediasoup worker / socket-map counts)
+ * before and after the run — they should return to ~baseline once connections
+ * settle. See  load-test/results/CONNECT-DISCONNECT-SOAK.md  for the procedure.
+ *
+ * Authored by Claude (Anthropic), via Claude Code — 2026-08-27.
+ */
+
+const { io } = require("socket.io-client");
+
+// ---------- CLI args ----------
+function arg(name, def) {
+  const i = process.argv.indexOf(`--${name}`);
+  if (i === -1) return def;
+  return process.argv[i + 1];
+}
+
+const URL = arg("url", "http://localhost:3000");
+const TOKEN = arg("token", null);
+const ROOM_ID = arg("room", null);
+const CYCLES = parseInt(arg("cycles", "1000"), 10); // total connect/disconnect cycles
+const CONCURRENCY = parseInt(arg("concurrency", "50"), 10); // parallel workers
+const HOLD_MS = parseInt(arg("holdMs", "0"), 10); // stay connected before disconnecting
+const JOIN = arg("join", "false") === "true"; // also joinRoom each cycle (exercises viewer cleanup)
+const SETTLE_MS = parseInt(arg("settleMs", "5000"), 10); // wait after churn for server cleanup
+const ACK_TIMEOUT_MS = parseInt(arg("timeout", "8000"), 10);
+const METRICS_URL = arg("metricsUrl", null); // optional health/metrics endpoint to sample
+
+if (!TOKEN) {
+  console.error(
+    "Missing --token <jwt>. The server's socket auth rejects unauthenticated connections."
+  );
+  process.exit(1);
+}
+
+if (JOIN && !ROOM_ID) {
+  console.error(
+    "--join true requires --room <roomId>. Create a room by starting a broadcast first."
+  );
+  process.exit(1);
+}
+
+// ---------- percentile / summarize (same style as signaling-latency.js) ----------
+function percentile(sortedArr, p) {
+  if (sortedArr.length === 0) return NaN;
+  const idx = Math.ceil((p / 100) * sortedArr.length) - 1;
+  return sortedArr[Math.min(Math.max(idx, 0), sortedArr.length - 1)];
+}
+
+function summarize(label, samples) {
+  const clean = samples.filter((n) => Number.isFinite(n)).sort((a, b) => a - b);
+  if (clean.length === 0) {
+    console.log(`${label}: no samples`);
+    return;
+  }
+  const avg = clean.reduce((a, b) => a + b, 0) / clean.length;
+  console.log(
+    `${label.padEnd(32)} ` +
+      `n=${clean.length.toString().padEnd(6)} ` +
+      `avg=${avg.toFixed(1)}ms  ` +
+      `p50=${percentile(clean, 50)}ms  ` +
+      `p90=${percentile(clean, 90)}ms  ` +
+      `p95=${percentile(clean, 95)}ms  ` +
+      `p99=${percentile(clean, 99)}ms  ` +
+      `max=${clean[clean.length - 1]}ms`
+  );
+}
+
+// ---------- Socket.IO ACK helper ----------
+function ackWithTimeout(socket, event, ...args) {
+  return new Promise((resolve, reject) => {
+    const timer = setTimeout(() => {
+      reject(new Error(`${event} ack timeout`));
+    }, ACK_TIMEOUT_MS);
+
+    const t0 = performance.now();
+
+    socket.emit(event, ...args, (response) => {
+      clearTimeout(timer);
+      resolve({ response, latencyMs: performance.now() - t0 });
+    });
+  });
+}
+
+function sleep(ms) {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+const results = {
+  connectMs: [],
+  joinMs: [],
+  cyclesDone: 0,
+  connectErrors: 0,
+  joinErrors: 0,
+  errors: [],
+};
+
+// ---------- one connect -> (join) -> disconnect cycle ----------
+async function oneCycle(workerId) {
+  const socket = io(URL, {
+    transports: ["websocket"],
+    reconnection: false,
+    forceNew: true,
+    extraHeaders: {
+      cookie: `accessToken=${TOKEN}`,
+    },
+  });
+
+  try {
+    const connectStart = performance.now();
+
+    await new Promise((resolve, reject) => {
+      const timer = setTimeout(
+        () => reject(new Error("connect timeout")),
+        ACK_TIMEOUT_MS
+      );
+      socket.once("connect", () => {
+        clearTimeout(timer);
+        resolve();
+      });
+      socket.once("connect_error", (err) => {
+        clearTimeout(timer);
+        reject(err);
+      });
+    });
+
+    results.connectMs.push(performance.now() - connectStart);
+
+    if (JOIN) {
+      try {
+        const { response, latencyMs } = await ackWithTimeout(
+          socket,
+          "joinRoom",
+          ROOM_ID
+        );
+        results.joinMs.push(latencyMs);
+        if (!response?.success) {
+          results.joinErrors++;
+          if (results.errors.length < 50) {
+            results.errors.push(
+              `worker ${workerId}: joinRoom failed: ${
+                response?.code || "unknown"
+              }`
+            );
+          }
+        }
+      } catch (e) {
+        results.joinErrors++;
+        if (results.errors.length < 50) {
+          results.errors.push(`worker ${workerId}: join ${e.message}`);
+        }
+      }
+    }
+
+    if (HOLD_MS > 0) await sleep(HOLD_MS);
+  } catch (err) {
+    results.connectErrors++;
+    if (results.errors.length < 50) {
+      results.errors.push(`worker ${workerId}: ${err?.message || String(err)}`);
+    }
+  } finally {
+    socket.disconnect();
+    results.cyclesDone++;
+  }
+}
+
+async function worker(workerId, cyclesForWorker) {
+  for (let i = 0; i < cyclesForWorker; i++) {
+    await oneCycle(workerId);
+  }
+}
+
+async function sampleMetrics(tag) {
+  if (!METRICS_URL) return;
+  if (typeof fetch !== "function") {
+    console.log(`[metrics @ ${tag}] fetch unavailable in this Node runtime — skipping`);
+    return;
+  }
+  try {
+    const res = await fetch(METRICS_URL, {
+      headers: { cookie: `accessToken=${TOKEN}` },
+    });
+    const text = await res.text();
+    console.log(
+      `\n[metrics @ ${tag}] ${METRICS_URL} -> ${res.status}\n${text.slice(0, 2000)}`
+    );
+  } catch (e) {
+    console.log(`[metrics @ ${tag}] fetch failed: ${e.message}`);
+  }
+}
+
+// ---------- main ----------
+async function main() {
+  console.log(
+    `Connect/disconnect soak: cycles=${CYCLES}, concurrency=${CONCURRENCY}, ` +
+      `join=${JOIN}, holdMs=${HOLD_MS}, url=${URL}`
+  );
+  console.log(
+    "Reminder: capture backend RSS + mediasoup worker / socket-map counts NOW (baseline). See the README."
+  );
+
+  await sampleMetrics("start");
+
+  const start = performance.now();
+
+  // distribute cycles as evenly as possible across the worker pool
+  const base = Math.floor(CYCLES / CONCURRENCY);
+  const extra = CYCLES % CONCURRENCY;
+  const workers = [];
+  for (let w = 0; w < CONCURRENCY; w++) {
+    const c = base + (w < extra ? 1 : 0);
+    if (c > 0) workers.push(worker(w, c));
+  }
+  await Promise.allSettled(workers);
+
+  const elapsedSec = (performance.now() - start) / 1000;
+
+  console.log(`\nSettling ${SETTLE_MS}ms so the server can finish disconnect cleanup...`);
+  await sleep(SETTLE_MS);
+  await sampleMetrics("end");
+
+  console.log("\n=== Results ===");
+  summarize("socket connect", results.connectMs);
+  if (JOIN) summarize("joinRoom ack", results.joinMs);
+  console.log(`cycles completed:      ${results.cyclesDone}/${CYCLES}`);
+  console.log(`connect errors:        ${results.connectErrors}`);
+  if (JOIN) console.log(`join errors:           ${results.joinErrors}`);
+  console.log(
+    `throughput:            ${(results.cyclesDone / elapsedSec).toFixed(1)} cycles/sec ` +
+      `over ${elapsedSec.toFixed(1)}s`
+  );
+
+  console.log(
+    "\nLEAK CHECK: compare backend RSS + mediasoup worker / socket-map counts to your baseline."
+  );
+  console.log(
+    "They should return to ~baseline after the settle window. A monotonic climb across repeated runs indicates a leak."
+  );
+
+  if (results.errors.length > 0) {
+    console.log(`\nSample errors (${results.errors.length}):`);
+    console.log(results.errors.slice(0, 10).join("\n"));
+    if (results.errors.length > 10) {
+      console.log(`...and ${results.errors.length - 10} more`);
+    }
+  }
+
+  process.exit(0);
+}
+
+main();

load-test/results/CONNECT-DISCONNECT-SOAK.md

@@ -0,0 +1,119 @@
+# Connect / Disconnect Soak Test
+
+> **Authored by Claude (Anthropic), via Claude Code — 2026-08-27.**
+> The test script (`load-test/connect-disconnect-soak.js`) and this document were written by Claude.
+
+Roadmap item: **Week 4 (State & reliability) — Day 5**
+_"Connect/disconnect soak shows flat resource counts (no leak)"_ and
+_"Clean client recovery after a simulated node loss."_
+
+## What this tests
+
+The soak driver hammers the signaling server with a large number of
+connect → (optional `joinRoom`) → disconnect cycles:
+
+```
+        ┌─────────────────────────────────────────┐
+        │  repeat CYCLES times, CONCURRENCY at a   │
+        │  time:                                   │
+        │                                          │
+        │   connect (cookie: accessToken=<jwt>)    │
+        │        ↓                                 │
+        │   joinRoom(roomId)        (if --join)    │
+        │        ↓                                 │
+        │   hold HOLD_MS            (if > 0)       │
+        │        ↓                                 │
+        │   disconnect  →  server handleDisconnect │
+        └─────────────────────────────────────────┘
+```
+
+It exercises the exact server paths that must clean up on disconnect:
+`handleDisconnect(socket)`, viewer/room map removal, and (with `--join`) the
+mediasoup viewer transport/consumer teardown.
+
+**What the script measures (client-side):** connect latency, `joinRoom` ack
+latency, cycle success/failure counts, and churn throughput (cycles/sec).
+
+**What you measure (server-side):** the actual leak signal. The script cannot
+read the server's memory, so you capture the backend process RSS and resource
+counts before/after — see [Detecting a leak](#detecting-a-leak).
+
+## Prerequisites
+
+- `socket.io-client` (already in `load-test/node_modules`).
+- A valid JWT `accessToken` (the socket auth middleware rejects anonymous connections).
+- **Only if using `--join true`:** a live room id — start a broadcast, then use its room id.
+  Without `--join`, the test needs only a token and exercises the pure connection lifecycle.
+
+## How to run
+
+```bash
+cd load-test
+
+# Pure connection-lifecycle churn (no room needed): 2000 cycles, 100 in flight
+node connect-disconnect-soak.js \
+  --url http://localhost:3000 \
+  --token "<JWT>" \
+  --cycles 2000 --concurrency 100
+
+# Full churn incl. join/leave cleanup (needs a live room):
+node connect-disconnect-soak.js \
+  --url http://localhost:3000 \
+  --token "<JWT>" \
+  --room "<ROOM_UUID>" \
+  --join true --cycles 2000 --concurrency 100 --holdMs 250
+```
+
+### Arguments
+
+| Flag | Default | Meaning |
+|---|---|---|
+| `--url` | `http://localhost:3000` | Signaling server URL |
+| `--token` | _(required)_ | JWT set as `accessToken` cookie |
+| `--room` | `null` | Live room id (required when `--join true`) |
+| `--cycles` | `1000` | Total connect/disconnect cycles |
+| `--concurrency` | `50` | Cycles in flight at once |
+| `--holdMs` | `0` | Time to stay connected before disconnecting |
+| `--join` | `false` | Also `joinRoom` each cycle (exercises viewer cleanup) |
+| `--settleMs` | `5000` | Wait after churn for server cleanup before final metrics |
+| `--metricsUrl` | `null` | Optional health/metrics endpoint sampled at start & end |
+| `--timeout` | `8000` | Per-op ack/connect timeout (ms) |
+
+## Detecting a leak
+
+The pass/fail signal is **server-side resource counts returning to baseline**.
+Recommended procedure:
+
+1. **Baseline** — with the server idle, record:
+   ```bash
+   ps -o rss= -p "$(pgrep -f 'src/index.ts' | head -1)"   # RSS in KB
+   ```
+   plus, if you expose them, mediasoup worker count and the sizes of the
+   socket/room/viewer maps.
+2. **Run** the soak (e.g. `--cycles 5000 --concurrency 100 --join true`).
+3. **After the settle window**, re-record the same numbers.
+4. **Interpret:** RSS and resource counts should return to ~baseline (allowing
+   for GC lag / connection keep-alive). Repeat the run 3–5×; a **monotonic
+   climb across runs** is the leak signal — a single elevated sample is not.
+
+If you wire up a `/metrics` or health endpoint that reports live counts, pass
+`--metricsUrl` and the script prints it at start and end for a quick delta.
+
+## Simulating node loss (recovery half of the roadmap item)
+
+To validate _clean client recovery after a simulated node loss_, run a small
+number of long-lived clients (`--concurrency 5 --cycles 5 --holdMs 60000`),
+then kill one signaling pod mid-run and confirm clients reconnect and re-join
+(the frontend uses Socket.IO auto-reconnect). This script intentionally uses
+`reconnection: false` for deterministic churn accounting, so drive recovery
+separately or add `reconnection: true` for that scenario.
+
+## Results
+
+> _Pending execution against a running backend. This environment had no live
+> server (no Mongo/Redis/mediasoup), so no numbers are recorded here yet —
+> they will be filled in after a real run rather than fabricated._
+
+| Cycles | Concurrency | Join | Connect p50 | Connect p99 | Errors | RSS Δ (baseline→settled) |
+|---:|---:|:---:|---:|---:|---:|---:|
+| _tbd_ | _tbd_ | _tbd_ | _tbd_ | _tbd_ | _tbd_ | _tbd_ |