heartbeat-limits.test.ts

heartbeat-limits.test.tstests/integration/realtime/heartbeat-limits.test.tsraw ← realtime8,464 B← Back to the app
1/**
2 * Realtime integration (TASK-0091, acceptance criteria 2 + 3): heartbeat
3 * / prune and the message/payload rate limits.
4 *
5 * Time model: the server runs on REAL time (the fixture's ping interval,
6 * prune threshold, and GC delay are shortened real timers — the same
7 * timers production uses, only faster). The heartbeat test uses a raw TCP
8 * socket that NEVER answers the server's protocol pings (the `ws` client
9 * auto-pongs, which would defeat the scenario).
10 */
11import { describe, expect, it } from "vitest";
12import * as Y from "yjs";
13import {
14  buildRealtimeFixture,
15  connectClient,
16  connectRawSocket,
17  makeTextUpdate,
18} from "./rt-helpers";
19import { WS_CLOSE, REALTIME_LIMITS } from "@factory/shared/realtime/frames";
20 
21describe("heartbeat + prune (R4)", () => {
22  it("a socket that never pongs is terminated with 4000 (heartbeat_timeout)", async () => {
23    // Short real cadences: ping every 300 ms, prune after 1.5 s silence.
24    const fx = await buildRealtimeFixture({
25      pages: new Set(["p-it"]),
26      workspaces: { "p-it": "w1" },
27      levels: { "page:p-it": "edit" },
28      pingIntervalMs: 300,
29      pruneSilenceMs: 1_500,
30    });
31    try {
32      const raw = await connectRawSocket(fx, { room: "page:p-it" });
33      // Wait for the prune: the raw socket never pongs → 90 s (here 1.5 s)
34      // of silence → close 4000. The raw scanner records the unmasked
35      // close frame code from the wire (it ignores the 101 response and
36      // the protocol ping frames — see rt-helpers' close-frame check).
37      const deadline = Date.now() + 8_000;
38      while (Date.now() < deadline && !raw.closeCodes().includes(WS_CLOSE.ProtocolOrHeartbeat)) {
39        await new Promise((r) => setTimeout(r, 100));
40      }
41      expect(raw.closeCodes(), "server should have sent the close frame").toContain(
42        WS_CLOSE.ProtocolOrHeartbeat,
43      );
44      raw.destroy();
45    } finally {
46      await fx.stop();
47    }
48  }, 30_000);
49 
50  it("a socket that pongs stays alive past the prune window (10× the window)", async () => {
51    const fx = await buildRealtimeFixture({
52      pages: new Set(["p-it"]),
53      workspaces: { "p-it": "w1" },
54      levels: { "page:p-it": "edit" },
55      pingIntervalMs: 300,
56      pruneSilenceMs: 1_500,
57    });
58    try {
59      const c = await connectClient(fx, { room: "page:p-it" });
60      await c.nextFrame("hello");
61      // The `ws` client auto-pongs every server ping — no app-level
62      // activity needed. Outlive 4× the prune window.
63      await new Promise((r) => setTimeout(r, 6_000));
64      expect(c.isOpen()).toBe(true);
65      c.close();
66    } finally {
67      await fx.stop();
68    }
69  });
70 
71  it("an application-level ping frame gets a ping-pong echo", async () => {
72    const fx = await buildRealtimeFixture({
73      pages: new Set(["p-it"]),
74      workspaces: { "p-it": "w1" },
75      levels: { "page:p-it": "edit" },
76    });
77    try {
78      const c = await connectClient(fx, { room: "page:p-it" });
79      await c.nextFrame("hello");
80      c.sendText({ type: "ping", ts: 12345 });
81      const pong = await c.nextFrame("ping-pong");
82      expect(pong.ts).toBe(12345);
83      c.close();
84    } finally {
85      await fx.stop();
86    }
87  });
88});
89 
90describe("rate limits (R1/R3)", () => {
91  it("a 300 KB frame kills the socket with 4009 payload_too_large + audit", async () => {
92    const fx = await buildRealtimeFixture({
93      pages: new Set(["p-it"]),
94      workspaces: { "p-it": "w1" },
95      levels: { "page:p-it": "edit" },
96    });
97    try {
98      const c = await connectClient(fx, { room: "page:p-it" });
99      await c.nextFrame("hello");
100      // 300 KB of raw bytes with the 0x01 update prefix (invalid Yjs —
101      // irrelevant: the payload check fires before validation).
102      const big = new Uint8Array(300 * 1024);
103      for (let i = 0; i < big.length; i += 1) big[i] = i & 0xff;
104      c.sendBinary(1, big);
105      const closed = await c.onClose(5_000);
106      expect(closed.code).toBe(WS_CLOSE.PayloadTooLarge);
107      expect(fx.audit.abuse("frame_too_large")).toBe(1);
108    } finally {
109      await fx.stop();
110    }
111  });
112 
113  it("120 msgs/s sustained kills the socket with 4008 rate_limited + audit", async () => {
114    const fx = await buildRealtimeFixture({
115      pages: new Set(["p-it"]),
116      workspaces: { "p-it": "w1" },
117      levels: { "page:p-it": "edit" },
118    });
119    try {
120      const c = await connectClient(fx, { room: "page:p-it" });
121      await c.nextFrame("hello");
122      const doc = new Y.Doc();
123      const u = makeTextUpdate(doc, "a"); // a valid small update (1–3 KB)
124      doc.destroy();
125 
126      // Fire 120 messages in a tight burst (well above the 100/s
127      // sustained budget; the bucket refills continuously, so a burst of
128      // >100 in ~100 ms is the deterministic overflow).
129      const closePromise = c.onClose(10_000);
130      for (let i = 0; i < 120; i += 1) {
131        if (!c.isOpen()) break;
132        c.sendBinary(1, u);
133        if (i % 20 === 0) await new Promise((r) => setTimeout(r, 10));
134      }
135      const closed = await closePromise;
136      expect(closed.code).toBe(WS_CLOSE.RateOrPolicy);
137      expect(fx.audit.abuse("msg_rate_limited")).toBe(1);
138    } finally {
139      await fx.stop();
140    }
141  });
142 
143  it("10 corrupt frames within the strike window → 4008 strike_limit + audit", async () => {
144    const fx = await buildRealtimeFixture({
145      pages: new Set(["p-it"]),
146      workspaces: { "p-it": "w1" },
147      levels: { "page:p-it": "edit" },
148    });
149    try {
150      const c = await connectClient(fx, { room: "page:p-it" });
151      await c.nextFrame("hello");
152      const garbage = new Uint8Array([0xde, 0xad, 0xbe, 0xef]);
153      const closePromise = c.onClose(10_000);
154      for (let i = 0; i < REALTIME_LIMITS.strikeLimit + 1; i += 1) {
155        if (!c.isOpen()) break;
156        c.sendBinary(1, garbage);
157        await new Promise((r) => setTimeout(r, 20));
158      }
159      const closed = await closePromise;
160      expect(closed.code).toBe(WS_CLOSE.RateOrPolicy);
161      expect(fx.audit.abuse()).toBeGreaterThanOrEqual(1);
162      expect(fx.audit.records.some((r) => String(r.payload.cause).startsWith("strikes:"))).toBe(true);
163    } finally {
164      await fx.stop();
165    }
166  });
167 
168  it("an unknown JSON frame type gets error unknown_frame (no close below the strike limit)", async () => {
169    const fx = await buildRealtimeFixture({
170      pages: new Set(["p-it"]),
171      workspaces: { "p-it": "w1" },
172      levels: { "page:p-it": "edit" },
173    });
174    try {
175      const c = await connectClient(fx, { room: "page:p-it" });
176      await c.nextFrame("hello");
177      c.sendText({ type: "teleport", payload: "nope" });
178      const err = await c.nextFrame("error");
179      expect(err.code).toBe("unknown_frame");
180      // One bad frame — the socket survives (the 10-strike policy).
181      expect(c.isOpen()).toBe(true);
182      c.close();
183    } finally {
184      await fx.stop();
185    }
186  });
187 
188  it("a corrupt Yjs binary (valid frame, invalid update) gets error bad_update", async () => {
189    const fx = await buildRealtimeFixture({
190      pages: new Set(["p-it"]),
191      workspaces: { "p-it": "w1" },
192      levels: { "page:p-it": "edit" },
193    });
194    try {
195      const c = await connectClient(fx, { room: "page:p-it" });
196      await c.nextFrame("hello");
197      // 0x01 prefix is applied by sendBinary; the payload is not a
198      // Yjs update → the throwaway applyUpdate probe throws.
199      c.sendBinary(1, new Uint8Array([1, 2, 3, 4, 5]));
200      const err = await c.nextFrame("error");
201      expect(err.code).toBe("bad_update");
202      c.close();
203    } finally {
204      await fx.stop();
205    }
206  });
207});
208 
209describe("awareness relay (R4 fan-out, the TASK-0093 seam)", () => {
210  it("an awareness frame is relayed to the other peers (not the sender)", async () => {
211    const fx = await buildRealtimeFixture({
212      pages: new Set(["p-it"]),
213      workspaces: { "p-it": "w1" },
214      levels: { "page:p-it": "edit" },
215    });
216    try {
217      const a = await connectClient(fx, { room: "page:p-it" });
218      const b = await connectClient(fx, { room: "page:p-it" });
219      await a.nextFrame("hello");
220      await b.nextFrame("hello");
221      const payload = new Uint8Array([0x00, 0x01, 0x02]);
222      a.sendBinary(2, payload);
223      const got = await b.nextBinary();
224      expect(got.kind).toBe("awareness");
225      expect(Buffer.from(got.bytes).equals(Buffer.from(payload))).toBe(true);
226      a.close();
227      b.close();
228    } finally {
229      await fx.stop();
230    }
231  });
232});
233