|
| 1 | +/** |
| 2 | + * @vitest-environment node |
| 3 | + * |
| 4 | + * The join's readiness contract, with the shared store ENABLED (`file-doc.test.ts` runs it disabled). |
| 5 | + * |
| 6 | + * A room loads its document from the file's Redis stream one entry at a time, into the same `Y.Doc` |
| 7 | + * that fans every update out to the room. So a client attached while that is happening is not sent the |
| 8 | + * document — it is sent the document's history, and it watches the history replay on screen (reload |
| 9 | + * right after moving a block and the block moves again in front of you). These tests pin the fix: the |
| 10 | + * join waits for the room to hold the whole document, so the client's first sync is authoritative. |
| 11 | + */ |
| 12 | +import { |
| 13 | + FILE_DOC_EVENTS, |
| 14 | + FILE_DOC_MESSAGE_TYPE, |
| 15 | + FILE_DOC_SEED, |
| 16 | +} from '@sim/realtime-protocol/file-doc' |
| 17 | +import * as decoding from 'lib0/decoding' |
| 18 | +import * as encoding from 'lib0/encoding' |
| 19 | +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest' |
| 20 | +import * as syncProtocol from 'y-protocols/sync' |
| 21 | +import * as Y from 'yjs' |
| 22 | +import type { IRoomManager } from '@/rooms' |
| 23 | + |
| 24 | +const { mockAuthorizeRoom, mockFetchFileDocSeed } = vi.hoisted(() => ({ |
| 25 | + mockAuthorizeRoom: vi.fn(), |
| 26 | + mockFetchFileDocSeed: vi.fn(), |
| 27 | +})) |
| 28 | + |
| 29 | +vi.mock('@sim/platform-authz/rooms', () => ({ authorizeRoom: mockAuthorizeRoom })) |
| 30 | + |
| 31 | +vi.mock('@/handlers/file-doc-app', () => ({ |
| 32 | + fetchFileDocSeed: mockFetchFileDocSeed, |
| 33 | + fetchFileDocMerge: vi.fn(), |
| 34 | + fetchFileDocPersist: vi.fn().mockResolvedValue({ status: 'persisted', version: 1 }), |
| 35 | +})) |
| 36 | + |
| 37 | +/** One in-memory Redis backing per test — only the stream/lock ops the store actually uses. */ |
| 38 | +const backing = vi.hoisted(() => ({ |
| 39 | + streams: new Map<string, { id: string; message: Record<string, string> }[]>(), |
| 40 | + kv: new Map<string, string>(), |
| 41 | + seq: 0, |
| 42 | + /** Ticks of event-loop delay each xRange takes, modelling a remote (cross-region) Redis. */ |
| 43 | + readDelayTicks: 0, |
| 44 | +})) |
| 45 | + |
| 46 | +const seqOf = (id: string) => Number(id.split('-')[0]) |
| 47 | + |
| 48 | +vi.mock('redis', () => { |
| 49 | + const makeClient = (): Record<string, unknown> => { |
| 50 | + const client: Record<string, unknown> = { |
| 51 | + connect: async () => {}, |
| 52 | + quit: async () => {}, |
| 53 | + on: () => client, |
| 54 | + duplicate: () => makeClient(), |
| 55 | + xAdd: async (key: string, _star: string, fields: Record<string, string>) => { |
| 56 | + const id = `${++backing.seq}-0` |
| 57 | + const arr = backing.streams.get(key) ?? [] |
| 58 | + arr.push({ id, message: { ...fields } }) |
| 59 | + backing.streams.set(key, arr) |
| 60 | + return id |
| 61 | + }, |
| 62 | + xRange: async (key: string) => { |
| 63 | + for (let i = 0; i < backing.readDelayTicks; i++) await Promise.resolve() |
| 64 | + return (backing.streams.get(key) ?? []).map((e) => ({ ...e })) |
| 65 | + }, |
| 66 | + xLen: async (key: string) => (backing.streams.get(key) ?? []).length, |
| 67 | + xRead: async (streams: { key: string; id: string }[]) => { |
| 68 | + const res: { name: string; messages: { id: string; message: Record<string, string> }[] }[] = |
| 69 | + [] |
| 70 | + for (const { key, id } of streams) { |
| 71 | + const after = (backing.streams.get(key) ?? []).filter((e) => seqOf(e.id) > seqOf(id)) |
| 72 | + if (after.length) res.push({ name: key, messages: after.map((e) => ({ ...e })) }) |
| 73 | + } |
| 74 | + if (res.length) return res |
| 75 | + await new Promise((r) => setTimeout(r, 5)) |
| 76 | + return null |
| 77 | + }, |
| 78 | + set: async (key: string, val: string, opts?: { NX?: boolean }) => { |
| 79 | + if (opts?.NX && backing.kv.has(key)) return null |
| 80 | + backing.kv.set(key, val) |
| 81 | + return 'OK' |
| 82 | + }, |
| 83 | + eval: async (script: string, opts: { keys: string[]; arguments: string[] }) => { |
| 84 | + const [key] = opts.keys |
| 85 | + if (script.includes('xlen')) { |
| 86 | + const [field, value] = opts.arguments |
| 87 | + const arr = backing.streams.get(key) ?? [] |
| 88 | + if (arr.length > 0) return 0 |
| 89 | + arr.push({ id: `${++backing.seq}-0`, message: { [field]: value } }) |
| 90 | + backing.streams.set(key, arr) |
| 91 | + return 1 |
| 92 | + } |
| 93 | + const [token] = opts.arguments |
| 94 | + if (backing.kv.get(key) === token) { |
| 95 | + backing.kv.delete(key) |
| 96 | + return 1 |
| 97 | + } |
| 98 | + return 0 |
| 99 | + }, |
| 100 | + expire: async () => 1, |
| 101 | + get: async (key: string) => backing.kv.get(key) ?? null, |
| 102 | + exists: async (key: string) => (backing.kv.has(key) ? 1 : 0), |
| 103 | + } |
| 104 | + return client |
| 105 | + } |
| 106 | + return { createClient: () => makeClient() } |
| 107 | +}) |
| 108 | + |
| 109 | +import { cleanupFileDocForSocket, setupWorkspaceFileDocHandlers } from '@/handlers/file-doc' |
| 110 | +import { getFileDocStore, initFileDocStore } from '@/handlers/file-doc-store' |
| 111 | + |
| 112 | +const FILE_ID = 'file-1' |
| 113 | +const ROOM_NAME = `workspace-file-doc:${FILE_ID}` |
| 114 | +const STREAM_KEY = `filedoc:stream:${ROOM_NAME}` |
| 115 | +const FIELD = 'default' |
| 116 | + |
| 117 | +type Handler = (payload?: unknown) => Promise<void> | void |
| 118 | + |
| 119 | +interface FakeSocket { |
| 120 | + id: string |
| 121 | + emit: (event: string, payload: unknown) => void |
| 122 | + rooms: Set<string> |
| 123 | +} |
| 124 | + |
| 125 | +/** |
| 126 | + * An `io` that actually DELIVERS: a room emit reaches every socket that joined that room, so a frame |
| 127 | + * the relay fans out mid-assembly lands on the joiner's `emit` exactly as it would in the browser. |
| 128 | + * Recording the emits without routing them would hide the very thing these tests are about. |
| 129 | + */ |
| 130 | +function createIo(sockets: FakeSocket[]) { |
| 131 | + const emitTo = (target: string, except: string | null, event: string, payload: unknown) => { |
| 132 | + for (const socket of sockets) { |
| 133 | + if (socket.id === except || !socket.rooms.has(target)) continue |
| 134 | + socket.emit(event, payload) |
| 135 | + } |
| 136 | + } |
| 137 | + const to = vi.fn((target: string) => ({ |
| 138 | + except: (exclude: string) => ({ |
| 139 | + emit: (event: string, payload: unknown) => emitTo(target, exclude, event, payload), |
| 140 | + }), |
| 141 | + emit: (event: string, payload: unknown) => emitTo(target, null, event, payload), |
| 142 | + })) |
| 143 | + return { |
| 144 | + to, |
| 145 | + in: vi.fn(() => ({ socketsLeave: () => {} })), |
| 146 | + local: { to }, |
| 147 | + } as unknown as IRoomManager['io'] |
| 148 | +} |
| 149 | + |
| 150 | +function setup(id: string, sockets: FakeSocket[]) { |
| 151 | + const handlers: Record<string, Handler> = {} |
| 152 | + const rooms = new Set<string>() |
| 153 | + const socket = { |
| 154 | + id, |
| 155 | + userId: 'user-1', |
| 156 | + userName: 'Test User', |
| 157 | + userImage: 'avatar.png', |
| 158 | + disconnected: false, |
| 159 | + rooms, |
| 160 | + on: vi.fn((event: string, handler: Handler) => { |
| 161 | + handlers[event] = handler |
| 162 | + }), |
| 163 | + emit: vi.fn(), |
| 164 | + join: vi.fn((name: string) => rooms.add(name)), |
| 165 | + leave: vi.fn((name: string) => rooms.delete(name)), |
| 166 | + } |
| 167 | + sockets.push(socket as unknown as FakeSocket) |
| 168 | + setupWorkspaceFileDocHandlers( |
| 169 | + socket as unknown as Parameters<typeof setupWorkspaceFileDocHandlers>[0], |
| 170 | + { isReady: () => true, io: createIo(sockets) } as unknown as IRoomManager |
| 171 | + ) |
| 172 | + return { socket, handlers } |
| 173 | +} |
| 174 | + |
| 175 | +/** Append a Yjs update to the file's stream, exactly as `publish`/`seedIfEmpty` would. */ |
| 176 | +function appendToStream(update: Uint8Array): void { |
| 177 | + const arr = backing.streams.get(STREAM_KEY) ?? [] |
| 178 | + arr.push({ id: `${++backing.seq}-0`, message: { u: Buffer.from(update).toString('base64') } }) |
| 179 | + backing.streams.set(STREAM_KEY, arr) |
| 180 | +} |
| 181 | + |
| 182 | +/** |
| 183 | + * A warm room's history: the seed, then a later edit — the "I moved a block, then reloaded" case. |
| 184 | + * Returns the markdown-equivalent text of each state. |
| 185 | + */ |
| 186 | +function seedWarmStreamHistory(): { intermediate: string; final: string } { |
| 187 | + const doc = new Y.Doc() |
| 188 | + doc.getText(FIELD).insert(0, 'AAA') |
| 189 | + doc.getMap(FILE_DOC_SEED.configMap).set(FILE_DOC_SEED.flag, true) |
| 190 | + appendToStream(Y.encodeStateAsUpdate(doc)) |
| 191 | + const afterSeed = Y.encodeStateVector(doc) |
| 192 | + doc.getText(FIELD).insert(0, 'BBB') |
| 193 | + appendToStream(Y.encodeStateAsUpdate(doc, afterSeed)) |
| 194 | + doc.destroy() |
| 195 | + return { intermediate: 'AAA', final: 'BBBAAA' } |
| 196 | +} |
| 197 | + |
| 198 | +/** Every document state this socket was ever shown, in order. */ |
| 199 | +function statesDeliveredTo(socket: { emit: ReturnType<typeof vi.fn> }): string[] { |
| 200 | + const clientDoc = new Y.Doc() |
| 201 | + const states: string[] = [] |
| 202 | + for (const [event, payload] of socket.emit.mock.calls) { |
| 203 | + if (event !== FILE_DOC_EVENTS.MESSAGE || !(payload instanceof Uint8Array)) continue |
| 204 | + const decoder = decoding.createDecoder(payload) |
| 205 | + if (decoding.readVarUint(decoder) !== FILE_DOC_MESSAGE_TYPE.SYNC) continue |
| 206 | + syncProtocol.readSyncMessage(decoder, encoding.createEncoder(), clientDoc, null) |
| 207 | + const text = clientDoc.getText(FIELD).toString() |
| 208 | + if (text !== (states.at(-1) ?? '')) states.push(text) |
| 209 | + } |
| 210 | + clientDoc.destroy() |
| 211 | + return states |
| 212 | +} |
| 213 | + |
| 214 | +/** Let anything the join left running (a catch-up, a seed) settle, so a frame it fans out afterwards |
| 215 | + * is counted — that late delivery IS the replay these tests exist to rule out. */ |
| 216 | +async function flushPendingWork(): Promise<void> { |
| 217 | + for (let i = 0; i < 20; i++) await Promise.resolve() |
| 218 | +} |
| 219 | + |
| 220 | +/** Ask the server for its state the way a client does after the join ack. */ |
| 221 | +function requestSyncStep2(handlers: Record<string, Handler>): void { |
| 222 | + const encoder = encoding.createEncoder() |
| 223 | + encoding.writeVarUint(encoder, FILE_DOC_MESSAGE_TYPE.SYNC) |
| 224 | + syncProtocol.writeSyncStep1(encoder, new Y.Doc()) |
| 225 | + handlers[FILE_DOC_EVENTS.MESSAGE](encoding.toUint8Array(encoder)) |
| 226 | +} |
| 227 | + |
| 228 | +describe('file-doc join readiness (shared store enabled)', () => { |
| 229 | + /** Every socket the test created, so a room emit can be routed to its members. */ |
| 230 | + const sockets: FakeSocket[] = [] |
| 231 | + |
| 232 | + // One store for the whole file: `initFileDocStore` is idempotent once enabled, so a per-test |
| 233 | + // shutdown would leave every later test running against a store with closed clients. |
| 234 | + beforeAll(async () => { |
| 235 | + await initFileDocStore('redis://fake') |
| 236 | + }) |
| 237 | + |
| 238 | + afterAll(async () => { |
| 239 | + await getFileDocStore().shutdown() |
| 240 | + }) |
| 241 | + |
| 242 | + beforeEach(() => { |
| 243 | + vi.clearAllMocks() |
| 244 | + backing.streams.clear() |
| 245 | + backing.kv.clear() |
| 246 | + backing.seq = 0 |
| 247 | + backing.readDelayTicks = 0 |
| 248 | + mockAuthorizeRoom.mockResolvedValue({ |
| 249 | + allowed: true, |
| 250 | + status: 200, |
| 251 | + workspaceId: 'ws-1', |
| 252 | + workspacePermission: 'write', |
| 253 | + }) |
| 254 | + mockFetchFileDocSeed.mockResolvedValue(null) |
| 255 | + }) |
| 256 | + |
| 257 | + afterEach(() => { |
| 258 | + cleanupFileDocForSocket('socket-1', createIo(sockets), true) |
| 259 | + sockets.length = 0 |
| 260 | + }) |
| 261 | + |
| 262 | + it('hands a joiner the final document, never the room history it was rebuilt from', async () => { |
| 263 | + const { intermediate, final } = seedWarmStreamHistory() |
| 264 | + // The catch-up read is not instantaneous — the case that made this visible is a cross-region Redis. |
| 265 | + backing.readDelayTicks = 6 |
| 266 | + const { socket, handlers } = setup('socket-1', sockets) |
| 267 | + |
| 268 | + await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: FILE_ID, clientId: 1 }) |
| 269 | + requestSyncStep2(handlers) |
| 270 | + await flushPendingWork() |
| 271 | + |
| 272 | + // One state, and it is the final one: the client never saw the pre-move document. |
| 273 | + expect(statesDeliveredTo(socket)).toEqual([final]) |
| 274 | + expect(statesDeliveredTo(socket)).not.toContain(intermediate) |
| 275 | + }) |
| 276 | + |
| 277 | + it('does not fetch a seed for a room the stream can already reconstruct', async () => { |
| 278 | + seedWarmStreamHistory() |
| 279 | + const { handlers } = setup('socket-1', sockets) |
| 280 | + |
| 281 | + await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: FILE_ID, clientId: 1 }) |
| 282 | + |
| 283 | + expect(mockFetchFileDocSeed).not.toHaveBeenCalled() |
| 284 | + }) |
| 285 | + |
| 286 | + it('pulls a seed another writer put in the stream instead of waiting for the tailer to push it', async () => { |
| 287 | + // The seed lock is held by a writer whose room has since been dropped (a fast open→close on a |
| 288 | + // freshly created file), and its seed lands in the stream. Waiting to be told about it is what |
| 289 | + // left a new file un-editable until the client's readiness deadline lapsed; the join reads it. |
| 290 | + backing.kv.set(`filedoc:seedlock:${ROOM_NAME}`, 'held-by-a-writer-that-is-gone') |
| 291 | + const doc = new Y.Doc() |
| 292 | + doc.getText(FIELD).insert(0, 'seeded by the writer that held the lock') |
| 293 | + doc.getMap(FILE_DOC_SEED.configMap).set(FILE_DOC_SEED.flag, true) |
| 294 | + appendToStream(Y.encodeStateAsUpdate(doc)) |
| 295 | + doc.destroy() |
| 296 | + |
| 297 | + const { socket, handlers } = setup('socket-1', sockets) |
| 298 | + await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: FILE_ID, clientId: 1 }) |
| 299 | + requestSyncStep2(handlers) |
| 300 | + await flushPendingWork() |
| 301 | + |
| 302 | + expect(mockFetchFileDocSeed).not.toHaveBeenCalled() |
| 303 | + expect(statesDeliveredTo(socket)).toEqual(['seeded by the writer that held the lock']) |
| 304 | + }) |
| 305 | +}) |
0 commit comments