fix(realtime): address post-merge review-comment findings

apps/realtime/src/handlers/connection.ts

@@ -21,6 +21,13 @@ export function setupConnectionHandlers(socket: AuthenticatedSocket, roomManager
  // evicted or TTL-expired (which would leave the manager's stored rooms empty).
  socket.on('disconnecting', async (reason) => {
  try {
+ // Snapshot the live Socket.IO room membership SYNCHRONOUSLY, before any
+ // await: Socket.IO clears `socket.rooms` via leaveAll() as soon as the
+ // synchronous portion of this `disconnecting` handler returns (i.e. at the
+ // first await below), so reading it afterwards would see an empty set and
+ // the eviction fallback would be dead.
+ const liveRoomNames = [...socket.rooms]
+
  // Clean up pending debounce entries for this socket to prevent memory leaks
  cleanupPendingSubblocksForSocket(socket.id)
  cleanupPendingVariablesForSocket(socket.id)
@@ -29,13 +36,13 @@ export function setupConnectionHandlers(socket: AuthenticatedSocket, roomManager
  // room the manager knows about.
  const removedRooms = await roomManager.removeSocketFromAllRooms(socket.id)
 
- // Union with the live Socket.IO membership (authoritative here, and it
+ // Union with the snapshotted Socket.IO membership (authoritative, and it
  // survives a Redis eviction/TTL lapse that would leave the manager's tracked
  // 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)
- for (const name of socket.rooms) {
+ for (const name of liveRoomNames) {
  // `wasInRooms.has(name)` already excludes every room the manager removed
  // (same room-name key via the roomName/parseRoomName bijection), so any
  // room reaching here was NOT in `removedRooms` and needs a removal attempt.

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

@@ -61,6 +61,7 @@ function createRoomManager(overrides?: Partial<IRoomManager>): IRoomManager {
  broadcastPresenceUpdate: vi.fn().mockResolvedValue(undefined),
  getRoomUsers: vi.fn().mockResolvedValue([]),
  hasRoom: vi.fn().mockResolvedValue(false),
+ deleteRoom: vi.fn().mockResolvedValue(undefined),
  addUserToRoom: vi.fn().mockResolvedValue(undefined),
  getUserSession: vi.fn().mockResolvedValue(null),
  updateUserActivity: vi.fn().mockResolvedValue(undefined),

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

@@ -55,6 +55,7 @@ function createRoomManager(overrides?: Partial<IRoomManager>): IRoomManager {
  broadcastPresenceUpdate: vi.fn().mockResolvedValue(undefined),
  getRoomUsers: vi.fn().mockResolvedValue([]),
  hasRoom: vi.fn().mockResolvedValue(false),
+ deleteRoom: vi.fn().mockResolvedValue(undefined),
  addUserToRoom: vi.fn().mockResolvedValue(undefined),
  getUserSession: vi.fn().mockResolvedValue(null),
  updateUserActivity: vi.fn().mockResolvedValue(undefined),

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

@@ -5,6 +5,7 @@ import { ROOM_TYPES, type RoomRef, roomName } from '@sim/realtime-protocol/rooms
 import { eq } from 'drizzle-orm'
 import type { AuthenticatedSocket } from '@/middleware/auth'
 import type { IRoomManager, UserPresence } from '@/rooms'
+import { filterVisiblePresence, sweepStalePresence } from '@/rooms/presence-visibility'
 
 const logger = createLogger('WorkspaceFilesHandlers')
 
@@ -77,6 +78,19 @@ export function setupWorkspaceFilesHandlers(
  return
  }
 
+ // Validate the client-supplied id before it reaches the DB query (matches
+ // the /api/workspace-files-changed guard; 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
+ }
+
  const room = filesRoom(workspaceId)
 
  let authorized: Awaited<ReturnType<typeof authorizeRoom>>
@@ -130,6 +144,10 @@ export function setupWorkspaceFilesHandlers(
  }
  }
 
+ // Reclaim any presence orphaned by an ungraceful disconnect (pod crash
+ // fires no `disconnecting` event; the room hashes have no TTL).
+ await sweepStalePresence(roomManager, room)
+
  socket.join(roomName(room))
 
  const presence: UserPresence = {
@@ -147,7 +165,13 @@ export function setupWorkspaceFilesHandlers(
 
  await roomManager.addUserToRoom(room, socket.id, presence)
 
- const presenceUsers = await roomManager.getRoomUsers(room)
+ // Filter the join ack to live members so a new joiner never briefly sees a
+ // ghost from an entry the sweep hasn't reclaimed yet.
+ const presenceUsers = await filterVisiblePresence(
+ roomManager.io,
+ room,
+ await roomManager.getRoomUsers(room)
+ )
  socket.emit('join-workspace-files-success', {
  workspaceId,
  socketId: socket.id,
@@ -159,6 +183,14 @@ export function setupWorkspaceFilesHandlers(
  logger.info(`User ${userId} (${userName}) joined files room for workspace ${workspaceId}`)
  } catch (error) {
  logger.error('Error joining workspace files room:', error)
+ // Roll back any partial join so a failed attempt can't leave the socket in
+ // the Socket.IO room or a stale presence entry behind (mirrors the workflow
+ // join's rollback), before signalling a retryable failure.
+ try {
+ const room = filesRoom(workspaceId)
+ socket.leave(roomName(room))
+ await roomManager.removeUserFromRoom(room, socket.id)
+ } catch {}
  socket.emit('join-workspace-files-error', {
  workspaceId,
  error: 'Failed to join workspace files',
@@ -169,14 +201,18 @@ export function setupWorkspaceFilesHandlers(
  }
  )
 
- socket.on('leave-workspace-files', async () => {
+ socket.on('leave-workspace-files', async (payload?: { workspaceId?: string }) => {
  try {
  if (!roomManager.isReady()) return
  const room = await roomManager.getRoomForSocket(socket.id, ROOM_TYPES.WORKSPACE_FILES)
  if (!room) return
+ // Scope the leave to a specific workspace when the client provides one: a
+ // deferred leave from a prior page must not evict the socket from a room it
+ // has since switched into (workspace A→B leaves A's leave targeting B).
+ if (payload?.workspaceId && payload.workspaceId !== room.id) return
  socket.leave(roomName(room))
  await roomManager.removeUserFromRoom(room, socket.id)
- await roomManager.broadcastPresenceUpdate(room)
+ await roomManager.broadcastPresenceUpdate(room, socket.id)
  } catch (error) {
  logger.error('Error leaving workspace files room:', error)
  }

apps/realtime/src/rooms/memory-manager.test.ts

@@ -8,6 +8,7 @@
 import { ROOM_TYPES, type RoomRef } from '@sim/realtime-protocol/rooms'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 import { MemoryRoomManager } from '@/rooms/memory-manager'
+import { sweepStalePresence } from '@/rooms/presence-visibility'
 import type { UserPresence } from '@/rooms/types'
 
 function fakeIo(liveSocketIds: string[] = []) {
@@ -120,6 +121,39 @@ describe('MemoryRoomManager multi-room', () => {
  expect(await manager.getUserSession('socket-2')).not.toBeNull()
  })
 
+ it('sweepStalePresence reclaims not-live stale entries but keeps live and fresh ones', async () => {
+ const { io } = fakeIo(['socket-live'])
+ const m = new MemoryRoomManager(io)
+ await m.initialize()
+
+ const staleMs = 76 * 60 * 1000
+ await m.addUserToRoom(FILES, 'socket-live', presence(FILES, 'socket-live', 'u1'))
+ await m.addUserToRoom(FILES, 'socket-dead', {
+ ...presence(FILES, 'socket-dead', 'u2'),
+ joinedAt: Date.now() - staleMs,
+ lastActivity: Date.now() - staleMs,
+ })
+ await m.addUserToRoom(FILES, 'socket-recent', presence(FILES, 'socket-recent', 'u3'))
+
+ await sweepStalePresence(m, FILES)
+
+ const remaining = (await m.getRoomUsers(FILES)).map((u) => u.socketId).sort()
+ // socket-dead: not live + stale → removed. socket-live: live → kept.
+ // socket-recent: not live but fresh (transient) → kept.
+ expect(remaining).toEqual(['socket-live', 'socket-recent'])
+ })
+
+ it('deleteRoom unconditionally drops all room state', async () => {
+ await manager.addUserToRoom(FILES, 'socket-1', presence(FILES, 'socket-1', 'user-1'))
+ await manager.addUserToRoom(FILES, 'socket-2', presence(FILES, 'socket-2', 'user-2'))
+ expect(await manager.hasRoom(FILES)).toBe(true)
+
+ await manager.deleteRoom(FILES)
+
+ expect(await manager.hasRoom(FILES)).toBe(false)
+ expect(await manager.getRoomUsers(FILES)).toHaveLength(0)
+ })
+
  it('ignores removal of a room the socket is not in (id-guarded)', async () => {
  await manager.addUserToRoom(FILES, 'socket-1', presence(FILES, 'socket-1', 'user-1'))
 

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

@@ -149,6 +149,10 @@ export class MemoryRoomManager implements IRoomManager {
  return this.rooms.has(roomKey(room))
  }
 
+ async deleteRoom(room: RoomRef): Promise<void> {
+ this.rooms.delete(roomKey(room))
+ }
+
  async updateUserActivity(
  room: RoomRef,
  socketId: string,

apps/realtime/src/rooms/presence-visibility.ts

@@ -1,5 +1,14 @@
 import { type RoomRef, roomName } from '@sim/realtime-protocol/rooms'
 import type { Server } from 'socket.io'
+import type { IRoomManager } from '@/rooms/types'
+
+/**
+ * How stale a not-live presence entry must be before a join-time sweep reclaims
+ * it. Kept above the 1h socket-key TTL so a normally-idle collaborator is never
+ * evicted; only genuinely-orphaned entries (e.g. a crashed pod that never fired
+ * `disconnecting`) are cleared. Matches the workflow join sweep.
+ */
+const STALE_PRESENCE_THRESHOLD_MS = 75 * 60 * 1000
 
 /**
  * Filters a room's stored presence down to what should actually be broadcast:
@@ -34,3 +43,32 @@ export async function filterVisiblePresence<T extends { socketId: string }>(
  return candidates
  }
 }
+
+/**
+ * Reclaims orphaned presence entries in a room: any stored socket that is no
+ * longer a live Socket.IO member AND has been idle past
+ * {@link STALE_PRESENCE_THRESHOLD_MS} is removed. This is how a room-users hash
+ * (which has no TTL) is bounded against ungraceful disconnects — a pod crash
+ * fires no `disconnecting` event, so its entries would otherwise persist forever.
+ * Run on join, like the workflow room does. No-op when the liveness lookup fails
+ * (so a transient adapter blip can't evict live collaborators).
+ */
+export async function sweepStalePresence(manager: IRoomManager, room: RoomRef): Promise<void> {
+ let liveIds: Set<string>
+ try {
+ const liveSockets = await manager.io.in(roomName(room)).fetchSockets()
+ liveIds = new Set(liveSockets.map((socket) => socket.id))
+ } catch {
+ return
+ }
+
+ const now = Date.now()
+ const users = await manager.getRoomUsers(room)
+ for (const user of users) {
+ if (liveIds.has(user.socketId)) continue
+ const lastSeen = user.lastActivity || user.joinedAt || 0
+ if (now - lastSeen > STALE_PRESENCE_THRESHOLD_MS) {
+ await manager.removeUserFromRoom(room, user.socketId)
+ }
+ }
+}

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

@@ -311,6 +311,17 @@ export class RedisRoomManager implements IRoomManager {
  return exists > 0
  }
 
+ async deleteRoom(room: RoomRef): Promise<void> {
+ // Log AND rethrow (like addUserToRoom): a failed wipe must not be reported as a
+ // clean deletion by the caller — the request surfaces it (and can be retried).
+ try {
+ await this.redis.del([KEYS.roomUsers(room), KEYS.roomMeta(room)])
+ } catch (error) {
+ logger.error(`Failed to delete room ${room.type}:${room.id}:`, error)
+ throw error
+ }
+ }
+
  async updateUserActivity(
  room: RoomRef,
  socketId: string,

apps/realtime/src/rooms/types.ts

@@ -101,6 +101,13 @@ export interface IRoomManager {
  /** Whether a room currently has any presence. */
  hasRoom(room: RoomRef): Promise<boolean>
 
+ /**
+ * Unconditionally drop all state for a room (presence + metadata). Used when a
+ * room's underlying resource is destroyed (e.g. a deleted workflow) to guarantee
+ * no state lingers even if per-socket removals failed or a socket joined mid-teardown.
+ */
+ deleteRoom(room: RoomRef): Promise<void>
+
  /** Update a socket's activity (cursor, selection, lastActivity) within a room. */
  updateUserActivity(
  room: RoomRef,

apps/realtime/src/rooms/workflow-room-service.ts

@@ -22,29 +22,44 @@ export class WorkflowRoomService {
  logger.info(`Handling workflow deletion notification for ${workflowId}`)
  const room = workflowRoom(workflowId)
 
- const users = await this.manager.getRoomUsers(room)
- if (users.length === 0) {
- logger.debug(`No active users found for deleted workflow ${workflowId}`)
- return
- }
+ const name = roomName(room)
 
+ // Always notify — reach every socket still in the Socket.IO room so the client
+ // clears the deleted workflow, even if that socket's Redis presence was evicted
+ // (in which case it would be missing from getRoomUsers). Emitting to an empty
+ // room is a harmless no-op.
  this.manager.emitToRoom(room, 'workflow-deleted', {
  workflowId,
  message: 'This workflow has been deleted',
  timestamp: Date.now(),
  })
 
+ // Clean per-socket state for every socket that is either a live Socket.IO member
+ // OR still has presence — so an evicted/late-joined socket's room mapping and
+ // session are dropped too, not just the presence-tracked ones.
+ const socketIds = new Set<string>()
+ try {
+ const liveSockets = await this.manager.io.in(name).fetchSockets()
+ for (const s of liveSockets) socketIds.add(s.id)
+ } catch (error) {
+ logger.warn(`Could not enumerate sockets for deleted workflow ${workflowId}`, error)
+ }
+ for (const user of await this.manager.getRoomUsers(room)) socketIds.add(user.socketId)
+
  // Remove every socket from the Socket.IO room (cross-pod via the Redis adapter).
- const name = roomName(room)
  await this.manager.io.in(name).socketsLeave(name)
 
- // Drop presence state for each socket; empty-room cleanup is handled by the manager.
- for (const user of users) {
- await this.manager.removeUserFromRoom(room, user.socketId)
+ for (const socketId of socketIds) {
+ await this.manager.removeUserFromRoom(room, socketId)
  }
 
+ // Final unconditional wipe — the workflow is gone, so no room state may linger
+ // even if a per-socket removal failed (matches the pre-refactor managers, which
+ // ended deletion with an unconditional room drop).
+ await this.manager.deleteRoom(room)
+
  logger.info(
- `Cleaned up workflow room ${workflowId} after deletion (${users.length} users disconnected)`
+ `Cleaned up workflow room ${workflowId} after deletion (${socketIds.size} sockets removed)`
  )
  }
 

apps/sim/app/workspace/[workspaceId]/files/hooks/use-workspace-files-room.ts

@@ -77,6 +77,12 @@ export function useWorkspaceFilesRoom(
  }) => {
  if (data.workspaceId !== workspaceId) return
  retries = 0
+ // Cancel any retry scheduled by a prior retryable error so it can't fire an
+ // extra join after we're already in.
+ if (retryTimer) {
+ clearTimeout(retryTimer)
+ retryTimer = null
+ }
  setPresenceUsers(data.presenceUsers ?? [])
  }
  const handleJoinError = (data: JoinErrorPayload) => {
@@ -103,13 +109,18 @@ export function useWorkspaceFilesRoom(
 
  return () => {
  if (retryTimer) clearTimeout(retryTimer)
- socket.emit('leave-workspace-files')
  socket.off('connect', join)
  socket.off('join-workspace-files-success', handleJoinSuccess)
  socket.off('join-workspace-files-error', handleJoinError)
  socket.off('workspace-files:presence-update', handlePresence)
  socket.off('workspace-files-changed', handleChanged)
  setPresenceUsers([])
+
+ // Leave the room, scoped to THIS workspace: the server no-ops if the socket
+ // has already switched to another workspace's files room (so a workspace
+ // A→B switch, where B's join runs first and auto-leaves A, can't have A's
+ // leave evict the fresh B membership).
+ socket.emit('leave-workspace-files', { workspaceId })
  }
  }, [socket, workspaceId, queryClient])
 

apps/sim/lib/realtime/notify.ts

@@ -11,9 +11,14 @@ const NOTIFY_TIMEOUT_MS = 2000
 /**
  * Best-effort fan-out to the realtime server that a workspace's file tree changed,
  * so every browser currently viewing that workspace's files refetches. File
- * mutations happen over the HTTP API (not the socket); this is the lossy liveness
- * signal — a dropped notification only degrades to stale-until-refetch, so it must
- * never throw or block the originating mutation.
+ * mutations happen over the HTTP API (not the socket); this is a lossy liveness
+ * signal — a dropped notification only degrades to stale-until-refetch.
+ *
+ * Never throws. Callers `await` it (rather than fire-and-forget) so the fetch is
+ * guaranteed to dispatch before a Node route handler returns — a floating promise
+ * can be dropped after the response is sent. It is a normally-sub-millisecond
+ * local call and is hard-bounded to {@link NOTIFY_TIMEOUT_MS}, so it adds that
+ * latency only when the socket pod is unreachable.
  */
 export async function notifyWorkspaceFilesChanged(workspaceId: string): Promise<void> {
  try {