realtime-1k.js · Source

realtime-1k.jstests/performance/realtime-1k.jsraw ← performance14,419 B← Back to the app
1/**
2 * AI Software Factory — k6: realtime 1k load baseline (TASK-0091 acceptance
3 * criterion "Load", first-run; TASK-0095 re-runs it as the final contract).
4 *
5 * Scenario (the task spec, verbatim):
6 *   - 1,000 connections over 30 s (100/s connect rate), held 2 min
7 *   - 200 msg/s mixed across the fleet: 50 % small doc updates,
8 *     30 % awareness, 20 % ping (per VU: 0.2 msg/s — one frame every 5 s)
9 *   - one node, p95 client→server→client round-trip < 200 ms,
10 *     0 protocol errors, 0 join failures
11 *
12 * RTT measurement (documented per the task spec):
13 *   The client sends the app-level `ping {ts}` frame; the server echoes
14 *   `ping-pong {ts}` with the SAME timestamp (server.ts `case "ping"`) —
15 *   so the round-trip is computed entirely on the client clock:
16 *   rtt = now - sentTs, immune to clock skew. The `hello.serverTime`
17 *   captured once at connect is recorded as the `clock_skew_ms` diagnostic
18 *   (server - client) but does NOT enter the RTT math.
19 *
20 * Protocol (the real TASK-0091 contract — packages/shared frames.ts):
21 *   connect:  <K6_REALTIME_URL>/realtime?token=<jwt>&room=page:<pageId>
22 *             header: X-Forwarded-For: 10.100.<a>.<b>   (per-VU identity)
23 *   hello:    {type:"hello", connId, canEdit, serverTime, roomPeers}
24 *   update:   binary 0x01 + YjsUpdate   (valid frames — generated with yjs;
25 *             an invalid update is a `bad_update` strike, 10 → 4008)
26 *   awareness: binary 0x02 + JSON state (the server relays, no content check)
27 *   ping:     {type:"ping", ts}                → {type:"ping-pong", ts}
28 *   ack:      {type:"ack", seq}                (logged updates)
29 *
30 * Why 1,000 distinct identities + 1,000 distinct source IPs (both MANDATORY):
31 *   - the server counts live sockets per user (100/user cap) and per-user
32 *     control sockets (10/user cap, the multi-tab guard) — ONE shared token
33 *     can never reach 1,000 connections (the 101st is rejected 429);
34 *   - the connect limiter is 30/min per source IP — the VUs present
35 *     per-socket `X-Forwarded-For` identities, so the server must run with
36 *     REALTIME_TRUST_PROXY=1 (the production reality: nginx terminates the
37 *     WS, the pod sees the proxy IP).
38 *   The seeder (helpers/seed-1k-load.mjs) creates the 1,000 users +
39 *   memberships + workspace-scoped EDIT grants; this script mints each VU's
40 *   HS256 JWT with k6/crypto (claims: sub, workspaces[], iat, exp, jti —
41 *   the exact shape @factory/auth/tokens verifies).
42 *
43 * Rooms: the 1,000 VUs are spread over K6_PAGE_IDS (default 10 pages ×
44 * 100 peers) so the update fan-out stays at ~100-way per room (a
45 * 1,000-peer single room would multiply every update 10×; the awareness
46 * fan-out cap of 50 peers/room in REALTIME_LIMITS is the design ceiling
47 * the server documents for presence).
48 *
49 * Thresholds (fail the run when violated):
50 *   rtt_ms              p(95) < 200 ms   (the DoD)
51 *   protocol_errors     count < 1        (the DoD: 0 protocol errors)
52 *   join_failures       rate  < 0.5 %
53 *   ws_unexpected_closes rate  < 0.5 %   (mid-hold server-side drops)
54 *
55 * Host-side DoD checks (NOT k6 thresholds — the runner run-1k.sh does them):
56 *   memory < 2 GB (RSS peak of the server process) and /health 200
57 *   throughout the run.
58 *
59 * Run (CI perf runner — see README.md):
60 *   node tests/performance/helpers/seed-1k-load.mjs      # once
61 *   K6_REALTIME_URL=ws://localhost:4001 K6_JWT_SECRET=… \
62 *   K6_WORKSPACE_ID=… K6_PAGE_IDS=p1,p2,…,p10 \
63 *   pnpm --dir tests/performance realtime:1k
64 *
65 * Artifact: realtime-1k-final.json (k6 --summary-export: all samples +
66 * percentiles; the VU/connection curve is the `vus` metric over time). The
67 * per-PR smoke tier (`realtime:smoke` — 100 conns / 30 s) reuses this script
68 * + the same seeder with a 2× flake-superset RTT threshold and a short ramp.
69 *
70 * k6 API note: this script uses the CURRENT k6/ws socket API
71 * (ws.connect(url, params, setup) + socket.setInterval / socket.setTimeout
72 * + socket.sendBinary — verified against internal/js/modules/k6/ws).
73 * The pre-existing realtime.js (10k) still targets the legacy `new WS()`
74 * + ws.wait() API; treat its frames/protocol as a placeholder until it is
75 * re-pointed at this contract (recorded TASK-0091 note).
76 */
77import ws from "k6/ws";
78import crypto from "k6/crypto";
79import encoding from "k6/encoding";
80import { Trend, Rate, Counter } from "k6/metrics";
81 
82const RTT_MS = new Trend("rtt_ms", true);
83const CLOCK_SKEW_MS = new Trend("clock_skew_ms", true);
84const joinFailures = new Rate("join_failures");
85const protocolErrors = new Counter("protocol_errors");
86const unexpectedCloses = new Rate("ws_unexpected_closes");
87const updateAcks = new Counter("update_acks");
88 
89const WS_URL = __ENV.K6_REALTIME_URL || "ws://localhost:4001";
90const SECRET = __ENV.K6_JWT_SECRET || "";
91const WORKSPACE_ID = __ENV.K6_WORKSPACE_ID || "";
92const PAGE_IDS = String(__ENV.K6_PAGE_IDS || "").split(",").filter(Boolean);
93const VUS = parseInt(__ENV.K6_VUS || "1000", 10);
94const RAMP_SECONDS = parseInt(__ENV.K6_RAMP_SECONDS || "30", 10);
95const HOLD_SECONDS = parseInt(__ENV.K6_HOLD_SECONDS || "120", 10);
96// RTT thresholds are env-overridable so the per-PR SMOKE tier (100 conns /
97// 30 s — see the `realtime:smoke` script) can run a flake-superset (2× the
98// DoD: p95<400 / p99<800) on noisy shared runners while the FULL 1k merge-
99// to-develop lane enforces the strict DoD (p95<200 / p99<400). The protocol-
100// correctness thresholds below (0 protocol errors, <0.5% joins/closes) are
101// NEVER flake-superseted — a protocol drop is a hard regression, not jitter.
102const RTT_P95_MS = parseInt(__ENV.K6_RTT_P95_MS || "200", 10);
103const RTT_P99_MS = parseInt(__ENV.K6_RTT_P99_MS || "400", 10);
104 
105/**
106 * Per-VU send cadence: 200 msg/s fleet / VUS. The spec pins the 200/VUS
107 * pair (1,000); keep the two in lockstep so the fleet rate stays 200/s.
108 */
109const MSG_INTERVAL_MS = Math.max(100, Math.round((VUS * 1000) / 200));
110 
111/**
112 * Valid Yjs updates (base64), each inserting a few bytes into Y.Text "k6"
113 * under a DEDICATED Yjs client id — re-applying the same frame is an
114 * idempotent no-op (the room never grows across the 2-min hold), and the
115 * server's throwaway-doc `applyUpdate` validation passes.
116 * Regenerate (not by hand!): see the TASK-0091 note in the seeder header.
117 */
118const YJS_UPDATES = [
119  "AQGCprqECwAEAQJrNhlrNiBsb2FkIHVwZGF0ZSB2YXJpYW50IDAgAA==",
120  "AQHgk6jsCgAEAQJrNhlrNiBsb2FkIHVwZGF0ZSB2YXJpYW50IDEgAA==",
121  "AQGf2MjCBAAEAQJrNhlrNiBsb2FkIHVwZGF0ZSB2YXJpYW50IDIgAA==",
122  "AQHzmP0+AAQBAms2GWs2IGxvYWQgdXBkYXRlIHZhcmlhbnQgMyAA",
123];
124 
125const BINARY_KIND = { Update: 0x01, Awareness: 0x02 }; // frames.ts BINARY_KIND
126 
127export const options = {
128  scenarios: {
129    realtime_1k: {
130      // NOTE: the executor is `ramping-vus` (the real k6 name in every
131      // version). An earlier draft spelled it `ramping-vusers` (unknown
132      // executor) — the script was never archive-validated; TASK-0095 S14
133      // caught + fixed it (and pinned k6 v0.57.0 in the CI perf lane, the
134      // first version where the `1_000` numeric separator below parses).
135      executor: "ramping-vus",
136      startVUs: 0,
137      stages: [
138        // The spec: 1,000 connections over 30 s (100/s connect rate).
139        { duration: `${RAMP_SECONDS}s`, target: VUS },
140        // …sustained 2 min at full fleet.
141        { duration: `${HOLD_SECONDS}s`, target: VUS },
142      ],
143      // A VU's iteration (connect + hold + clean close) takes exactly the
144      // scenario duration, so no iteration overlap.
145      gracefulStop: "10s",
146    },
147  },
148  thresholds: {
149    // The RTT DoD (full lane: p95<200 / p99<400); the smoke tier overrides
150    // via K6_RTT_P95_MS/K6_RTT_P99_MS (2× flake-superset — see above).
151    rtt_ms: [`p(95)<${RTT_P95_MS}`, `p(99)<${RTT_P99_MS}`],
152    protocol_errors: ["count<1"],
153    join_failures: ["rate<0.005"],
154    ws_unexpected_closes: ["rate<0.005"],
155  },
156};
157 
158/** base64url for JWT segments (no padding). */
159function b64url(input) {
160  return input.replace(/\+/g, "-").replace(/\//g, "_").replace(/=+$/, "");
161}
162 
163/**
164 * Mint this VU's access JWT (HS256, shared secret) — the exact claims
165 * shape @factory/auth/tokens `verifyAccessToken` + the handshake verify
166 * (apps/realtime/src/auth.ts) accept: sub (string), exp (unix seconds),
167 * iat (unix seconds, not in the future), jti, workspaces (string[]).
168 * `iss` is optional — omitted (checked only when present, note 13).
169 */
170function mintToken(sub, workspaces) {
171  const nowSec = Math.floor(Date.now() / 1000);
172  const header = b64url(encoding.b64encode(JSON.stringify({ alg: "HS256", typ: "JWT" }), "base64"));
173  const claims = b64url(
174    encoding.b64encode(
175      JSON.stringify({
176        sub: sub,
177        workspaces: workspaces,
178        iat: nowSec,
179        exp: nowSec + 900, // 15 min (the API's ACCESS_TOKEN lifetime)
180        jti: `k6-${sub}-${nowSec}`,
181      }),
182      "base64",
183    ),
184  );
185  const sig = b64url(crypto.hmac("sha256", SECRET, `${header}.${claims}`, "base64"));
186  return `${header}.${claims}.${sig}`;
187}
188 
189/** binary 0x01 + a valid Yjs update → ArrayBuffer for ws.sendBinary. */
190function updateFrame(variant) {
191  const bytes = encoding.b64decode(YJS_UPDATES[variant % YJS_UPDATES.length], "base64");
192  const out = new Uint8Array(bytes.length + 1);
193  out[0] = BINARY_KIND.Update;
194  out.set(bytes, 1);
195  return out.buffer;
196}
197 
198/** binary 0x02 + a small awareness state (the server relays it as-is). */
199function awarenessFrame(userId) {
200  const payload = JSON.stringify({
201    user: { id: userId, name: "k6", color: "slate" },
202    cursor: { x: (userId.length * 7) % 800, y: (userId.length * 13) % 600 },
203  });
204  const data = encoding.b64encode(payload, "utf8");
205  const out = new Uint8Array(data.length + 1);
206  out[0] = BINARY_KIND.Awareness;
207  out.set(data, 1);
208  return out.buffer;
209}
210 
211export function setup() {
212  if (!SECRET) throw new Error("setup: K6_JWT_SECRET is required (the shared HS256 secret)");
213  if (!WORKSPACE_ID) throw new Error("setup: K6_WORKSPACE_ID is required (the seeder prints it)");
214  if (PAGE_IDS.length === 0) throw new Error("setup: K6_PAGE_IDS is required (the seeder prints them)");
215  return { ok: true };
216}
217 
218export default function () {
219  // One identity per VU (1-based __VU → stable across the ramp):
220  // distinct user (100/user socket cap), distinct IP (30/min/IP connect
221  // limiter — the server must run REALTIME_TRUST_PROXY=1), page round-robin
222  // (100 peers/room fan-out).
223  const vu = __VU;
224  const pad4 = String(vu - 1).padStart(4, "0");
225  const userId = `k6-load-u-${pad4}`;
226  const ip = `10.100.${Math.floor((vu - 1) / 256)}.${(vu - 1) % 256}`;
227  const pageId = PAGE_IDS[(vu - 1) % PAGE_IDS.length];
228  const token = mintToken(userId, [WORKSPACE_ID]);
229 
230  const url = `${WS_URL}/realtime?token=${encodeURIComponent(token)}&room=${encodeURIComponent(`page:${pageId}`)}`;
231  // Each VU connects when its ramp slot opens, then HOLDS until the
232  // scenario end (__END_TIME) — so the fleet holds the FULL 2 minutes at
233  // 1,000 (the ramp's 30 s stagger would otherwise taper the last
234  // 30 s as early VUs finish first). ws.connect blocks the iteration
235  // until the socket closes, so the hold is the iteration's duration.
236  const holdMs = Math.max(1_000, __END_TIME - (Date.now() - __START_TIME));
237  const deadline = Date.now() + holdMs;
238 
239  let joined = false;
240  const sentTs = {}; // ping ts → true (cleared on the echo)
241 
242  const resp = ws.connect(
243    url,
244    { headers: { "x-forwarded-for": ip } },
245    (socket) => {
246      socket.on("message", (raw) => {
247        let msg;
248        try {
249          msg = JSON.parse(raw);
250        } catch {
251          return; // non-JSON (shouldn't happen for text frames)
252        }
253        switch (msg.type) {
254          case "hello":
255            joined = true;
256            if (msg.canEdit !== true) {
257              // The seeded identity has a workspace-scoped EDIT grant; a
258              // VIEW hello means the seed/permission wiring is wrong —
259              // every update this VU sends would be a FORBIDDEN protocol
260              // error, so count it NOW (the 0-protocol-errors DoD).
261              protocolErrors.add(1);
262            }
263            if (typeof msg.serverTime === "number") {
264              // One-time skew diagnostic (server - client), per the spec.
265              CLOCK_SKEW_MS.add(msg.serverTime - Date.now());
266            }
267            break;
268          case "ping-pong":
269            if (typeof msg.ts === "number" && sentTs[msg.ts]) {
270              RTT_MS.add(Date.now() - msg.ts);
271              delete sentTs[msg.ts];
272            }
273            break;
274          case "ack":
275            updateAcks.add(1);
276            break;
277          case "error":
278            // A server-side protocol error frame (FORBIDDEN / bad_update /
279            // unknown_frame / …). The DoD: zero of these.
280            protocolErrors.add(1);
281            break;
282          // sync.state / doc.persisted / comment.event / … — ignore.
283        }
284      });
285      // Binary frames (relayed updates / awareness) — not RTT samples.
286      socket.on("binaryMessage", () => {});
287      socket.on("close", () => {
288        // A close BEFORE the hold window ends is a mid-test disconnect.
289        if (!joined || Date.now() < deadline - 5000) unexpectedCloses.add(1);
290      });
291      socket.on("error", () => {
292        protocolErrors.add(1);
293      });
294 
295      // The mixed 200 msg/s fleet (0.2 msg/s per VU), spec mix 50/30/20.
296      // socket.setInterval runs the callback in the socket event loop.
297      let tick = 0;
298      socket.setInterval(() => {
299        const r = Math.random();
300        if (r < 0.5) {
301          socket.sendBinary(updateFrame(tick));
302        } else if (r < 0.8) {
303          socket.sendBinary(awarenessFrame(userId));
304        } else {
305          const ts = Date.now();
306          sentTs[ts] = true;
307          socket.send(JSON.stringify({ type: "ping", ts: ts }));
308        }
309        tick += 1;
310      }, MSG_INTERVAL_MS);
311 
312      // End the hold with a clean 1000 close (the close handler's window
313      // check keeps a deadline close from counting as an unexpected close).
314      socket.setTimeout(() => {
315        if (joined) socket.close(1000);
316      }, holdMs);
317    },
318  );
319 
320  if (resp.Status !== 101) {
321    // The handshake was rejected at the upgrade (401/403/429/400 — the
322    // pre-join rejections are HTTP statuses, no WS close frame exists).
323    joinFailures.add(1);
324  }
325}
326