lifecycle.test.ts

lifecycle.test.tstests/integration/realtime/lifecycle.test.tsraw ← realtime7,430 B← Back to the app
1/**
2 * Realtime integration (TASK-0091, acceptance criteria 5 + 7): room GC,
3 * the SIGTERM drain, and the `/health` + `/readyz` probes.
4 */
5import { afterAll, beforeAll, describe, expect, it } from "vitest";
6import {
7  buildRealtimeFixture,
8  connectClient,
9  fetchJson,
10  makeTextUpdate,
11  type RealtimeFixture,
12} from "./rt-helpers";
13import * as Y from "yjs";
14import { WS_CLOSE } from "@factory/shared/realtime/frames";
15 
16/** Join, send one update, leave — leaves the room dirty with 1 update. */
17async function touch(fx: RealtimeFixture, room: string, text: string): Promise<void> {
18  const c = await connectClient(fx, { room });
19  try {
20    await c.nextFrame("hello");
21    const doc = new Y.Doc();
22    c.sendBinary(1, makeTextUpdate(doc, text));
23    await c.nextFrame("ack");
24    doc.destroy();
25  } finally {
26    c.close();
27  }
28}
29 
30describe("room GC (R2)", () => {
31  it("a dirty room (unsaved updates) that loses its last peer flushes, then evicts", async () => {
32    // Short GC window (the 5 min warm room, compressed for the test).
33    const fx = await buildRealtimeFixture({
34      pages: new Set(["p-gc"]),
35      workspaces: { "p-gc": "w1" },
36      levels: { "page:p-gc": "edit" },
37      roomGcMs: 500,
38    });
39    try {
40      await touch(fx, "page:p-gc", "gc-update");
41      const room = fx.server.manager.get("page:p-gc");
42      expect(room).toBeDefined();
43      expect(room!.isDirty).toBe(true);
44 
45      // The GC fires after 500 ms of emptiness: flush + evict.
46      const deadline = Date.now() + 5_000;
47      let gone = false;
48      while (Date.now() < deadline && !gone) {
49        await new Promise((r) => setTimeout(r, 100));
50        gone = fx.server.manager.get("page:p-gc") === undefined;
51      }
52      expect(gone, "room should have been evicted").toBe(true);
53      // The dirty state was persisted BEFORE eviction (the store holds it
54      // and the log was cleared with the room).
55      const blocks = await fx.store.hydrate("p-gc");
56      expect(blocks).not.toBeNull();
57      expect(fx.updateLog.size("page:p-gc")).toBe(0);
58    } finally {
59      await fx.stop();
60    }
61  });
62 
63  it("a peer rejoining before the GC window cancels the eviction", async () => {
64    const fx = await buildRealtimeFixture({
65      pages: new Set(["p-gc2"]),
66      workspaces: { "p-gc2": "w1" },
67      levels: { "page:p-gc2": "edit" },
68      roomGcMs: 400,
69    });
70    try {
71      await touch(fx, "page:p-gc2", "one");
72      // Rejoin within the 400 ms window → the GC timer is cancelled.
73      const c = await connectClient(fx, { room: "page:p-gc2" });
74      try {
75        await c.nextFrame("hello");
76        await new Promise((r) => setTimeout(r, 900)); // past 2× the GC window
77        expect(fx.server.manager.get("page:p-gc2")).toBeDefined();
78      } finally {
79        c.close();
80      }
81    } finally {
82      await fx.stop();
83    }
84  });
85});
86 
87describe("SIGTERM drain (R6)", () => {
88  it("drain() closes every socket with 4010 + retryAfterMs, flushes all dirty rooms, and stops", async () => {
89    const fx = await buildRealtimeFixture({
90      pages: new Set(["p-d1", "p-d2", "p-d3"]),
91      workspaces: { "p-d1": "w1", "p-d2": "w1", "p-d3": "w1" },
92      levels: { "page:p-d1": "edit", "page:p-d2": "edit", "page:p-d3": "edit" },
93    });
94    try {
95      // 3 dirty rooms (the drain must flush ALL of them).
96      await touch(fx, "page:p-d1", "drain-1");
97      await touch(fx, "page:p-d2", "drain-2");
98      await touch(fx, "page:p-d3", "drain-3");
99      for (const k of ["page:p-d1", "page:p-d2", "page:p-d3"]) {
100        expect(fx.server.manager.get(k)?.isDirty).toBe(true);
101      }
102 
103      // Live clients on two of the rooms — they must see the 4010.
104      const a = await connectClient(fx, { room: "page:p-d1" });
105      const b = await connectClient(fx, { room: "page:p-d2" });
106      await a.nextFrame("hello");
107      await b.nextFrame("hello");
108 
109      const started = Date.now();
110      await fx.server.drain();
111      const elapsed = Date.now() - started;
112      expect(elapsed).toBeLessThanOrEqual(10_000); // the drain budget
113 
114      // Every live socket closed with the shutdown code + retry hint.
115      const ca = await a.onClose(3_000);
116      const cb = await b.onClose(3_000);
117      expect(ca.code).toBe(WS_CLOSE.ServerShutdown);
118      expect(cb.code).toBe(WS_CLOSE.ServerShutdown);
119      expect(ca.reason).toContain("retryAfterMs");
120      // The error notice frame (server_shutdown) precedes the close.
121      expect(a.textFrames().some((f) => f.type === "error" && f.code === "server_shutdown")).toBe(true);
122 
123      // ALL 3 dirty rooms were persisted (the flush happened during the
124      // drain, not after it).
125      for (const id of ["p-d1", "p-d2", "p-d3"]) {
126        const blocks = await fx.store.hydrate(id);
127        expect(blocks, `room ${id} not flushed`).not.toBeNull();
128      }
129      // The manager is empty (every room evicted).
130      expect(fx.server.manager.rooms()).toHaveLength(0);
131    } finally {
132      await fx.stop().catch(() => undefined); // already stopped by the drain
133    }
134  });
135});
136 
137describe("/health + /readyz (R6)", () => {
138  let fx: RealtimeFixture;
139  beforeAll(async () => {
140    fx = await buildRealtimeFixture({
141      pages: new Set(["p-it"]),
142      workspaces: { "p-it": "w1" },
143      levels: { "page:p-it": "edit" },
144    });
145  });
146  afterAll(async () => {
147    await fx.stop();
148  });
149 
150  it("/health returns the documented shape (status ok, connections, rooms, uptime, memoryRss)", async () => {
151    const r = await fetchJson(`${fx.healthUrl}/health`);
152    expect(r.status).toBe(200);
153    expect(r.body.status).toBe("ok");
154    expect(r.body).toHaveProperty("connections");
155    expect(r.body).toHaveProperty("rooms");
156    expect(r.body).toHaveProperty("uptime");
157    expect(Number(r.body.memoryRss)).toBeGreaterThan(0);
158  });
159 
160  it("/readyz is 200 ready with the checks object while the seams are up", async () => {
161    const r = await fetchJson(`${fx.healthUrl}/readyz`);
162    expect(r.status).toBe(200);
163    expect(r.body.status).toBe("ready");
164    expect(r.body.checks).toMatchObject({ ws: "ok", database: "ok" });
165  });
166 
167  it("/readyz is 503 not_ready when the database seam reports down", async () => {
168    fx.store.setPingOk(false);
169    try {
170      const r = await fetchJson(`${fx.healthUrl}/readyz`);
171      expect(r.status).toBe(503);
172      expect(r.body.status).toBe("not_ready");
173      expect(r.body.checks).toMatchObject({ database: "error" });
174      // /health stays 200 (liveness ≠ readiness — the probe contract).
175      const h = await fetchJson(`${fx.healthUrl}/health`);
176      expect(h.status).toBe(200);
177    } finally {
178      fx.store.setPingOk(true);
179    }
180  });
181 
182  it("the legacy /health/ready path serves the same readiness (Phase-0 compose)", async () => {
183    const r = await fetchJson(`${fx.healthUrl}/health/ready`);
184    expect(r.status).toBe(200);
185    expect(r.body.status).toBe("ready");
186  });
187 
188  it("/metrics serves the Prometheus text format", async () => {
189    const r = await fetch(`${fx.healthUrl}/metrics`);
190    expect(r.status).toBe(200);
191    const text = await r.text();
192    expect(text).toContain("ws_connections");
193    expect(text).toContain("# HELP");
194  });
195 
196  it("the ws port ALSO serves the health shape (note 15 — both ports probed)", async () => {
197    const httpUrl = fx.wsUrl.replace(/^ws:\/\//, "http://");
198    const r = await fetchJson(`${httpUrl}/health`);
199    expect(r.status).toBe(200);
200    expect(r.body.status).toBe("ok");
201  });
202});
203