resume.test.ts · Source

resume.test.tstests/integration/realtime/resume.test.tsraw ← realtime10,007 B← Back to the app
1/**
2 * Realtime integration (TASK-0091, acceptance criterion 4): the resume
3 * protocol — the load-bearing test.
4 *
5 * Wire behavior under test (server.ts `handleSyncRequest`, R5):
6 *  - reconnect with `lastSeq` covered by the update log → the server
7 *    replays the logged updates as binary `update` frames, acks the last
8 *    seq, THEN always finishes with the state-vector delta (idempotent
9 *    convergence — the log is a speed optimization, the state vector is
10 *    the correctness guarantee).
11 *  - log gap (truncated log) → state-vector path only.
12 *  - invariant: both paths converge to the byte-identical doc.
13 *
14 * The "server doc" is observed through a fresh client's full sync
15 * (sync.request without a state vector → the complete doc), never
16 * through the in-memory store (which only sees the debounced flush).
17 */
18import { afterAll, beforeAll, describe, expect, it } from "vitest";
19import * as Y from "yjs";
20import {
21  buildRealtimeFixture,
22  connectClient,
23  makeTextUpdate,
24  type RealtimeFixture,
25  type RtClient,
26} from "./rt-helpers";
27 
28/** Apply every Yjs update to `doc` (sync.state chunks + binary replays). */
29function applyAll(doc: Y.Doc, c: RtClient): void {
30  for (const f of c.textFrames()) {
31    if (f.type === "sync.state" && typeof f.chunk === "string" && f.chunk.length > 0) {
32      Y.applyUpdate(doc, Buffer.from(f.chunk, "base64"), "it");
33    }
34  }
35  for (const b of c.binaryFrames()) {
36    if (b.kind === "update") Y.applyUpdate(doc, b.bytes, "it");
37  }
38}
39 
40/** A fresh client that full-syncs the room (the authoritative server view). */
41async function serverDoc(fx: RealtimeFixture, room = "page:p-it"): Promise<Y.Doc> {
42  const c = await connectClient(fx, { room });
43  try {
44    await c.nextFrame("hello");
45    c.sendText({ type: "sync.request" });
46    await c.nextFrameWhere((f) => f.type === "sync.state" && f.done === true, 8_000);
47    const doc = new Y.Doc();
48    applyAll(doc, c);
49    return doc;
50  } finally {
51    c.close();
52  }
53}
54 
55/**
56 * Join a page room, full-sync, send `n` one-char updates (awaiting each
57 * ack so the log seqs are settled), return the client + the last acked
58 * seq + the local doc.
59 */
60async function editSession(
61  fx: RealtimeFixture,
62  room: string,
63  n: number,
64  prefix: string,
65  localDoc: Y.Doc,
66): Promise<{ client: RtClient; lastSeq: number }> {
67  const client = await connectClient(fx, { room });
68  await client.nextFrame("hello");
69  client.sendText({ type: "sync.request" });
70  await client.nextFrameWhere((f) => f.type === "sync.state" && f.done === true, 8_000);
71  applyAll(localDoc, client);
72  let lastSeq = 0;
73  for (let i = 0; i < n; i += 1) {
74    const bytes = makeTextUpdate(localDoc, `${prefix}${i}`);
75    client.sendBinary(1, bytes);
76    const ack = await client.nextFrame("ack");
77    lastSeq = Number(ack.seq);
78  }
79  return { client, lastSeq };
80}
81 
82describe("resume protocol (R5)", () => {
83  let fx: RealtimeFixture;
84  beforeAll(async () => {
85    fx = await buildRealtimeFixture({
86      pages: new Set(["p-it", "p-seed"]),
87      workspaces: { "p-it": "w1", "p-seed": "w1" },
88      levels: { "page:p-it": "edit", "page:p-seed": "edit" },
89    });
90  });
91  afterAll(async () => {
92    await fx.stop();
93  });
94 
95  it("a client that was briefly offline resumes from the update log (exactly the missed frames) and converges", async () => {
96    // Client A: joins, receives the full doc, applies 20 updates, leaves.
97    const docA = new Y.Doc();
98    const a = await editSession(fx, "page:p-it", 20, "A", docA);
99    a.client.close();
100    const aLastSeq = a.lastSeq;
101    expect(aLastSeq).toBe(20); // the log is per-room, this is the only writer so far
102 
103    // Client B: joins, applies 5 updates (the "server applies 5 more").
104    const docB = new Y.Doc();
105    const b = await editSession(fx, "page:p-it", 5, "B", docB);
106    const serverSeq = b.lastSeq;
107    expect(serverSeq).toBe(25);
108    b.client.close();
109 
110    // The reference: what the server's doc looks like right now.
111    const ref = await serverDoc(fx);
112    const refVec = Y.encodeStateVector(ref);
113 
114    // A reconnects with its last-ack seq → it must receive EXACTLY the 5
115    // logged updates (binary frames before the state-delta sync.state),
116    // the ack of the last logged seq, then the (idempotent) state delta.
117    const docA2 = new Y.Doc();
118    // Re-apply A's local state (its editor kept the doc in memory).
119    const docA3 = new Y.Doc();
120    Y.applyUpdate(docA3, Y.encodeStateAsUpdate(docA), "it");
121    void docA2;
122 
123    const a2 = await connectClient(fx, { room: "page:p-it" });
124    try {
125      await a2.nextFrame("hello");
126      a2.sendText({ type: "sync.request", lastSeq: aLastSeq });
127      await a2.nextFrameWhere((f) => f.type === "sync.state" && f.done === true, 8_000);
128 
129      // Exactly 5 binary update frames were replayed from the log
130      // (the state delta arrives as sync.state TEXT frames, not binary).
131      const replayed = a2.binaryFrames().filter((b2) => b2.kind === "update");
132      expect(replayed).toHaveLength(5);
133      // The ack covers the last logged seq.
134      const acks = a2.textFrames().filter((f) => f.type === "ack");
135      expect(acks.length).toBeGreaterThanOrEqual(1);
136      expect(Number(acks[acks.length - 1]!.seq)).toBe(serverSeq);
137 
138      // Converge the local doc: replay + the state delta.
139      applyAll(docA3, a2);
140      expect(Y.encodeStateVector(docA3)).toEqual(refVec); // byte-identical
141      const txt = (d: Y.Doc) => d.getText("yText").toString();
142      expect(txt(docA3)).toBe(txt(ref));
143    } finally {
144      a2.close();
145      ref.destroy();
146      docA3.destroy();
147    }
148  });
149 
150  it("with the log truncated (a gap), the state-vector path returns the same final doc", async () => {
151    // Same room, current log maxSeq is 25 — truncate it (the "Redis log
152    // killed" scenario; InMemoryUpdateLog.clear is the seam).
153    void (await fx.updateLog.maxSeq("page:p-it"));
154    await fx.updateLog.clear("page:p-it");
155 
156    // A fresh client reconnecting with a lastSeq the log no longer covers
157    // → no binary replay, the state-vector delta alone.
158    const doc = new Y.Doc();
159    const c = await connectClient(fx, { room: "page:p-it" });
160    try {
161      await c.nextFrame("hello");
162      c.sendText({ type: "sync.request", lastSeq: 25 });
163      await c.nextFrameWhere((f) => f.type === "sync.state" && f.done === true, 8_000);
164      expect(c.binaryFrames().filter((b2) => b2.kind === "update")).toHaveLength(0);
165      applyAll(doc, c);
166 
167      const ref = await serverDoc(fx);
168      expect(Y.encodeStateVector(doc)).toEqual(Y.encodeStateVector(ref));
169      expect(doc.getText("yText").toString()).toBe(ref.getText("yText").toString());
170      ref.destroy();
171    } finally {
172      c.close();
173      doc.destroy();
174    }
175  });
176 
177  it("invariant (property): 20 random disconnect/reconnect windows always converge", async () => {
178    // Deterministic pseudo-random (mulberry32) — the property is over
179    // random window shapes, not the RNG itself.
180    let seed = 0x20260926;
181    const rnd = (): number => {
182      seed |= 0;
183      seed = (seed + 0x6d2b79f5) | 0;
184      let t = Math.imul(seed ^ (seed >>> 15), 1 | seed);
185      t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t;
186      return ((t ^ (t >>> 14)) >>> 0) / 4294967296;
187    };
188 
189    const room = "page:p-seed";
190    const local = new Y.Doc();
191    let lastSeq = 0;
192    let useLogResume = true;
193 
194    for (let i = 0; i < 20; i += 1) {
195      // Window shape: 1–4 local edits before the disconnect.
196      const client = await connectClient(fx, { room });
197      const edits = 1 + Math.floor(rnd() * 4);
198      for (let e = 0; e < edits; e += 1) {
199        const bytes = makeTextUpdate(local, `w${i}-${e}`);
200        client.sendBinary(1, bytes);
201        const ack = await client.nextFrame("ack");
202        lastSeq = Number(ack.seq);
203      }
204      // The client drops (no clean close — a network partition).
205      client.terminate();
206      await new Promise((r) => setTimeout(r, 30));
207 
208      // Meanwhile the room advances 1–3 updates from another peer.
209      const other = await connectClient(fx, { room });
210      const remoteEdits = 1 + Math.floor(rnd() * 3);
211      const otherDoc = new Y.Doc();
212      for (let e = 0; e < remoteEdits; e += 1) {
213        // otherDoc is empty — its updates are fresh content, still
214        // valid CRDT merges.
215        const bytes = makeTextUpdate(otherDoc, `r${i}-${e}`);
216        other.sendBinary(1, bytes);
217        await other.nextFrame("ack");
218      }
219      other.close();
220 
221      // Reconnect: 50/50 log-resume vs state-vector resume.
222      useLogResume = rnd() > 0.5;
223      const back = await connectClient(fx, { room });
224      back.sendText(useLogResume ? { type: "sync.request", lastSeq } : { type: "sync.request" });
225      await back.nextFrameWhere((f) => f.type === "sync.state" && f.done === true, 8_000);
226      applyAll(local, back);
227 
228      const ref = await serverDoc(fx, room);
229      expect(Y.encodeStateVector(local), `window ${i} diverged`).toEqual(
230        Y.encodeStateVector(ref),
231      );
232      ref.destroy();
233      back.close();
234    }
235    local.destroy();
236  });
237 
238  it("a resume for a different room's log is rejected to the state-vector path (tenant scoping)", async () => {
239    // The log is keyed per room: lastSeq from room X on room Y is a gap
240    // (the security requirement — a client can only ever resume ITS room).
241    const doc = new Y.Doc();
242    const c = await connectClient(fx, { room: "page:p-it" });
243    try {
244      await c.nextFrame("hello");
245      // 999 does not exist in p-it's log (it's at ~25) → gap → SV path.
246      c.sendText({ type: "sync.request", lastSeq: 999 });
247      await c.nextFrameWhere((f) => f.type === "sync.state" && f.done === true, 8_000);
248      expect(c.binaryFrames().filter((b2) => b2.kind === "update")).toHaveLength(0);
249      applyAll(doc, c);
250      const ref = await serverDoc(fx);
251      expect(Y.encodeStateVector(doc)).toEqual(Y.encodeStateVector(ref));
252      ref.destroy();
253    } finally {
254      c.close();
255      doc.destroy();
256    }
257  });
258});
259