fix(realtime): guard table join commit + rollback against supersession; drop no-op eviction cleanup

Review round on #5991:
- Table join re-checked the generation only once after authorize, then awaited
  leave/sweep/avatar before joining + registering presence. A table switch or
  leave in that window stranded the socket in the wrong room, and the failure
  catch could tear down a newer successful join. Resolve the avatar up-front,
  re-check generation immediately before the membership commit (matching the
  file-doc join), and skip the rollback/error for a superseded join. + a
  post-authorize-window regression test.
- access-revalidation cleanup treated removeUserFromRoom's no-op false as a
  transport failure and re-enqueued a still-connected socket forever. Only retry
  when the socket is still mapped to the room (a healthy null mapping means the
  entry is already gone). Repurposed the expired-mapping test to lock it.

Author: waleedlatif1

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

@@ -226,25 +226,25 @@ describe('access-revalidation sweep', () => {
     expect(manager.broadcastPresenceUpdate).toHaveBeenCalledWith({ type: 'workflow', id: 'wf-1' })
   })
 
-  it('defers cleanup when removal fails with expired socket mappings', async () => {
+  it('drops eviction cleanup when the socket is no longer mapped to the room (no infinite retry)', async () => {
     const socket = makeSocket('sock-1', 'user-1', 'wf-1')
     const manager = makeManager([socket], [{ socketId: 'sock-1', role: 'read' }])
-    // Mapping keys already expired (lookup resolves null) AND the removal fails
-    // (the Redis manager swallows the transport error into null) — the failed
-    // removal must still defer instead of reading as success.
-    manager.removeUserFromRoom.mockResolvedValueOnce(false)
+    // A healthy lookup shows the socket is no longer mapped to any workflow room (its presence
+    // is already gone), and removeUserFromRoom reports a no-op `false`. This is "already clean",
+    // not a deferrable failure — the cleanup must drop it, never re-enqueue a still-connected
+    // socket forever. (A genuine failure — still mapped + false — is covered by the next test.)
+    manager.getRoomForSocket.mockResolvedValue(null)
+    manager.removeUserFromRoom.mockResolvedValue(false)
     mockResolveRole.mockResolvedValue(null)
 
     const sweep = startAccessRevalidationSweep(manager)
     await sweep.runOnce()
-
-    expect(manager.broadcastPresenceUpdate).not.toHaveBeenCalled()
-
     await sweep.runOnce()
     sweep.stop()
 
-    expect(manager.removeUserFromRoom).toHaveBeenCalledTimes(2)
-    expect(manager.broadcastPresenceUpdate).toHaveBeenCalledWith({ type: 'workflow', id: 'wf-1' })
+    // Attempted once, then dropped — not re-enqueued across passes, and no broadcast.
+    expect(manager.removeUserFromRoom).toHaveBeenCalledTimes(1)
+    expect(manager.broadcastPresenceUpdate).not.toHaveBeenCalled()
   })
 
   it('defers cleanup when the manager swallows a removal failure into null', async () => {

apps/realtime/src/access-revalidation.ts

@@ -172,11 +172,18 @@ export function startAccessRevalidationSweep(roomManager: IRoomManager): AccessR
       // entry from the known target room via the explicit ref below.
       const removed = await roomManager.removeUserFromRoom(wf(workflowId), socketId)
       if (!removed) {
-        // The socket is still connected and hasn't moved/re-joined, yet the removal
-        // wasn't confirmed — a transport error the manager swallowed into false, or a
-        // lost race. Defer and retry next sweep rather than leave a revoked
-        // collaborator's stale presence entry behind.
-        throw new Error('room-state removal not confirmed')
+        // `false` conflates two outcomes: the entry was already gone (a no-op), or a
+        // transport error the manager swallowed. Only retry when the socket is still mapped
+        // to THIS room — then a false result is a genuine, deferrable failure. When a healthy
+        // getRoomForSocket above returned no workflow mapping (`currentWorkflowId === null`),
+        // the presence entry is already gone, so the cleanup is complete: dropping it avoids
+        // re-enqueuing a still-connected socket forever. (A real Redis outage throws at
+        // getRoomForSocket and is deferred by the outer catch, never reaching here.)
+        if (currentWorkflowId === workflowId) {
+          throw new Error('room-state removal not confirmed')
+        }
+        pendingCleanups.delete(key)
+        return
       }
 
       await roomManager.broadcastPresenceUpdate(wf(workflowId))

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

@@ -284,6 +284,53 @@ describe('setupTablesHandlers', () => {
     )
   })
 
+  it('aborts a join superseded during the post-authorize leave/sweep window', async () => {
+    const { socket, handlers } = createSocket()
+    // Hold A hung on its leave-prior lookup (which runs AFTER the post-authorize recheck), then
+    // fire a newer join B. When A resumes, the final generation guard before the membership
+    // commit must abort it — the post-authorize awaits are no longer an unguarded window.
+    let aReachedLookup: () => void = () => {}
+    const aAtLookup = new Promise<void>((resolve) => {
+      aReachedLookup = resolve
+    })
+    let releaseLookup: (value: unknown) => void = () => {}
+    const pendingLookup = new Promise((resolve) => {
+      releaseLookup = resolve
+    })
+    let lookupCalls = 0
+    const roomManager = createRoomManager({
+      getRoomForSocket: vi.fn(() => {
+        lookupCalls += 1
+        if (lookupCalls === 1) {
+          aReachedLookup()
+          return pendingLookup
+        }
+        return Promise.resolve(null)
+      }),
+    })
+    mockAuthorizeRoom.mockResolvedValue({
+      allowed: true,
+      status: 200,
+      workspaceId: 'ws-1',
+      workspacePermission: 'admin',
+    })
+    setupTablesHandlers(socket as unknown as SetupArg, roomManager)
+
+    const joinA = handlers[TABLE_PRESENCE_EVENTS.JOIN]({ tableId: 'table-A' })
+    await aAtLookup // A has passed its post-authorize recheck and is hung on the leave-prior lookup
+    await handlers[TABLE_PRESENCE_EVENTS.JOIN]({ tableId: 'table-B' }) // bumps generation, completes
+    releaseLookup(null)
+    await joinA
+
+    expect(socket.join).toHaveBeenCalledWith('table:table-B')
+    expect(socket.join).not.toHaveBeenCalledWith('table:table-A')
+    expect(roomManager.addUserToRoom).not.toHaveBeenCalledWith(
+      { type: ROOM_TYPES.TABLE, id: 'table-A' },
+      expect.anything(),
+      expect.anything()
+    )
+  })
+
   it('leaves the table room on leave', async () => {
     const { socket, handlers } = createSocket()
     const roomManager = createRoomManager({

apps/realtime/src/handlers/tables.ts

@@ -131,6 +131,12 @@ export function setupTablesHandlers(socket: AuthenticatedSocket, roomManager: IR
 
       // A newer JOIN started on this socket during authorize (or the socket dropped):
       // abort so a stale join can't leave the room the client has since moved to.
+      // Server-authenticated avatar for the presence roster. Resolved up-front so the guard
+      // below also covers this await (mirrors the file-doc join).
+      const avatarUrl = await resolveAvatarUrl(socket, userId)
+
+      // Abort a JOIN superseded during authorize/avatar resolution — a newer JOIN (table
+      // switch), a LEAVE, or a disconnect. Registering below would strand the socket.
       if (joinGeneration !== joinAttempt || socket.disconnected) return
 
       // Leave a previously-joined table room if switching tables.
@@ -162,6 +168,12 @@ export function setupTablesHandlers(socket: AuthenticatedSocket, roomManager: IR
         }
       }
 
+      // Final re-check immediately before the membership commit: a newer JOIN (table switch), a
+      // LEAVE, or a disconnect during the leave/sweep awaits above must abort here — otherwise
+      // this superseded join would join the room and register presence, stranding the socket in
+      // the wrong table. No await sits between this guard and addUserToRoom (the commit).
+      if (joinGeneration !== joinAttempt || socket.disconnected) return
+
       socket.join(roomName(room))
 
       const presence: UserPresence = {
@@ -173,7 +185,7 @@ export function setupTablesHandlers(socket: AuthenticatedSocket, roomManager: IR
         joinedAt: Date.now(),
         lastActivity: Date.now(),
         role: authorized.workspacePermission ?? 'read',
-        avatarUrl: await resolveAvatarUrl(socket, userId),
+        avatarUrl,
       }
 
       await roomManager.addUserToRoom(room, socket.id, presence)
@@ -196,6 +208,10 @@ export function setupTablesHandlers(socket: AuthenticatedSocket, roomManager: IR
       logger.info(`User ${userId} (${userName}) joined table room ${tableId}`)
     } catch (error) {
       logger.error('Error joining table room:', error)
+      // A superseded join (a newer join/leave bumped the generation) must NOT roll back — it
+      // would tear down room state a newer successful join to the same table now holds — nor
+      // signal an error for a table the client already left.
+      if (joinGeneration !== joinAttempt) return
       // 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, before signalling a retry.
       try {