konnosDashboard
realtime-subs.ts219 lines · main
1/**
2 * Realtime subscriptions load test — opens N concurrent WebSocket subs
3 * against the realtime service, exercises NOTIFY fan-out, reports
4 * latency + connection-failure stats.
5 *
6 * Phase 2 target: 1,000 concurrent subs on KVM4 with p99 fan-out latency
7 * under 500ms and zero connection failures.
8 *
9 * Usage:
10 * bun run infra/load-tests/realtime-subs.ts \
11 * --url ws://localhost:3004/v1/subscribe \
12 * --secret "$BRIVEN_RUNTIME_SHARED_SECRET" \
13 * --project p_01HZ... \
14 * --function poolStats \
15 * --subs 1000 \
16 * --duration 60
17 *
18 * `--ws-impl` defaults to bun's built-in WebSocket. The harness expects
19 * to be run with bun, not node.
20 *
21 * Stops after `--duration` seconds; emits a summary on stdout.
22 */
23
24interface Args {
25 url: string;
26 secret: string;
27 projectId: string;
28 functionName: string;
29 subs: number;
30 durationSec: number;
31 rampMs: number;
32}
33
34interface Stats {
35 opened: number;
36 failed: number;
37 closed: number;
38 framesReceived: number;
39 firstFrameLatencyMs: number[];
40 errors: Map<string, number>;
41}
42
43function parseArgs(argv: readonly string[]): Args {
44 const out: Partial<Args> = {};
45 for (let i = 0; i < argv.length; i++) {
46 const a = argv[i];
47 const next = argv[i + 1];
48 if (a === '--url' && next) out.url = next;
49 else if (a === '--secret' && next) out.secret = next;
50 else if (a === '--project' && next) out.projectId = next;
51 else if (a === '--function' && next) out.functionName = next;
52 else if (a === '--subs' && next) out.subs = Number(next);
53 else if (a === '--duration' && next) out.durationSec = Number(next);
54 else if (a === '--ramp' && next) out.rampMs = Number(next);
55 }
56 return {
57 url: out.url ?? 'ws://localhost:3004/v1/subscribe',
58 secret: out.secret ?? process.env.BRIVEN_RUNTIME_SHARED_SECRET ?? '',
59 projectId: out.projectId ?? '',
60 functionName: out.functionName ?? 'poolStats',
61 subs: out.subs ?? 100,
62 durationSec: out.durationSec ?? 30,
63 rampMs: out.rampMs ?? 10,
64 };
65}
66
67function help(): void {
68 process.stdout.write(`realtime-subs — load test the briven realtime service
69
70flags:
71 --url URL ws[s]:// endpoint (default: ws://localhost:3004/v1/subscribe)
72 --secret HEX runtime shared secret (default: \$BRIVEN_RUNTIME_SHARED_SECRET)
73 --project ID target project id, p_…
74 --function NAME function to subscribe to (default: poolStats)
75 --subs N concurrent subs to open (default: 100)
76 --duration SEC how long to hold open after ramp completes (default: 30)
77 --ramp MS delay between successive subscribe frames (default: 10)
78`);
79}
80
81async function openOne(
82 args: Args,
83 stats: Stats,
84 index: number,
85 signal: AbortSignal,
86): Promise<void> {
87 // Bun's WebSocket supports the `headers` option via a second-arg trick.
88 // The realtime service expects Authorization on the upgrade request.
89 // We use a custom protocol prefix to convey the bearer because the
90 // browser-style WebSocket constructor doesn't accept headers; the
91 // realtime side currently reads from the upgrade handler. Workaround:
92 // include the token in the URL as a query string, OR run this script
93 // with an env var the harness understands. We use a `?bearer=` param
94 // that the test harness can recognise — for production the dashboard
95 // SDK sends a real Authorization header during the upgrade.
96 const wsUrl = `${args.url}?bearer=${encodeURIComponent(args.secret)}`;
97 const ws = new WebSocket(wsUrl);
98 const t0 = performance.now();
99 let firstFrameSeen = false;
100
101 return new Promise((resolve) => {
102 const cleanup = (): void => {
103 try {
104 ws.close();
105 } catch {
106 /* ignore */
107 }
108 resolve();
109 };
110 signal.addEventListener('abort', cleanup, { once: true });
111
112 ws.addEventListener('open', () => {
113 stats.opened++;
114 ws.send(
115 JSON.stringify({
116 type: 'subscribe',
117 subscriptionId: `s_${index}_${Date.now()}`,
118 projectId: args.projectId,
119 functionName: args.functionName,
120 args: {},
121 }),
122 );
123 });
124 ws.addEventListener('message', (ev: MessageEvent) => {
125 stats.framesReceived++;
126 if (!firstFrameSeen) {
127 firstFrameSeen = true;
128 stats.firstFrameLatencyMs.push(performance.now() - t0);
129 }
130 const data = typeof ev.data === 'string' ? ev.data : '';
131 try {
132 const frame = JSON.parse(data) as { type?: string; code?: string };
133 if (frame.type === 'error' && frame.code) {
134 stats.errors.set(frame.code, (stats.errors.get(frame.code) ?? 0) + 1);
135 }
136 } catch {
137 /* ignore */
138 }
139 });
140 ws.addEventListener('error', () => {
141 stats.failed++;
142 cleanup();
143 });
144 ws.addEventListener('close', () => {
145 stats.closed++;
146 resolve();
147 });
148 });
149}
150
151function percentile(values: number[], p: number): number {
152 if (values.length === 0) return 0;
153 const sorted = [...values].sort((a, b) => a - b);
154 const idx = Math.min(sorted.length - 1, Math.floor((p / 100) * sorted.length));
155 return sorted[idx] ?? 0;
156}
157
158async function main(): Promise<number> {
159 const args = parseArgs(process.argv.slice(2));
160 if (process.argv.includes('--help') || process.argv.includes('-h')) {
161 help();
162 return 0;
163 }
164 if (!args.projectId) {
165 process.stderr.write('error: --project is required\n');
166 help();
167 return 1;
168 }
169 if (!args.secret) {
170 process.stderr.write('error: --secret or BRIVEN_RUNTIME_SHARED_SECRET is required\n');
171 return 1;
172 }
173
174 const stats: Stats = {
175 opened: 0,
176 failed: 0,
177 closed: 0,
178 framesReceived: 0,
179 firstFrameLatencyMs: [],
180 errors: new Map(),
181 };
182
183 process.stdout.write(
184 `opening ${args.subs} subscriptions against ${args.url} (ramp ${args.rampMs}ms)\n`,
185 );
186 const controller = new AbortController();
187 const inflight: Promise<void>[] = [];
188 for (let i = 0; i < args.subs; i++) {
189 inflight.push(openOne(args, stats, i, controller.signal));
190 if (args.rampMs > 0) await new Promise((r) => setTimeout(r, args.rampMs));
191 }
192
193 process.stdout.write(`ramp complete. holding for ${args.durationSec}s…\n`);
194 await new Promise((r) => setTimeout(r, args.durationSec * 1000));
195 controller.abort();
196 await Promise.all(inflight);
197
198 const summary = {
199 opened: stats.opened,
200 failed: stats.failed,
201 closed: stats.closed,
202 framesReceived: stats.framesReceived,
203 firstFrameLatency: {
204 n: stats.firstFrameLatencyMs.length,
205 p50: Math.round(percentile(stats.firstFrameLatencyMs, 50)),
206 p99: Math.round(percentile(stats.firstFrameLatencyMs, 99)),
207 max: Math.round(Math.max(0, ...stats.firstFrameLatencyMs)),
208 },
209 errorsByCode: Object.fromEntries(stats.errors.entries()),
210 };
211
212 process.stdout.write(`\n${JSON.stringify(summary, null, 2)}\n`);
213
214 // Exit non-zero if anything failed — useful in CI.
215 return stats.failed > 0 ? 1 : 0;
216}
217
218const code = await main();
219process.exit(code);