1#!/usr/bin/env node
2
43import crypto from "node:crypto";
44import { createRequire } from "node:module";
45
46
47const require = createRequire(import.meta.url);
48const Y = require("yjs");
49const WebSocket = require("ws");
50
51const WS_URL = process.env.K6_REALTIME_URL || "ws://localhost:4001";
52const SECRET = process.env.K6_JWT_SECRET || "";
53const WORKSPACE_ID = process.env.K6_WORKSPACE_ID || "";
54const VERIFY_PAGE_ID = process.env.K6_VERIFY_PAGE_ID || "load-p-verify";
55
56
57
58
59const YJS_UPDATES = [
60 "AQGCprqECwAEAQJrNhlrNiBsb2FkIHVwZGF0ZSB2YXJpYW50IDAgAA==",
61 "AQHgk6jsCgAEAQJrNhlrNiBsb2FkIHVwZGF0ZSB2YXJpYW50IDEgAA==",
62 "AQGf2MjCBAAEAQJrNhlrNiBsb2FkIHVwZGF0ZSB2YXJpYW50IDIgAA==",
63 "AQHzmP0+AAQBAms2GWs2IGxvYWQgdXBkYXRlIHZhcmlhbnQgMyAA",
64];
65
66function b64url(input) {
67 return input.replace(/\+/g, "-").replace(/\
68}
69
70
71function mintToken(sub) {
72 const nowSec = Math.floor(Date.now() / 1000);
73 const header = b64url(Buffer.from(JSON.stringify({ alg: "HS256", typ: "JWT" })).toString("base64"));
74 const claims = b64url(
75 Buffer.from(
76 JSON.stringify({ sub, workspaces: [WORKSPACE_ID], iat: nowSec, exp: nowSec + 600, jti: `verify-${sub}-${nowSec}` }),
77 ).toString("base64"),
78 );
79 const sig = b64url(crypto.createHmac("sha256", SECRET).update(`${header}.${claims}`).digest("base64"));
80 return `${header}.${claims}.${sig}`;
81}
82
83function sha256(b64) {
84 return crypto.createHash("sha256").update(Buffer.from(b64, "base64")).digest("hex");
85}
86
87function fullStateB64(doc) {
88 return Buffer.from(Y.encodeStateAsUpdate(doc)).toString("base64");
89}
90
91
95function fetchRoomState(token) {
96 return new Promise((resolve, reject) => {
97 const url = `${WS_URL}/realtime?token=${encodeURIComponent(token)}&room=${encodeURIComponent(`page:${VERIFY_PAGE_ID}`)}`;
98 const ws = new WebSocket(url, { headers: { "x-forwarded-for": "10.100.250.250" } });
99 let chunks = [];
100 let gotHello = false;
101 const timer = setTimeout(() => {
102 reject(new Error(`timeout: no full sync within 30 s (joined=${gotHello})`));
103 ws.close();
104 }, 30_000);
105 ws.on("open", () => {
106
107
108 });
109 ws.on("message", (data) => {
110 const buf = Buffer.isBuffer(data) ? data : Buffer.from(data);
111 if (buf.length === 0) return;
112 const kind = buf[0];
113 if (kind === 0x01 || kind === 0x02) return;
114 let msg;
115 try {
116 msg = JSON.parse(buf.toString("utf8"));
117 } catch {
118 return;
119 }
120 if (msg.type === "hello") {
121 gotHello = true;
122
123
124 ws.send(JSON.stringify({ type: "sync.request" }));
125 } else if (msg.type === "sync.state") {
126 if (msg.chunk) chunks.push(msg.chunk);
127 if (msg.done === true) {
128 clearTimeout(timer);
129 ws.close();
130 resolve(Buffer.concat(chunks.map((c) => Buffer.from(c, "base64"))));
131 }
132 } else if (msg.type === "error") {
133 clearTimeout(timer);
134 ws.close();
135 reject(new Error(`server error frame: ${JSON.stringify(msg)}`));
136 }
137 });
138 ws.on("error", (err) => {
139 clearTimeout(timer);
140 reject(err);
141 });
142 });
143}
144
145async function main() {
146 if (!SECRET) throw new Error("K6_JWT_SECRET is required");
147 if (!WORKSPACE_ID) throw new Error("K6_WORKSPACE_ID is required");
148
149
150 const stateBytes = await fetchRoomState(mintToken("k6-load-u-0000"));
151 const serverDoc = new Y.Doc();
152 Y.applyUpdate(serverDoc, stateBytes, "remote");
153 const serverState = fullStateB64(serverDoc);
154
155
156
157
158 const expectedDoc = new Y.Doc();
159 expectedDoc.getMap("yMeta").set("pageId", VERIFY_PAGE_ID);
160 for (const b64 of YJS_UPDATES) {
161 Y.applyUpdate(expectedDoc, Buffer.from(b64, "base64"), "remote");
162 }
163 const expectedState = fullStateB64(expectedDoc);
164
165 const serverHash = sha256(serverState);
166 const expectedHash = sha256(expectedState);
167 const serverText = serverDoc.getText("k6").toString();
168 const expectedText = expectedDoc.getText("k6").toString();
169 const missing = YJS_UPDATES.map((_, i) => `k6 load update variant ${i}`).filter(
170 (marker) => !serverText.includes(marker),
171 );
172
173 console.log("zero-op-loss verification:");
174 console.log(` room page:${VERIFY_PAGE_ID}`);
175 console.log(` server state ${stateBytes.byteLength} bytes, sha256=${serverHash}`);
176 console.log(` replay state ${Buffer.from(expectedState, "base64").byteLength} bytes, sha256=${expectedHash}`);
177 console.log(` server "k6" ${JSON.stringify(serverText)}`);
178 console.log(` replay "k6" ${JSON.stringify(expectedText)}`);
179
180 serverDoc.destroy();
181 expectedDoc.destroy();
182
183 if (serverHash === expectedHash) {
184 console.log("PASS — final doc state == sequential-replay hash (zero dropped Yjs ops)");
185 process.exit(0);
186 }
187
188 console.error("FAIL — final doc state != sequential-replay hash (dropped or extra ops)");
189 if (missing.length) console.error(` missing markers: ${missing.join(", ")}`);
190 console.error(
191 " hint: if the room was evicted + re-hydrated from the doc store (GC warm window, ~5 min),\n" +
192 " the 'k6' text is not represented there — re-run this check within 5 min of the k6 run.",
193 );
194 process.exit(1);
195}
196
197main().catch((err) => {
198 console.error("verify-zero-op-loss failed:", err.message || err);
199 process.exit(1);
200});
201
