From ece7c9e9d39bd2bfb44e5a8ffc30ed4e459e3c86 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 24 Jul 2026 15:09:03 -0700 Subject: [PATCH 1/9] feat(realtime): Yjs relay server for collaborative document editing [4/N] MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the server-side backend for live carets + text selection in a file's rich-text editor (Google Docs style), on the shared realtime rooms abstraction. Server-only and inert until the client provider lands (PR 5): a client must opt in by emitting `join-file-doc`, so this is a no-op for every existing flow. - Foundation: new `WORKSPACE_FILE_DOC` room type (`roomName`/parse codec) and its authz resolver (fileId -> workspace, honoring soft-delete and null- workspace copilot uploads). One-line enum entry + one resolver, exactly the extension point the room spine was built for. - `@sim/realtime-protocol/file-doc`: the shared wire protocol (event names, Yjs binary message tags, seed-flag keys, join payloads) so server and the future client can never drift. - `handlers/file-doc.ts`: a faithful y-websocket-style relay carried over the already-authenticated Socket.IO connection rather than a separate ws server — an authoritative in-memory Y.Doc + Awareness per file, `doc.on('update')` broadcast to the room except origin, `readSyncMessage` reply to sender, and per-connection controlled-clientID tracking cleared on disconnect (departed caret vanishes). Join requires workspace write. - Seed-once: the server elects exactly one seeder per empty document (kills the cold-start double-seed race deterministically), re-electing if the elected seeder disconnects before seeding; the client additionally guards on the CRDT `initialContentLoaded` flag. No durable Yjs snapshot yet (PR 6 follow-up): a document lives only while an editor is connected and re-seeds from the file's stored markdown — the markdown, saved through the content API by a client, remains the durable source of truth. Single-writer/single-replica assumption is documented (Helm pins realtime.replicaCount: 1). Tests: 10 relay tests with real Yjs sync/awareness round-trips (election, re-election, relay-except-sender, teardown) + room codec tests; full realtime suite (138) green; boundary + prune-graph gates pass. --- apps/realtime/package.json | 3 + apps/realtime/src/handlers/connection.ts | 4 + apps/realtime/src/handlers/file-doc.test.ts | 275 +++++++++++++++ apps/realtime/src/handlers/file-doc.ts | 353 +++++++++++++++++++ apps/realtime/src/handlers/index.ts | 2 + bun.lock | 11 + packages/platform-authz/src/rooms.ts | 22 +- packages/realtime-protocol/package.json | 4 + packages/realtime-protocol/src/file-doc.ts | 83 +++++ packages/realtime-protocol/src/rooms.test.ts | 15 + packages/realtime-protocol/src/rooms.ts | 7 + 11 files changed, 778 insertions(+), 1 deletion(-) create mode 100644 apps/realtime/src/handlers/file-doc.test.ts create mode 100644 apps/realtime/src/handlers/file-doc.ts create mode 100644 packages/realtime-protocol/src/file-doc.ts diff --git a/apps/realtime/package.json b/apps/realtime/package.json index 633ffb60827..35e45fc29e6 100644 --- a/apps/realtime/package.json +++ b/apps/realtime/package.json @@ -33,9 +33,12 @@ "@sim/workflow-types": "workspace:*", "@socket.io/redis-adapter": "8.3.0", "drizzle-orm": "^0.45.2", + "lib0": "0.2.117", "postgres": "^3.4.5", "redis": "5.10.0", "socket.io": "^4.8.1", + "y-protocols": "1.0.7", + "yjs": "13.6.31", "zod": "4.3.6" }, "devDependencies": { diff --git a/apps/realtime/src/handlers/connection.ts b/apps/realtime/src/handlers/connection.ts index 8b0d646ee73..aa134d8a771 100644 --- a/apps/realtime/src/handlers/connection.ts +++ b/apps/realtime/src/handlers/connection.ts @@ -1,5 +1,6 @@ import { createLogger } from '@sim/logger' import { parseRoomName, type RoomRef, roomName } from '@sim/realtime-protocol/rooms' +import { cleanupFileDocForSocket } from '@/handlers/file-doc' import { cleanupPendingSubblocksForSocket } from '@/handlers/subblocks' import { cleanupPendingVariablesForSocket } from '@/handlers/variables' import type { AuthenticatedSocket } from '@/middleware/auth' @@ -31,6 +32,9 @@ export function setupConnectionHandlers(socket: AuthenticatedSocket, roomManager // Clean up pending debounce entries for this socket to prevent memory leaks cleanupPendingSubblocksForSocket(socket.id) cleanupPendingVariablesForSocket(socket.id) + // Clear the socket's collaborative-document awareness (removes its caret for + // everyone else) and drop the room if it was the last editor. + cleanupFileDocForSocket(socket.id, roomManager.io) // A socket may occupy multiple rooms (one per type). Remove it from every // room the manager knows about. diff --git a/apps/realtime/src/handlers/file-doc.test.ts b/apps/realtime/src/handlers/file-doc.test.ts new file mode 100644 index 00000000000..38ab9433e5c --- /dev/null +++ b/apps/realtime/src/handlers/file-doc.test.ts @@ -0,0 +1,275 @@ +/** + * @vitest-environment node + */ +import { FILE_DOC_EVENTS, FILE_DOC_MESSAGE_TYPE } from '@sim/realtime-protocol/file-doc' +import * as encoding from 'lib0/encoding' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import * as awarenessProtocol from 'y-protocols/awareness' +import * as syncProtocol from 'y-protocols/sync' +import * as Y from 'yjs' +import type { IRoomManager } from '@/rooms' + +const { mockAuthorizeRoom } = vi.hoisted(() => ({ + mockAuthorizeRoom: vi.fn(), +})) + +vi.mock('@sim/platform-authz/rooms', () => ({ + authorizeRoom: mockAuthorizeRoom, +})) + +import { cleanupFileDocForSocket, setupWorkspaceFileDocHandlers } from '@/handlers/file-doc' + +type Handler = (payload?: unknown) => Promise | void + +const ROOM_NAME = 'workspace-file-doc:file-1' + +/** A shared Socket.IO `io` mock whose `to().except().emit()` chain is inspectable. */ +function createIo() { + const emit = vi.fn() + const except = vi.fn().mockReturnValue({ emit }) + const to = vi.fn().mockReturnValue({ except, emit }) + return { io: { to } as unknown as IRoomManager['io'], to, except, emit } +} + +function createSocket(id: string, overrides?: Record) { + const handlers: Record = {} + const socket = { + id, + userId: 'user-1', + userName: 'Test User', + on: vi.fn((event: string, handler: Handler) => { + handlers[event] = handler + }), + emit: vi.fn(), + join: vi.fn(), + leave: vi.fn(), + ...overrides, + } + return { handlers, socket } +} + +function createRoomManager( + io: IRoomManager['io'], + overrides?: Partial +): IRoomManager { + return { + isReady: vi.fn().mockReturnValue(true), + io, + ...overrides, + } as unknown as IRoomManager +} + +function setup(id: string, io: IRoomManager['io'], socketOverrides?: Record) { + const { socket, handlers } = createSocket(id, socketOverrides) + setupWorkspaceFileDocHandlers( + socket as unknown as Parameters[0], + createRoomManager(io) + ) + return { socket, handlers } +} + +/** Frame a Yjs message with its type tag, exactly as the client provider would. */ +function frame(type: number, write: (encoder: encoding.Encoder) => void): Uint8Array { + const encoder = encoding.createEncoder() + encoding.writeVarUint(encoder, type) + write(encoder) + return encoding.toUint8Array(encoder) +} + +/** The last `JOIN_SUCCESS` payload emitted on a socket. */ +function joinSuccess(socket: { emit: ReturnType }) { + const calls = socket.emit.mock.calls.filter( + (call: unknown[]) => call[0] === FILE_DOC_EVENTS.JOIN_SUCCESS + ) + const last = calls[calls.length - 1] + return last?.[1] as { fileId: string; shouldSeed: boolean } | undefined +} + +describe('setupWorkspaceFileDocHandlers', () => { + beforeEach(() => { + vi.clearAllMocks() + mockAuthorizeRoom.mockResolvedValue({ + allowed: true, + status: 200, + workspaceId: 'ws-1', + workspacePermission: 'write', + }) + }) + + afterEach(() => { + // The room store is module-global; drop any room the test's sockets left open. + const { io } = createIo() + for (const id of ['socket-1', 'socket-a', 'socket-b', 'socket-c']) { + cleanupFileDocForSocket(id, io) + } + }) + + it('rejects join when the socket is not authenticated', async () => { + const { io } = createIo() + const { socket, handlers } = setup('socket-1', io, { userId: undefined, userName: undefined }) + + await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) + + expect(socket.emit).toHaveBeenCalledWith( + FILE_DOC_EVENTS.JOIN_ERROR, + expect.objectContaining({ code: 'AUTHENTICATION_REQUIRED', retryable: false }) + ) + }) + + it('rejects join with a retryable error when realtime is unavailable', async () => { + const { io } = createIo() + const { socket, handlers } = createSocket('socket-1') + setupWorkspaceFileDocHandlers( + socket as unknown as Parameters[0], + createRoomManager(io, { isReady: vi.fn().mockReturnValue(false) }) + ) + + await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) + + expect(socket.emit).toHaveBeenCalledWith( + FILE_DOC_EVENTS.JOIN_ERROR, + expect.objectContaining({ code: 'ROOM_MANAGER_UNAVAILABLE', retryable: true }) + ) + }) + + it('rejects an invalid file id before authorizing', async () => { + const { io } = createIo() + const { socket, handlers } = setup('socket-1', io) + + await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: '' }) + + expect(socket.emit).toHaveBeenCalledWith( + FILE_DOC_EVENTS.JOIN_ERROR, + expect.objectContaining({ code: 'INVALID_PAYLOAD', retryable: false }) + ) + expect(mockAuthorizeRoom).not.toHaveBeenCalled() + }) + + it('requires write permission and reports 404 as NOT_FOUND', async () => { + mockAuthorizeRoom.mockResolvedValue({ allowed: false, status: 404, workspacePermission: null }) + const { io } = createIo() + const { socket, handlers } = setup('socket-1', io) + + await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) + + expect(mockAuthorizeRoom).toHaveBeenCalledWith( + expect.objectContaining({ + action: 'write', + room: { type: 'workspace-file-doc', id: 'file-1' }, + }) + ) + expect(socket.emit).toHaveBeenCalledWith( + FILE_DOC_EVENTS.JOIN_ERROR, + expect.objectContaining({ code: 'NOT_FOUND', retryable: false }) + ) + }) + + it('joins the room, elects the first client as seeder, and sends sync step 1', async () => { + const { io, to } = createIo() + const { socket, handlers } = setup('socket-1', io) + + await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) + + expect(socket.join).toHaveBeenCalledWith(ROOM_NAME) + expect(joinSuccess(socket)).toEqual({ fileId: 'file-1', shouldSeed: true }) + + // A binary sync-step-1 message (type tag 0) is sent to kick off the handshake. + const syncMessage = socket.emit.mock.calls.find( + ([event, payload]) => event === FILE_DOC_EVENTS.MESSAGE && payload instanceof Uint8Array + ) + expect(syncMessage).toBeDefined() + expect((syncMessage?.[1] as Uint8Array)[0]).toBe(FILE_DOC_MESSAGE_TYPE.SYNC) + // A fresh join with no doc updates yet performs no room broadcast. + expect(to).not.toHaveBeenCalled() + }) + + it('elects only one seeder across concurrent joiners of the same file', async () => { + const { io } = createIo() + const a = setup('socket-a', io) + const b = setup('socket-b', io) + + await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) + await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) + + expect(joinSuccess(a.socket)?.shouldSeed).toBe(true) + expect(joinSuccess(b.socket)?.shouldSeed).toBe(false) + }) + + it('relays a document update to the rest of the room, excluding the sender', async () => { + const { io, to, except, emit } = createIo() + const a = setup('socket-a', io) + const b = setup('socket-b', io) + await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) + await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) + to.mockClear() + except.mockClear() + emit.mockClear() + + // A real client edit, framed as a sync update. + const clientDoc = new Y.Doc() + clientDoc.getText('default').insert(0, 'hello') + const update = Y.encodeStateAsUpdate(clientDoc) + a.handlers[FILE_DOC_EVENTS.MESSAGE]( + frame(FILE_DOC_MESSAGE_TYPE.SYNC, (e) => syncProtocol.writeUpdate(e, update)) + ) + + expect(to).toHaveBeenCalledWith(ROOM_NAME) + expect(except).toHaveBeenCalledWith('socket-a') + const relayed = emit.mock.calls.find(([event]) => event === FILE_DOC_EVENTS.MESSAGE) + expect(relayed).toBeDefined() + expect((relayed?.[1] as Uint8Array)[0]).toBe(FILE_DOC_MESSAGE_TYPE.SYNC) + }) + + it('relays awareness (cursor/selection) updates to the room, excluding the sender', async () => { + const { io, except, emit } = createIo() + const a = setup('socket-a', io) + const b = setup('socket-b', io) + await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) + await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) + except.mockClear() + emit.mockClear() + + const clientDoc = new Y.Doc() + const clientAwareness = new awarenessProtocol.Awareness(clientDoc) + clientAwareness.setLocalStateField('user', { name: 'Ada', color: '#f783ac' }) + const awarenessUpdate = awarenessProtocol.encodeAwarenessUpdate(clientAwareness, [ + clientDoc.clientID, + ]) + a.handlers[FILE_DOC_EVENTS.MESSAGE]( + frame(FILE_DOC_MESSAGE_TYPE.AWARENESS, (e) => encoding.writeVarUint8Array(e, awarenessUpdate)) + ) + + expect(except).toHaveBeenCalledWith('socket-a') + const relayed = emit.mock.calls.find(([event]) => event === FILE_DOC_EVENTS.MESSAGE) + expect((relayed?.[1] as Uint8Array)[0]).toBe(FILE_DOC_MESSAGE_TYPE.AWARENESS) + }) + + it('re-elects a seeder when the elected one disconnects before seeding', async () => { + const { io } = createIo() + const a = setup('socket-a', io) + const b = setup('socket-b', io) + await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) + await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) + expect(joinSuccess(a.socket)?.shouldSeed).toBe(true) + + // The seeder leaves before it ever seeded; b remains so the room survives. + cleanupFileDocForSocket('socket-a', io) + + const c = setup('socket-c', io) + await c.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) + expect(joinSuccess(c.socket)?.shouldSeed).toBe(true) + }) + + it('drops the document when the last editor leaves, re-seeding a fresh joiner', async () => { + const { io } = createIo() + const a = setup('socket-a', io) + await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) + cleanupFileDocForSocket('socket-a', io) + + // With the room dropped, the next joiner starts a brand-new document and is + // elected to seed again. + const b = setup('socket-b', io) + await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) + expect(joinSuccess(b.socket)?.shouldSeed).toBe(true) + }) +}) diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts new file mode 100644 index 00000000000..e626b672982 --- /dev/null +++ b/apps/realtime/src/handlers/file-doc.ts @@ -0,0 +1,353 @@ +import { createLogger } from '@sim/logger' +import { authorizeRoom } from '@sim/platform-authz/rooms' +import { + FILE_DOC_EVENTS, + FILE_DOC_MESSAGE_TYPE, + FILE_DOC_SEED, + type JoinFileDocPayload, + type LeaveFileDocPayload, +} from '@sim/realtime-protocol/file-doc' +import { ROOM_TYPES, type RoomRef, roomName } from '@sim/realtime-protocol/rooms' +import * as decoding from 'lib0/decoding' +import * as encoding from 'lib0/encoding' +import type { Server } from 'socket.io' +import * as awarenessProtocol from 'y-protocols/awareness' +import * as syncProtocol from 'y-protocols/sync' +import * as Y from 'yjs' +import type { AuthenticatedSocket } from '@/middleware/auth' +import type { IRoomManager } from '@/rooms' + +const logger = createLogger('FileDocHandlers') + +/** + * Collaborative document editing (live carets + text selection) for a single + * file's rich-text editor. This is the standard Yjs "websocket server" relay — + * an authoritative in-memory {@link Y.Doc} + {@link awarenessProtocol.Awareness} + * per file — carried over the shared, already-authenticated Socket.IO connection + * and the room abstraction, rather than a separate ws server. Clients speak the + * `y-protocols` sync + awareness protocols; the server applies and relays them. + * + * No durable Yjs state is kept yet: the document lives only while at least one + * collaborator is connected, and is re-seeded from the file's stored markdown on + * the next cold open (the markdown, saved by a client through the content API, is + * the durable source of truth). Durable Yjs snapshots are a separate follow-up. + * + * Single-writer assumption: the authoritative {@link Y.Doc} is held in this + * process's memory, so correctness assumes one realtime replica per file (Helm + * pins `realtime.replicaCount: 1`). Horizontal scaling would need a shared Yjs + * backend (y-redis / Hocuspocus) — out of scope here. + */ +interface FileDocRoom { + doc: Y.Doc + awareness: awarenessProtocol.Awareness + /** socketId → the awareness clientIDs it controls, cleared on its disconnect. */ + controlledIds: Map> + /** The socket elected to seed initial content, or `null` if unseeded/unelected. */ + seederSocketId: string | null +} + +/** Live documents keyed by Socket.IO room name. Module-global: one Y.Doc per file. */ +const fileDocRooms = new Map() +/** socketId → its current file-doc room name (a socket edits at most one doc). */ +const socketToRoomName = new Map() + +interface AwarenessChange { + added: number[] + updated: number[] + removed: number[] +} + +const fileDocRoom = (fileId: string): RoomRef => ({ + type: ROOM_TYPES.WORKSPACE_FILE_DOC, + id: fileId, +}) + +/** + * A `y-protocols` transaction/awareness origin is the emitting socket id (a + * string) when it came from a client, and something else (`null` / `'local'` / + * `'timeout'`) for server-internal changes. Returns the socket id to exclude + * from a relay, or `null` to broadcast to the whole room. + */ +function originSocketId(origin: unknown): string | null { + return typeof origin === 'string' ? origin : null +} + +function broadcast(io: Server, name: string, payload: Uint8Array, exceptSocketId: string | null) { + const channel = exceptSocketId ? io.to(name).except(exceptSocketId) : io.to(name) + channel.emit(FILE_DOC_EVENTS.MESSAGE, payload) +} + +/** Whether the client has recorded that it seeded the document's initial content. */ +function isDocSeeded(doc: Y.Doc): boolean { + return doc.getMap(FILE_DOC_SEED.configMap).get(FILE_DOC_SEED.flag) === true +} + +function toUint8Array(data: unknown): Uint8Array | null { + if (data instanceof Uint8Array) return data + if (data instanceof ArrayBuffer) return new Uint8Array(data) + return null +} + +/** + * Get (or lazily create) the authoritative document for a room, wiring the two + * relay handlers exactly once: document updates and awareness changes are + * broadcast to the room, excluding the origin socket (it already applied them). + */ +function getOrCreateRoom(io: Server, name: string): FileDocRoom { + const existing = fileDocRooms.get(name) + if (existing) return existing + + const doc = new Y.Doc() + const awareness = new awarenessProtocol.Awareness(doc) + // The server holds no cursor of its own; it only relays clients' awareness. + awareness.setLocalState(null) + + const room: FileDocRoom = { doc, awareness, controlledIds: new Map(), seederSocketId: null } + fileDocRooms.set(name, room) + + doc.on('update', (update: Uint8Array, origin: unknown) => { + const encoder = encoding.createEncoder() + encoding.writeVarUint(encoder, FILE_DOC_MESSAGE_TYPE.SYNC) + syncProtocol.writeUpdate(encoder, update) + broadcast(io, name, encoding.toUint8Array(encoder), originSocketId(origin)) + }) + + awareness.on('update', ({ added, updated, removed }: AwarenessChange, origin: unknown) => { + const from = originSocketId(origin) + if (from) { + const controlled = room.controlledIds.get(from) + if (controlled) { + for (const id of added) controlled.add(id) + for (const id of removed) controlled.delete(id) + } + } + const changed = added.concat(updated, removed) + if (changed.length === 0) return + const encoder = encoding.createEncoder() + encoding.writeVarUint(encoder, FILE_DOC_MESSAGE_TYPE.AWARENESS) + encoding.writeVarUint8Array( + encoder, + awarenessProtocol.encodeAwarenessUpdate(awareness, changed) + ) + broadcast(io, name, encoding.toUint8Array(encoder), from) + }) + + return room +} + +function emitJoinError( + socket: AuthenticatedSocket, + fileId: unknown, + error: string, + code: string, + retryable: boolean +) { + socket.emit(FILE_DOC_EVENTS.JOIN_ERROR, { + fileId: typeof fileId === 'string' ? fileId : '', + error, + code, + retryable, + }) +} + +function handleMessage(socket: AuthenticatedSocket, data: unknown) { + const name = socketToRoomName.get(socket.id) + if (!name) return + const room = fileDocRooms.get(name) + if (!room) return + + const bytes = toUint8Array(data) + if (!bytes) return + + const decoder = decoding.createDecoder(bytes) + const messageType = decoding.readVarUint(decoder) + + switch (messageType) { + case FILE_DOC_MESSAGE_TYPE.SYNC: { + const encoder = encoding.createEncoder() + encoding.writeVarUint(encoder, FILE_DOC_MESSAGE_TYPE.SYNC) + // `socket.id` is the transaction origin, so the doc's `update` handler + // excludes this sender when relaying the applied update to the room. + syncProtocol.readSyncMessage(decoder, encoder, room.doc, socket.id) + // A reply longer than the 1-byte type tag is a sync step 2 (or step 1) + // destined for the sender only; applied updates fan out via `doc.on`. + if (encoding.length(encoder) > 1) { + socket.emit(FILE_DOC_EVENTS.MESSAGE, encoding.toUint8Array(encoder)) + } + break + } + case FILE_DOC_MESSAGE_TYPE.AWARENESS: { + awarenessProtocol.applyAwarenessUpdate( + room.awareness, + decoding.readVarUint8Array(decoder), + socket.id + ) + break + } + default: + logger.warn('Unknown file-doc message type', { messageType }) + } +} + +/** + * Remove a socket from its file-doc room: clear its awareness state (so its caret + * disappears for everyone else), free the seeder election if it left before + * seeding, and drop the room's document when the last collaborator leaves. + * Exported for the disconnect handler; safe to call for a socket in no room. + */ +export function cleanupFileDocForSocket(socketId: string, io: Server): void { + const name = socketToRoomName.get(socketId) + if (!name) return + socketToRoomName.delete(socketId) + + const room = fileDocRooms.get(name) + if (!room) return + + const controlled = room.controlledIds.get(socketId) + room.controlledIds.delete(socketId) + if (controlled && controlled.size > 0) { + // Fires the awareness `update` handler with a non-socket origin → the removal + // is broadcast to every remaining client, so the departed caret vanishes. + awarenessProtocol.removeAwarenessStates(room.awareness, Array.from(controlled), null) + } + + // If the elected seeder left before it seeded, free the election so the next + // joiner seeds — otherwise the document would stay permanently empty. + if (room.seederSocketId === socketId && !isDocSeeded(room.doc)) { + room.seederSocketId = null + } + + // Drop the document + awareness once idle so no memory is held for a file with + // no active editors; a later joiner re-creates and re-seeds it. + if (room.controlledIds.size === 0) { + room.awareness.destroy() + room.doc.destroy() + fileDocRooms.delete(name) + } +} + +/** + * Registers the collaborative file-document handlers on a socket. Room id is the + * file id; joining requires workspace `write` (editing a document). Mirrors the + * workspace-files join shape (auth → readiness → validate → authorize → join), + * then runs the Yjs sync/awareness handshake. + */ +export function setupWorkspaceFileDocHandlers( + socket: AuthenticatedSocket, + roomManager: IRoomManager +) { + const io = roomManager.io + + socket.on(FILE_DOC_EVENTS.JOIN, async ({ fileId }: JoinFileDocPayload) => { + try { + const userId = socket.userId + const userName = socket.userName + + if (!userId || !userName) { + emitJoinError(socket, fileId, 'Authentication required', 'AUTHENTICATION_REQUIRED', false) + return + } + if (!roomManager.isReady()) { + emitJoinError(socket, fileId, 'Realtime unavailable', 'ROOM_MANAGER_UNAVAILABLE', true) + return + } + if (typeof fileId !== 'string' || fileId.length === 0) { + emitJoinError(socket, fileId, 'Invalid file id', 'INVALID_PAYLOAD', false) + return + } + + const room = fileDocRoom(fileId) + const name = roomName(room) + + let authorized: Awaited> + try { + authorized = await authorizeRoom({ userId, room, action: 'write' }) + } catch (error) { + logger.warn(`Error authorizing file-doc room for ${userId}:`, error) + emitJoinError( + socket, + fileId, + 'Failed to verify workspace access', + 'VERIFY_ACCESS_FAILED', + true + ) + return + } + if (!authorized.allowed) { + emitJoinError( + socket, + fileId, + authorized.status === 404 ? 'File not found' : 'Access denied to file', + authorized.status === 404 ? 'NOT_FOUND' : 'ACCESS_DENIED', + false + ) + return + } + + // Switched documents on the same socket — leave the previous one first (a + // socket edits at most one document). A duplicate join of the SAME room + // falls through and simply re-runs the sync handshake, idempotently. + const currentName = socketToRoomName.get(socket.id) + if (currentName && currentName !== name) { + socket.leave(currentName) + cleanupFileDocForSocket(socket.id, io) + } + + const entry = getOrCreateRoom(io, name) + if (!entry.controlledIds.has(socket.id)) entry.controlledIds.set(socket.id, new Set()) + socketToRoomName.set(socket.id, name) + socket.join(name) + + // Elect exactly one seeder for an unseeded document, so its initial content + // is imported from the stored markdown once and never duplicated. + const shouldSeed = !isDocSeeded(entry.doc) && entry.seederSocketId === null + if (shouldSeed) entry.seederSocketId = socket.id + + socket.emit(FILE_DOC_EVENTS.JOIN_SUCCESS, { fileId, shouldSeed }) + + // Begin the sync handshake: send the server's state (sync step 1). The + // client replies with its updates and requests the server's in return. + const syncEncoder = encoding.createEncoder() + encoding.writeVarUint(syncEncoder, FILE_DOC_MESSAGE_TYPE.SYNC) + syncProtocol.writeSyncStep1(syncEncoder, entry.doc) + socket.emit(FILE_DOC_EVENTS.MESSAGE, encoding.toUint8Array(syncEncoder)) + + // Send existing awareness so the new client immediately sees others' carets. + const states = entry.awareness.getStates() + if (states.size > 0) { + const awarenessEncoder = encoding.createEncoder() + encoding.writeVarUint(awarenessEncoder, FILE_DOC_MESSAGE_TYPE.AWARENESS) + encoding.writeVarUint8Array( + awarenessEncoder, + awarenessProtocol.encodeAwarenessUpdate(entry.awareness, Array.from(states.keys())) + ) + socket.emit(FILE_DOC_EVENTS.MESSAGE, encoding.toUint8Array(awarenessEncoder)) + } + + logger.info(`User ${userId} joined file-doc room ${fileId} (seed=${shouldSeed})`) + } catch (error) { + logger.error('Error joining file-doc room:', error) + try { + socket.leave(roomName(fileDocRoom(fileId))) + cleanupFileDocForSocket(socket.id, io) + } catch {} + emitJoinError(socket, fileId, 'Failed to join file document', 'JOIN_FAILED', true) + } + }) + + socket.on(FILE_DOC_EVENTS.MESSAGE, (data: unknown) => handleMessage(socket, data)) + + socket.on(FILE_DOC_EVENTS.LEAVE, (payload?: LeaveFileDocPayload) => { + try { + const name = socketToRoomName.get(socket.id) + if (!name) return + // Scope the leave to the named file when provided: a deferred leave from a + // prior document must not evict the socket from one it has since opened. + if (payload?.fileId && roomName(fileDocRoom(payload.fileId)) !== name) return + socket.leave(name) + cleanupFileDocForSocket(socket.id, io) + } catch (error) { + logger.error('Error leaving file-doc room:', error) + } + }) +} diff --git a/apps/realtime/src/handlers/index.ts b/apps/realtime/src/handlers/index.ts index 91573f1a1fc..9e034d77db1 100644 --- a/apps/realtime/src/handlers/index.ts +++ b/apps/realtime/src/handlers/index.ts @@ -1,4 +1,5 @@ import { setupConnectionHandlers } from '@/handlers/connection' +import { setupWorkspaceFileDocHandlers } from '@/handlers/file-doc' import { setupOperationsHandlers } from '@/handlers/operations' import { setupPresenceHandlers } from '@/handlers/presence' import { setupSubblocksHandlers } from '@/handlers/subblocks' @@ -15,5 +16,6 @@ export function setupAllHandlers(socket: AuthenticatedSocket, roomManager: IRoom setupVariablesHandlers(socket, roomManager) setupPresenceHandlers(socket, roomManager) setupWorkspaceFilesHandlers(socket, roomManager) + setupWorkspaceFileDocHandlers(socket, roomManager) setupConnectionHandlers(socket, roomManager) } diff --git a/bun.lock b/bun.lock index 033da3635b8..f1e8d8bde90 100644 --- a/bun.lock +++ b/bun.lock @@ -84,9 +84,12 @@ "@sim/workflow-types": "workspace:*", "@socket.io/redis-adapter": "8.3.0", "drizzle-orm": "^0.45.2", + "lib0": "0.2.117", "postgres": "^3.4.5", "redis": "5.10.0", "socket.io": "^4.8.1", + "y-protocols": "1.0.7", + "yjs": "13.6.31", "zod": "4.3.6", }, "devDependencies": { @@ -2955,6 +2958,8 @@ "isomorphic-ws": ["isomorphic-ws@5.0.0", "", { "peerDependencies": { "ws": "*" } }, "sha512-muId7Zzn9ywDsyXgTIafTry2sV3nySZeUDe6YedVd1Hvuuep5AsIlqK+XefWpYTyJG5e503F2xIuT2lcU6rCSw=="], + "isomorphic.js": ["isomorphic.js@0.2.5", "", {}, "sha512-PIeMbHqMt4DnUP3MA/Flc0HElYjMXArsw1qwJZcm9sqR8mq3l8NYizFMty0pWwE/tzIGH3EKK5+jes5mAr85yw=="], + "istanbul-lib-coverage": ["istanbul-lib-coverage@3.2.2", "", {}, "sha512-O8dpsF+r0WV/8MNRKfnmrtCWhuKjxrq2w+jpzBL5UZKTi2LeVWnWOmWRxFlesJONmc+wLAGvKQZEOanko0LFTg=="], "istanbul-lib-report": ["istanbul-lib-report@3.0.1", "", { "dependencies": { "istanbul-lib-coverage": "^3.0.0", "make-dir": "^4.0.0", "supports-color": "^7.1.0" } }, "sha512-GCfE1mtsHGOELCU8e/Z7YWzpmybrx/+dSTfLrvY8qRmaY6zXTKWn6WQIjaAFw069icm6GVMNkgu0NzI4iPZUNw=="], @@ -3017,6 +3022,8 @@ "leac": ["leac@0.6.0", "", {}, "sha512-y+SqErxb8h7nE/fiEX07jsbuhrpO9lL8eca7/Y1nuWV2moNlXhyd59iDGcRf6moVyDMbmTNzL40SUyrFU/yDpg=="], + "lib0": ["lib0@0.2.117", "", { "dependencies": { "isomorphic.js": "^0.2.4" }, "bin": { "0serve": "bin/0serve.js", "0gentesthtml": "bin/gentesthtml.js", "0ecdsa-generate-keypair": "bin/0ecdsa-generate-keypair.js" } }, "sha512-DeXj9X5xDCjgKLU/7RR+/HQEVzuuEUiwldwOGsHK/sfAfELGWEyTcf0x+uOvCvK3O2zPmZePXWL85vtia6GyZw=="], + "libbase64": ["libbase64@1.3.0", "", {}, "sha512-GgOXd0Eo6phYgh0DJtjQ2tO8dc0IVINtZJeARPeiIJqge+HdsWSuaDTe8ztQ7j/cONByDZ3zeB325AHiv5O0dg=="], "libmime": ["libmime@5.3.7", "", { "dependencies": { "encoding-japanese": "2.2.0", "iconv-lite": "0.6.3", "libbase64": "1.3.0", "libqp": "2.1.1" } }, "sha512-FlDb3Wtha8P01kTL3P9M+ZDNDWPKPmKHWaU/cG/lg5pfuAwdflVpZE+wm9m7pKmC5ww6s+zTxBKS1p6yl3KpSw=="], @@ -4147,6 +4154,8 @@ "xpath": ["xpath@0.0.34", "", {}, "sha512-FxF6+rkr1rNSQrhUNYrAFJpRXNzlDoMxeXN5qI84939ylEv3qqPFKa85Oxr6tDaJKqwW6KKyo2v26TSv3k6LeA=="], + "y-protocols": ["y-protocols@1.0.7", "", { "dependencies": { "lib0": "^0.2.85" }, "peerDependencies": { "yjs": "^13.0.0" } }, "sha512-YSVsLoXxO67J6eE/nV4AtFtT3QEotZf5sK5BHxFBXso7VDUT3Tx07IfA6hsu5Q5OmBdMkQVmFZ9QOA7fikWvnw=="], + "y18n": ["y18n@5.0.8", "", {}, "sha512-0pfFzegeDWJHJIAmTLRP2DwHjdF5s7jo9tuztdQxAhINCdvS+3nGINqPd00AphqJR/0LhANUS6/+7SCb98YOfA=="], "yallist": ["yallist@5.0.0", "", {}, "sha512-YgvUTfwqyc7UXVMrB+SImsVYSmTS8X/tSrtdNZMImM+n7+QTriRXyXim0mBrTXNeqzVF0KWGgHPeiyViFFrNDw=="], @@ -4159,6 +4168,8 @@ "yauzl": ["yauzl@3.4.0", "", { "dependencies": { "pend": "~1.2.0" } }, "sha512-jIH9yLR9wqr0wOS0TpBvo/g/2UgZH5qePVbjgRliiF0BYvOZyaBknKsF+x9Iht0O6sqgnB93rCICdOZFecJuDw=="], + "yjs": ["yjs@13.6.31", "", { "dependencies": { "lib0": "^0.2.99" } }, "sha512-Eq+5BRfbeGyqGVrTJL3bEcr8gKkxPuyuoHmAwpk52fDb8kOVMrfVSTRPd6yiGgX5Fskb96qCRjzjbRjrL4YEnw=="], + "yocto-queue": ["yocto-queue@0.1.0", "", {}, "sha512-rVksvsnNCdJ/ohGc6xgPwyN8eheCxsiLM8mxuE/t/mOVqJewPuO1miLpTHQiRgTKCLexL4MeAFVagts7HmNZ2Q=="], "yoctocolors": ["yoctocolors@2.1.2", "", {}, "sha512-CzhO+pFNo8ajLM2d2IW/R93ipy99LWjtwblvC1RsoSUMZgyLbYFr221TnSNT7GjGdYui6P459mw9JH/g/zW2ug=="], diff --git a/packages/platform-authz/src/rooms.ts b/packages/platform-authz/src/rooms.ts index 1fd9949333e..f8676cfde45 100644 --- a/packages/platform-authz/src/rooms.ts +++ b/packages/platform-authz/src/rooms.ts @@ -1,4 +1,4 @@ -import { db, workspace } from '@sim/db' +import { db, workspace, workspaceFiles } from '@sim/db' import { ROOM_TYPES, type RoomRef, type RoomType } from '@sim/realtime-protocol/rooms' import { and, eq, isNull } from 'drizzle-orm' import { getActiveWorkflowContext } from './workflow' @@ -37,6 +37,24 @@ async function resolveWorkspaceRoomWorkspace(workspaceId: string): Promise { + const [file] = await db + .select({ workspaceId: workspaceFiles.workspaceId }) + .from(workspaceFiles) + .where(and(eq(workspaceFiles.id, fileId), isNull(workspaceFiles.deletedAt))) + .limit(1) + + if (!file?.workspaceId) return null + return resolveWorkspaceRoomWorkspace(file.workspaceId) +} + /** * Single source of truth mapping each room type to its resource→workspace * lookup. Every realtime room is workspace-scoped and authorizes through the @@ -54,6 +72,8 @@ const ROOM_WORKSPACE_RESOLVERS: Record = { }, // A workspace-files room is addressed directly by its workspace id. [ROOM_TYPES.WORKSPACE_FILES]: resolveWorkspaceRoomWorkspace, + // A file-doc room is addressed by file id; resolve it to its workspace. + [ROOM_TYPES.WORKSPACE_FILE_DOC]: resolveFileDocWorkspace, } /** Resolves a room's owning workspace, or `null` if the room resource is gone. */ diff --git a/packages/realtime-protocol/package.json b/packages/realtime-protocol/package.json index 4f0ea685ebe..7023ec6a12a 100644 --- a/packages/realtime-protocol/package.json +++ b/packages/realtime-protocol/package.json @@ -25,6 +25,10 @@ "./rooms": { "types": "./src/rooms.ts", "default": "./src/rooms.ts" + }, + "./file-doc": { + "types": "./src/file-doc.ts", + "default": "./src/file-doc.ts" } }, "scripts": { diff --git a/packages/realtime-protocol/src/file-doc.ts b/packages/realtime-protocol/src/file-doc.ts new file mode 100644 index 00000000000..0829af27264 --- /dev/null +++ b/packages/realtime-protocol/src/file-doc.ts @@ -0,0 +1,83 @@ +/** + * Wire protocol for the collaborative file-document room + * ({@link ROOM_TYPES.WORKSPACE_FILE_DOC}). Live carets and text selection ride + * Yjs document sync + awareness over the shared Socket.IO connection. These are + * the event names, binary message tags, and join payloads that the server + * (`apps/realtime/src/handlers/file-doc.ts`) and the client provider + * (`apps/sim/.../file-doc`) must agree on exactly — the single source of truth + * for both sides so they can never drift. + * + * The binary channel uses the standard Yjs "websocket" framing: every + * {@link FILE_DOC_EVENTS.MESSAGE} payload is a `Uint8Array` whose first varUint + * is a {@link FILE_DOC_MESSAGE_TYPE} tag (sync protocol vs awareness protocol), + * so the client provider can reuse `y-protocols` verbatim. + */ + +/** Socket.IO event names for the file-document collaboration channel. */ +export const FILE_DOC_EVENTS = { + /** Client → server: join a file's collaborative session ({@link JoinFileDocPayload}). */ + JOIN: 'join-file-doc', + /** Server → client: join accepted ({@link JoinFileDocSuccess}). */ + JOIN_SUCCESS: 'join-file-doc-success', + /** Server → client: join rejected ({@link JoinFileDocError}). */ + JOIN_ERROR: 'join-file-doc-error', + /** Client → server: leave the session ({@link LeaveFileDocPayload}). */ + LEAVE: 'leave-file-doc', + /** Both directions: a framed Yjs message (binary), tagged by {@link FILE_DOC_MESSAGE_TYPE}. */ + MESSAGE: 'file-doc-message', +} as const + +/** + * The tag carried in the first varUint of a {@link FILE_DOC_EVENTS.MESSAGE} + * payload — the standard Yjs websocket framing distinguishing a document-sync + * message from an awareness (cursor/selection) message. + */ +export const FILE_DOC_MESSAGE_TYPE = { + SYNC: 0, + AWARENESS: 1, +} as const + +export type FileDocMessageType = (typeof FILE_DOC_MESSAGE_TYPE)[keyof typeof FILE_DOC_MESSAGE_TYPE] + +/** + * Where the client records that it has seeded the document's initial content, + * stored inside the Yjs document as `doc.getMap(configMap).get(flag) === true`. + * Because it lives in the CRDT it merges across clients and the server can read + * it — the server uses it to decide whether to re-elect a seeder when an elected + * one disconnects before seeding. Client and server MUST use these exact keys. + */ +export const FILE_DOC_SEED = { + configMap: 'config', + flag: 'initialContentLoaded', +} as const + +/** Client → server join request. `fileId` is the `workspace_files.id`. */ +export interface JoinFileDocPayload { + fileId: string +} + +/** Server → client acceptance of a {@link FILE_DOC_EVENTS.JOIN}. */ +export interface JoinFileDocSuccess { + fileId: string + /** + * Whether this client was elected to seed the document's initial content from + * the file's stored markdown. The server elects exactly one client per empty + * document, so seeding never duplicates even under a cold-start race; the + * client additionally guards on the CRDT `initialContentLoaded` flag to cover + * the case where an elected seeder disconnects before it seeds. + */ + shouldSeed: boolean +} + +/** Server → client rejection of a {@link FILE_DOC_EVENTS.JOIN}. */ +export interface JoinFileDocError { + fileId: string + error: string + code: string + retryable?: boolean +} + +/** Client → server leave request. */ +export interface LeaveFileDocPayload { + fileId: string +} diff --git a/packages/realtime-protocol/src/rooms.test.ts b/packages/realtime-protocol/src/rooms.test.ts index 6ab07de542e..cb37c3fca45 100644 --- a/packages/realtime-protocol/src/rooms.test.ts +++ b/packages/realtime-protocol/src/rooms.test.ts @@ -18,6 +18,20 @@ describe('roomName', () => { expect(roomName({ type: ROOM_TYPES.WORKSPACE_FILES, id: 'ws-123' })).toBe( 'workspace-files:ws-123' ) + expect(roomName({ type: ROOM_TYPES.WORKSPACE_FILE_DOC, id: 'file-123' })).toBe( + 'workspace-file-doc:file-123' + ) + }) + + it('keeps the file-doc and file-browser namespaces distinct for the same id', () => { + // `workspace-file-doc` must not be parsed as the `workspace-files` browser + // room (or vice versa): the prefix match is the whole segment before `:`. + const id = 'a1b2c3d4-uuid' + const doc = roomName({ type: ROOM_TYPES.WORKSPACE_FILE_DOC, id }) + const browser = roomName({ type: ROOM_TYPES.WORKSPACE_FILES, id }) + expect(doc).not.toBe(browser) + expect(parseRoomName(doc)).toEqual({ type: ROOM_TYPES.WORKSPACE_FILE_DOC, id }) + expect(parseRoomName(browser)).toEqual({ type: ROOM_TYPES.WORKSPACE_FILES, id }) }) it('never collides a namespaced room with a bare workflow id for real ids', () => { @@ -35,6 +49,7 @@ describe('parseRoomName', () => { const refs: RoomRef[] = [ { type: ROOM_TYPES.WORKFLOW, id: 'wf-123' }, { type: ROOM_TYPES.WORKSPACE_FILES, id: 'ws-456' }, + { type: ROOM_TYPES.WORKSPACE_FILE_DOC, id: 'file-789' }, ] for (const ref of refs) { expect(parseRoomName(roomName(ref))).toEqual(ref) diff --git a/packages/realtime-protocol/src/rooms.ts b/packages/realtime-protocol/src/rooms.ts index c03ad621d45..b81b65e72bb 100644 --- a/packages/realtime-protocol/src/rooms.ts +++ b/packages/realtime-protocol/src/rooms.ts @@ -22,6 +22,13 @@ export const ROOM_TYPES = { WORKFLOW: 'workflow', /** The workspace file browser (one room per workspace). */ WORKSPACE_FILES: 'workspace-files', + /** + * A single collaborative file document — the rich-text editor for one file + * (one room per file). Carries Yjs document sync + awareness (live carets and + * text selection), so its id space is the file id, distinct from the + * workspace-scoped {@link ROOM_TYPES.WORKSPACE_FILES} browser room. + */ + WORKSPACE_FILE_DOC: 'workspace-file-doc', } as const export type RoomType = (typeof ROOM_TYPES)[keyof typeof ROOM_TYPES] From a16ad4f74a2339ea2cb4b9bc2c1054de5ddf97e1 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 24 Jul 2026 15:22:03 -0700 Subject: [PATCH 2/9] fix(realtime): harden file-doc relay per review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses the review of the collaborative-document relay: - Seeder handoff: drive seeding via an explicit `SEED_REQUEST` event so that when the elected seeder disconnects before it seeds, the role is handed to a remaining client immediately — previously a co-opening pair could leave the survivor with a permanently empty document until an unrelated new joiner arrived. - Awareness ownership: clients now declare their Yjs `clientID` at join; the server binds it and rejects awareness frames for any other client id, so an authenticated peer can no longer forge or clear another collaborator's caret. - Malformed-frame containment: the binary message handler decodes inside a try/catch, dropping a bad frame with a warning instead of letting it escape as a process-level exception. Note on post-join permission revocation: consistent with the existing realtime authz model (join-time gate), and the durable markdown-mirror save re-checks write permission on every write, so a revoked user cannot persist changes. Periodic in-session revalidation is a follow-up. Tests: +spoofed-awareness-dropped, +malformed-frame-contained, +seeder-handoff-to-remaining; full realtime suite (140) green. --- apps/realtime/src/handlers/file-doc.test.ts | 189 +++++++++++++------- apps/realtime/src/handlers/file-doc.ts | 168 ++++++++++------- packages/realtime-protocol/src/file-doc.ts | 31 +++- 3 files changed, 251 insertions(+), 137 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc.test.ts b/apps/realtime/src/handlers/file-doc.test.ts index 38ab9433e5c..df72c1c3841 100644 --- a/apps/realtime/src/handlers/file-doc.test.ts +++ b/apps/realtime/src/handlers/file-doc.test.ts @@ -23,12 +23,24 @@ type Handler = (payload?: unknown) => Promise | void const ROOM_NAME = 'workspace-file-doc:file-1' -/** A shared Socket.IO `io` mock whose `to().except().emit()` chain is inspectable. */ +interface SentMessage { + target: string + except?: string + event: string + payload: unknown +} + +/** An `io` mock that records every server-originated emit with its target/except. */ function createIo() { - const emit = vi.fn() - const except = vi.fn().mockReturnValue({ emit }) - const to = vi.fn().mockReturnValue({ except, emit }) - return { io: { to } as unknown as IRoomManager['io'], to, except, emit } + const sent: SentMessage[] = [] + const to = vi.fn((target: string) => ({ + except: (exclude: string) => ({ + emit: (event: string, payload: unknown) => + sent.push({ target, except: exclude, event, payload }), + }), + emit: (event: string, payload: unknown) => sent.push({ target, event, payload }), + })) + return { io: { to } as unknown as IRoomManager['io'], sent } } function createSocket(id: string, overrides?: Record) { @@ -76,13 +88,26 @@ function frame(type: number, write: (encoder: encoding.Encoder) => void): Uint8A return encoding.toUint8Array(encoder) } -/** The last `JOIN_SUCCESS` payload emitted on a socket. */ -function joinSuccess(socket: { emit: ReturnType }) { +/** Build a real awareness frame carrying a single client's state. */ +function awarenessFrame(clientId: number, name: string): { frame: Uint8Array; clientId: number } { + const doc = new Y.Doc() + // Force a specific clientID so the test can bind/spoof deliberately. + doc.clientID = clientId + const awareness = new awarenessProtocol.Awareness(doc) + awareness.setLocalStateField('user', { name }) + const update = awarenessProtocol.encodeAwarenessUpdate(awareness, [clientId]) + return { + frame: frame(FILE_DOC_MESSAGE_TYPE.AWARENESS, (e) => encoding.writeVarUint8Array(e, update)), + clientId, + } +} + +function joinSuccessFileId(socket: { emit: ReturnType }) { const calls = socket.emit.mock.calls.filter( (call: unknown[]) => call[0] === FILE_DOC_EVENTS.JOIN_SUCCESS ) const last = calls[calls.length - 1] - return last?.[1] as { fileId: string; shouldSeed: boolean } | undefined + return (last?.[1] as { fileId: string } | undefined)?.fileId } describe('setupWorkspaceFileDocHandlers', () => { @@ -108,7 +133,7 @@ describe('setupWorkspaceFileDocHandlers', () => { const { io } = createIo() const { socket, handlers } = setup('socket-1', io, { userId: undefined, userName: undefined }) - await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) + await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 }) expect(socket.emit).toHaveBeenCalledWith( FILE_DOC_EVENTS.JOIN_ERROR, @@ -124,7 +149,7 @@ describe('setupWorkspaceFileDocHandlers', () => { createRoomManager(io, { isReady: vi.fn().mockReturnValue(false) }) ) - await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) + await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 }) expect(socket.emit).toHaveBeenCalledWith( FILE_DOC_EVENTS.JOIN_ERROR, @@ -132,11 +157,12 @@ describe('setupWorkspaceFileDocHandlers', () => { ) }) - it('rejects an invalid file id before authorizing', async () => { + it('rejects a payload missing the file id or client id before authorizing', async () => { const { io } = createIo() const { socket, handlers } = setup('socket-1', io) - await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: '' }) + await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: '', clientId: 1 }) + await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) expect(socket.emit).toHaveBeenCalledWith( FILE_DOC_EVENTS.JOIN_ERROR, @@ -150,7 +176,7 @@ describe('setupWorkspaceFileDocHandlers', () => { const { io } = createIo() const { socket, handlers } = setup('socket-1', io) - await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) + await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 }) expect(mockAuthorizeRoom).toHaveBeenCalledWith( expect.objectContaining({ @@ -164,48 +190,51 @@ describe('setupWorkspaceFileDocHandlers', () => { ) }) - it('joins the room, elects the first client as seeder, and sends sync step 1', async () => { - const { io, to } = createIo() + it('joins the room, sends sync step 1, and asks the first client to seed', async () => { + const { io, sent } = createIo() const { socket, handlers } = setup('socket-1', io) - await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) + await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 }) expect(socket.join).toHaveBeenCalledWith(ROOM_NAME) - expect(joinSuccess(socket)).toEqual({ fileId: 'file-1', shouldSeed: true }) + expect(joinSuccessFileId(socket)).toBe('file-1') // A binary sync-step-1 message (type tag 0) is sent to kick off the handshake. const syncMessage = socket.emit.mock.calls.find( ([event, payload]) => event === FILE_DOC_EVENTS.MESSAGE && payload instanceof Uint8Array ) - expect(syncMessage).toBeDefined() expect((syncMessage?.[1] as Uint8Array)[0]).toBe(FILE_DOC_MESSAGE_TYPE.SYNC) - // A fresh join with no doc updates yet performs no room broadcast. - expect(to).not.toHaveBeenCalled() + + // The lone joiner is elected to seed the empty document. + const seed = sent.find((m) => m.event === FILE_DOC_EVENTS.SEED_REQUEST) + expect(seed).toEqual({ + target: 'socket-1', + event: FILE_DOC_EVENTS.SEED_REQUEST, + payload: { fileId: 'file-1' }, + }) }) - it('elects only one seeder across concurrent joiners of the same file', async () => { - const { io } = createIo() + it('asks only one client to seed across concurrent joiners of the same file', async () => { + const { io, sent } = createIo() const a = setup('socket-a', io) const b = setup('socket-b', io) - await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) - await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) + await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 }) + await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 2 }) - expect(joinSuccess(a.socket)?.shouldSeed).toBe(true) - expect(joinSuccess(b.socket)?.shouldSeed).toBe(false) + const seeds = sent.filter((m) => m.event === FILE_DOC_EVENTS.SEED_REQUEST) + expect(seeds).toHaveLength(1) + expect(seeds[0].target).toBe('socket-a') }) it('relays a document update to the rest of the room, excluding the sender', async () => { - const { io, to, except, emit } = createIo() + const { io, sent } = createIo() const a = setup('socket-a', io) const b = setup('socket-b', io) - await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) - await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) - to.mockClear() - except.mockClear() - emit.mockClear() + await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 }) + await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 2 }) + sent.length = 0 - // A real client edit, framed as a sync update. const clientDoc = new Y.Doc() clientDoc.getText('default').insert(0, 'hello') const update = Y.encodeStateAsUpdate(clientDoc) @@ -213,63 +242,85 @@ describe('setupWorkspaceFileDocHandlers', () => { frame(FILE_DOC_MESSAGE_TYPE.SYNC, (e) => syncProtocol.writeUpdate(e, update)) ) - expect(to).toHaveBeenCalledWith(ROOM_NAME) - expect(except).toHaveBeenCalledWith('socket-a') - const relayed = emit.mock.calls.find(([event]) => event === FILE_DOC_EVENTS.MESSAGE) - expect(relayed).toBeDefined() - expect((relayed?.[1] as Uint8Array)[0]).toBe(FILE_DOC_MESSAGE_TYPE.SYNC) + const relayed = sent.find((m) => m.event === FILE_DOC_EVENTS.MESSAGE) + expect(relayed?.target).toBe(ROOM_NAME) + expect(relayed?.except).toBe('socket-a') + expect((relayed?.payload as Uint8Array)[0]).toBe(FILE_DOC_MESSAGE_TYPE.SYNC) }) - it('relays awareness (cursor/selection) updates to the room, excluding the sender', async () => { - const { io, except, emit } = createIo() + it('relays an owned awareness update to the room, excluding the sender', async () => { + const { io, sent } = createIo() + const { frame: awFrame, clientId } = awarenessFrame(4242, 'Ada') const a = setup('socket-a', io) const b = setup('socket-b', io) - await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) - await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) - except.mockClear() - emit.mockClear() + await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId }) + await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 2 }) + sent.length = 0 - const clientDoc = new Y.Doc() - const clientAwareness = new awarenessProtocol.Awareness(clientDoc) - clientAwareness.setLocalStateField('user', { name: 'Ada', color: '#f783ac' }) - const awarenessUpdate = awarenessProtocol.encodeAwarenessUpdate(clientAwareness, [ - clientDoc.clientID, - ]) - a.handlers[FILE_DOC_EVENTS.MESSAGE]( - frame(FILE_DOC_MESSAGE_TYPE.AWARENESS, (e) => encoding.writeVarUint8Array(e, awarenessUpdate)) + a.handlers[FILE_DOC_EVENTS.MESSAGE](awFrame) + + const relayed = sent.find( + (m) => + m.event === FILE_DOC_EVENTS.MESSAGE && + (m.payload as Uint8Array)[0] === FILE_DOC_MESSAGE_TYPE.AWARENESS ) + expect(relayed?.except).toBe('socket-a') + }) + + it('drops an awareness frame that spoofs another client id', async () => { + const { io, sent } = createIo() + // socket-a binds client id 100 at join, but sends awareness for client 999. + const { frame: spoof } = awarenessFrame(999, 'Mallory') + const a = setup('socket-a', io) + await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 100 }) + sent.length = 0 + + a.handlers[FILE_DOC_EVENTS.MESSAGE](spoof) - expect(except).toHaveBeenCalledWith('socket-a') - const relayed = emit.mock.calls.find(([event]) => event === FILE_DOC_EVENTS.MESSAGE) - expect((relayed?.[1] as Uint8Array)[0]).toBe(FILE_DOC_MESSAGE_TYPE.AWARENESS) + const relayed = sent.find( + (m) => + m.event === FILE_DOC_EVENTS.MESSAGE && + (m.payload as Uint8Array)[0] === FILE_DOC_MESSAGE_TYPE.AWARENESS + ) + expect(relayed).toBeUndefined() }) - it('re-elects a seeder when the elected one disconnects before seeding', async () => { + it('drops a malformed frame without throwing', async () => { const { io } = createIo() const a = setup('socket-a', io) + await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 }) + + expect(() => + a.handlers[FILE_DOC_EVENTS.MESSAGE](new Uint8Array([255, 254, 253, 200])) + ).not.toThrow() + }) + + it('hands the seeder role to a remaining client when the elected one leaves before seeding', async () => { + const { io, sent } = createIo() + const a = setup('socket-a', io) const b = setup('socket-b', io) - await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) - await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) - expect(joinSuccess(a.socket)?.shouldSeed).toBe(true) + await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 }) + await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 2 }) + sent.length = 0 - // The seeder leaves before it ever seeded; b remains so the room survives. + // The seeder leaves before it ever seeded; b remains. cleanupFileDocForSocket('socket-a', io) - const c = setup('socket-c', io) - await c.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) - expect(joinSuccess(c.socket)?.shouldSeed).toBe(true) + const seed = sent.find((m) => m.event === FILE_DOC_EVENTS.SEED_REQUEST) + expect(seed?.target).toBe('socket-b') }) it('drops the document when the last editor leaves, re-seeding a fresh joiner', async () => { - const { io } = createIo() + const { io, sent } = createIo() const a = setup('socket-a', io) - await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) + await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 }) cleanupFileDocForSocket('socket-a', io) + sent.length = 0 - // With the room dropped, the next joiner starts a brand-new document and is - // elected to seed again. const b = setup('socket-b', io) - await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1' }) - expect(joinSuccess(b.socket)?.shouldSeed).toBe(true) + await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 2 }) + + const seed = sent.find((m) => m.event === FILE_DOC_EVENTS.SEED_REQUEST) + expect(seed?.target).toBe('socket-b') }) }) diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index e626b672982..cee52dbb901 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -38,11 +38,17 @@ const logger = createLogger('FileDocHandlers') * backend (y-redis / Hocuspocus) — out of scope here. */ interface FileDocRoom { + /** The `workspace_files.id` this room edits (for seed-request payloads). */ + fileId: string doc: Y.Doc awareness: awarenessProtocol.Awareness - /** socketId → the awareness clientIDs it controls, cleared on its disconnect. */ - controlledIds: Map> - /** The socket elected to seed initial content, or `null` if unseeded/unelected. */ + /** + * socketId → the awareness clientID it declared at join. A socket owns exactly + * one clientID and may only publish/remove awareness for that one, so an + * authenticated peer cannot forge or clear another collaborator's presence. + */ + ownedClientId: Map + /** The socket currently elected to seed initial content, or `null`. */ seederSocketId: string | null } @@ -88,12 +94,46 @@ function toUint8Array(data: unknown): Uint8Array | null { return null } +/** + * Decode the client IDs an awareness update carries, without applying it, to + * check a frame only touches its sender's own presence. Mirrors the wire format + * of `awarenessProtocol.encodeAwarenessUpdate`: a count, then per client a + * varUint id, a varUint clock, and a varString state. + */ +function awarenessUpdateClientIds(update: Uint8Array): number[] { + const decoder = decoding.createDecoder(update) + const count = decoding.readVarUint(decoder) + const ids: number[] = [] + for (let i = 0; i < count; i++) { + ids.push(decoding.readVarUint(decoder)) + decoding.readVarUint(decoder) // clock + decoding.readVarString(decoder) // state json + } + return ids +} + +/** + * Elect a client to seed an unseeded document and ask it to import the stored + * markdown, if one is needed and not already assigned. Called after a join (to + * elect the newcomer) and after the elected seeder leaves (to hand the role to a + * remaining client, so the document never stays permanently empty). A no-op once + * the document is seeded or a seeder is already assigned. + */ +function electSeederIfNeeded(io: Server, room: FileDocRoom) { + if (room.seederSocketId !== null || isDocSeeded(room.doc)) return + const next = room.ownedClientId.keys().next() + if (next.done) return + room.seederSocketId = next.value + io.to(next.value).emit(FILE_DOC_EVENTS.SEED_REQUEST, { fileId: room.fileId }) +} + /** * Get (or lazily create) the authoritative document for a room, wiring the two * relay handlers exactly once: document updates and awareness changes are * broadcast to the room, excluding the origin socket (it already applied them). */ -function getOrCreateRoom(io: Server, name: string): FileDocRoom { +function getOrCreateRoom(io: Server, ref: RoomRef): FileDocRoom { + const name = roomName(ref) const existing = fileDocRooms.get(name) if (existing) return existing @@ -102,7 +142,13 @@ function getOrCreateRoom(io: Server, name: string): FileDocRoom { // The server holds no cursor of its own; it only relays clients' awareness. awareness.setLocalState(null) - const room: FileDocRoom = { doc, awareness, controlledIds: new Map(), seederSocketId: null } + const room: FileDocRoom = { + fileId: ref.id, + doc, + awareness, + ownedClientId: new Map(), + seederSocketId: null, + } fileDocRooms.set(name, room) doc.on('update', (update: Uint8Array, origin: unknown) => { @@ -113,14 +159,6 @@ function getOrCreateRoom(io: Server, name: string): FileDocRoom { }) awareness.on('update', ({ added, updated, removed }: AwarenessChange, origin: unknown) => { - const from = originSocketId(origin) - if (from) { - const controlled = room.controlledIds.get(from) - if (controlled) { - for (const id of added) controlled.add(id) - for (const id of removed) controlled.delete(id) - } - } const changed = added.concat(updated, removed) if (changed.length === 0) return const encoder = encoding.createEncoder() @@ -129,7 +167,7 @@ function getOrCreateRoom(io: Server, name: string): FileDocRoom { encoder, awarenessProtocol.encodeAwarenessUpdate(awareness, changed) ) - broadcast(io, name, encoding.toUint8Array(encoder), from) + broadcast(io, name, encoding.toUint8Array(encoder), originSocketId(origin)) }) return room @@ -159,39 +197,50 @@ function handleMessage(socket: AuthenticatedSocket, data: unknown) { const bytes = toUint8Array(data) if (!bytes) return - const decoder = decoding.createDecoder(bytes) - const messageType = decoding.readVarUint(decoder) - - switch (messageType) { - case FILE_DOC_MESSAGE_TYPE.SYNC: { - const encoder = encoding.createEncoder() - encoding.writeVarUint(encoder, FILE_DOC_MESSAGE_TYPE.SYNC) - // `socket.id` is the transaction origin, so the doc's `update` handler - // excludes this sender when relaying the applied update to the room. - syncProtocol.readSyncMessage(decoder, encoder, room.doc, socket.id) - // A reply longer than the 1-byte type tag is a sync step 2 (or step 1) - // destined for the sender only; applied updates fan out via `doc.on`. - if (encoding.length(encoder) > 1) { - socket.emit(FILE_DOC_EVENTS.MESSAGE, encoding.toUint8Array(encoder)) + // A malformed frame from any client must never escape as a process-level + // exception; drop it and keep the relay running. + try { + const decoder = decoding.createDecoder(bytes) + const messageType = decoding.readVarUint(decoder) + + switch (messageType) { + case FILE_DOC_MESSAGE_TYPE.SYNC: { + const encoder = encoding.createEncoder() + encoding.writeVarUint(encoder, FILE_DOC_MESSAGE_TYPE.SYNC) + // `socket.id` is the transaction origin, so the doc's `update` handler + // excludes this sender when relaying the applied update to the room. + syncProtocol.readSyncMessage(decoder, encoder, room.doc, socket.id) + // A reply longer than the 1-byte type tag is a sync step 2 (or step 1) + // destined for the sender only; applied updates fan out via `doc.on`. + if (encoding.length(encoder) > 1) { + socket.emit(FILE_DOC_EVENTS.MESSAGE, encoding.toUint8Array(encoder)) + } + break } - break - } - case FILE_DOC_MESSAGE_TYPE.AWARENESS: { - awarenessProtocol.applyAwarenessUpdate( - room.awareness, - decoding.readVarUint8Array(decoder), - socket.id - ) - break + case FILE_DOC_MESSAGE_TYPE.AWARENESS: { + const update = decoding.readVarUint8Array(decoder) + // Enforce presence ownership: a socket may only publish/remove awareness + // for the clientID it bound at join, so a peer cannot spoof or clear + // another collaborator's caret. + const owned = room.ownedClientId.get(socket.id) + if (owned === undefined || awarenessUpdateClientIds(update).some((id) => id !== owned)) { + logger.warn('Dropping awareness frame for an unowned client id', { socketId: socket.id }) + return + } + awarenessProtocol.applyAwarenessUpdate(room.awareness, update, socket.id) + break + } + default: + logger.warn('Unknown file-doc message type', { messageType }) } - default: - logger.warn('Unknown file-doc message type', { messageType }) + } catch (error) { + logger.warn('Dropping malformed file-doc frame', { socketId: socket.id, error }) } } /** * Remove a socket from its file-doc room: clear its awareness state (so its caret - * disappears for everyone else), free the seeder election if it left before + * disappears for everyone else), hand off the seeder role if it left before * seeding, and drop the room's document when the last collaborator leaves. * Exported for the disconnect handler; safe to call for a socket in no room. */ @@ -203,23 +252,24 @@ export function cleanupFileDocForSocket(socketId: string, io: Server): void { const room = fileDocRooms.get(name) if (!room) return - const controlled = room.controlledIds.get(socketId) - room.controlledIds.delete(socketId) - if (controlled && controlled.size > 0) { + const owned = room.ownedClientId.get(socketId) + room.ownedClientId.delete(socketId) + if (owned !== undefined) { // Fires the awareness `update` handler with a non-socket origin → the removal // is broadcast to every remaining client, so the departed caret vanishes. - awarenessProtocol.removeAwarenessStates(room.awareness, Array.from(controlled), null) + awarenessProtocol.removeAwarenessStates(room.awareness, [owned], null) } - // If the elected seeder left before it seeded, free the election so the next - // joiner seeds — otherwise the document would stay permanently empty. - if (room.seederSocketId === socketId && !isDocSeeded(room.doc)) { + // Hand off the seeder role: if the elected seeder left before it seeded, elect + // a remaining client so the document doesn't stay permanently empty. + if (room.seederSocketId === socketId) { room.seederSocketId = null + electSeederIfNeeded(io, room) } // Drop the document + awareness once idle so no memory is held for a file with // no active editors; a later joiner re-creates and re-seeds it. - if (room.controlledIds.size === 0) { + if (room.ownedClientId.size === 0) { room.awareness.destroy() room.doc.destroy() fileDocRooms.delete(name) @@ -238,7 +288,7 @@ export function setupWorkspaceFileDocHandlers( ) { const io = roomManager.io - socket.on(FILE_DOC_EVENTS.JOIN, async ({ fileId }: JoinFileDocPayload) => { + socket.on(FILE_DOC_EVENTS.JOIN, async ({ fileId, clientId }: JoinFileDocPayload) => { try { const userId = socket.userId const userName = socket.userName @@ -251,8 +301,8 @@ export function setupWorkspaceFileDocHandlers( emitJoinError(socket, fileId, 'Realtime unavailable', 'ROOM_MANAGER_UNAVAILABLE', true) return } - if (typeof fileId !== 'string' || fileId.length === 0) { - emitJoinError(socket, fileId, 'Invalid file id', 'INVALID_PAYLOAD', false) + if (typeof fileId !== 'string' || fileId.length === 0 || typeof clientId !== 'number') { + emitJoinError(socket, fileId, 'Invalid join payload', 'INVALID_PAYLOAD', false) return } @@ -293,17 +343,12 @@ export function setupWorkspaceFileDocHandlers( cleanupFileDocForSocket(socket.id, io) } - const entry = getOrCreateRoom(io, name) - if (!entry.controlledIds.has(socket.id)) entry.controlledIds.set(socket.id, new Set()) + const entry = getOrCreateRoom(io, room) + entry.ownedClientId.set(socket.id, clientId) socketToRoomName.set(socket.id, name) socket.join(name) - // Elect exactly one seeder for an unseeded document, so its initial content - // is imported from the stored markdown once and never duplicated. - const shouldSeed = !isDocSeeded(entry.doc) && entry.seederSocketId === null - if (shouldSeed) entry.seederSocketId = socket.id - - socket.emit(FILE_DOC_EVENTS.JOIN_SUCCESS, { fileId, shouldSeed }) + socket.emit(FILE_DOC_EVENTS.JOIN_SUCCESS, { fileId }) // Begin the sync handshake: send the server's state (sync step 1). The // client replies with its updates and requests the server's in return. @@ -324,7 +369,10 @@ export function setupWorkspaceFileDocHandlers( socket.emit(FILE_DOC_EVENTS.MESSAGE, encoding.toUint8Array(awarenessEncoder)) } - logger.info(`User ${userId} joined file-doc room ${fileId} (seed=${shouldSeed})`) + // Ask a client to seed the initial content if the document is still empty. + electSeederIfNeeded(io, entry) + + logger.info(`User ${userId} joined file-doc room ${fileId}`) } catch (error) { logger.error('Error joining file-doc room:', error) try { diff --git a/packages/realtime-protocol/src/file-doc.ts b/packages/realtime-protocol/src/file-doc.ts index 0829af27264..240b0d9a6b6 100644 --- a/packages/realtime-protocol/src/file-doc.ts +++ b/packages/realtime-protocol/src/file-doc.ts @@ -21,6 +21,13 @@ export const FILE_DOC_EVENTS = { JOIN_SUCCESS: 'join-file-doc-success', /** Server → client: join rejected ({@link JoinFileDocError}). */ JOIN_ERROR: 'join-file-doc-error', + /** + * Server → client: this client is elected to seed the document's initial + * content from the file's stored markdown ({@link SeedRequestPayload}). Sent to + * exactly one client of an unseeded document — at join, or later to a remaining + * client if the previously-elected seeder disconnects before it seeds. + */ + SEED_REQUEST: 'file-doc-seed-request', /** Client → server: leave the session ({@link LeaveFileDocPayload}). */ LEAVE: 'leave-file-doc', /** Both directions: a framed Yjs message (binary), tagged by {@link FILE_DOC_MESSAGE_TYPE}. */ @@ -54,19 +61,27 @@ export const FILE_DOC_SEED = { /** Client → server join request. `fileId` is the `workspace_files.id`. */ export interface JoinFileDocPayload { fileId: string + /** + * The joining Yjs document's `clientID`. The server binds it to this socket so + * a client can only publish/remove awareness (cursor/selection) for its own + * client — an authenticated peer cannot forge or clear another's presence. + */ + clientId: number } /** Server → client acceptance of a {@link FILE_DOC_EVENTS.JOIN}. */ export interface JoinFileDocSuccess { fileId: string - /** - * Whether this client was elected to seed the document's initial content from - * the file's stored markdown. The server elects exactly one client per empty - * document, so seeding never duplicates even under a cold-start race; the - * client additionally guards on the CRDT `initialContentLoaded` flag to cover - * the case where an elected seeder disconnects before it seeds. - */ - shouldSeed: boolean +} + +/** + * Server → client seed election ({@link FILE_DOC_EVENTS.SEED_REQUEST}). The + * recipient imports the file's stored markdown into the (empty) document. The + * client still guards on the CRDT `initialContentLoaded` flag so a re-election + * that races an in-flight seed can never duplicate content. + */ +export interface SeedRequestPayload { + fileId: string } /** Server → client rejection of a {@link FILE_DOC_EVENTS.JOIN}. */ From 1bf1a1b4761fc9e777f3a598783762d5b68bfca7 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 24 Jul 2026 15:30:10 -0700 Subject: [PATCH 3/9] fix(realtime): enforce client-id uniqueness + clear ghost carets MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round 2 of review on the file-doc relay: - Duplicate client-id spoofing bypass: the per-socket ownership binding did not ensure a client id is owned by at most one LIVE socket, so a writer could bind an active peer's id and pass the per-frame ownership check. Join now rejects a duplicate whose current owner is still connected (CLIENT_ID_IN_USE), and reclaims a stale binding whose prior owner is gone — so a reconnect reusing the same Yjs client id still works (single-replica registry, matching the existing in-memory Y.Doc assumption). - Ghost carets: a same-socket rejoin with a changed client id now clears the old client's awareness before rebinding, instead of leaving a lingering caret. Tests: +reject-live-duplicate, +reclaim-stale-on-reconnect, +clear-ghost-caret. --- apps/realtime/src/handlers/file-doc.test.ts | 57 ++++++++++++++++++++- apps/realtime/src/handlers/file-doc.ts | 24 +++++++++ 2 files changed, 80 insertions(+), 1 deletion(-) diff --git a/apps/realtime/src/handlers/file-doc.test.ts b/apps/realtime/src/handlers/file-doc.test.ts index df72c1c3841..46ecc62e54a 100644 --- a/apps/realtime/src/handlers/file-doc.test.ts +++ b/apps/realtime/src/handlers/file-doc.test.ts @@ -33,6 +33,8 @@ interface SentMessage { /** An `io` mock that records every server-originated emit with its target/except. */ function createIo() { const sent: SentMessage[] = [] + // Socket ids the mock considers currently connected (for client-id ownership). + const connected = new Set() const to = vi.fn((target: string) => ({ except: (exclude: string) => ({ emit: (event: string, payload: unknown) => @@ -40,7 +42,11 @@ function createIo() { }), emit: (event: string, payload: unknown) => sent.push({ target, event, payload }), })) - return { io: { to } as unknown as IRoomManager['io'], sent } + const io = { + to, + sockets: { sockets: { has: (id: string) => connected.has(id) } }, + } as unknown as IRoomManager['io'] + return { io, sent, connected } } function createSocket(id: string, overrides?: Record) { @@ -285,6 +291,55 @@ describe('setupWorkspaceFileDocHandlers', () => { expect(relayed).toBeUndefined() }) + it("rejects a join that binds a live peer's client id", async () => { + const { io, connected } = createIo() + connected.add('socket-a') + const a = setup('socket-a', io) + const b = setup('socket-b', io) + + await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 7 }) + await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 7 }) + + expect(b.socket.emit).toHaveBeenCalledWith( + FILE_DOC_EVENTS.JOIN_ERROR, + expect.objectContaining({ code: 'CLIENT_ID_IN_USE' }) + ) + expect(b.socket.join).not.toHaveBeenCalled() + }) + + it('reclaims a client id whose prior owner is no longer connected (reconnect)', async () => { + const { io } = createIo() + // socket-a joined with client id 7 but is NOT in the connected set (its + // disconnect cleanup has not run yet); socket-b reconnects reusing id 7. + const a = setup('socket-a', io) + await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 7 }) + const b = setup('socket-b', io) + + await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 7 }) + + expect(joinSuccessFileId(b.socket)).toBe('file-1') + expect(b.socket.join).toHaveBeenCalledWith(ROOM_NAME) + }) + + it('clears a departed caret when a socket rejoins the room with a new client id', async () => { + const { io, sent } = createIo() + const { frame: awFrame } = awarenessFrame(500, 'A') + const a = setup('socket-a', io) + await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 500 }) + a.handlers[FILE_DOC_EVENTS.MESSAGE](awFrame) + sent.length = 0 + + await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 501 }) + + // The old client (500) caret removal is broadcast to the room. + const removal = sent.find( + (m) => + m.event === FILE_DOC_EVENTS.MESSAGE && + (m.payload as Uint8Array)[0] === FILE_DOC_MESSAGE_TYPE.AWARENESS + ) + expect(removal).toBeDefined() + }) + it('drops a malformed frame without throwing', async () => { const { io } = createIo() const a = setup('socket-a', io) diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index cee52dbb901..c2eefe0c45e 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -344,6 +344,30 @@ export function setupWorkspaceFileDocHandlers( } const entry = getOrCreateRoom(io, room) + + // A same socket rejoining with a NEW client id: clear its old caret so it + // doesn't linger as a ghost after the binding is overwritten below. + const previousClientId = entry.ownedClientId.get(socket.id) + if (previousClientId !== undefined && previousClientId !== clientId) { + awarenessProtocol.removeAwarenessStates(entry.awareness, [previousClientId], null) + } + + // A client id must be owned by at most one LIVE socket, or a peer could bind + // an active collaborator's id and pass the per-frame ownership check to + // spoof/clear its caret. Reject a live duplicate; reclaim a stale binding + // (a reconnect reuses the same Yjs client id, and its prior socket may not + // be cleaned up yet). `io.sockets.sockets` is this replica's registry — the + // same single-replica assumption the in-memory Y.Doc already relies on. + for (const [otherSid, otherCid] of entry.ownedClientId) { + if (otherCid !== clientId || otherSid === socket.id) continue + if (io.sockets.sockets.has(otherSid)) { + emitJoinError(socket, fileId, 'Client id already in use', 'CLIENT_ID_IN_USE', false) + return + } + entry.ownedClientId.delete(otherSid) + awarenessProtocol.removeAwarenessStates(entry.awareness, [otherCid], null) + } + entry.ownedClientId.set(socket.id, clientId) socketToRoomName.set(socket.id, name) socket.join(name) From 56af96454d42e78d106330644f4a814ebdf9c946 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 24 Jul 2026 15:44:16 -0700 Subject: [PATCH 4/9] fix(realtime): key client-id ownership by user to fix reconnect lockout Independent audit found that the client-id uniqueness check (round 2) could lock a user out of a document for up to the ping-timeout window: on a silent network drop the old socket lingers in `io.sockets.sockets`, so the reconnecting socket (same reused Yjs client id) was rejected CLIENT_ID_IN_USE (non-retryable) until the dead socket timed out. Fix: discriminate reconnect from spoof by the OWNING USER, not socket liveness. A client id already held is reclaimed when the joining user is the same (a reconnect reusing its Yjs client id) and rejected only when a DIFFERENT user tries to bind it. This is race-free and drops the io.sockets registry dependency. Also: document the seed contract (content + initialContentLoaded flag must be written in one Yjs transaction, or a re-election could duplicate content) and the reserved 'config' Y.Map name. Tests updated: spoof = different user rejected; reconnect = same user reclaims. --- apps/realtime/src/handlers/file-doc.test.ts | 23 +++---- apps/realtime/src/handlers/file-doc.ts | 67 ++++++++++++--------- packages/realtime-protocol/src/file-doc.ts | 7 +++ 3 files changed, 53 insertions(+), 44 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc.test.ts b/apps/realtime/src/handlers/file-doc.test.ts index 46ecc62e54a..97abca8e7eb 100644 --- a/apps/realtime/src/handlers/file-doc.test.ts +++ b/apps/realtime/src/handlers/file-doc.test.ts @@ -33,8 +33,6 @@ interface SentMessage { /** An `io` mock that records every server-originated emit with its target/except. */ function createIo() { const sent: SentMessage[] = [] - // Socket ids the mock considers currently connected (for client-id ownership). - const connected = new Set() const to = vi.fn((target: string) => ({ except: (exclude: string) => ({ emit: (event: string, payload: unknown) => @@ -42,11 +40,7 @@ function createIo() { }), emit: (event: string, payload: unknown) => sent.push({ target, event, payload }), })) - const io = { - to, - sockets: { sockets: { has: (id: string) => connected.has(id) } }, - } as unknown as IRoomManager['io'] - return { io, sent, connected } + return { io: { to } as unknown as IRoomManager['io'], sent } } function createSocket(id: string, overrides?: Record) { @@ -291,11 +285,10 @@ describe('setupWorkspaceFileDocHandlers', () => { expect(relayed).toBeUndefined() }) - it("rejects a join that binds a live peer's client id", async () => { - const { io, connected } = createIo() - connected.add('socket-a') + it("rejects a DIFFERENT user binding a peer's client id (spoof)", async () => { + const { io } = createIo() const a = setup('socket-a', io) - const b = setup('socket-b', io) + const b = setup('socket-b', io, { userId: 'attacker' }) await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 7 }) await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 7 }) @@ -307,13 +300,13 @@ describe('setupWorkspaceFileDocHandlers', () => { expect(b.socket.join).not.toHaveBeenCalled() }) - it('reclaims a client id whose prior owner is no longer connected (reconnect)', async () => { + it('reclaims a client id for the SAME user reconnecting (reused Yjs client id)', async () => { const { io } = createIo() - // socket-a joined with client id 7 but is NOT in the connected set (its - // disconnect cleanup has not run yet); socket-b reconnects reusing id 7. + // The same user's dropped socket still owns client id 7 (its disconnect + // cleanup has not run yet) when it reconnects on a new socket reusing id 7. const a = setup('socket-a', io) await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 7 }) - const b = setup('socket-b', io) + const b = setup('socket-b', io) // same default userId 'user-1' await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 7 }) diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index c2eefe0c45e..60e743291b8 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -37,17 +37,26 @@ const logger = createLogger('FileDocHandlers') * pins `realtime.replicaCount: 1`). Horizontal scaling would need a shared Yjs * backend (y-redis / Hocuspocus) — out of scope here. */ +/** A socket's presence ownership within a room. */ +interface FileDocOwner { + /** + * The awareness clientID the socket declared at join. It owns exactly this one + * and may only publish/remove awareness for it, so an authenticated peer cannot + * forge or clear another collaborator's presence. + */ + clientId: number + /** The owning user — used to tell a reconnect (same user reusing its Yjs client + * id) from a spoof (a different user binding a peer's id). */ + userId: string +} + interface FileDocRoom { /** The `workspace_files.id` this room edits (for seed-request payloads). */ fileId: string doc: Y.Doc awareness: awarenessProtocol.Awareness - /** - * socketId → the awareness clientID it declared at join. A socket owns exactly - * one clientID and may only publish/remove awareness for that one, so an - * authenticated peer cannot forge or clear another collaborator's presence. - */ - ownedClientId: Map + /** socketId → its presence ownership. */ + owners: Map /** The socket currently elected to seed initial content, or `null`. */ seederSocketId: string | null } @@ -121,7 +130,7 @@ function awarenessUpdateClientIds(update: Uint8Array): number[] { */ function electSeederIfNeeded(io: Server, room: FileDocRoom) { if (room.seederSocketId !== null || isDocSeeded(room.doc)) return - const next = room.ownedClientId.keys().next() + const next = room.owners.keys().next() if (next.done) return room.seederSocketId = next.value io.to(next.value).emit(FILE_DOC_EVENTS.SEED_REQUEST, { fileId: room.fileId }) @@ -146,7 +155,7 @@ function getOrCreateRoom(io: Server, ref: RoomRef): FileDocRoom { fileId: ref.id, doc, awareness, - ownedClientId: new Map(), + owners: new Map(), seederSocketId: null, } fileDocRooms.set(name, room) @@ -222,7 +231,7 @@ function handleMessage(socket: AuthenticatedSocket, data: unknown) { // Enforce presence ownership: a socket may only publish/remove awareness // for the clientID it bound at join, so a peer cannot spoof or clear // another collaborator's caret. - const owned = room.ownedClientId.get(socket.id) + const owned = room.owners.get(socket.id)?.clientId if (owned === undefined || awarenessUpdateClientIds(update).some((id) => id !== owned)) { logger.warn('Dropping awareness frame for an unowned client id', { socketId: socket.id }) return @@ -252,12 +261,12 @@ export function cleanupFileDocForSocket(socketId: string, io: Server): void { const room = fileDocRooms.get(name) if (!room) return - const owned = room.ownedClientId.get(socketId) - room.ownedClientId.delete(socketId) - if (owned !== undefined) { + const owner = room.owners.get(socketId) + room.owners.delete(socketId) + if (owner !== undefined) { // Fires the awareness `update` handler with a non-socket origin → the removal // is broadcast to every remaining client, so the departed caret vanishes. - awarenessProtocol.removeAwarenessStates(room.awareness, [owned], null) + awarenessProtocol.removeAwarenessStates(room.awareness, [owner.clientId], null) } // Hand off the seeder role: if the elected seeder left before it seeded, elect @@ -269,7 +278,7 @@ export function cleanupFileDocForSocket(socketId: string, io: Server): void { // Drop the document + awareness once idle so no memory is held for a file with // no active editors; a later joiner re-creates and re-seeds it. - if (room.ownedClientId.size === 0) { + if (room.owners.size === 0) { room.awareness.destroy() room.doc.destroy() fileDocRooms.delete(name) @@ -347,28 +356,28 @@ export function setupWorkspaceFileDocHandlers( // A same socket rejoining with a NEW client id: clear its old caret so it // doesn't linger as a ghost after the binding is overwritten below. - const previousClientId = entry.ownedClientId.get(socket.id) - if (previousClientId !== undefined && previousClientId !== clientId) { - awarenessProtocol.removeAwarenessStates(entry.awareness, [previousClientId], null) + const previous = entry.owners.get(socket.id) + if (previous !== undefined && previous.clientId !== clientId) { + awarenessProtocol.removeAwarenessStates(entry.awareness, [previous.clientId], null) } - // A client id must be owned by at most one LIVE socket, or a peer could bind - // an active collaborator's id and pass the per-frame ownership check to - // spoof/clear its caret. Reject a live duplicate; reclaim a stale binding - // (a reconnect reuses the same Yjs client id, and its prior socket may not - // be cleaned up yet). `io.sockets.sockets` is this replica's registry — the - // same single-replica assumption the in-memory Y.Doc already relies on. - for (const [otherSid, otherCid] of entry.ownedClientId) { - if (otherCid !== clientId || otherSid === socket.id) continue - if (io.sockets.sockets.has(otherSid)) { + // A client id must be owned by at most one user, or a peer could bind an + // active collaborator's id and pass the per-frame ownership check to + // spoof/clear its caret. Distinguish a reconnect from a spoof by the owning + // user: the same user reclaiming its own client id (a dropped socket + // reconnecting reuses the Yjs client id, and its prior socket may not be + // cleaned up yet) takes over the stale binding; a DIFFERENT user is rejected. + for (const [otherSid, owner] of entry.owners) { + if (owner.clientId !== clientId || otherSid === socket.id) continue + if (owner.userId !== userId) { emitJoinError(socket, fileId, 'Client id already in use', 'CLIENT_ID_IN_USE', false) return } - entry.ownedClientId.delete(otherSid) - awarenessProtocol.removeAwarenessStates(entry.awareness, [otherCid], null) + entry.owners.delete(otherSid) + awarenessProtocol.removeAwarenessStates(entry.awareness, [owner.clientId], null) } - entry.ownedClientId.set(socket.id, clientId) + entry.owners.set(socket.id, { clientId, userId }) socketToRoomName.set(socket.id, name) socket.join(name) diff --git a/packages/realtime-protocol/src/file-doc.ts b/packages/realtime-protocol/src/file-doc.ts index 240b0d9a6b6..ca86253793e 100644 --- a/packages/realtime-protocol/src/file-doc.ts +++ b/packages/realtime-protocol/src/file-doc.ts @@ -52,6 +52,13 @@ export type FileDocMessageType = (typeof FILE_DOC_MESSAGE_TYPE)[keyof typeof FIL * Because it lives in the CRDT it merges across clients and the server can read * it — the server uses it to decide whether to re-elect a seeder when an elected * one disconnects before seeding. Client and server MUST use these exact keys. + * + * The seeding client MUST write the imported content AND set this flag in a + * SINGLE Yjs transaction (`doc.transact(...)`). Otherwise a seeder that dies + * between the two writes can leave content with no flag, and a re-elected client + * would seed again and duplicate it. `configMap` is a reserved top-level Y.Map + * name; the editor must not use a top-level type of the same name (TipTap uses + * `getXmlFragment('default')`, so there is no collision today). */ export const FILE_DOC_SEED = { configMap: 'config', From ee4aed7795f89c3d8d523106fb463feb1a685c04 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 24 Jul 2026 15:50:24 -0700 Subject: [PATCH 5/9] fix(realtime): check client-id ownership before clearing prior caret A rejected rebind (a socket asking to bind a client id owned by a DIFFERENT user) removed its own prior caret before returning CLIENT_ID_IN_USE, so peers lost the socket's caret until it published again. Run the uniqueness check before any state mutation, so a rejected rebind leaves the existing binding and caret untouched. +test. --- apps/realtime/src/handlers/file-doc.test.ts | 27 +++++++++++++++++++++ apps/realtime/src/handlers/file-doc.ts | 18 ++++++++------ 2 files changed, 37 insertions(+), 8 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc.test.ts b/apps/realtime/src/handlers/file-doc.test.ts index 97abca8e7eb..ee47be49c47 100644 --- a/apps/realtime/src/handlers/file-doc.test.ts +++ b/apps/realtime/src/handlers/file-doc.test.ts @@ -333,6 +333,33 @@ describe('setupWorkspaceFileDocHandlers', () => { expect(removal).toBeDefined() }) + it('preserves the existing caret when a rebind to a foreign client id is rejected', async () => { + const { io, sent } = createIo() + const { frame: awFrame } = awarenessFrame(10, 'A') + const a = setup('socket-a', io) + const b = setup('socket-b', io, { userId: 'user-b' }) + await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 10 }) + a.handlers[FILE_DOC_EVENTS.MESSAGE](awFrame) // a publishes its caret for client 10 + await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 20 }) + sent.length = 0 + + // socket-a (owns 10) tries to rebind to 20, owned by a different user → reject. + await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 20 }) + + expect(a.socket.emit).toHaveBeenCalledWith( + FILE_DOC_EVENTS.JOIN_ERROR, + expect.objectContaining({ code: 'CLIENT_ID_IN_USE' }) + ) + // The rejected rebind must NOT have removed a's existing caret (no awareness + // removal broadcast fires). + const removal = sent.find( + (m) => + m.event === FILE_DOC_EVENTS.MESSAGE && + (m.payload as Uint8Array)[0] === FILE_DOC_MESSAGE_TYPE.AWARENESS + ) + expect(removal).toBeUndefined() + }) + it('drops a malformed frame without throwing', async () => { const { io } = createIo() const a = setup('socket-a', io) diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index 60e743291b8..f3e24f632f9 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -354,19 +354,14 @@ export function setupWorkspaceFileDocHandlers( const entry = getOrCreateRoom(io, room) - // A same socket rejoining with a NEW client id: clear its old caret so it - // doesn't linger as a ghost after the binding is overwritten below. - const previous = entry.owners.get(socket.id) - if (previous !== undefined && previous.clientId !== clientId) { - awarenessProtocol.removeAwarenessStates(entry.awareness, [previous.clientId], null) - } - // A client id must be owned by at most one user, or a peer could bind an // active collaborator's id and pass the per-frame ownership check to // spoof/clear its caret. Distinguish a reconnect from a spoof by the owning // user: the same user reclaiming its own client id (a dropped socket // reconnecting reuses the Yjs client id, and its prior socket may not be - // cleaned up yet) takes over the stale binding; a DIFFERENT user is rejected. + // cleaned up yet) takes over the stale binding; a DIFFERENT user is + // rejected. This runs BEFORE any state mutation below, so a rejected rebind + // leaves the socket's existing binding and caret untouched. for (const [otherSid, owner] of entry.owners) { if (owner.clientId !== clientId || otherSid === socket.id) continue if (owner.userId !== userId) { @@ -377,6 +372,13 @@ export function setupWorkspaceFileDocHandlers( awarenessProtocol.removeAwarenessStates(entry.awareness, [owner.clientId], null) } + // Accepted: a same socket rebinding to a NEW client id clears its old caret + // so it doesn't linger as a ghost after the binding is overwritten. + const previous = entry.owners.get(socket.id) + if (previous !== undefined && previous.clientId !== clientId) { + awarenessProtocol.removeAwarenessStates(entry.awareness, [previous.clientId], null) + } + entry.owners.set(socket.id, { clientId, userId }) socketToRoomName.set(socket.id, name) socket.join(name) From a161659790238941b1173464e7157a44feedb4a3 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 24 Jul 2026 16:09:27 -0700 Subject: [PATCH 6/9] fix(realtime): guard concurrent-JOIN race in file-doc relay MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An independent 4-lens audit (correctness + security both ranked it #1) found the async JOIN handler had no post-await guard, so across the `authorizeRoom` await: - a socket disconnecting mid-authorize would register a DEAD socket, permanently leaking its Y.Doc/Awareness (+30s timer) and, if it was the elected seeder, stalling seeding for that file for the whole process; - two JOINs on a fast document switch could complete out of order, binding the socket to the WRONG doc and applying one file's edits into another's; - a LEAVE racing ahead of an in-flight JOIN was lost. Fix: a per-socket monotonic join generation, bumped on JOIN/LEAVE and dropped on cleanup. After the await, a JOIN proceeds only if it is still the socket's most recent intent (and the socket is still connected) — otherwise it aborts before mutating any state. Also drop a freshly-created empty room in the JOIN catch. Document the editor-hook contracts the relay's safety depends on (autosave gated on synced+seeded to prevent overwriting real content, atomic seed-or-leave, one provider per socket, re-mint clientID on CLIENT_ID_IN_USE). Tests: +out-of-order-join, +disconnect-during-authorize, +LEAVE-scoping, +sync-step-2-reply, +no-reseed-when-seeded. Full realtime suite (149) green. --- apps/realtime/src/handlers/file-doc.test.ts | 95 +++++++++++++++++++++ apps/realtime/src/handlers/file-doc.ts | 39 ++++++++- packages/realtime-protocol/src/file-doc.ts | 17 ++++ 3 files changed, 150 insertions(+), 1 deletion(-) diff --git a/apps/realtime/src/handlers/file-doc.test.ts b/apps/realtime/src/handlers/file-doc.test.ts index ee47be49c47..aade23abd0e 100644 --- a/apps/realtime/src/handlers/file-doc.test.ts +++ b/apps/realtime/src/handlers/file-doc.test.ts @@ -49,6 +49,7 @@ function createSocket(id: string, overrides?: Record) { id, userId: 'user-1', userName: 'Test User', + disconnected: false, on: vi.fn((event: string, handler: Handler) => { handlers[event] = handler }), @@ -398,4 +399,98 @@ describe('setupWorkspaceFileDocHandlers', () => { const seed = sent.find((m) => m.event === FILE_DOC_EVENTS.SEED_REQUEST) expect(seed?.target).toBe('socket-b') }) + + it('aborts a join superseded by a newer join during authorization (no cross-binding)', async () => { + const { io } = createIo() + let resolveFirst: (v: unknown) => void = () => {} + mockAuthorizeRoom + .mockReturnValueOnce(new Promise((resolve) => (resolveFirst = resolve))) + .mockResolvedValueOnce({ allowed: true, status: 200, workspacePermission: 'write' }) + const s = setup('socket-a', io) + + const pending = s.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 }) + await s.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-2', clientId: 1 }) + resolveFirst({ allowed: true, status: 200, workspacePermission: 'write' }) + await pending + + // The socket is bound only to the newer file, never cross-bound to file-1. + expect(s.socket.join).toHaveBeenCalledWith('workspace-file-doc:file-2') + expect(s.socket.join).not.toHaveBeenCalledWith('workspace-file-doc:file-1') + }) + + it('does not register a socket that disconnected during authorization', async () => { + const { io, sent } = createIo() + let resolveAuth: (v: unknown) => void = () => {} + mockAuthorizeRoom.mockReturnValueOnce(new Promise((resolve) => (resolveAuth = resolve))) + const s = setup('socket-a', io) + + const pending = s.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 }) + s.socket.disconnected = true + cleanupFileDocForSocket('socket-a', io) // disconnect cleanup — no-op, nothing registered yet + resolveAuth({ allowed: true, status: 200, workspacePermission: 'write' }) + await pending + + expect(s.socket.join).not.toHaveBeenCalled() + // No room leaked: a fresh joiner starts a new document and is elected to seed. + const b = setup('socket-b', io) + await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 2 }) + expect(sent.find((m) => m.event === FILE_DOC_EVENTS.SEED_REQUEST)?.target).toBe('socket-b') + }) + + it('scopes LEAVE to the named file (a leave for a different file is a no-op)', async () => { + const { io } = createIo() + const a = setup('socket-a', io) + await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 }) + + a.handlers[FILE_DOC_EVENTS.LEAVE]({ fileId: 'other' }) + expect(a.socket.leave).not.toHaveBeenCalledWith(ROOM_NAME) + + a.handlers[FILE_DOC_EVENTS.LEAVE]({ fileId: 'file-1' }) + expect(a.socket.leave).toHaveBeenCalledWith(ROOM_NAME) + }) + + it('replies with a sync step 2 to the sender on a sync step 1 frame', async () => { + const { io } = createIo() + const a = setup('socket-a', io) + await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 }) + // Give the server doc some content so a step-1 request yields a non-empty step 2. + const seeded = new Y.Doc() + seeded.getText('default').insert(0, 'hi') + a.handlers[FILE_DOC_EVENTS.MESSAGE]( + frame(FILE_DOC_MESSAGE_TYPE.SYNC, (e) => + syncProtocol.writeUpdate(e, Y.encodeStateAsUpdate(seeded)) + ) + ) + a.socket.emit.mockClear() + + a.handlers[FILE_DOC_EVENTS.MESSAGE]( + frame(FILE_DOC_MESSAGE_TYPE.SYNC, (e) => syncProtocol.writeSyncStep1(e, new Y.Doc())) + ) + + const reply = a.socket.emit.mock.calls.find( + ([event, payload]) => event === FILE_DOC_EVENTS.MESSAGE && payload instanceof Uint8Array + ) + expect((reply?.[1] as Uint8Array)[0]).toBe(FILE_DOC_MESSAGE_TYPE.SYNC) + }) + + it('does not re-elect a seeder once the document is marked seeded', async () => { + const { io, sent } = createIo() + const a = setup('socket-a', io) + await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 }) // a elected seeder + // a seeds: set the CRDT initialContentLoaded flag on the server doc. + const seeded = new Y.Doc() + seeded.getMap('config').set('initialContentLoaded', true) + a.handlers[FILE_DOC_EVENTS.MESSAGE]( + frame(FILE_DOC_MESSAGE_TYPE.SYNC, (e) => + syncProtocol.writeUpdate(e, Y.encodeStateAsUpdate(seeded)) + ) + ) + const b = setup('socket-b', io) + await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 2 }) + sent.length = 0 + + // The seeder leaves a SEEDED doc → no re-election (no duplicate seed). + cleanupFileDocForSocket('socket-a', io) + expect(sent.find((m) => m.event === FILE_DOC_EVENTS.SEED_REQUEST)).toBeUndefined() + }) }) diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index f3e24f632f9..935f591f8fe 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -65,6 +65,15 @@ interface FileDocRoom { const fileDocRooms = new Map() /** socketId → its current file-doc room name (a socket edits at most one doc). */ const socketToRoomName = new Map() +/** + * socketId → a monotonic join generation. A JOIN bumps it on arrival and, after + * the async authorization, proceeds only if the generation is still its own — so + * a disconnect, a LEAVE, or a newer JOIN (a fast document switch) that occurred + * during authorization aborts the now-stale JOIN. Without this, an out-of-order + * authorize completion could bind the socket to the wrong document, or a + * disconnect-during-authorize could register a dead socket and leak its room. + */ +const joinGeneration = new Map() interface AwarenessChange { added: number[] @@ -254,6 +263,10 @@ function handleMessage(socket: AuthenticatedSocket, data: unknown) { * Exported for the disconnect handler; safe to call for a socket in no room. */ export function cleanupFileDocForSocket(socketId: string, io: Server): void { + // Drop the join generation so an in-flight JOIN for this socket aborts after + // its authorize resolves, and the map never leaks across the socket's life. + joinGeneration.delete(socketId) + const name = socketToRoomName.get(socketId) if (!name) return socketToRoomName.delete(socketId) @@ -315,6 +328,10 @@ export function setupWorkspaceFileDocHandlers( return } + // Claim this JOIN's generation before the async authorize below. + const generation = (joinGeneration.get(socket.id) ?? 0) + 1 + joinGeneration.set(socket.id, generation) + const room = fileDocRoom(fileId) const name = roomName(room) @@ -343,6 +360,11 @@ export function setupWorkspaceFileDocHandlers( return } + // Abort a JOIN superseded during authorization: the socket disconnected, or + // a LEAVE / newer JOIN bumped the generation. Registering here would leak a + // dead socket's room or bind the socket to a document it has left behind. + if (socket.disconnected || joinGeneration.get(socket.id) !== generation) return + // Switched documents on the same socket — leave the previous one first (a // socket edits at most one document). A duplicate join of the SAME room // falls through and simply re-runs the sync handshake, idempotently. @@ -411,8 +433,18 @@ export function setupWorkspaceFileDocHandlers( } catch (error) { logger.error('Error joining file-doc room:', error) try { - socket.leave(roomName(fileDocRoom(fileId))) + const name = roomName(fileDocRoom(fileId)) + socket.leave(name) cleanupFileDocForSocket(socket.id, io) + // If the failure happened after `getOrCreateRoom` but before the socket + // registered as an owner, `cleanupFileDocForSocket` (which keys off + // `socketToRoomName`) can't drop the freshly-created empty room — do it here. + const room = fileDocRooms.get(name) + if (room && room.owners.size === 0) { + room.awareness.destroy() + room.doc.destroy() + fileDocRooms.delete(name) + } } catch {} emitJoinError(socket, fileId, 'Failed to join file document', 'JOIN_FAILED', true) } @@ -422,6 +454,11 @@ export function setupWorkspaceFileDocHandlers( socket.on(FILE_DOC_EVENTS.LEAVE, (payload?: LeaveFileDocPayload) => { try { + // Invalidate any in-flight JOIN for this socket: a LEAVE arriving while a + // JOIN is still authorizing means the client no longer wants the room by + // the time that JOIN resolves. + joinGeneration.set(socket.id, (joinGeneration.get(socket.id) ?? 0) + 1) + const name = socketToRoomName.get(socket.id) if (!name) return // Scope the leave to the named file when provided: a deferred leave from a diff --git a/packages/realtime-protocol/src/file-doc.ts b/packages/realtime-protocol/src/file-doc.ts index ca86253793e..85c79b1a9bb 100644 --- a/packages/realtime-protocol/src/file-doc.ts +++ b/packages/realtime-protocol/src/file-doc.ts @@ -86,6 +86,23 @@ export interface JoinFileDocSuccess { * recipient imports the file's stored markdown into the (empty) document. The * client still guards on the CRDT `initialContentLoaded` flag so a re-election * that races an in-flight seed can never duplicate content. + * + * Consumer (editor hook) contract — the relay depends on these to be safe: + * 1. **Never overwrite content with an unseeded doc.** Autosave (the markdown + * mirror written through the content API) MUST be gated on the document being + * both synced AND seeded (`initialContentLoaded === true`). Otherwise an empty + * or still-syncing doc could be saved over the real file — the one true + * data-loss path, and the reason a withholding seeder is only a liveness + * nuisance rather than destructive. + * 2. **Seed atomically, or leave.** Write content + the flag in ONE + * `doc.transact(...)`; if seeding fails, emit {@link FILE_DOC_EVENTS.LEAVE} + * (or destroy the provider) so the server re-elects another client. + * 3. **One provider per socket.** Destroy the previous {@link FILE_DOC_EVENTS} + * provider before creating the next (document switch), so a stale provider's + * binary-frame listener can't apply another document's updates. + * 4. **Re-mint on CLIENT_ID_IN_USE.** On a `CLIENT_ID_IN_USE` join error, + * recreate the Yjs doc (fresh `clientID`) and rejoin rather than giving up — + * the id is transiently held by another socket of the same user. */ export interface SeedRequestPayload { fileId: string From 4913c4104634312195ea10551c37459d435ffce6 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 24 Jul 2026 16:19:04 -0700 Subject: [PATCH 7/9] improvement(realtime): seed deadline re-election + audit polish Addresses the remaining audit improvement areas on the file-doc relay: - Seed deadline: a withholding/stuck elected seeder previously left the document empty for the whole room until it disconnected (flagged by both the security and simplicity audits). The server now arms a deadline on election; if the seeder doesn't complete the seed in time it is passed over and a remaining client is elected. A successful seed cancels the deadline; tried-and-failed seeders are skipped so one client can't block the room. - Hoist the duplicated Uint8Array coercion into the shared protocol module (toFileDocBytes), used by both the server relay and the client provider. - Test hygiene: afterEach now drops rooms for every socket id a test created (tracked set) instead of a hardcoded list that could drift; fake timers drive the deadline and prevent a real timer firing into a later test. Tests: +seed-deadline-reelection, +deadline-cancelled-on-seed, +file-switch-leaves-previous. Full realtime suite (152) green. --- apps/realtime/src/handlers/file-doc.test.ts | 73 +++++++++++++++++++-- apps/realtime/src/handlers/file-doc.ts | 70 ++++++++++++++++---- packages/realtime-protocol/src/file-doc.ts | 12 ++++ 3 files changed, 137 insertions(+), 18 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc.test.ts b/apps/realtime/src/handlers/file-doc.test.ts index aade23abd0e..0fda4b68ebe 100644 --- a/apps/realtime/src/handlers/file-doc.test.ts +++ b/apps/realtime/src/handlers/file-doc.test.ts @@ -43,7 +43,12 @@ function createIo() { return { io: { to } as unknown as IRoomManager['io'], sent } } +/** Every socket id a test created, so `afterEach` can drop their rooms without a + * hardcoded list drifting out of sync with the tests. */ +const createdSocketIds = new Set() + function createSocket(id: string, overrides?: Record) { + createdSocketIds.add(id) const handlers: Record = {} const socket = { id, @@ -114,6 +119,9 @@ function joinSuccessFileId(socket: { emit: ReturnType }) { describe('setupWorkspaceFileDocHandlers', () => { beforeEach(() => { vi.clearAllMocks() + // The seed deadline uses setTimeout; fake it so tests can drive it and so a + // real timer can never fire into a later test. + vi.useFakeTimers() mockAuthorizeRoom.mockResolvedValue({ allowed: true, status: 200, @@ -123,11 +131,12 @@ describe('setupWorkspaceFileDocHandlers', () => { }) afterEach(() => { - // The room store is module-global; drop any room the test's sockets left open. + // The room store is module-global; drop every room the test's sockets opened. const { io } = createIo() - for (const id of ['socket-1', 'socket-a', 'socket-b', 'socket-c']) { - cleanupFileDocForSocket(id, io) - } + for (const id of createdSocketIds) cleanupFileDocForSocket(id, io) + createdSocketIds.clear() + vi.clearAllTimers() + vi.useRealTimers() }) it('rejects join when the socket is not authenticated', async () => { @@ -493,4 +502,60 @@ describe('setupWorkspaceFileDocHandlers', () => { cleanupFileDocForSocket('socket-a', io) expect(sent.find((m) => m.event === FILE_DOC_EVENTS.SEED_REQUEST)).toBeUndefined() }) + + it('leaves the previous document when a socket switches files', async () => { + const { io, sent } = createIo() + const s = setup('socket-a', io) + await s.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 }) + await s.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-2', clientId: 1 }) + + expect(s.socket.leave).toHaveBeenCalledWith('workspace-file-doc:file-1') + expect(s.socket.join).toHaveBeenCalledWith('workspace-file-doc:file-2') + + // file-1's room was dropped (socket-a was its only owner): a fresh joiner of + // file-1 starts a new document and is elected to seed. + sent.length = 0 + const b = setup('socket-b', io) + await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 2 }) + expect(sent.find((m) => m.event === FILE_DOC_EVENTS.SEED_REQUEST)?.target).toBe('socket-b') + }) + + it('re-elects a new seeder when the elected one misses the seed deadline', async () => { + const { io, sent } = createIo() + const a = setup('socket-a', io) + const b = setup('socket-b', io) + await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 }) + await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 2 }) + expect(sent.find((m) => m.event === FILE_DOC_EVENTS.SEED_REQUEST)?.target).toBe('socket-a') + sent.length = 0 + + // socket-a never seeds; the deadline lapses. + vi.advanceTimersByTime(10_000) + + // The remaining un-tried client is asked to seed instead. + expect(sent.find((m) => m.event === FILE_DOC_EVENTS.SEED_REQUEST)?.target).toBe('socket-b') + }) + + it('cancels the seed deadline once the document is seeded', async () => { + const { io, sent } = createIo() + const a = setup('socket-a', io) + const b = setup('socket-b', io) + await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 }) + await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 2 }) + + // socket-a seeds within the deadline. + const seeded = new Y.Doc() + seeded.getMap('config').set('initialContentLoaded', true) + a.handlers[FILE_DOC_EVENTS.MESSAGE]( + frame(FILE_DOC_MESSAGE_TYPE.SYNC, (e) => + syncProtocol.writeUpdate(e, Y.encodeStateAsUpdate(seeded)) + ) + ) + sent.length = 0 + + vi.advanceTimersByTime(10_000) + + // No re-election: the successful seed cancelled the deadline. + expect(sent.find((m) => m.event === FILE_DOC_EVENTS.SEED_REQUEST)).toBeUndefined() + }) }) diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index 935f591f8fe..7768343f54a 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -6,6 +6,7 @@ import { FILE_DOC_SEED, type JoinFileDocPayload, type LeaveFileDocPayload, + toFileDocBytes, } from '@sim/realtime-protocol/file-doc' import { ROOM_TYPES, type RoomRef, roomName } from '@sim/realtime-protocol/rooms' import * as decoding from 'lib0/decoding' @@ -19,6 +20,14 @@ import type { IRoomManager } from '@/rooms' const logger = createLogger('FileDocHandlers') +/** + * How long an elected seeder has to import the initial content before the server + * re-elects another client. Seeding is a local, synchronous editor write, so this + * only trips when a seeder never completes it (a bug, or a client withholding the + * seed) — it keeps the document from staying empty for the rest of the room. + */ +const SEED_DEADLINE_MS = 10_000 + /** * Collaborative document editing (live carets + text selection) for a single * file's rich-text editor. This is the standard Yjs "websocket server" relay — @@ -59,6 +68,11 @@ interface FileDocRoom { owners: Map /** The socket currently elected to seed initial content, or `null`. */ seederSocketId: string | null + /** Deadline timer for the current seeder to complete the seed, or `null`. */ + seedTimer: ReturnType | null + /** Sockets that were elected but failed to seed within the deadline; skipped + * on re-election so a single stuck/withholding client can't block the room. */ + triedSeeders: Set } /** Live documents keyed by Socket.IO room name. Module-global: one Y.Doc per file. */ @@ -106,12 +120,6 @@ function isDocSeeded(doc: Y.Doc): boolean { return doc.getMap(FILE_DOC_SEED.configMap).get(FILE_DOC_SEED.flag) === true } -function toUint8Array(data: unknown): Uint8Array | null { - if (data instanceof Uint8Array) return data - if (data instanceof ArrayBuffer) return new Uint8Array(data) - return null -} - /** * Decode the client IDs an awareness update carries, without applying it, to * check a frame only touches its sender's own presence. Mirrors the wire format @@ -130,19 +138,48 @@ function awarenessUpdateClientIds(update: Uint8Array): number[] { return ids } +/** Cancel a pending seed-deadline timer, if any. */ +function clearSeedTimer(room: FileDocRoom) { + if (room.seedTimer !== null) { + clearTimeout(room.seedTimer) + room.seedTimer = null + } +} + /** * Elect a client to seed an unseeded document and ask it to import the stored * markdown, if one is needed and not already assigned. Called after a join (to - * elect the newcomer) and after the elected seeder leaves (to hand the role to a - * remaining client, so the document never stays permanently empty). A no-op once - * the document is seeded or a seeder is already assigned. + * elect the newcomer), after the elected seeder leaves (to hand the role to a + * remaining client), and after a seed deadline lapses (to pass over a client that + * never seeded). Skips clients that already failed to seed, and arms a deadline so + * a stuck or withholding seeder can't leave the document empty for the whole room. + * A no-op once the document is seeded, a seeder is already assigned, or no + * un-tried client remains. */ function electSeederIfNeeded(io: Server, room: FileDocRoom) { if (room.seederSocketId !== null || isDocSeeded(room.doc)) return - const next = room.owners.keys().next() - if (next.done) return - room.seederSocketId = next.value - io.to(next.value).emit(FILE_DOC_EVENTS.SEED_REQUEST, { fileId: room.fileId }) + + let elected: string | null = null + for (const socketId of room.owners.keys()) { + if (!room.triedSeeders.has(socketId)) { + elected = socketId + break + } + } + if (elected === null) return + + room.seederSocketId = elected + io.to(elected).emit(FILE_DOC_EVENTS.SEED_REQUEST, { fileId: room.fileId }) + + clearSeedTimer(room) + room.seedTimer = setTimeout(() => { + room.seedTimer = null + // Only act if this election is still the pending one and it never seeded. + if (room.seederSocketId !== elected || isDocSeeded(room.doc)) return + room.triedSeeders.add(elected) + room.seederSocketId = null + electSeederIfNeeded(io, room) + }, SEED_DEADLINE_MS) } /** @@ -166,10 +203,14 @@ function getOrCreateRoom(io: Server, ref: RoomRef): FileDocRoom { awareness, owners: new Map(), seederSocketId: null, + seedTimer: null, + triedSeeders: new Set(), } fileDocRooms.set(name, room) doc.on('update', (update: Uint8Array, origin: unknown) => { + // Once the document is seeded, the seed deadline is moot — cancel it. + if (room.seedTimer !== null && isDocSeeded(doc)) clearSeedTimer(room) const encoder = encoding.createEncoder() encoding.writeVarUint(encoder, FILE_DOC_MESSAGE_TYPE.SYNC) syncProtocol.writeUpdate(encoder, update) @@ -212,7 +253,7 @@ function handleMessage(socket: AuthenticatedSocket, data: unknown) { const room = fileDocRooms.get(name) if (!room) return - const bytes = toUint8Array(data) + const bytes = toFileDocBytes(data) if (!bytes) return // A malformed frame from any client must never escape as a process-level @@ -292,6 +333,7 @@ export function cleanupFileDocForSocket(socketId: string, io: Server): void { // Drop the document + awareness once idle so no memory is held for a file with // no active editors; a later joiner re-creates and re-seeds it. if (room.owners.size === 0) { + clearSeedTimer(room) room.awareness.destroy() room.doc.destroy() fileDocRooms.delete(name) diff --git a/packages/realtime-protocol/src/file-doc.ts b/packages/realtime-protocol/src/file-doc.ts index 85c79b1a9bb..738af5bc123 100644 --- a/packages/realtime-protocol/src/file-doc.ts +++ b/packages/realtime-protocol/src/file-doc.ts @@ -120,3 +120,15 @@ export interface JoinFileDocError { export interface LeaveFileDocPayload { fileId: string } + +/** + * Coerce a Socket.IO binary payload to a `Uint8Array`, or `null` if it is + * neither. Shared by the server relay and the client provider so the two agree + * on how an inbound {@link FILE_DOC_EVENTS.MESSAGE} frame is read (Socket.IO may + * deliver a `Uint8Array`/`Buffer` or an `ArrayBuffer` depending on runtime). + */ +export function toFileDocBytes(data: unknown): Uint8Array | null { + if (data instanceof Uint8Array) return data + if (data instanceof ArrayBuffer) return new Uint8Array(data) + return null +} From 050d34831571a8f938dbcfadb5b83210e5cf2fb1 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 24 Jul 2026 16:27:40 -0700 Subject: [PATCH 8/9] cleanup(realtime): dedupe idle-room teardown + skip awareness state decode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit /simplify pass (4 parallel agents: reuse/simplification/efficiency/altitude — reuse and altitude found nothing to change): - Extract destroyRoomIfIdle() — the idle-room teardown was duplicated between cleanupFileDocForSocket and the JOIN catch. - awarenessUpdateClientIds now advances past the presence-state bytes instead of UTF-8-decoding a throwaway string, on the (moderately hot) awareness path. --- apps/realtime/src/handlers/file-doc.ts | 31 +++++++++++++------------- 1 file changed, 16 insertions(+), 15 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index 7768343f54a..b27bdaab3c8 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -133,7 +133,7 @@ function awarenessUpdateClientIds(update: Uint8Array): number[] { for (let i = 0; i < count; i++) { ids.push(decoding.readVarUint(decoder)) decoding.readVarUint(decoder) // clock - decoding.readVarString(decoder) // state json + decoding.readVarUint8Array(decoder) // state bytes — advanced past, only ids matter } return ids } @@ -146,6 +146,19 @@ function clearSeedTimer(room: FileDocRoom) { } } +/** + * Drop a room's document + awareness (and its seed timer) once it has no owners, + * so an idle file holds no memory. A later joiner re-creates and re-seeds it. + */ +function destroyRoomIfIdle(name: string) { + const room = fileDocRooms.get(name) + if (!room || room.owners.size > 0) return + clearSeedTimer(room) + room.awareness.destroy() + room.doc.destroy() + fileDocRooms.delete(name) +} + /** * Elect a client to seed an unseeded document and ask it to import the stored * markdown, if one is needed and not already assigned. Called after a join (to @@ -330,14 +343,7 @@ export function cleanupFileDocForSocket(socketId: string, io: Server): void { electSeederIfNeeded(io, room) } - // Drop the document + awareness once idle so no memory is held for a file with - // no active editors; a later joiner re-creates and re-seeds it. - if (room.owners.size === 0) { - clearSeedTimer(room) - room.awareness.destroy() - room.doc.destroy() - fileDocRooms.delete(name) - } + destroyRoomIfIdle(name) } /** @@ -481,12 +487,7 @@ export function setupWorkspaceFileDocHandlers( // If the failure happened after `getOrCreateRoom` but before the socket // registered as an owner, `cleanupFileDocForSocket` (which keys off // `socketToRoomName`) can't drop the freshly-created empty room — do it here. - const room = fileDocRooms.get(name) - if (room && room.owners.size === 0) { - room.awareness.destroy() - room.doc.destroy() - fileDocRooms.delete(name) - } + destroyRoomIfIdle(name) } catch {} emitJoinError(socket, fileId, 'Failed to join file document', 'JOIN_FAILED', true) } From 5a47caac1da7a9078d7a886a8dad3938446b8f17 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 24 Jul 2026 16:32:39 -0700 Subject: [PATCH 9/9] fix(realtime): stop LEAVE from aborting an unrelated in-flight join Cursor review caught that bumping the join generation in the LEAVE handler could silently abort a legitimate in-flight JOIN for a DIFFERENT file (a stale or cross-file leave during a document switch), leaving the socket bound to no room and the client with no ack. The generation guard only needs the JOIN bump + socket.disconnected check; the LEAVE coupling is removed. The case it covered (a same-file join-then-leave with no remount) is a bounded, self-healing ghost that cleans up on the next file open or disconnect. Also apply the /simplify + comment-audit tidy-ups (drop two self-documenting comments; align the generation-guard comments with the removed LEAVE coupling). +test: a different-file leave does not abort an in-flight join. --- apps/realtime/src/handlers/file-doc.test.ts | 16 ++++++++++++++ apps/realtime/src/handlers/file-doc.ts | 24 ++++++++++----------- 2 files changed, 27 insertions(+), 13 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc.test.ts b/apps/realtime/src/handlers/file-doc.test.ts index 0fda4b68ebe..48da76f22ac 100644 --- a/apps/realtime/src/handlers/file-doc.test.ts +++ b/apps/realtime/src/handlers/file-doc.test.ts @@ -446,6 +446,22 @@ describe('setupWorkspaceFileDocHandlers', () => { expect(sent.find((m) => m.event === FILE_DOC_EVENTS.SEED_REQUEST)?.target).toBe('socket-b') }) + it('does not abort an in-flight join when a leave for a different file arrives', async () => { + const { io } = createIo() + let resolveAuth: (v: unknown) => void = () => {} + mockAuthorizeRoom.mockReturnValueOnce(new Promise((resolve) => (resolveAuth = resolve))) + const s = setup('socket-a', io) + + const pending = s.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-2', clientId: 1 }) + // A stale leave for a DIFFERENT file must not invalidate the in-flight join. + s.handlers[FILE_DOC_EVENTS.LEAVE]({ fileId: 'file-1' }) + resolveAuth({ allowed: true, status: 200, workspacePermission: 'write' }) + await pending + + expect(joinSuccessFileId(s.socket)).toBe('file-2') + expect(s.socket.join).toHaveBeenCalledWith('workspace-file-doc:file-2') + }) + it('scopes LEAVE to the named file (a leave for a different file is a no-op)', async () => { const { io } = createIo() const a = setup('socket-a', io) diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index b27bdaab3c8..a6c43c00cfa 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -82,10 +82,11 @@ const socketToRoomName = new Map() /** * socketId → a monotonic join generation. A JOIN bumps it on arrival and, after * the async authorization, proceeds only if the generation is still its own — so - * a disconnect, a LEAVE, or a newer JOIN (a fast document switch) that occurred - * during authorization aborts the now-stale JOIN. Without this, an out-of-order - * authorize completion could bind the socket to the wrong document, or a - * disconnect-during-authorize could register a dead socket and leak its room. + * a newer JOIN (a fast document switch) or a disconnect (which drops the entry in + * cleanup) that occurred during authorization aborts the now-stale JOIN. Without + * this, an out-of-order authorize completion could bind the socket to the wrong + * document, or a disconnect-during-authorize could register a dead socket and + * leak its room. */ const joinGeneration = new Map() @@ -138,7 +139,6 @@ function awarenessUpdateClientIds(update: Uint8Array): number[] { return ids } -/** Cancel a pending seed-deadline timer, if any. */ function clearSeedTimer(room: FileDocRoom) { if (room.seedTimer !== null) { clearTimeout(room.seedTimer) @@ -409,8 +409,8 @@ export function setupWorkspaceFileDocHandlers( } // Abort a JOIN superseded during authorization: the socket disconnected, or - // a LEAVE / newer JOIN bumped the generation. Registering here would leak a - // dead socket's room or bind the socket to a document it has left behind. + // a newer JOIN (a document switch) bumped the generation. Registering here + // would leak a dead socket's room or bind the socket to the wrong document. if (socket.disconnected || joinGeneration.get(socket.id) !== generation) return // Switched documents on the same socket — leave the previous one first (a @@ -474,7 +474,6 @@ export function setupWorkspaceFileDocHandlers( socket.emit(FILE_DOC_EVENTS.MESSAGE, encoding.toUint8Array(awarenessEncoder)) } - // Ask a client to seed the initial content if the document is still empty. electSeederIfNeeded(io, entry) logger.info(`User ${userId} joined file-doc room ${fileId}`) @@ -497,11 +496,10 @@ export function setupWorkspaceFileDocHandlers( socket.on(FILE_DOC_EVENTS.LEAVE, (payload?: LeaveFileDocPayload) => { try { - // Invalidate any in-flight JOIN for this socket: a LEAVE arriving while a - // JOIN is still authorizing means the client no longer wants the room by - // the time that JOIN resolves. - joinGeneration.set(socket.id, (joinGeneration.get(socket.id) ?? 0) + 1) - + // Only affect a REGISTERED room; never touch the join generation here. A + // leave that raced ahead of an in-flight join (no room registered yet) is a + // no-op — bumping the generation would silently abort an unrelated join for + // a different file (a document switch), leaving the socket bound to nothing. const name = socketToRoomName.get(socket.id) if (!name) return // Scope the leave to the named file when provided: a deferred leave from a