fix(realtime): harden seeder recovery, join-generation, and misc robustness

Final line-by-line audit follow-ups (all LOW/MED, no P0/P1):

- file-doc: a sole client whose seed FETCH fails was added to triedSeeders,
  re-election found nobody, and the document stayed permanently empty until
  reload. Re-offer seeding a bounded number of rounds (MAX_SEED_ROUNDS) before
  giving up. Also bound clientId to a non-negative integer (it is an ownership key).
- tables + workspace-files: validate the room id BEFORE advancing joinGeneration,
  so a malformed/rejected join can't cancel a legitimate in-flight join.
- workflow + tables + workspace-files: suppress the client-facing join error when
  the op was already superseded (a retryable error naming the abandoned room could
  make a client re-join and cancel its newer join). The rollback still runs.
- redis-manager: set isConnected=true only after scriptLoad succeeds (and reset it
  on failure) so isReady() can't report ready while the Lua SHAs are null.
- connection: apply the presence-bearing filter to the manager-removed set too
  (symmetry with the fallback path).
- http.ts: validate workflowId on the four workflow endpoints (matching the files one).
- client: clear a pending join-retry timer before rescheduling (reconnect churn no
  longer orphans a stray extra join); clear caret fade timers on plugin destroy;
  seed-effect cleanup reports NOT-ready (safe direction).
- Tests: bounded seeder recovery, cell-selection strip-junk, TABLE round-trip,
  presenceEventName.

Gates: tsc (sim/realtime/packages) 0, 208 realtime + 12 protocol tests, biome,
api-validation, boundaries, prune.

Author: waleedlatif1

apps/realtime/src/handlers/connection.ts

