fix(realtime): close join/eviction race and make sweep cleanups independent

Re-authorize immediately before socket.join so an in-flight join cannot
reverse a sweep eviction, and run the sweep's best-effort cleanups
independently so a room-state failure cannot skip the presence broadcast.

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

Author: icecrasher321

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

@@ -163,6 +163,20 @@ describe('access-revalidation sweep', () => {
     expect(noRoom.leave).not.toHaveBeenCalled()
   })
 
+  it('still broadcasts presence when room-state removal fails', async () => {
+    const socket = makeSocket('sock-1', 'user-1', 'wf-1')
+    const manager = makeManager([socket], [{ socketId: 'sock-1', role: 'read' }])
+    manager.removeUserFromRoom.mockRejectedValue(new Error('redis down'))
+    mockResolveRole.mockResolvedValue(null)
+
+    const sweep = startAccessRevalidationSweep(manager)
+    await sweep.runOnce()
+    sweep.stop()
+
+    expect(socket.leave).toHaveBeenCalledWith('wf-1')
+    expect(manager.broadcastPresenceUpdate).toHaveBeenCalledWith('wf-1')
+  })
+
   it('still evaluates access when presence lookup fails (falls back safely)', async () => {
     const socket = makeSocket('sock-1', 'user-1', 'wf-1')
     const manager = makeManager([socket])

apps/realtime/src/access-revalidation.ts

@@ -87,9 +87,24 @@ export function startAccessRevalidationSweep(roomManager: IRoomManager): AccessR
       `Revoked live access for user ${socket.userId} on workflow ${workflowId} (socket ${socket.id})`
     )
 
-    // Best-effort room-state cleanup; failure here does not restore access.
-    await roomManager.removeUserFromRoom(socket.id, workflowId)
-    await roomManager.broadcastPresenceUpdate(workflowId)
+    // Best-effort cleanups; each is independent so one failure neither restores
+    // access nor prevents the other from running.
+    try {
+      await roomManager.removeUserFromRoom(socket.id, workflowId)
+    } catch (error) {
+      logger.warn(
+        `Failed to remove evicted socket ${socket.id} from room state for ${workflowId}`,
+        error
+      )
+    }
+    try {
+      await roomManager.broadcastPresenceUpdate(workflowId)
+    } catch (error) {
+      logger.warn(
+        `Failed to broadcast presence after evicting socket ${socket.id} from ${workflowId}`,
+        error
+      )
+    }
   }
 
   async function runOnce(): Promise<void> {

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

@@ -4,10 +4,12 @@
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 import type { IRoomManager } from '@/rooms'
 
-const { mockGetWorkflowState, mockVerifyWorkflowAccess } = vi.hoisted(() => ({
-  mockGetWorkflowState: vi.fn(),
-  mockVerifyWorkflowAccess: vi.fn(),
-}))
+const { mockGetWorkflowState, mockVerifyWorkflowAccess, mockResolveCurrentWorkflowRole } =
+  vi.hoisted(() => ({
+    mockGetWorkflowState: vi.fn(),
+    mockVerifyWorkflowAccess: vi.fn(),
+    mockResolveCurrentWorkflowRole: vi.fn(),
+  }))
 
 vi.mock('@sim/db', () => ({
   db: { select: vi.fn() },
@@ -20,6 +22,7 @@ vi.mock('@/database/operations', () => ({
 
 vi.mock('@/middleware/permissions', () => ({
   verifyWorkflowAccess: mockVerifyWorkflowAccess,
+  resolveCurrentWorkflowRole: mockResolveCurrentWorkflowRole,
 }))
 
 import { setupWorkflowHandlers } from '@/handlers/workflow'
@@ -86,6 +89,7 @@ describe('setupWorkflowHandlers', () => {
     vi.clearAllMocks()
     mockGetWorkflowState.mockResolvedValue({ id: 'workflow-1', state: {} })
     mockVerifyWorkflowAccess.mockResolvedValue({ hasAccess: true, role: 'admin' })
+    mockResolveCurrentWorkflowRole.mockResolvedValue('admin')
   })
 
   it('includes workflowId when authentication is missing', async () => {
@@ -149,6 +153,52 @@ describe('setupWorkflowHandlers', () => {
     })
   })
 
+  it('denies the join when access is revoked while the join is in flight', async () => {
+    mockResolveCurrentWorkflowRole.mockResolvedValue(null)
+
+    const { socket, handlers } = createSocket()
+    const roomManager = createRoomManager()
+
+    setupWorkflowHandlers(
+      socket as unknown as Parameters<typeof setupWorkflowHandlers>[0],
+      roomManager
+    )
+
+    await handlers['join-workflow']({ workflowId: 'workflow-1', tabSessionId: 'tab-1' })
+
+    expect(socket.emit).toHaveBeenCalledWith('join-workflow-error', {
+      workflowId: 'workflow-1',
+      error: 'Access denied to workflow',
+      code: 'ACCESS_DENIED',
+      retryable: false,
+    })
+    expect(socket.join).not.toHaveBeenCalled()
+    expect(roomManager.addUserToRoom).not.toHaveBeenCalled()
+  })
+
+  it('joins with the re-validated role, passing the join-time role as fallback', async () => {
+    mockVerifyWorkflowAccess.mockResolvedValue({ hasAccess: true, role: 'write' })
+    mockResolveCurrentWorkflowRole.mockResolvedValue('read')
+
+    const { socket, handlers } = createSocket()
+    const roomManager = createRoomManager()
+
+    setupWorkflowHandlers(
+      socket as unknown as Parameters<typeof setupWorkflowHandlers>[0],
+      roomManager
+    )
+
+    await handlers['join-workflow']({ workflowId: 'workflow-1', tabSessionId: 'tab-1' })
+
+    expect(mockResolveCurrentWorkflowRole).toHaveBeenCalledWith('user-1', 'workflow-1', 'write')
+    expect(socket.join).toHaveBeenCalledWith('workflow-1')
+    expect(roomManager.addUserToRoom).toHaveBeenCalledWith(
+      'workflow-1',
+      'socket-1',
+      expect.objectContaining({ role: 'read' })
+    )
+  })
+
   it('marks workflow access verification failures as retryable', async () => {
     mockVerifyWorkflowAccess.mockRejectedValue(new Error('database unavailable'))
 

apps/realtime/src/handlers/workflow.ts

@@ -3,7 +3,7 @@ import { createLogger } from '@sim/logger'
 import { eq } from 'drizzle-orm'
 import { getWorkflowState } from '@/database/operations'
 import type { AuthenticatedSocket } from '@/middleware/auth'
-import { verifyWorkflowAccess } from '@/middleware/permissions'
+import { resolveCurrentWorkflowRole, verifyWorkflowAccess } from '@/middleware/permissions'
 import type { IRoomManager, UserPresence } from '@/rooms'
 
 const logger = createLogger('WorkflowHandlers')
@@ -131,6 +131,27 @@ export function setupWorkflowHandlers(socket: AuthenticatedSocket, roomManager:
         }
       }
 
+      // Re-authorize immediately before joining: the access-revalidation sweep
+      // may have evicted this socket while the awaits above were in flight, and
+      // its eviction is recorded in the shared role cache before it runs — so a
+      // revoked user resolves to null here. No awaits sit between this check
+      // and socket.join, so a sweep eviction cannot interleave after it and be
+      // reversed by this join.
+      const currentRole = await resolveCurrentWorkflowRole(userId, workflowId, userRole)
+      if (currentRole === null) {
+        logger.warn(
+          `User ${userId} (${userName}) lost access to workflow ${workflowId} before join completed`
+        )
+        socket.emit('join-workflow-error', {
+          workflowId,
+          error: 'Access denied to workflow',
+          code: 'ACCESS_DENIED',
+          retryable: false,
+        })
+        return
+      }
+      userRole = currentRole
+
       // Join the new room
       socket.join(workflowId)