fix(realtime): scope room removal to the target workflow and detect swallowed cleanup failures

Honor the workflowIdHint as the target room in both room managers so
removing a stale room cannot clobber the mapping of a room the socket has
since moved to, and confirm eviction cleanup via the returned workflowId
plus an unswallowed mapping read so Redis failures actually defer into the
retry queue.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: icecrasher321

apps/realtime/src/access-revalidation.test.ts

@@ -39,11 +39,15 @@ function makeManager(sockets: FakeSocket[], presence: Partial<UserPresence>[] =
     io: { sockets: { sockets: socketMap } },
     isReady: () => true,
     getWorkflowUsers: vi.fn().mockResolvedValue(presence),
-    removeUserFromRoom: vi.fn().mockResolvedValue(null),
+    getWorkflowIdForSocket: vi.fn().mockResolvedValue(null),
+    removeUserFromRoom: vi
+      .fn()
+      .mockImplementation(async (_socketId: string, workflowId?: string) => workflowId ?? null),
     broadcastPresenceUpdate: vi.fn().mockResolvedValue(undefined),
   }
   return manager as unknown as IRoomManager & {
     getWorkflowUsers: ReturnType<typeof vi.fn>
+    getWorkflowIdForSocket: ReturnType<typeof vi.fn>
     removeUserFromRoom: ReturnType<typeof vi.fn>
     broadcastPresenceUpdate: ReturnType<typeof vi.fn>
   }
@@ -185,6 +189,47 @@ describe('access-revalidation sweep', () => {
     expect(manager.broadcastPresenceUpdate).toHaveBeenCalledWith('wf-1')
   })
 