@@ -54,7 +54,13 @@ export function setupConnectionHandlers(socket: AuthenticatedSocket, roomManager
       // rooms empty). Attempt removal for any room the manager didn't already
       // remove — best-effort, since a transient Redis error can't be recovered here.
       const wasInRooms = new Map<string, RoomRef>()
-      for (const room of removedRooms) wasInRooms.set(roomName(room), room)
+      // Only presence-bearing rooms get a corrective broadcast. Manager-removed rooms are
+      // presence-bearing by construction today (only workflow/table write the socket→room hash),
+      // but filter symmetrically with the fallback path below so a future room type that ever
+      // tracks presence here can't emit a bogus presence-update no client listens to.
+      for (const room of removedRooms) {
+        if (PRESENCE_BEARING_TYPES.has(room.type)) wasInRooms.set(roomName(room), room)
+      }
       for (const name of liveRoomNames) {
         // `wasInRooms.has(name)` already excludes every room the manager removed (same
         // room-name key via the roomName/parseRoomName bijection). Skip room types with no

apps/realtime/src/handlers/file-doc.test.ts

@@ -665,6 +665,22 @@ describe('setupWorkspaceFileDocHandlers', () => {
     expect(sent.find((m) => m.event === FILE_DOC_EVENTS.SEED_REQUEST)?.target).toBe('socket-b')
   })
 
+  it('re-offers seeding to a sole client that missed the deadline, then gives up after a bound', async () => {
+    const { io, sent } = createIo()
+    const a = setup('socket-a', io)
+    await a.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
+    expect(sent.filter((m) => m.event === FILE_DOC_EVENTS.SEED_REQUEST).length).toBe(1)
+
+    // The sole client keeps missing the deadline. Without recovery it would be permanently excluded
+    // (empty document forever); instead it is re-offered a bounded number of rounds.
+    for (let i = 0; i < 6; i++) vi.advanceTimersByTime(10_000)
+
+    // Initial offer + MAX_SEED_ROUNDS (3) re-offers = 4 total, all to socket-a; then it stops.
+    const requests = sent.filter((m) => m.event === FILE_DOC_EVENTS.SEED_REQUEST)
+    expect(requests.length).toBe(4)
+    expect(requests.every((m) => m.target === 'socket-a')).toBe(true)
+  })
+
   it('cancels the seed deadline once the document is seeded', async () => {
     const { io, sent } = createIo()
     const a = setup('socket-a', io)

apps/realtime/src/handlers/file-doc.ts

@@ -50,6 +50,16 @@ const logger = createLogger('FileDocHandlers')
  */
 const SEED_DEADLINE_MS = 10_000
 
+/**
+ * Max times the room re-offers seeding to its owners after every one of them has
+ * missed a deadline. Without this, a sole client whose seed *fetch fails* (not just
+ * slow) is added to `triedSeeders`, re-election finds no un-tried owner, and the
+ * document is left permanently empty until that client reloads. Each exhausted round
+ * clears `triedSeeders` and re-offers, giving a transient failure a bounded number of
+ * retries (~MAX × deadline) before the room genuinely gives up.
+ */
+const MAX_SEED_ROUNDS = 3
+
 /** A socket's presence ownership within a room. */
 interface FileDocOwner {
   /**
@@ -81,6 +91,10 @@ interface FileDocRoom {
   /** 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<string>
+  /** Count of full re-offer rounds spent after every owner missed a deadline;
+   * bounds the retry loop (see {@link MAX_SEED_ROUNDS}) so a permanently-broken
+   * seed doesn't reset forever. */
+  seedRounds: number
 }
 
 /** Live documents keyed by Socket.IO room name. Module-global: one Y.Doc per file. */
@@ -222,6 +236,19 @@ function electSeederIfNeeded(io: Server, room: FileDocRoom) {
     room.triedSeeders.add(elected)
     room.seederSocketId = null
     electSeederIfNeeded(io, room)
+    // If that left no seeder (every current owner has now been tried) while the doc is still
+    // unseeded and owners remain, re-offer the whole room a bounded number of times — otherwise a
+    // sole client whose seed fetch failed leaves the document permanently empty.
+    if (
+      room.seederSocketId === null &&
+      !isDocSeeded(room.doc) &&
+      room.owners.size > 0 &&
+      room.seedRounds < MAX_SEED_ROUNDS
+    ) {
+      room.seedRounds += 1
+      room.triedSeeders.clear()
+      electSeederIfNeeded(io, room)
+    }
   }, SEED_DEADLINE_MS)
 }
 
@@ -248,6 +275,7 @@ function getOrCreateRoom(io: Server, ref: RoomRef): FileDocRoom {
     seederSocketId: null,
     seedTimer: null,
     triedSeeders: new Set(),
+    seedRounds: 0,
   }
   fileDocRooms.set(name, room)
 
@@ -418,7 +446,14 @@ export function setupWorkspaceFileDocHandlers(
         emitJoinError(socket, fileId, 'Realtime unavailable', 'ROOM_MANAGER_UNAVAILABLE', true)
         return
       }
-      if (typeof fileId !== 'string' || fileId.length === 0 || typeof clientId !== 'number') {
+      if (
+        typeof fileId !== 'string' ||
+        fileId.length === 0 ||
+        // A Yjs clientID is a uint32; reject NaN/Infinity/negative/non-integer so a malformed id
+        // can't become a bogus ownership key.
+        !Number.isInteger(clientId) ||
+        clientId < 0
+      ) {
         emitJoinError(socket, fileId, 'Invalid join payload', 'INVALID_PAYLOAD', false)
         return
       }

apps/realtime/src/handlers/tables.test.ts

@@ -187,6 +187,37 @@ describe('setupTablesHandlers', () => {
     expect(toEmit).not.toHaveBeenCalled()
   })
 
+  it('strips unknown/oversized fields from an otherwise-valid selection before storing or relaying', async () => {
+    const { socket, handlers, toEmit } = createSocket()
+    const roomManager = createRoomManager({
+      getRoomForSocket: vi.fn().mockResolvedValue(TABLE_ROOM),
+    })
+    setupTablesHandlers(socket as unknown as SetupArg, roomManager)
+
+    await handlers[TABLE_PRESENCE_EVENTS.CELL_SELECTION]({
+      cell: {
+        anchor: { rowId: 'row-1', columnId: 'col-a', junk: 'x'.repeat(10_000) },
+        focus: { rowId: 'row-1', columnId: 'col-a' },
+        editing: true,
+        bloat: 'x'.repeat(100_000),
+      },
+    })
+
+    // Only the whitelisted fields survive — a hostile peer can't amplify an oversized object.
+    const expected = {
+      anchor: { rowId: 'row-1', columnId: 'col-a' },
+      focus: { rowId: 'row-1', columnId: 'col-a' },
+      editing: true,
+    }
+    expect(roomManager.updateUserActivity).toHaveBeenCalledWith(TABLE_ROOM, 'socket-1', {
+      cell: expected,
+    })
+    expect(toEmit).toHaveBeenCalledWith(TABLE_PRESENCE_EVENTS.CELL_SELECTION, {
+      socketId: 'socket-1',
+      cell: expected,
+    })
+  })
+
   it('skips a superseded queued join on a fast table switch', async () => {
     const { socket, handlers } = createSocket()
     const roomManager = createRoomManager()

apps/realtime/src/handlers/tables.ts

@@ -78,6 +78,17 @@ export function setupTablesHandlers(socket: AuthenticatedSocket, roomManager: IR
   let opChain: Promise<void> = Promise.resolve()
 
   socket.on(TABLE_PRESENCE_EVENTS.JOIN, ({ tableId, tabSessionId }: JoinTablePayload) => {
+    // Validate the id BEFORE claiming a generation, so a malformed join can't advance
+    // joinGeneration and cancel a legitimate in-flight join for another table.
+    if (typeof tableId !== 'string' || tableId.length === 0) {
+      socket.emit(TABLE_PRESENCE_EVENTS.JOIN_ERROR, {
+        tableId: typeof tableId === 'string' ? tableId : '',
+        error: 'Invalid table id',
+        code: 'INVALID_PAYLOAD',
+        retryable: false,
+      })
+      return
+    }
     const joinAttempt = (joinGeneration += 1)
     currentTableId = tableId
     opChain = opChain
@@ -119,17 +130,6 @@ export function setupTablesHandlers(socket: AuthenticatedSocket, roomManager: IR
         return
       }
 
-      // Validate the client-supplied id before it reaches the DB query.
-      if (typeof tableId !== 'string' || tableId.length === 0) {
-        socket.emit(TABLE_PRESENCE_EVENTS.JOIN_ERROR, {
-          tableId: typeof tableId === 'string' ? tableId : '',
-          error: 'Invalid table id',
-          code: 'INVALID_PAYLOAD',
-          retryable: false,
-        })
-        return
-      }
-
       const room = tableRoom(tableId)
 
       const authorized = await resolveRoomJoinAuth({
@@ -244,6 +244,10 @@ export function setupTablesHandlers(socket: AuthenticatedSocket, roomManager: IR
         // Best-effort rollback — the original join failure is the one surfaced below, so a
         // secondary cleanup error must not mask it or throw out of the error handler.
       }
+      // Suppress the client-facing error when this join was already superseded: the client has moved
+      // to a newer table, and a retryable error naming the abandoned one could make it re-join and
+      // supersede the newer join. The rollback above still runs.
+      if (superseded()) return
       socket.emit(TABLE_PRESENCE_EVENTS.JOIN_ERROR, {
         tableId,
         error: 'Failed to join table',

apps/realtime/src/handlers/workflow.ts

@@ -266,6 +266,10 @@ export function setupWorkflowHandlers(socket: AuthenticatedSocket, roomManager:
       // room state, never the newer op's.
       socket.leave(workflowId)
       await roomManager.removeUserFromRoom(wf(workflowId), socket.id)
+      // Suppress the client-facing error when this join was already superseded: the client has moved
+      // to a newer workflow, and a retryable error naming the abandoned one could make it re-join and
+      // supersede the newer join (an A/B flicker). The rollback above still runs.
+      if (superseded()) return
       const isReady = roomManager.isReady()
       socket.emit('join-workflow-error', {
         workflowId,

apps/realtime/src/handlers/workspace-files.ts

@@ -44,41 +44,43 @@ export function setupWorkspaceFilesHandlers(
   let currentWorkspace: string | null = null
 
   socket.on('join-workspace-files', async ({ workspaceId }: JoinPayload) => {
-    const joinAttempt = (joinGeneration += 1)
-    currentWorkspace = workspaceId
-    try {
-      if (!socket.userId || !socket.userName) {
-        socket.emit('join-workspace-files-error', {
-          workspaceId,
-          error: 'Authentication required',
-          code: 'AUTHENTICATION_REQUIRED',
-          retryable: false,
-        })
-        return
-      }
+    // Validate synchronously BEFORE claiming a generation, so a rejected/malformed join can't
+    // advance joinGeneration and cancel a legitimate in-flight join for another workspace.
+    if (!socket.userId || !socket.userName) {
+      socket.emit('join-workspace-files-error', {
+        workspaceId,
+        error: 'Authentication required',
+        code: 'AUTHENTICATION_REQUIRED',
+        retryable: false,
+      })
+      return
+    }
 
-      if (!roomManager.isReady()) {
-        socket.emit('join-workspace-files-error', {
-          workspaceId,
-          error: 'Realtime unavailable',
-          code: 'ROOM_MANAGER_UNAVAILABLE',
-          retryable: true,
-        })
-        return
-      }
+    if (!roomManager.isReady()) {
+      socket.emit('join-workspace-files-error', {
+        workspaceId,
+        error: 'Realtime unavailable',
+        code: 'ROOM_MANAGER_UNAVAILABLE',
+        retryable: true,
+      })
+      return
+    }
 
-      // Validate the client-supplied id before it reaches the DB query (join payloads are
-      // otherwise raw client input).
-      if (typeof workspaceId !== 'string' || workspaceId.length === 0) {
-        socket.emit('join-workspace-files-error', {
-          workspaceId: typeof workspaceId === 'string' ? workspaceId : '',
-          error: 'Invalid workspace id',
-          code: 'INVALID_PAYLOAD',
-          retryable: false,
-        })
-        return
-      }
+    // Validate the client-supplied id before it reaches the DB query (join payloads are
+    // otherwise raw client input) and before advancing the generation.
+    if (typeof workspaceId !== 'string' || workspaceId.length === 0) {
+      socket.emit('join-workspace-files-error', {
+        workspaceId: typeof workspaceId === 'string' ? workspaceId : '',
+        error: 'Invalid workspace id',
+        code: 'INVALID_PAYLOAD',
+        retryable: false,
+      })
+      return
+    }
 
+    const joinAttempt = (joinGeneration += 1)
+    currentWorkspace = workspaceId
+    try {
       const room = filesRoom(workspaceId)
 
       const authorized = await resolveRoomJoinAuth({
@@ -115,6 +117,10 @@ export function setupWorkspaceFilesHandlers(
       try {
         socket.leave(roomName(filesRoom(workspaceId)))
       } catch {}
+      // Suppress the client-facing error when this join was already superseded: the client has
+      // switched to a newer workspace, and a retryable error naming the abandoned one could make it
+      // re-join and cancel the newer join. The leave above still runs.
+      if (joinGeneration !== joinAttempt || socket.disconnected) return
       socket.emit('join-workspace-files-error', {
         workspaceId,
         error: 'Failed to join workspace files',

apps/realtime/src/rooms/redis-manager.ts

@@ -168,13 +168,18 @@ export class RedisRoomManager implements IRoomManager {
 
     try {
       await this.redis.connect()
-      this.isConnected = true
 
       this.removeRoomScriptSha = await this.redis.scriptLoad(REMOVE_ROOM_SCRIPT)
       this.updateActivityScriptSha = await this.redis.scriptLoad(UPDATE_ACTIVITY_SCRIPT)
 
+      // Mark ready only after the scripts load — isReady() gates removeUserFromRoom/updateUserActivity,
+      // which silently no-op without a script SHA. Setting the flag before scriptLoad would make
+      // isReady() lie if scriptLoad threw.
+      this.isConnected = true
+
       logger.info('RedisRoomManager connected to Redis and scripts loaded')
     } catch (error) {
+      this.isConnected = false
       logger.error('Failed to connect to Redis:', error)
       throw error
     }

apps/realtime/src/routes/http.ts

@@ -42,6 +42,10 @@ function readRequestBody(req: IncomingMessage): Promise<string> {
   })
 }
 
+function isNonEmptyString(value: unknown): value is string {
+  return typeof value === 'string' && value.length > 0
+}
+
 function sendSuccess(res: ServerResponse): void {
   res.writeHead(200, { 'Content-Type': 'application/json' })
   res.end(JSON.stringify({ success: true }))
@@ -102,6 +106,7 @@ export function createHttpHandler(roomManager: IRoomManager, logger: Logger) {
       try {
         const body = await readRequestBody(req)
         const { workflowId } = JSON.parse(body)
+        if (!isNonEmptyString(workflowId)) return sendError(res, 'Invalid workflowId', 400)
         await workflowRoomService.handleWorkflowDeletion(workflowId)
         sendSuccess(res)
       } catch (error) {
@@ -116,6 +121,7 @@ export function createHttpHandler(roomManager: IRoomManager, logger: Logger) {
       try {
         const body = await readRequestBody(req)
         const { workflowId } = JSON.parse(body)
+        if (!isNonEmptyString(workflowId)) return sendError(res, 'Invalid workflowId', 400)
         await workflowRoomService.handleWorkflowUpdate(workflowId)
         sendSuccess(res)
       } catch (error) {
@@ -130,6 +136,7 @@ export function createHttpHandler(roomManager: IRoomManager, logger: Logger) {
       try {
         const body = await readRequestBody(req)
         const { workflowId } = JSON.parse(body)
+        if (!isNonEmptyString(workflowId)) return sendError(res, 'Invalid workflowId', 400)
         await workflowRoomService.handleWorkflowDeployed(workflowId)
         sendSuccess(res)
       } catch (error) {
@@ -144,6 +151,7 @@ export function createHttpHandler(roomManager: IRoomManager, logger: Logger) {
       try {
         const body = await readRequestBody(req)
         const { workflowId, timestamp } = JSON.parse(body)
+        if (!isNonEmptyString(workflowId)) return sendError(res, 'Invalid workflowId', 400)
         await workflowRoomService.handleWorkflowRevert(workflowId, timestamp)
         sendSuccess(res)
       } catch (error) {

apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/caret-presence.ts

@@ -135,6 +135,17 @@ export function createCaretActivityExtension(awareness: Awareness): Extension {
               destroy: () => {
                 awareness.off('change', onChange)
                 if (raf) cancelAnimationFrame(raf)
+                // Clear any pending fade timers for this editor's carets so they don't fire on
+                // detached nodes after unmount (harmless no-op, but a genuine leaked timer).
+                for (const caret of editorView.dom.querySelectorAll<HTMLElement>(
+                  '.collaboration-carets__caret'
+                )) {
+                  const timer = caretFadeTimers.get(caret)
+                  if (timer) {
+                    clearTimeout(timer)
+                    caretFadeTimers.delete(caret)
+                  }
+                }
               },
             }
           },