diff --git a/load-test/redis-adapter-throughput.js b/load-test/redis-adapter-throughput.js new file mode 100644 index 0000000..0995819 --- /dev/null +++ b/load-test/redis-adapter-throughput.js @@ -0,0 +1,452 @@ +/** + * redis-adapter-throughput.js + * + * CrowdStream — Redis adapter (cross-pod pub/sub) throughput load test. + * + * Purpose: stress the @socket.io/redis-adapter cross-pod fan-out path and + * expose Redis pub/sub ops + CPU as the scaling bottleneck. Many receivers are + * spread round-robin across >=2 pod URLs that share one (sharded) Redis, so + * every chat broadcast MUST traverse Redis pub/sub to reach receivers on other + * pods. Senders (a subset of receivers) emit chat:message at a fixed rate; each + * body carries an in-payload send timestamp so receivers can compute fan-out + * latency against the shared process clock. + * + * Style matches load-test/signaling-latency.js (socket.io-client, arg(), + * percentile(), summarize(), ackWithTimeout()). + * + * SECURITY: never hardcode secrets. The JWT is passed via --token at runtime + * and sent as the accessToken cookie, exactly like the signaling test. + * + * 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 = arg("urls", null); +const URL_SINGLE = arg("url", null); +const TOKEN = arg("token", null); +const ROOM_ID = arg("room", null); +const NUM_RECEIVERS = parseInt(arg("receivers", "300"), 10); +let NUM_SENDERS = parseInt(arg("senders", "30"), 10); +const RATE = parseFloat(arg("rate", "5")); +const DURATION_MS = parseInt(arg("durationMs", "30000"), 10); +const RAMP_MS = parseInt(arg("rampMs", "10000"), 10); +const ACK_TIMEOUT_MS = parseInt(arg("timeout", "8000"), 10); + +// A per-run tag so receivers only count THIS run's broadcasts and ignore any +// unrelated chat traffic already flowing in the live room. +const RUN_ID = Math.random().toString(36).slice(2, 8); + +// ---------- URL resolution ---------- +function resolveUrls() { + if (URLS_ARG) { + return URLS_ARG.split(",") + .map((s) => s.trim()) + .filter(Boolean); + } + if (URL_SINGLE) { + return [URL_SINGLE.trim()]; + } + return []; +} + +const URLS = resolveUrls(); + +// ---------- arg validation ---------- +const argErrors = []; + +if (URLS.length === 0) { + argErrors.push( + "Missing --urls (or --url ). Provide >=2 pod " + + "URLs sharing one Redis to exercise cross-pod fan-out." + ); +} + +if (!TOKEN) { + argErrors.push( + "Missing --token . The socket JWT middleware rejects " + + "unauthenticated connections." + ); +} + +if (!ROOM_ID) { + argErrors.push( + "Missing --room . Must be a live room; senders join it to " + + "broadcast chat. roomId must be a UUID." + ); +} + +if (!Number.isFinite(NUM_RECEIVERS) || NUM_RECEIVERS < 1) { + argErrors.push("--receivers must be a positive integer."); +} + +if (!Number.isFinite(NUM_SENDERS) || NUM_SENDERS < 1) { + argErrors.push("--senders must be a positive integer."); +} + +if (!Number.isFinite(RATE) || RATE <= 0) { + argErrors.push("--rate must be a positive number (messages/sec per sender)."); +} + +if (!Number.isFinite(DURATION_MS) || DURATION_MS < 1) { + argErrors.push("--durationMs must be a positive integer."); +} + +if (!Number.isFinite(RAMP_MS) || RAMP_MS < 0) { + argErrors.push("--rampMs must be a non-negative integer."); +} + +if (argErrors.length > 0) { + console.error( + "Argument errors:\n" + argErrors.map((e) => ` - ${e}`).join("\n") + ); + process.exit(1); +} + +// ---------- warnings (non-fatal) ---------- +if (NUM_SENDERS > NUM_RECEIVERS) { + console.warn( + `--senders (${NUM_SENDERS}) > --receivers (${NUM_RECEIVERS}); clamping ` + + `senders to ${NUM_RECEIVERS} (senders are a subset of receivers).` + ); + NUM_SENDERS = NUM_RECEIVERS; +} + +if (URLS.length === 1) { + console.warn( + "WARNING: only one pod URL supplied — this does NOT exercise cross-pod " + + "Redis fan-out. Pass --urls with >=2 pod URLs that share the same " + + "Redis for a real adapter test." + ); +} + +// ---------- percentile ---------- +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(8)} ` + + `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)); +} + +// ---------- shared state ---------- +const state = { + connectErrors: [], + joinErrors: [], + rateLimited: 0, + moderated: 0, + totalSent: 0, + totalDeliveries: 0, + fanOutLatencyMs: [], + firstDeliveryAt: null, + lastDeliveryAt: null, +}; + +// Only count deliveries that belong to THIS run and carry our send timestamp. +// Body format: cslt|--| +function onChatMessage(payload) { + if (payload?.roomId && payload.roomId !== ROOM_ID) return; + + const body = payload?.message; + if (typeof body !== "string" || !body.startsWith("cslt|")) return; + + const parts = body.split("|"); + if (parts.length < 3) return; + + const nonce = parts[1]; + if (!nonce.startsWith(`${RUN_ID}-`)) return; + + const sentAt = Number(parts[2]); + if (!Number.isFinite(sentAt)) return; + + const now = Date.now(); + + state.totalDeliveries += 1; + state.fanOutLatencyMs.push(now - sentAt); + + if (state.firstDeliveryAt === null) state.firstDeliveryAt = now; + state.lastDeliveryAt = now; +} + +// ---------- receiver: connect (round-robin) + joinRoom + listen ---------- +async function connectReceiver(idx) { + const url = URLS[idx % URLS.length]; + + const socket = io(url, { + transports: ["websocket"], + reconnection: false, + forceNew: true, + extraHeaders: { + cookie: `accessToken=${TOKEN}`, + }, + }); + + // Attach listeners immediately so we never miss an early broadcast. + socket.on("chat:message", onChatMessage); + socket.on("chat:rateLimited", () => { + state.rateLimited += 1; + }); + socket.on("chat:moderated", () => { + state.moderated += 1; + }); + + try { + // 1. CONNECT + 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); + }); + }); + + // 2. JOIN ROOM (required to send/receive that room's chat) + const { response } = await ackWithTimeout(socket, "joinRoom", ROOM_ID); + + if (!response?.success) { + throw new Error(`joinRoom failed: ${response?.code || "unknown error"}`); + } + + return socket; + } catch (err) { + socket.disconnect(); + throw err; + } +} + +// ---------- sender: emit chat:message (NO ack) at --rate for --durationMs ---------- +function startSender(socket, senderIdx) { + let seq = 0; + const intervalMs = 1000 / RATE; + + return setInterval(() => { + seq += 1; + + // cslt|--| (1..500 chars, well under cap) + const message = `cslt|${RUN_ID}-${senderIdx}-${seq}|${Date.now()}`; + + socket.emit("chat:message", { roomId: ROOM_ID, message }); + state.totalSent += 1; + }, intervalMs); +} + +// ---------- report ---------- +function report(senderCount, joinedReceivers) { + console.log("\n=== Redis adapter cross-pod fan-out results ==="); + + summarize("fan-out latency (recv-sent)", state.fanOutLatencyMs); + + const sendWindowSec = DURATION_MS / 1000; + const expected = state.totalSent * joinedReceivers; // sent × receivers + const deliveryPct = + expected > 0 ? (state.totalDeliveries / expected) * 100 : 0; + + const spanMs = + state.firstDeliveryAt !== null && state.lastDeliveryAt !== null + ? Math.max(1, state.lastDeliveryAt - state.firstDeliveryAt) + : null; + + const broadcastRate = spanMs + ? state.totalDeliveries / (spanMs / 1000) + : 0; + + const sendRate = sendWindowSec > 0 ? state.totalSent / sendWindowSec : 0; + + console.log(""); + console.log(`pods (Redis fan-out spread): ${URLS.length}`); + console.log( + `receivers: ${joinedReceivers} joined / ` + + `${NUM_RECEIVERS} requested` + ); + console.log(`senders: ${senderCount} @ ${RATE} msg/s`); + console.log(`messages sent (emit count): ${state.totalSent}`); + console.log(`send rate into system: ${sendRate.toFixed(1)} msg/s`); + console.log(`expected deliveries (sent×recv): ${expected}`); + console.log(`actual deliveries: ${state.totalDeliveries}`); + console.log(`delivery ratio: ${deliveryPct.toFixed(1)}%`); + console.log( + `aggregate broadcast rate: ${broadcastRate.toFixed(0)} deliveries/s` + + (spanMs ? ` over ${(spanMs / 1000).toFixed(1)}s observed span` : "") + ); + console.log( + " ^ cross-pod fan-out load Redis pub/sub carried (deliveries/sec)" + ); + console.log(""); + console.log(`rate-limited events: ${state.rateLimited}`); + console.log(`moderated events: ${state.moderated}`); + console.log(`connect errors: ${state.connectErrors.length}`); + console.log(`join errors: ${state.joinErrors.length}`); + + if (URLS.length === 1) { + console.log( + "\nNOTE: single pod — broadcasts did not traverse Redis between pods. " + + "Re-run with --urls for a real cross-pod test." + ); + } + + const allErrs = [...state.connectErrors, ...state.joinErrors]; + if (allErrs.length > 0) { + console.log("\nSample errors:"); + console.log(allErrs.slice(0, 10).join("\n")); + if (allErrs.length > 10) { + console.log(`...and ${allErrs.length - 10} more`); + } + } +} + +// ---------- main ---------- +async function main() { + console.log( + "=== CrowdStream Redis adapter (cross-pod pub/sub) throughput test ===" + ); + console.log(`pods: ${URLS.length} (${URLS.join(", ")})`); + console.log(`room: ${ROOM_ID}`); + console.log(`receivers: ${NUM_RECEIVERS} (round-robin across pods)`); + console.log(`senders: ${NUM_SENDERS} @ ${RATE} msg/s`); + console.log( + `duration: ${DURATION_MS}ms ramp: ${RAMP_MS}ms ` + + `timeout: ${ACK_TIMEOUT_MS}ms` + ); + console.log(`run id: ${RUN_ID}`); + console.log(""); + + // Phase 1 — ramped connect + join of all receivers. + const receivers = new Array(NUM_RECEIVERS).fill(null); + const delayBetween = NUM_RECEIVERS > 0 ? RAMP_MS / NUM_RECEIVERS : 0; + const connectRuns = []; + + for (let i = 0; i < NUM_RECEIVERS; i++) { + const run = connectReceiver(i) + .then((socket) => { + receivers[i] = socket; + }) + .catch((err) => { + const msg = err?.message || String(err); + if (msg.startsWith("joinRoom")) { + state.joinErrors.push(`receiver ${i}: ${msg}`); + } else { + state.connectErrors.push(`receiver ${i}: ${msg}`); + } + }); + + connectRuns.push(run); + + if (delayBetween > 0) { + await sleep(delayBetween); + } + } + + await Promise.allSettled(connectRuns); + + const connected = receivers.filter(Boolean); + + console.log( + `Connected + joined ${connected.length}/${NUM_RECEIVERS} receivers ` + + `across ${URLS.length} pod(s).` + ); + + if (connected.length === 0) { + console.error("No receivers connected/joined; aborting."); + report(0, 0); + process.exit(1); + } + + // Phase 2 — senders are the first N joined receivers. + const senderSockets = connected.slice( + 0, + Math.min(NUM_SENDERS, connected.length) + ); + + console.log( + `Starting ${senderSockets.length} senders at ${RATE} msg/s for ` + + `${DURATION_MS}ms ...` + ); + + const timers = senderSockets.map((socket, i) => startSender(socket, i)); + + await sleep(DURATION_MS); + timers.forEach((t) => clearInterval(t)); + + // Phase 3 — drain: let in-flight cross-pod broadcasts land before tallying. + console.log( + `Send window closed; draining ${ACK_TIMEOUT_MS}ms for in-flight fan-out ...` + ); + await sleep(ACK_TIMEOUT_MS); + + // Teardown. + connected.forEach((socket) => socket.disconnect()); + + report(senderSockets.length, connected.length); + + // Force exit — socket.io-client engine timers can otherwise keep the loop alive. + process.exit(0); +} + +main().catch((err) => { + console.error("\nTEST FAILED:"); + console.error(err); + process.exit(1); +}); diff --git a/load-test/results/REDIS-ADAPTER-THROUGHPUT.md b/load-test/results/REDIS-ADAPTER-THROUGHPUT.md new file mode 100644 index 0000000..f963275 --- /dev/null +++ b/load-test/results/REDIS-ADAPTER-THROUGHPUT.md @@ -0,0 +1,179 @@ +# Redis Adapter Cross-Pod Pub/Sub Throughput Test + +> Authored by Claude (Anthropic), via Claude Code — 2026-08-27. + +**Roadmap gap this closes:** CrowdStream scales horizontally by running many +pods behind a load balancer, with Socket.IO rooms glued together by +`@socket.io/redis-adapter` (sharded across 3 Redis ports). Every chat broadcast +to a room whose members are spread across pods must fan out through Redis +pub/sub. At high cross-pod fan-out (`receivers × senders × rate`), **Redis pub/sub +ops and CPU become the bottleneck** — and that ceiling is currently **untested**. +This test drives that path directly so we can watch Redis fall over before users +do. + +--- + +## What it tests + +- Spreads `--receivers` socket.io clients **round-robin across `>=2` pod URLs**, + each joining the same live room. +- A subset (`--senders`) emit `chat:message` at `--rate` msg/s for + `--durationMs`. Because senders and receivers sit on *different* pods, the + server cannot deliver a broadcast locally — it **must publish to Redis**, and + every other pod's adapter must receive and re-emit to its local room members. +- Each message body carries an in-payload send timestamp + per-run nonce + (`cslt|--|`). Receivers parse it and compute + **fan-out latency = `Date.now() - sentAt`**. Single process = one shared clock, + so no clock-skew correction is needed. +- Only messages tagged with this run's `runId` are counted, so unrelated chat + already flowing in the live room does not pollute the numbers. + +### Why `>=2` pods sharing one Redis is required + +If all receivers land on a single pod, Socket.IO delivers broadcasts **in-process** +and Redis is never touched for fan-out — you would be load-testing one Node event +loop, not the adapter. To exercise the adapter you need **at least two pods that +share the same Redis (sharded) cluster**, so that a broadcast originating on pod A +has to travel A → Redis → pod B to reach B's receivers. Pass every pod URL to +`--urls`; the receivers are distributed round-robin so a large fraction of every +broadcast crosses a pod boundary. If you pass only one URL the script **warns** +that it is not exercising cross-pod fan-out. + +--- + +## Prerequisites + +- **`socket.io-client`** must be resolvable. The repo's `load-test/` directory + already has it installed (`socket.io-client ^4.8.3`), so either: + - copy `redis-adapter-throughput.js` into `load-test/` and run it there, or + - point Node at that `node_modules` via `NODE_PATH` (see below). +- **A live room.** Start a broadcast first and grab its room UUID for `--room`. +- **A valid JWT** for an authenticated user, passed via `--token` (sent as the + `accessToken` cookie). Never hardcode it. +- **`>=2` pod URLs** that share the same Redis, reachable from the test host + (e.g. individual pod addresses / node-ports, bypassing sticky-session LB so you + control the distribution). +- Read access to the Redis host(s) for the sampling commands below. + +--- + +## How to run + +```bash +# Option A: from the load-test directory (socket.io-client already installed) +node redis-adapter-throughput.js \ + --urls "http://pod-a:3000,http://pod-b:3000,http://pod-c:3000" \ + --token "$CS_JWT" \ + --room "" \ + --receivers 300 --senders 30 --rate 5 --durationMs 30000 + +# Option B: run in place, borrowing the installed dependency +NODE_PATH=/home/harshit/CrowdStream/load-test/node_modules \ +node /tmp/cs-tests/redis-adapter-throughput/redis-adapter-throughput.js \ + --urls "http://pod-a:3000,http://pod-b:3000" \ + --token "$CS_JWT" --room "" +``` + +Pass the JWT via an environment variable (as above) rather than typing it inline, +so it does not leak into your shell history. + +### Arguments + +| Arg | Default | Description | +|----------------|-------------------------|-------------| +| `--urls` | *(required for x-pod)* | Comma-separated pod URLs. Receivers connect round-robin. Use `>=2` pods sharing one Redis for a real cross-pod test. | +| `--url` | — | Fallback single URL if `--urls` is omitted. One URL only → warns it is not exercising cross-pod fan-out. | +| `--token` | *(required)* | JWT for an authenticated user; sent as the `accessToken` cookie. | +| `--room` | *(required)* | Live room UUID. Receivers/senders join it; must be a UUID. | +| `--receivers` | `300` | Total receiver sockets (fan-out targets), spread across `--urls`. | +| `--senders` | `30` | How many receivers also send (subset of receivers). Clamped to `--receivers`. | +| `--rate` | `5` | Messages/sec **per sender**. | +| `--durationMs` | `30000` | Active send window. | +| `--rampMs` | `10000` | Spread receiver connect+join over this window to avoid a thundering herd. | +| `--timeout` | `8000` | Connect/ack timeout; also used as the post-send drain window for in-flight deliveries. | + +### What it reports + +Fan-out latency percentiles (p50/p90/p95/p99/max), messages sent, expected vs +actual deliveries (`sent × receivers`) and **delivery %**, the computed +**aggregate broadcast rate (deliveries/sec)** — i.e. the load Redis carried — +plus rate-limited / moderated event counts and connect / join error counts. + +--- + +## Redis-side sampling (run while the test is in flight) + +The client-side deliveries/sec is the demand; these commands show what Redis is +actually paying to serve it. Sample each Redis shard (the sharded adapter spreads +traffic across the 3 ports) once per second or so during the run: + +```bash +# Ops throughput + bytes pushed to subscribers — the core adapter cost signal. +# Watch: instantaneous_ops_per_sec, total_net_output_bytes, expired/keyspace churn. +redis-cli -h -p INFO stats + +# CPU burned by the Redis event loop serving PUBLISH/SUBSCRIBE fan-out. +# Watch: used_cpu_sys, used_cpu_user (delta per second). +redis-cli -h -p INFO cpu + +# Connected clients / subscriber channels — one adapter connection per pod. +# Watch: connected_clients, blocked_clients, pubsub_channels, pubsub_patterns. +redis-cli -h -p INFO clients + +# See the actual PUBLISH firehose. USE SPARINGLY — MONITOR itself is a heavy +# load on Redis and will distort your measurement. Sample for a second, not the +# whole run; pipe through head so you don't drown. +redis-cli -h -p MONITOR | head -n 200 +``` + +Repeat against each of the 3 sharded Redis ports; the sharded adapter distributes +channels across them, so no single port sees the full firehose. + +--- + +## Interpretation + +- **Fan-out latency and Redis CPU climb together as `receivers × senders × rate` + grows.** Low fan-out: p99 latency is a few ms and `used_cpu` barely moves. As + you scale receivers and senders, `instantaneous_ops_per_sec` and + `total_net_output_bytes` rise, Redis CPU approaches a core's ceiling, and p95/p99 + fan-out latency inflates — that inflection is the **adapter ceiling**. +- **Delivery % dropping below ~100%** while Redis CPU is saturated (or the pod + event loops are backed up) indicates the fan-out pipeline can no longer keep up + with offered load — messages are delayed past the drain window or dropped. +- **The sharded adapter spreads publish/subscribe load across the 3 Redis ports.** + If one port shows disproportionately higher ops/CPU, channel distribution is + skewed and that shard becomes the true bottleneck — worth noting separately. +- Cross-reference the client-side **aggregate broadcast rate (deliveries/sec)** + with the sum of `instantaneous_ops_per_sec` across shards: they should track + each other. A growing gap means Redis is the constraint, not the clients. + +--- + +## Caveats + +- **The chat rate limiter is active** (per-user + per-IP token buckets). Many + emitted messages will be rejected server-side and surface as `chat:rateLimited` + rather than broadcasts — so **delivery % can be well below 100% for reasons + unrelated to Redis**. Read the rate-limited count alongside delivery %. +- **A shared `--token` means a shared user bucket.** All senders authenticated + with the same JWT draw from one per-user token bucket, which throttles + aggregate send throughput hard. For a true Redis stress test use **distinct + tokens per sender** (or raise/disable the limiter in a staging environment) so + the offered load actually reaches the adapter. As-is, you are partly measuring + the rate limiter, not Redis. +- Per-IP limiting means running all clients from one host also shares an IP + bucket; distribute load generators if that becomes the binding constraint. +- Single-process design keeps one shared clock (accurate latency) but caps how + much load one machine can generate; scale out with multiple test hosts for + higher fan-out. + +--- + +## Results + +_Pending execution — not fabricated._ + +| Pods | Receivers | Senders | Rate (msg/s) | Sent | Deliveries | Delivery % | Fan-out p50 | Fan-out p95 | Fan-out p99 | Aggregate broadcast rate (del/s) | Redis ops/s (sum shards) | Redis CPU | Rate-limited | +|------|-----------|---------|--------------|------|------------|------------|-------------|-------------|-------------|----------------------------------|--------------------------|-----------|--------------| +| | | | | | | | | | | | | | |