realtime.js · Source

realtime.jstests/performance/realtime.jsraw ← performance5,483 B← Back to the app
1/**
2 * AI Software Factory — k6: realtime WebSocket scale (10,000 connections).
3 *
4 * Scenario (PRODUCT-SPEC §Performance, realtime < 200 ms):
5 *   - 10,000 concurrent WebSocket connections (one per VU)
6 *   - every VU joins the shared room "load" (configurable via K6_ROOM)
7 *   - every 500 ms each VU sends a Yjs update (small base64 doc delta)
8 *   - the realtime server acks each update (echoing the client seq); the
9 *     round-trip time is the measured metric
10 *
11 * Pass criteria (thresholds below, fail the run when violated):
12 *   rtt_ms          p(95) < 200 ms
13 *   ws_disconnects  rate  < 0.5% during the hold phase
14 *   join_failures   count < 1
15 *
16 * Protocol assumption (Phase 7 realtime contract — keep in sync with
17 * apps/realtime):
18 *   connect:  <K6_REALTIME_URL>/?token=<jwt>
19 *   join:     {"type":"join","roomId":"load"}            → {"type":"join:ack","roomId":"load"}
20 *   update:   {"type":"yjs-update","roomId":"load","docId":"<pageId>","seq":N,"ts":<ms>}
21 *   ack:      {"type":"yjs-update:ack","roomId":"load","docId":"<pageId>","seq":N,"ts":<ms>}
22 */
23import http from "k6/http";
24import { check } from "k6";
25import WS from "k6/ws";
26import { Trend, Rate } from "k6/metrics";
27 
28const RTT_MS = new Trend("rtt_ms", true);
29const disconnects = new Rate("ws_disconnects");
30const joinFailures = new Rate("join_failures");
31 
32const API_URL = __ENV.K6_API_URL || "http://localhost:4000";
33const WS_URL = __ENV.K6_REALTIME_URL || "ws://localhost:4001";
34const ROOM = __ENV.K6_ROOM || "load";
35const EMAIL = __ENV.K6_EMAIL || "demo@factory.local";
36const PASSWORD = __ENV.K6_PASSWORD || "factory-demo-123";
37 
38const HOLD_SECONDS = parseInt(__ENV.K6_DURATION || "300", 10); // default 5 min
39 
40export const options = {
41  scenarios: {
42    realtime_scale: {
43      executor: "constant-vusers",
44      vus: parseInt(__ENV.K6_VUS || "10000", 10),
45      duration: `${HOLD_SECONDS}s`,
46    },
47  },
48  thresholds: {
49    rtt_ms: ["p(95)<200", "p(99)<400"],
50    ws_disconnects: ["rate<0.005"],
51    join_failures: ["count<1"],
52  },
53};
54 
55export function setup() {
56  // Authenticate (seeded demo user; register as fallback) and mint a token
57  // shared by all VUs (the token is a constant load attribute, not a
58  // per-connection identity — the server's presence count is what matters).
59  let login = http.post(`${API_URL}/api/v1/auth/login`, JSON.stringify({ email: EMAIL, password: PASSWORD }), {
60    headers: { "Content-Type": "application/json" },
61  });
62  if (login.status !== 200) {
63    const reg = http.post(`${API_URL}/api/v1/auth/register`, JSON.stringify({ email: EMAIL, password: PASSWORD, name: "K6 Realtime" }), {
64      headers: { "Content-Type": "application/json" },
65    });
66    if (reg.status === 201 || reg.status === 200) {
67      login = http.post(`${API_URL}/api/v1/auth/login`, JSON.stringify({ email: EMAIL, password: PASSWORD }), {
68        headers: { "Content-Type": "application/json" },
69      });
70    }
71  }
72  if (login.status !== 200) throw new Error(`setup: login failed (${login.status})`);
73  const body = login.json();
74  const token = body.accessToken || body.access_token || body.token;
75  if (!token) throw new Error("setup: no access token in login response");
76  return { token, docId: `k6-realtime-doc-${Date.now()}` };
77}
78 
79export default function (data) {
80  const url = `${WS_URL}/?token=${encodeURIComponent(data.token)}`;
81  const deadline = Date.now() + HOLD_SECONDS * 1000;
82  let seq = 0;
83 
84  // In-flight update → sent-at map (for RTT measurement on ack).
85  const pending = new Map();
86  let joined = false;
87 
88  const ws = new WS(url, {
89    // The shared token authenticates the socket; no extra headers needed.
90    compression: false,
91  });
92 
93  ws.on("open", () => {
94    ws.send(JSON.stringify({ type: "join", roomId: ROOM }));
95  });
96 
97  ws.on("message", (raw) => {
98    let msg;
99    try {
100      msg = JSON.parse(raw.toString());
101    } catch {
102      return; // non-JSON frame (e.g. binary Yjs sync) — ignore for RTT.
103    }
104    if (msg.type === "join:ack" || msg.type === "welcome") {
105      joined = true;
106      return;
107    }
108    if (msg.type === "yjs-update:ack" && typeof msg.seq === "number" && pending.has(msg.seq)) {
109      const sentAt = pending.get(msg.seq);
110      pending.delete(msg.seq);
111      RTT_MS.add(Date.now() - sentAt);
112    }
113  });
114 
115  ws.on("close", (e) => {
116    // A close BEFORE the deadline is a mid-test disconnect (a clean close
117    // at the deadline is expected and must not count).
118    if (Date.now() < deadline - 1000) disconnects.add(1);
119  });
120 
121  // Block the VU (yielding the event loop) until the hold phase ends.
122  while (Date.now() < deadline) {
123    // Wait ~500 ms between updates (ws.wait yields to the k6 runtime and
124    // processes queued socket events).
125    ws.wait(500);
126    if (ws.closed) break;
127    if (!joined) {
128      // Give the join a bounded number of attempts; after that, count it as
129      // a join failure (the threshold fails the run).
130      ws.send(JSON.stringify({ type: "join", roomId: ROOM }));
131      continue;
132    }
133    seq += 1;
134    pending.set(seq, Date.now());
135    ws.send(
136      JSON.stringify({
137        type: "yjs-update",
138        roomId: ROOM,
139        docId: data.docId,
140        seq,
141        ts: Date.now(),
142        update: "AA==", // tiny Yjs delta (server acks; content is irrelevant)
143      }),
144    );
145  }
146 
147  // Join must have succeeded by the end of the hold phase.
148  if (!joined) joinFailures.add(1);
149 
150  // Drain: let in-flight acks land, then close cleanly.
151  ws.wait(1500);
152  ws.close();
153}
154