rooms.test.ts · Source

rooms.test.tsapps/realtime/test/rooms.test.tsraw ← test22,375 B← Back to the app
1import { describe, it, expect, beforeEach } from "vitest";
2import * as Y from "yjs";
3import {
4  InMemoryDocStore,
5  RoomManager,
6  makePeer,
7  realClock,
8  type Room,
9  type RoomClock,
10  type RoomManagerEvents,
11  type DocStore,
12  type PersistResult,
13} from "../src/rooms.js";
14import {
15  blocksToYDoc,
16  yDocToBlockList,
17  FRAGMENT_NAME,
18  type BlockLike,
19  type BlockOperation,
20} from "@factory/collaboration/codec";
21 
22// --- deterministic clock -----------------------------------------------------------
23 
24/**
25 * A fake clock where `advance(ms)` fires every scheduled timer in
26 * chronological order up to `t + ms`. Timers registered with
27 * `setTimeout(fn, delay)` fire at `now + delay`.
28 */
29function driveClock(): { clock: RoomClock; run: (ms: number) => void; now: () => number } {
30  let t = 0;
31  const timers = new Map<number, { at: number; fn: () => void }>();
32  let nextId = 1;
33  const clock: RoomClock = {
34    now: () => t,
35    setTimeout: (fn: () => void, ms: number) => {
36      const id = nextId++;
37      timers.set(id, { at: t + Math.max(0, ms), fn });
38      return id;
39    },
40    clearTimeout: (h: unknown) => {
41      if (typeof h === "number") timers.delete(h);
42    },
43  };
44  return {
45    clock,
46    now: () => t,
47    run: (ms: number) => {
48      const end = t + ms;
49      while (true) {
50        let earliestId = -1;
51        let earliestAt = Infinity;
52        for (const [id, entry] of timers) {
53          if (entry.at <= end && entry.at < earliestAt) {
54            earliestAt = entry.at;
55            earliestId = id;
56          }
57        }
58        if (earliestId === -1) break;
59        const entry = timers.get(earliestId)!;
60        t = entry.at;
61        timers.delete(earliestId);
62        entry.fn();
63      }
64      t = end;
65    },
66  };
67}
68 
69// --- fixtures ----------------------------------------------------------------------
70 
71const PARA = (id: string, text: string, position: string, parentId: string | null = null): BlockLike => ({
72  id,
73  type: "paragraph",
74  content: { text },
75  parentId,
76  position,
77});
78 
79describe("InMemoryDocStore (the persistence seam, note 17)", () => {
80  it("seeds + hydrates blocks + version (a copy — mutations don't leak back)", async () => {
81    const store = new InMemoryDocStore();
82    store.seed("p1", [PARA("a", "hello", "1")]);
83    const hydrated = await store.hydrate("p1");
84    expect(hydrated.blocks).toHaveLength(1);
85    expect(hydrated.blocks![0].id).toBe("a");
86    expect(hydrated.version).toBe(1); // seed = version 1
87    // mutating the hydrated copy must not corrupt the store
88    (hydrated.blocks![0] as BlockLike).content = { text: "changed" };
89    const again = await store.hydrate("p1");
90    expect(again.blocks![0].content).toEqual({ text: "hello" });
91  });
92 
93  it("hydrates { blocks: null, version: 1 } for an unknown page", async () => {
94    const store = new InMemoryDocStore();
95    expect(await store.hydrate("missing")).toEqual({ blocks: null, version: 1 });
96  });
97 
98  it("persist bumps the version from the CAS base", async () => {
99    const store = new InMemoryDocStore();
100    const r1 = await store.persist("p1", [PARA("a", "x", "1")], 1);
101    expect(r1).toEqual({ version: 2 });
102    const r2 = await store.persist("p1", [PARA("a", "y", "1")], 2);
103    expect(r2).toEqual({ version: 3 });
104    expect(store.versionOf("p1")).toBe(3);
105  });
106 
107  it("persist with a stale base is a NO-OP and returns the external state (the CAS conflict)", async () => {
108    const store = new InMemoryDocStore();
109    store.seed("p1", [PARA("a", "base", "1")]);
110    // A concurrent REST writer replaces the blocks + bumps the version.
111    const externalVersion = store.writeExternal("p1", [PARA("a", "rest-wrote", "1")]);
112    expect(externalVersion).toBe(2);
113    // The room's flush, based on the version it hydrated at (1), conflicts.
114    const r = await store.persist("p1", [PARA("a", "room-edit", "1")], 1);
115    expect(r.version).toBe(2); // the store's CURRENT version
116    expect(r.conflict).toBeDefined();
117    expect(r.conflict!.externalBlocks.map((b) => b.content)).toEqual([{ text: "rest-wrote" }]);
118    // The persist was a no-op — the store still holds the external writer's state.
119    expect((await store.hydrate("p1")).blocks![0].content).toEqual({ text: "rest-wrote" });
120  });
121 
122  it("ping reflects setPingOk (drives /readyz)", async () => {
123    const store = new InMemoryDocStore();
124    expect(await store.ping()).toBe(true);
125    store.setPingOk(false);
126    expect(await store.ping()).toBe(false);
127  });
128});
129 
130describe("RoomManager — cold hydration (R2)", () => {
131  let store: InMemoryDocStore;
132  let events: RoomManagerEvents;
133  let manager: RoomManager;
134 
135  beforeEach(() => {
136    store = new InMemoryDocStore();
137    events = { onPersisted: () => {}, onRoomEvicted: () => {} };
138    manager = new RoomManager({ store, clock: realClock() }, events);
139  });
140 
141  it("hydrates a cold page room from the store before publishing it", async () => {
142    store.seed("p1", [PARA("a", "seeded", "1"), PARA("b", "two", "2")]);
143    const room = await manager.getOrCreate("page:p1", "page", "p1");
144    const blocks = yDocToBlockList(room.doc);
145    expect(blocks.map((b) => b.id)).toEqual(["a", "b"]);
146    expect(blocks[0].content).toEqual({ text: "seeded" });
147  });
148 
149  it("a warm room is returned as-is (no re-hydration)", async () => {
150    store.seed("p1", [PARA("a", "seeded", "1")]);
151    const r1 = await manager.getOrCreate("page:p1", "page", "p1");
152    const r2 = await manager.getOrCreate("page:p1", "page", "p1");
153    expect(r1).toBe(r2);
154  });
155 
156  it("get returns the live room; remove drops it from the map", async () => {
157    const room = await manager.getOrCreate("page:p1", "page", "p1");
158    expect(manager.get("page:p1")).toBe(room);
159    manager.remove("page:p1");
160    expect(manager.get("page:p1")).toBeUndefined();
161  });
162});
163 
164describe("Room — updates, dirty state, debounced flush (R2)", () => {
165  let drive: ReturnType<typeof driveClock>;
166  let store: InMemoryDocStore;
167  let evicted: string[];
168  let persisted: Array<{ roomKey: string; version: number }>;
169  let manager: RoomManager;
170 
171  beforeEach(() => {
172    drive = driveClock();
173    store = new InMemoryDocStore();
174    evicted = [];
175    persisted = [];
176    manager = new RoomManager(
177      { store, clock: drive.clock, persistDebounceMs: 1000, persistHardCapMs: 10_000 },
178      {
179        onPersisted: (roomKey, _resourceId, version) => persisted.push({ roomKey, version }),
180        onRoomEvicted: (roomKey) => evicted.push(roomKey),
181      },
182    );
183  });
184 
185  /** Simulate a client's update: build it in a second doc (via the codec)
186   *  and apply the encoded update into the room's authoritative doc — the
187   *  same path the server takes for an inbound 0x01 frame. */
188  function clientUpdate(room: Awaited<ReturnType<RoomManager["getOrCreate"]>>): void {
189    const clientDoc = blocksToYDoc([PARA("newblock", "from client", "9")], { pageId: "p1" });
190    const update = Y.encodeStateAsUpdate(clientDoc);
191    clientDoc.destroy();
192    expect(room.applyUpdate(update)).toBe(true); // applied now (not hydrating)
193  }
194 
195  it("applying an update marks the room dirty", async () => {
196    const room = await manager.getOrCreate("page:p1", "page", "p1");
197    expect(room.isDirty).toBe(false);
198    clientUpdate(room);
199    expect(room.isDirty).toBe(true);
200  });
201 
202  it("persists after the debounce (1 s) — the store sees the new block", async () => {
203    const room = await manager.getOrCreate("page:p1", "page", "p1");
204    clientUpdate(room);
205    expect(persisted).toHaveLength(0); // not yet
206    drive.run(1000); // fire the debounce timer
207    await new Promise((r) => setTimeout(r, 0)); // the flush is async (store round-trip)
208    expect(persisted).toHaveLength(1);
209    const stored = await store.hydrate("p1");
210    expect(stored.blocks?.some((b) => b.id === "newblock")).toBe(true);
211    expect(room.isDirty).toBe(false);
212  });
213 
214  it("coalesces repeated updates into one flush", async () => {
215    const room = await manager.getOrCreate("page:p1", "page", "p1");
216    clientUpdate(room);
217    clientUpdate(room);
218    clientUpdate(room);
219    drive.run(1000);
220    await new Promise((r) => setTimeout(r, 0));
221    expect(persisted).toHaveLength(1); // one flush, not three
222  });
223 
224  it("stateDelta with no state vector returns the full doc", async () => {
225    store.seed("p1", [PARA("a", "full", "1")]);
226    const room = await manager.getOrCreate("page:p1", "page", "p1");
227    const full = room.stateDelta(null);
228    expect(full.byteLength).toBeGreaterThan(0);
229    // applying the full update to a fresh doc reproduces the blocks
230    const fresh = new Y.Doc();
231    Y.applyUpdate(fresh, full, "tester");
232    expect(yDocToBlockList(fresh).map((b) => b.id)).toEqual(["a"]);
233    fresh.destroy();
234  });
235 
236  it("stateDelta with a client state vector returns only the missing update (R5 resume)", async () => {
237    store.seed("p1", [PARA("a", "seed", "1")]);
238    const room = await manager.getOrCreate("page:p1", "page", "p1");
239    // The client joins, receives the full doc, and applies it locally — its
240    // state vector now matches the room's current state (the real join path).
241    const clientDoc = new Y.Doc();
242    Y.applyUpdate(clientDoc, room.stateDelta(null), "server");
243    const clientVector = Y.encodeStateVector(clientDoc);
244    // The server then receives one more update (another client).
245    clientUpdate(room);
246    const delta = room.stateDelta(clientVector);
247    const full = room.stateDelta(null);
248    // The delta is non-empty (the new block) and smaller than the full doc.
249    expect(delta.byteLength).toBeGreaterThan(0);
250    expect(delta.byteLength).toBeLessThan(full.byteLength);
251    // Applying the delta to the client doc converges it to the server state.
252    Y.applyUpdate(clientDoc, delta, "server");
253    expect(yDocToBlockList(clientDoc).some((b) => b.id === "newblock")).toBe(true);
254    clientDoc.destroy();
255  });
256 
257  it("evict flushes a dirty room and reports the eviction", async () => {
258    const room = await manager.getOrCreate("page:p1", "page", "p1");
259    clientUpdate(room);
260    await room.evict();
261    expect(evicted).toEqual(["page:p1"]);
262    expect(persisted).toHaveLength(1); // flushed before eviction
263    const stored = await store.hydrate("p1");
264    expect(stored.blocks?.some((b) => b.id === "newblock")).toBe(true);
265  });
266 
267  it("evict of a clean room does not persist", async () => {
268    const room = await manager.getOrCreate("page:p1", "page", "p1");
269    await room.evict();
270    expect(evicted).toEqual(["page:p1"]);
271    expect(persisted).toHaveLength(0);
272  });
273 
274  it("empty-room GC evicts after the warm lifetime (R2)", async () => {
275    const gcManager = new RoomManager(
276      { store, clock: drive.clock, roomGcMs: 100 },
277      { onPersisted: () => {}, onRoomEvicted: (k) => evicted.push(k) },
278    );
279    const room = await gcManager.getOrCreate("page:p1", "page", "p1");
280    const peer = makePeer("c1", "u1", true, "Ada");
281    room.peers.set("c1", peer);
282    room.notePeersChanged(); // has a peer → GC armed off
283    room.peers.delete("c1");
284    room.notePeersChanged(); // empty → GC armed for 100 ms
285    drive.run(99);
286    expect(evicted).toHaveLength(0);
287    drive.run(1);
288    expect(evicted).toEqual(["page:p1"]);
289  });
290 
291  it("a peer re-joining cancels the pending GC", async () => {
292    const gcManager = new RoomManager(
293      { store, clock: drive.clock, roomGcMs: 100 },
294      { onPersisted: () => {}, onRoomEvicted: (k) => evicted.push(k) },
295    );
296    const room = await gcManager.getOrCreate("page:p1", "page", "p1");
297    room.notePeersChanged(); // empty → GC armed
298    room.peers.set("c1", makePeer("c1", "u1", true));
299    room.notePeersChanged(); // re-join → disarmed
300    drive.run(1000);
301    expect(evicted).toHaveLength(0);
302    expect(room.isEvicted).toBe(false);
303  });
304 
305  it("drainAll flushes every dirty room and evicts all (R6 SIGTERM)", async () => {
306    const r1 = await manager.getOrCreate("page:p1", "page", "p1");
307    void (await manager.getOrCreate("page:p2", "page", "p2"));
308    clientUpdate(r1);
309    const result = await manager.drainAll(10_000);
310    expect(result).toEqual({ flushed: 1, evicted: 2 });
311    expect(evicted.sort()).toEqual(["page:p1", "page:p2"]);
312    expect(persisted).toHaveLength(1);
313    expect(manager.rooms()).toHaveLength(0);
314  });
315});
316 
317describe("Room — flush-rebase (TASK-0092 R2, the 3-way merge)", () => {
318  let drive: ReturnType<typeof driveClock>;
319  let store: InMemoryDocStore;
320  let events: {
321    persisted: Array<{ version: number; blockDelta: BlockOperation[] }>;
322    rebased: number;
323  };
324  let manager: RoomManager;
325 
326  beforeEach(() => {
327    drive = driveClock();
328    store = new InMemoryDocStore();
329    events = { persisted: [], rebased: 0 };
330    manager = new RoomManager(
331      { store, clock: drive.clock, persistDebounceMs: 1000, persistHardCapMs: 10_000 },
332      {
333        onPersisted: (_roomKey, _resourceId, version, _logSeq, blockDelta) =>
334          events.persisted.push({ version, blockDelta }),
335        onRebased: () => {
336          events.rebased += 1;
337        },
338        onRoomEvicted: () => {},
339      },
340    );
341  });
342 
343  /**
344   * Edit a block's text content in the room's authoritative doc, the way a
345   * client's 0x01 update would (a real Yjs text edit, not a codec round-
346   * trip): finds the block element by its `id` attr and appends to its
347   * `text` slot. Also marks the room dirty (the server's applyUpdate path
348   * does this; the test bypasses the frame decode, so it calls the room's
349   * dirty bookkeeping via a subsequent flush — the assertion is on the
350   * persisted content, which only lands if the edit is in the doc).
351   */
352  function editBlockText(room: Room, blockId: string, append: string): void {
353    const fragment = room.doc.getXmlFragment(FRAGMENT_NAME);
354    const walk = (els: Y.XmlElement[]): Y.XmlElement | null => {
355      for (const el of els) {
356        if (el.getAttribute("id") === blockId) return el;
357        const nested = el.toArray().filter((c): c is Y.XmlElement => c instanceof Y.XmlElement);
358        const found = walk(nested);
359        if (found !== null) return found;
360      }
361      return null;
362    };
363    const el = walk(fragment.toArray() as Y.XmlElement[]);
364    expect(el, `block "${blockId}" found in the doc`).not.toBeNull();
365    const text = el
366      .toArray()
367      .find((c): c is Y.XmlText => c instanceof Y.XmlText && c.getAttribute("name") === "text");
368    expect(text, "a paragraph has a `text` slot").toBeDefined();
369    text.insert(text.length, append);
370  }
371 
372  it("concurrent REST write + room edit on DIFFERENT blocks → both survive (no clobber), next flush is a no-op", async () => {
373    store.seed("p1", [PARA("a", "base a", "1"), PARA("b", "base b", "2")]);
374    const room = await manager.getOrCreate("page:p1", "page", "p1");
375    // The room's client edits block "a".
376    editBlockText(room, "a", "+room edit on a");
377    // A concurrent REST write (bypassing the room) edits block "b" and bumps
378    // the version — exactly the race the flush's CAS detects.
379    store.writeExternal(
380      "p1",
381      [PARA("a", "base a", "1"), PARA("b", "base b+rest edit on b", "2")],
382    );
383    await room.flush();
384 
385    // The rebase ran once.
386    expect(events.rebased).toBe(1);
387    // The FINAL store state has BOTH edits (the acceptance criterion).
388    const final = (await store.hydrate("p1")).blocks!;
389    const a = final.find((b) => b.id === "a")!;
390    const b = final.find((b) => b.id === "b")!;
391    expect(a.content).toEqual({ text: "base a+room edit on a" });
392    expect(b.content).toEqual({ text: "base b+rest edit on b" });
393 
394    // The page version advanced past the external writer (the persist
395    // retried at the external version and won).
396    expect(store.versionOf("p1")).toBe(3); // seed=1, rest=2, flush=3
397    // The persisted blockDelta is non-trivial (real changes landed).
398    const last = events.persisted[events.persisted.length - 1]!;
399    expect(last.blockDelta.length).toBeGreaterThan(0);
400 
401    // The rebase converged: the doc now matches the store exactly, so a
402    // further persist of the same blocks at the current version is content-
403    // identical (the idempotency assert — no clobber, nothing to re-merge).
404    const blocksNow = yDocToBlockList(room.doc);
405    const noop = await store.persist("p1", blocksNow, store.versionOf("p1"));
406    expect(noop.conflict).toBeUndefined();
407    expect(noop.version).toBe(4); // a version bump (the store's CAS)
408    const after = (await store.hydrate("p1")).blocks!;
409    expect(after).toEqual(blocksNow); // same content — the rebase converged
410  });
411 
412  it("BOTH sides edit the SAME block → the FLUSH (room) wins (the spec's conflict rule)", async () => {
413    store.seed("p1", [PARA("a", "base a", "1"), PARA("b", "keep b", "2")]);
414    const room = await manager.getOrCreate("page:p1", "page", "p1");
415    // The room's client edits block "a".
416    editBlockText(room, "a", "+room edit WINS");
417    // A concurrent REST write edits the SAME block "a" (and touches "b").
418    store.writeExternal(
419      "p1",
420      [PARA("a", "base a+rest edit loses", "1"), PARA("b", "keep b+rest", "2")],
421    );
422    await room.flush();
423    const final = (await store.hydrate("p1")).blocks!;
424    const a = final.find((b) => b.id === "a")!;
425    const b = final.find((b) => b.id === "b")!;
426    // Flush wins on the shared block; REST's non-shared edit survives.
427    expect(a.content).toEqual({ text: "base a+room edit WINS" });
428    expect(b.content).toEqual({ text: "keep b+rest" });
429  });
430 
431  it("one rebase absorbs MULTIPLE concurrent external writes (all survive, single persist)", async () => {
432    store.seed("p1", [PARA("a", "base", "1")]);
433    const room = await manager.getOrCreate("page:p1", "page", "p1");
434    editBlockText(room, "a", "+room");
435    // Two external writers land before the flush — the rebase must absorb
436    // BOTH (each is a real change vs the room's base) in one pass.
437    store.writeExternal("p1", [PARA("a", "base", "1"), PARA("e1", "external 1", "3")]);
438    store.writeExternal("p1", [PARA("a", "base", "1"), PARA("e1", "external 1", "3"), PARA("e2", "external 2", "4")]);
439    await room.flush();
440    expect(events.rebased).toBe(1);
441    expect(events.persisted).toHaveLength(1);
442    const final = (await store.hydrate("p1")).blocks!;
443    expect(final.map((b) => b.id).sort()).toEqual(["a", "e1", "e2"]);
444    expect(final.find((b) => b.id === "a")!.content).toEqual({ text: "base+room" });
445    expect(final.find((b) => b.id === "e1")!.content).toEqual({ text: "external 1" });
446    expect(final.find((b) => b.id === "e2")!.content).toEqual({ text: "external 2" });
447  });
448 
449  it("a version bump DURING the retry → stays dirty (no clobber), converges on the next flush", async () => {
450    // A purpose-built store that simulates a SECOND concurrent writer firing
451    // between the conflicted persist and the rebase retry — the only way to
452    // reach the second-conflict branch synchronously (a real race is
453    // asynchronous; this deterministically exercises the single-retry limit).
454    const base = new InMemoryDocStore();
455    base.seed("p1", [PARA("a", "base", "1")]);
456    let persistCalls = 0;
457    const raceStore: DocStore = {
458      hydrate: () => base.hydrate("p1"),
459      ping: () => base.ping(),
460      persist: async (pageId, blocks, bVersion): Promise<PersistResult> => {
461        persistCalls += 1;
462        const r = await base.persist(pageId, blocks, bVersion);
463        if (r.conflict !== undefined && persistCalls === 1) {
464          // The second external writer lands right after the first conflict
465          // is reported — before the rebase retry re-checks the CAS.
466          base.writeExternal("p1", [
467            PARA("a", "base", "1"),
468            PARA("e1", "external 1", "3"),
469            PARA("e2", "external 2", "4"),
470          ]);
471        }
472        return r;
473      },
474    };
475    const raceManager = new RoomManager(
476      { store: raceStore, clock: drive.clock, persistDebounceMs: 1000, persistHardCapMs: 10_000 },
477      {
478        onPersisted: (_k, _r, _v, _s, delta) => events.persisted.push({ version: _v, blockDelta: delta }),
479        onRebased: () => {
480          events.rebased += 1;
481        },
482        onRoomEvicted: () => {},
483      },
484    );
485    const room = await raceManager.getOrCreate("page:p1", "page", "p1");
486    editBlockText(room, "a", "+room");
487    // External write #1 (conflicts the first persist).
488    base.writeExternal("p1", [PARA("a", "base", "1"), PARA("e1", "external 1", "3")]);
489    await room.flush();
490    // The single retry was used up (the second writer landed during it):
491    // no persist landed, the room keeps its user edits, no clobber.
492    expect(events.persisted).toHaveLength(0);
493    expect(base.versionOf("p1")).toBe(3); // e1 then e2 → v2, v3
494    // The next flush converges (absorbs both externals + the room's edit).
495    await room.flush();
496    expect(events.persisted).toHaveLength(1);
497    const final = (await base.hydrate("p1")).blocks!;
498    expect(final.map((b) => b.id).sort()).toEqual(["a", "e1", "e2"]);
499    expect(final.find((b) => b.id === "a")!.content).toEqual({ text: "base+room" });
500  });
501 
502  it("the doc's origin for the rebase update is 'realtime-flush-rebase' (observable)", async () => {
503    store.seed("p1", [PARA("a", "base", "1")]);
504    const room = await manager.getOrCreate("page:p1", "page", "p1");
505    const origins: string[] = [];
506    room.doc.on("update", (_u: Uint8Array, origin: string | null) => {
507      if (typeof origin === "string") origins.push(origin);
508    });
509    store.writeExternal("p1", [PARA("a", "base", "1"), PARA("x", "ext", "2")]);
510    await room.flush();
511    expect(origins).toContain("realtime-flush-rebase");
512  });
513});
514 
515describe("makePeer (roster entry)", () => {
516  it("builds a peer with a deterministic per-user color + the provided name", () => {
517    const a = makePeer("c1", "u1", true, "Ada");
518    const b = makePeer("c2", "u1", false, "Ada");
519    expect(a.userId).toBe("u1");
520    expect(a.canEdit).toBe(true);
521    expect(b.canEdit).toBe(false);
522    expect(a.roster.user.id).toBe("u1");
523    expect(a.roster.user.name).toBe("Ada");
524    // same user → same color (the shared 20-color palette)
525    expect(a.roster.user.color).toBe(b.roster.user.color);
526    expect(typeof a.roster.user.color).toBe("string");
527  });
528 
529  it("defaults the name when none is provided", () => {
530    const p = makePeer("c1", "u9", true);
531    expect(p.roster.user.name.length).toBeGreaterThan(0);
532  });
533});
534