diff --git a/load-test/chat-load-fanout.js b/load-test/chat-load-fanout.js new file mode 100644 index 0000000..76e34d1 --- /dev/null +++ b/load-test/chat-load-fanout.js @@ -0,0 +1,323 @@ +/** + * chat-load-fanout.js + * ----------------------------------------------------------------------------- + * Week 5 (Real-time messaging) — chat fan-out load test. + * + * Spins up many viewers that join a room and listen for `chat:message` + * broadcasts. A subset of them send messages. Measures end-to-end fan-out + * latency (sender emit -> receiver delivery), delivery completeness, and how + * often the server's rate limiter / moderation engage under load. + * + * Cross-pod: pass --urls "http://pod1,http://pod2" to spread clients across + * signaling pods so fan-out exercises the Redis adapter (the Week 3/5 goal). + * + * Latency method: each message body embeds a nonce + the sender's wall-clock + * send time ("cslt||"). Every receiver parses it on + * delivery and computes (Date.now() - sendEpochMs). All sockets run in ONE + * process, so Date.now() is a shared clock and the number is true end-to-end + * fan-out latency (server validate + moderate + rate-limit + Redis publish + + * broadcast). `chat:message` has no ack, which is why latency is carried in + * the payload rather than measured with an ack round-trip. + * + * 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 URLS = arg("urls", arg("url", "http://localhost:3000")) + .split(",") + .map((s) => s.trim()) + .filter(Boolean); +const TOKEN = arg("token", null); +const ROOM_ID = arg("room", null); +const RECEIVERS = parseInt(arg("receivers", "200"), 10); // viewers that join + listen +const SENDERS = parseInt(arg("senders", "20"), 10); // subset that also send +const MESSAGES = parseInt(arg("messages", "10"), 10); // messages per sender +const SEND_INTERVAL_MS = parseInt(arg("sendIntervalMs", "1500"), 10); // spacing per sender (respects rate limit) +const RAMP_MS = parseInt(arg("rampMs", "10000"), 10); // ramp for connect+join +const ACK_TIMEOUT_MS = parseInt(arg("timeout", "8000"), 10); +const DRAIN_MS = parseInt(arg("drainMs", "5000"), 10); // wait for in-flight broadcasts after last send + +if (!TOKEN) { + console.error( + "Missing --token . The server's socket auth rejects unauthenticated connections." + ); + process.exit(1); +} +if (!ROOM_ID) { + console.error( + "Missing --room . Start a broadcast to create a room, then pass its id." + ); + process.exit(1); +} +if (SENDERS > RECEIVERS) { + console.error( + `--senders (${SENDERS}) cannot exceed --receivers (${RECEIVERS}); senders are a subset of receivers.` + ); + 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 = { + joinMs: [], + fanoutMs: [], // one sample per (message x receiver) delivery + sent: 0, + delivered: 0, // total deliveries observed during the send phase + rateLimited: 0, + moderated: 0, + connectErrors: 0, + joinErrors: 0, + errors: [], +}; + +const nonceDeliveries = new Map(); // nonce -> how many receivers got it + +let seq = 0; +function makeNonce(senderIdx, msgIdx) { + return `s${senderIdx}-m${msgIdx}-${seq++}`; +} + +// ---------- connect a receiver and join the room ---------- +async function connectAndJoin(idx) { + const url = URLS[idx % URLS.length]; + const socket = io(url, { + transports: ["websocket"], + reconnection: false, + forceNew: true, + extraHeaders: { + cookie: `accessToken=${TOKEN}`, + }, + }); + + // Every joined socket is a receiver: measure fan-out latency + delivery count. + socket.on("chat:message", (msg) => { + results.delivered++; + const body = typeof msg?.message === "string" ? msg.message : ""; + if (body.startsWith("cslt|")) { + const parts = body.split("|"); + const nonce = parts[1]; + const sentAt = Number(parts[2]); + if (Number.isFinite(sentAt)) { + results.fanoutMs.push(Date.now() - sentAt); + } + if (nonce) { + nonceDeliveries.set(nonce, (nonceDeliveries.get(nonce) || 0) + 1); + } + } + }); + + try { + 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); + }); + }); + + 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( + `client ${idx}: joinRoom failed: ${response?.code || "unknown"}` + ); + } + socket.disconnect(); + return null; + } + + return socket; + } catch (err) { + results.connectErrors++; + if (results.errors.length < 50) { + results.errors.push(`client ${idx}: ${err?.message || String(err)}`); + } + socket.disconnect(); + return null; + } +} + +// ---------- sender loop ---------- +async function runSender(socket, senderIdx) { + socket.on("chat:rateLimited", () => { + results.rateLimited++; + }); + socket.on("chat:moderated", () => { + results.moderated++; + }); + + for (let m = 0; m < MESSAGES; m++) { + const nonce = makeNonce(senderIdx, m); + const body = `cslt|${nonce}|${Date.now()}`; + socket.emit("chat:message", { roomId: ROOM_ID, message: body }); + results.sent++; + if (m < MESSAGES - 1) await sleep(SEND_INTERVAL_MS); + } +} + +// ---------- main ---------- +async function main() { + console.log( + `Chat fan-out load test: receivers=${RECEIVERS}, senders=${SENDERS}, ` + + `messages/sender=${MESSAGES}, sendInterval=${SEND_INTERVAL_MS}ms` + ); + console.log(`pods=${URLS.length} (${URLS.join(", ")}), room=${ROOM_ID}`); + + // Ramp connect + join. + const delayBetween = RECEIVERS > 0 ? RAMP_MS / RECEIVERS : 0; + const conns = []; + for (let i = 0; i < RECEIVERS; i++) { + conns.push(connectAndJoin(i)); + if (delayBetween > 0) await sleep(delayBetween); + } + const sockets = await Promise.all(conns); + const joined = sockets.filter(Boolean); + console.log( + `\nJoined ${joined.length}/${RECEIVERS} receivers. ` + + `Starting ${Math.min(SENDERS, joined.length)} senders...` + ); + + if (joined.length === 0) { + console.error( + "No receivers joined — is the room live and the token valid? Aborting." + ); + process.exit(1); + } + + // Only count deliveries for the send phase (ignore anything before senders start). + results.delivered = 0; + results.fanoutMs.length = 0; + nonceDeliveries.clear(); + + const senderCount = Math.min(SENDERS, joined.length); + const senderRuns = []; + for (let s = 0; s < senderCount; s++) { + senderRuns.push(runSender(joined[s], s)); + } + await Promise.all(senderRuns); + + console.log( + `\nAll messages sent (${results.sent}). Draining ${DRAIN_MS}ms for delivery...` + ); + await sleep(DRAIN_MS); + + // Delivery completeness. Each message that passes the server gate should + // reach every joined receiver. Rate-limited / moderated messages never + // broadcast, so they simply won't appear in nonceDeliveries. + const totalExpected = results.sent * joined.length; + const totalReceived = results.delivered; + let fullyDelivered = 0; + for (const [, count] of nonceDeliveries) { + if (count >= joined.length) fullyDelivered++; + } + + console.log("\n=== Results ==="); + summarize("joinRoom ack", results.joinMs); + summarize("chat fan-out (emit->deliver)", results.fanoutMs); + console.log(`messages sent: ${results.sent}`); + console.log(`receivers joined: ${joined.length}`); + console.log( + `deliveries received: ${totalReceived} / ${totalExpected} expected ` + + `(${totalExpected ? ((100 * totalReceived) / totalExpected).toFixed(1) : "0"}%)` + ); + console.log( + `messages broadcast: ${nonceDeliveries.size} / ${results.sent} sent ` + + `(rest blocked by rate-limit/moderation)` + ); + console.log( + `fully fanned-out: ${fullyDelivered} / ${nonceDeliveries.size} broadcast messages reached all receivers` + ); + console.log(`rate-limited events: ${results.rateLimited}`); + console.log(`moderated events: ${results.moderated}`); + console.log(`connect errors: ${results.connectErrors}`); + console.log(`join errors: ${results.joinErrors}`); + + if (results.rateLimited > 0) { + console.log( + `\nNote: ${results.rateLimited} messages were rate-limited by the server (expected under load).` + ); + console.log( + "Raise --sendIntervalMs or lower --senders/--messages to stay under the token bucket," + ); + console.log("or relax the limit server-side for a pure throughput test."); + } + + 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`); + } + } + + for (const s of joined) s.disconnect(); + process.exit(0); +} + +main(); diff --git a/load-test/results/CHAT-LOAD.md b/load-test/results/CHAT-LOAD.md new file mode 100644 index 0000000..7a66bf3 --- /dev/null +++ b/load-test/results/CHAT-LOAD.md @@ -0,0 +1,110 @@ +# Chat Fan-out Load Test + +> **Authored by Claude (Anthropic), via Claude Code — 2026-08-27.** +> The test script (`load-test/chat-load-fanout.js`) and this document were written by Claude. + +Roadmap item: **Week 5 (Real-time messaging) — Day 5** +_"Load test chat: many senders across ≥2 pods; confirm fan-out latency stays +acceptable"_ and _"Messaging stays bounded under load."_ + +## What this tests + +Many viewers join one room and listen for `chat:message` broadcasts; a subset +also send messages. The test measures how fast and how completely messages +fan out to everyone — optionally across multiple pods via the Redis adapter. + +``` + senders ──emit chat:message {roomId, message}──► pod + │ validate → moderate → rate-limit + │ publishMessage() → Redis pub/sub + ▼ + io.to(room:).emit("chat:message") ──► every receiver + (fans out across ALL pods via @socket.io/redis-adapter) +``` + +### How latency is measured + +`chat:message` is fire-and-forget (no ack), so latency is carried **in the +payload**. Each message body is `cslt||`. On delivery every +receiver computes `Date.now() - sendEpochMs`. All sockets live in one Node +process, so the clock is shared and the result is true end-to-end fan-out +latency: server validate + moderate + rate-limit + Redis publish + broadcast. + +### Metrics reported + +- **chat fan-out (emit→deliver)** — latency percentiles across every delivery. +- **deliveries received / expected** — completeness (`sent × receivers`). +- **messages broadcast / sent** — how many passed the server gate (the rest + were rate-limited or moderated and never broadcast). +- **fully fanned-out** — messages that reached *all* receivers (cross-pod check). +- **rate-limited / moderated events** — how often the server throttled senders. + +## Prerequisites + +- `socket.io-client` (already in `load-test/node_modules`). +- A valid JWT `accessToken`. +- A **live room** — start a broadcast, then pass its room id (viewers must + join an existing room to receive its chat). +- For the cross-pod goal: **≥2 signaling pods** behind your load balancer, + sharing Redis. Pass them via `--urls`. + +## How to run + +```bash +cd load-test + +# Single pod: 300 receivers, 30 senders, 10 msgs each, spaced 1.5s +node chat-load-fanout.js \ + --url http://localhost:3000 \ + --token "" \ + --room "" \ + --receivers 300 --senders 30 --messages 10 --sendIntervalMs 1500 + +# Cross-pod fan-out (Week 5 goal): clients spread across two pods +node chat-load-fanout.js \ + --urls "http://localhost:3001,http://localhost:3002" \ + --token "" \ + --room "" \ + --receivers 500 --senders 50 +``` + +### Arguments + +| Flag | Default | Meaning | +|---|---|---| +| `--url` | `http://localhost:3000` | Single signaling URL | +| `--urls` | _(uses `--url`)_ | Comma-separated pod URLs; clients spread round-robin | +| `--token` | _(required)_ | JWT set as `accessToken` cookie | +| `--room` | _(required)_ | Live room id | +| `--receivers` | `200` | Viewers that join and listen | +| `--senders` | `20` | Subset that also send (must be ≤ receivers) | +| `--messages` | `10` | Messages per sender | +| `--sendIntervalMs` | `1500` | Spacing between a sender's messages (respects rate limit) | +| `--rampMs` | `10000` | Ramp window for connect + join | +| `--drainMs` | `5000` | Wait after last send for in-flight broadcasts | +| `--timeout` | `8000` | Per-op ack/connect timeout (ms) | + +## Interpreting results & caveats + +- **Rate limiting is active** for `chat:message` (per-user and per-IP token + buckets). Under load some messages *will* be throttled — that's the + moderation/backpressure the Week 5 item wants to observe, not a bug. Watch + `rate-limited events` and `messages broadcast / sent`. For a pure throughput + number, raise `--sendIntervalMs` or relax the limit server-side. +- **Same token for all sockets:** by default every simulated client uses the + same JWT, so the per-user bucket is shared. To test many *distinct* users, + extend the script to accept a token list — noted as a follow-up. +- **"Bounded under load"** = fan-out latency percentiles stay flat and + `deliveries received ≈ (broadcast messages × receivers)` as you scale + receivers/senders up. A latency blow-up or delivery shortfall is the signal. + +## Results + +> _Pending execution against a running backend. This environment had no live +> server (no Mongo/Redis/mediasoup and no ≥2-pod deployment), so no numbers are +> recorded here yet — they will be filled in after a real run rather than +> fabricated._ + +| Pods | Receivers | Senders | Msgs/sender | Fan-out p50 | Fan-out p99 | Delivery % | Rate-limited | +|---:|---:|---:|---:|---:|---:|---:|---:| +| _tbd_ | _tbd_ | _tbd_ | _tbd_ | _tbd_ | _tbd_ | _tbd_ | _tbd_ |