+  it('defers cleanup when the manager swallows a removal failure into null', async () => {
+    const socket = makeSocket('sock-1', 'user-1', 'wf-1')
+    const manager = makeManager([socket], [{ socketId: 'sock-1', role: 'read' }])
+    // Live mapping but the removal reports nothing removed — the Redis manager
+    // swallows transport errors into null, so this is the only failure signal.
+    manager.getWorkflowIdForSocket.mockResolvedValue('wf-1')
+    manager.removeUserFromRoom.mockResolvedValueOnce(null)
+    mockResolveRole.mockResolvedValue(null)
+
+    const sweep = startAccessRevalidationSweep(manager)
+    await sweep.runOnce()
+
+    expect(socket.leave).toHaveBeenCalledWith('wf-1')
+    expect(manager.broadcastPresenceUpdate).not.toHaveBeenCalled()
+
+    // Next pass: the socket left the room, and the removal now succeeds.
+    socket.rooms = new Set(['sock-1'])
+    await sweep.runOnce()
+    sweep.stop()
+
+    expect(manager.removeUserFromRoom).toHaveBeenCalledTimes(2)
+    expect(manager.broadcastPresenceUpdate).toHaveBeenCalledWith('wf-1')
+  })
+
+  it('skips removal when the socket has since moved to a different workflow', async () => {
+    const socket = makeSocket('sock-1', 'user-1', 'wf-1')
+    const manager = makeManager([socket], [{ socketId: 'sock-1', role: 'read' }])
+    // Between the membership snapshot and cleanup, the socket switched to a
+    // workflow it can still access — removal must not touch its new presence.
+    manager.getWorkflowIdForSocket.mockResolvedValue('wf-2')
+    mockResolveRole.mockResolvedValue(null)
+
+    const sweep = startAccessRevalidationSweep(manager)
+    await sweep.runOnce()
+    sweep.stop()
+
+    expect(socket.leave).toHaveBeenCalledWith('wf-1')
+    expect(manager.removeUserFromRoom).not.toHaveBeenCalled()
+    expect(manager.broadcastPresenceUpdate).not.toHaveBeenCalled()
+  })
+
   it('drops a deferred cleanup when the socket legitimately re-joined the room', async () => {
     const socket = makeSocket('sock-1', 'user-1', 'wf-1')
     const manager = makeManager([socket], [{ socketId: 'sock-1', role: 'read' }])

apps/realtime/src/access-revalidation.ts

@@ -84,7 +84,24 @@ export function startAccessRevalidationSweep(roomManager: IRoomManager): AccessR
   async function cleanupEvictedSocket(socketId: string, workflowId: string): Promise<void> {
     const key = `${socketId}:${workflowId}`
     try {
-      await roomManager.removeUserFromRoom(socketId, workflowId)
+      // Unlike removeUserFromRoom, this read does not swallow transport errors,
+      // so a Redis outage lands in the catch below and defers the cleanup.
+      const currentWorkflowId = await roomManager.getWorkflowIdForSocket(socketId)
+      if (currentWorkflowId !== null && currentWorkflowId !== workflowId) {
+        // The socket has since moved to a different workflow it can still
+        // access; that join's room switch already removed this room's presence
+        // entry, so there is nothing stale left to clean here.
+        pendingCleanups.delete(key)
+        return
+      }
+
+      const removed = await roomManager.removeUserFromRoom(socketId, workflowId)
+      if (removed === null && currentWorkflowId !== null) {
+        // The Redis manager swallows transport errors into null — a live
+        // mapping with no reported removal means the removal did not happen.
+        throw new Error('room-state removal not confirmed')
+      }
+
       await roomManager.broadcastPresenceUpdate(workflowId)
       pendingCleanups.delete(key)
     } catch (error) {

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

@@ -66,17 +66,19 @@ export class MemoryRoomManager implements IRoomManager {
     logger.debug(`Added user ${presence.userId} to workflow ${workflowId} (socket: ${socketId})`)
   }
 
-  async removeUserFromRoom(socketId: string, _workflowIdHint?: string): Promise<string | null> {
-    const workflowId = this.socketToWorkflow.get(socketId)
+  async removeUserFromRoom(socketId: string, workflowIdHint?: string): Promise<string | null> {
+    const currentWorkflowId = this.socketToWorkflow.get(socketId) ?? null
+    const workflowId = workflowIdHint ?? currentWorkflowId
 
     if (!workflowId) {
       return null
     }
 
     const room = this.workflowRooms.get(workflowId)
     if (room) {
-      room.users.delete(socketId)
-      room.activeConnections = Math.max(0, room.activeConnections - 1)
+      if (room.users.delete(socketId)) {
+        room.activeConnections = Math.max(0, room.activeConnections - 1)
+      }
 
       // Clean up empty rooms
       if (room.activeConnections === 0) {
@@ -85,8 +87,13 @@ export class MemoryRoomManager implements IRoomManager {
       }
     }
 
-    this.socketToWorkflow.delete(socketId)
-    this.userSessions.delete(socketId)
+    // Only clear the socket's own mappings when it is not mapped to a different
+    // room — removing a stale room's entry must not destroy the mapping of a
+    // room the socket has since moved to.
+    if (currentWorkflowId === null || currentWorkflowId === workflowId) {
+      this.socketToWorkflow.delete(socketId)
+      this.userSessions.delete(socketId)
+    }
 
     logger.debug(`Removed socket ${socketId} from workflow ${workflowId}`)
     return workflowId

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

@@ -18,7 +18,11 @@ const SOCKET_PRESENCE_WORKFLOW_KEY_TTL = 24 * 60 * 60
 
 /**
  * Lua script for atomic user removal from room.
- * Returns workflowId if user was removed, null otherwise.
+ * The hint, when provided, is the target room to remove membership from; the
+ * socket's current mapping is only the fallback. Socket-level keys are cleared
+ * only when the socket is not mapped to a different room, so removing a stale
+ * room cannot destroy the mapping of a room the socket has since moved to.
+ * Returns the target workflowId, or null when no target could be resolved.
  * Handles room cleanup atomically to prevent race conditions.
  */
 const REMOVE_USER_SCRIPT = `
@@ -30,12 +34,13 @@ local workflowMetaPrefix = ARGV[2]
 local socketId = ARGV[3]
 local workflowIdHint = ARGV[4]
 
-local workflowId = redis.call('GET', socketWorkflowKey)
-if not workflowId then
-  workflowId = redis.call('GET', socketPresenceWorkflowKey)
+local currentWorkflowId = redis.call('GET', socketWorkflowKey)
+if not currentWorkflowId then
+  currentWorkflowId = redis.call('GET', socketPresenceWorkflowKey)
 end
 
-if not workflowId and workflowIdHint ~= '' then
+local workflowId = currentWorkflowId
+if workflowIdHint ~= '' then
   workflowId = workflowIdHint
 end
 
@@ -47,7 +52,10 @@ local workflowUsersKey = workflowUsersPrefix .. workflowId .. ':users'
 local workflowMetaKey = workflowMetaPrefix .. workflowId .. ':meta'
 
 redis.call('HDEL', workflowUsersKey, socketId)
-redis.call('DEL', socketWorkflowKey, socketSessionKey, socketPresenceWorkflowKey)
+
+if (not currentWorkflowId) or currentWorkflowId == workflowId then
+  redis.call('DEL', socketWorkflowKey, socketSessionKey, socketPresenceWorkflowKey)
+end
 
 local remaining = redis.call('HLEN', workflowUsersKey)
 if remaining == 0 then

apps/realtime/src/rooms/types.ts

@@ -64,9 +64,14 @@ export interface IRoomManager {
   addUserToRoom(workflowId: string, socketId: string, presence: UserPresence): Promise<void>
 
   /**
-   * Remove a user from their current room
-   * Optional workflowIdHint is used when socket mapping keys are missing/expired.
-   * Returns the workflowId they were in, or null if not in any room.
+   * Remove a user's membership of a workflow room.
+   * When workflowIdHint is provided it is the target room; the socket's current
+   * mapping is only the fallback (and covers missing/expired mapping keys).
+   * Socket-level mappings are cleared only when the socket is not mapped to a
+   * different room, so removing a stale room cannot destroy the mapping of a
+   * room the socket has since moved to.
+   * Returns the target workflowId, or null when no target could be resolved
+   * (or, for the Redis manager, when the removal failed).
    */
   removeUserFromRoom(socketId: string, workflowIdHint?: string): Promise<string | null>