fix(realtime): single-flight role resolution and retry failed eviction cleanup

Coalesce concurrent role resolutions per (user, workflow) so a slow stale
read can never overwrite a recorded revocation, and defer failed eviction
room-state cleanups into a per-sweep retry queue so collaborators are not
left with a stale presence entry.

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

Author: icecrasher321

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

@@ -163,20 +163,48 @@ describe('access-revalidation sweep', () => {
     expect(noRoom.leave).not.toHaveBeenCalled()
   })
 
-  it('still broadcasts presence when room-state removal fails', async () => {
+  it('defers failed room-state cleanup and retries it on the next pass', 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'))
+    manager.removeUserFromRoom.mockRejectedValueOnce(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).not.toHaveBeenCalled()
+
+    // The evicted socket is out of the room now, so membership scans no longer
+    // see it — the retry queue must drive the cleanup to completion.
+    socket.rooms = new Set(['sock-1'])
+    await sweep.runOnce()
+    sweep.stop()
+
+    expect(manager.removeUserFromRoom).toHaveBeenCalledTimes(2)
     expect(manager.broadcastPresenceUpdate).toHaveBeenCalledWith('wf-1')
   })
 
+  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' }])
+    manager.removeUserFromRoom.mockRejectedValueOnce(new Error('redis down'))
+    mockResolveRole.mockResolvedValueOnce(null)
+
+    const sweep = startAccessRevalidationSweep(manager)
+    await sweep.runOnce()
+    expect(socket.leave).toHaveBeenCalledWith('wf-1')
+
+    // Access restored and the socket re-joined (still in the room in this
+    // fake): the retry must NOT remove the fresh presence entry.
+    mockResolveRole.mockResolvedValue('read')
+    await sweep.runOnce()
+    sweep.stop()
+
+    expect(manager.removeUserFromRoom).toHaveBeenCalledTimes(1)
+    expect(manager.broadcastPresenceUpdate).not.toHaveBeenCalled()
+  })
+
   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

@@ -72,6 +72,44 @@ export function startAccessRevalidationSweep(roomManager: IRoomManager): AccessR
   const io = roomManager.io
   let running = false
 
+  /**
+   * Evictions whose room-state cleanup failed transiently, keyed
+   * `${socketId}:${workflowId}`. The evicted socket has already left the
+   * Socket.IO room, so membership scans will never see it again — these are
+   * retried at the start of every sweep pass until they succeed, so remaining
+   * collaborators do not keep a stale presence entry for the evicted socket.
+   */
+  const pendingCleanups = new Map<string, { socketId: string; workflowId: string }>()
+
+  async function cleanupEvictedSocket(socketId: string, workflowId: string): Promise<void> {
+    const key = `${socketId}:${workflowId}`
+    try {
+      await roomManager.removeUserFromRoom(socketId, workflowId)
+      await roomManager.broadcastPresenceUpdate(workflowId)
+      pendingCleanups.delete(key)
+    } catch (error) {
+      pendingCleanups.set(key, { socketId, workflowId })
+      logger.warn(
+        `Room-state cleanup failed for evicted socket ${socketId} on ${workflowId}; will retry next sweep`,
+        error
+      )
+    }
+  }
+
+  async function retryPendingCleanups(): Promise<void> {
+    for (const [key, { socketId, workflowId }] of pendingCleanups) {
+      const liveSocket = io.sockets.sockets.get(socketId)
+      if (liveSocket?.rooms.has(workflowId)) {
+        // The socket re-joined legitimately after the eviction (access was
+        // restored); that join re-added its presence entry, so there is
+        // nothing stale left to clean and removal would erase live presence.
+        pendingCleanups.delete(key)
+        continue
+      }
+      await cleanupEvictedSocket(socketId, workflowId)
+    }
+  }
+
   async function revokeSocket(socket: AuthenticatedSocket, workflowId: string): Promise<void> {
     // Security-critical, pod-local, and synchronous: stop this socket receiving
     // room broadcasts immediately, before any async bookkeeping that could fail.
@@ -87,27 +125,14 @@ export function startAccessRevalidationSweep(roomManager: IRoomManager): AccessR
       `Revoked live access for user ${socket.userId} on workflow ${workflowId} (socket ${socket.id})`
     )
 
-    // 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
-      )
-    }
+    // Cleanup failure never restores access (the socket already left the room);
+    // it defers to pendingCleanups and is retried on subsequent passes.
+    await cleanupEvictedSocket(socket.id, workflowId)
   }
 
   async function runOnce(): Promise<void> {
+    await retryPendingCleanups()
+
     const memberships = collectLocalMemberships(io)
 
     for (const [workflowId, sockets] of memberships) {

apps/realtime/src/handlers/workflow.ts

@@ -134,9 +134,11 @@ 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.
+      // revoked user resolves to null here. The resolver is single-flighted per
+      // (user, workflow), so this read cannot race the sweep's and overwrite a
+      // recorded revocation with a stale role; and 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(

apps/realtime/src/middleware/permissions.test.ts

@@ -23,7 +23,11 @@ vi.mock('@sim/platform-authz/workflow', () => ({
   authorizeWorkflowByWorkspacePermission: mockAuthorize,
 }))
 
-import { checkRolePermission, checkWorkflowOperationPermission } from '@/middleware/permissions'
+import {
+  checkRolePermission,
+  checkWorkflowOperationPermission,
+  resolveCurrentWorkflowRole,
+} from '@/middleware/permissions'
 
 describe('checkRolePermission', () => {
   describe('admin role', () => {
@@ -414,3 +418,66 @@ describe('checkWorkflowOperationPermission', () => {
     }
   })
 })
+
+describe('resolveCurrentWorkflowRole single-flight', () => {
+  const userId = 'sf-user-1'
+  let workflowCounter = 0
+  let workflowId: string
+
+  beforeEach(() => {
+    vi.clearAllMocks()
+    // Unique workflowId per test so the module-level role cache never leaks across tests
+    workflowCounter += 1
+    workflowId = `sf-wf-${workflowCounter}`
+  })
+
+  it('coalesces concurrent resolutions into a single authorization query', async () => {
+    let resolveAuthorize!: (value: { allowed: boolean; workspacePermission: string | null }) => void
+    mockAuthorize.mockReturnValue(
+      new Promise((resolve) => {
+        resolveAuthorize = resolve
+      })
+    )
+
+    // Both callers race the same expired/cold cache entry; they must share one
+    // in-flight query so a slower duplicate can never overwrite a newer
+    // decision (e.g. a revocation recorded by the eviction sweep).
+    const first = resolveCurrentWorkflowRole(userId, workflowId, 'read')
+    const second = resolveCurrentWorkflowRole(userId, workflowId, 'read')
+
+    resolveAuthorize({ allowed: true, workspacePermission: 'write' })
+
+    expect(await first).toBe('write')
+    expect(await second).toBe('write')
+    expect(mockAuthorize).toHaveBeenCalledTimes(1)
+  })
+
+  it('does not coalesce resolutions for different workflows', async () => {
+    mockAuthorize.mockResolvedValue({ allowed: true, workspacePermission: 'read' })
+
+    const [first, second] = await Promise.all([
+      resolveCurrentWorkflowRole(userId, workflowId, 'read'),
+      resolveCurrentWorkflowRole(userId, `${workflowId}-other`, 'read'),
+    ])
+
+    expect(first).toBe('read')
+    expect(second).toBe('read')
+    expect(mockAuthorize).toHaveBeenCalledTimes(2)
+  })
+
+  it('starts a fresh query after an in-flight resolution settles and its cache entry expires', async () => {
+    vi.useFakeTimers()
+    try {
+      mockAuthorize.mockResolvedValue({ allowed: true, workspacePermission: 'write' })
+      expect(await resolveCurrentWorkflowRole(userId, workflowId, 'read')).toBe('write')
+
+      vi.advanceTimersByTime(31_000)
+      mockAuthorize.mockResolvedValue({ allowed: false, workspacePermission: null })
+
+      expect(await resolveCurrentWorkflowRole(userId, workflowId, 'read')).toBeNull()
+      expect(mockAuthorize).toHaveBeenCalledTimes(2)
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+})

apps/realtime/src/middleware/permissions.ts

@@ -112,6 +112,15 @@ interface CachedRole {
  */
 const roleCache = new Map<string, CachedRole>()
 
+/**
+ * In-flight resolutions keyed like {@link roleCache}. Concurrent callers for the same
+ * (user, workflow) share one authorization query instead of racing independent ones, so
+ * cache writes per key are serialized — a slow, stale pre-revocation read can never
+ * overwrite a newer recorded decision (e.g. the revocation the eviction sweep just
+ * cached before kicking the socket).
+ */
+const inFlightRoleResolutions = new Map<string, Promise<string | null>>()
+
 function purgeExpiredRoles(now: number): void {
   for (const [key, entry] of roleCache) {
     if (entry.expiresAt <= now) {
@@ -120,35 +129,20 @@ function purgeExpiredRoles(now: number): void {
   }
 }
 
-/**
- * Resolves a user's current workspace role for a workflow, re-reading the `permissions`
- * table at most once per {@link ROLE_REVALIDATION_TTL_MS} per pod.
- *
- * Returns `null` when the user genuinely has no access (removed/revoked). On a transient
- * DB failure it reuses the last recorded decision for this (user, workflow) — including a
- * previously recorded revocation (`null`) — and only falls back to `fallbackRole` when no
- * decision has been recorded yet, so a blip neither blocks legitimate editors nor
- * resurrects already-revoked access.
- */
-export async function resolveCurrentWorkflowRole(
+async function resolveRoleUncached(
+  key: string,
   userId: string,
   workflowId: string,
   fallbackRole: string
 ): Promise<string | null> {
-  const now = Date.now()
-  const key = `${userId}:${workflowId}`
-  const cached = roleCache.get(key)
-  if (cached && cached.expiresAt > now) {
-    return cached.role
-  }
-
   try {
     const authorization = await authorizeWorkflowByWorkspacePermission({
       workflowId,
       userId,
       action: 'read',
     })
     const role = authorization.allowed ? (authorization.workspacePermission ?? null) : null
+    const now = Date.now()
     if (roleCache.size >= MAX_ROLE_CACHE_ENTRIES) {
       purgeExpiredRoles(now)
     }
@@ -168,6 +162,41 @@ export async function resolveCurrentWorkflowRole(
   }
 }
 
+/**
+ * Resolves a user's current workspace role for a workflow, re-reading the `permissions`
+ * table at most once per {@link ROLE_REVALIDATION_TTL_MS} per pod. Concurrent calls for
+ * the same (user, workflow) coalesce onto a single in-flight query (single-flight), so
+ * out-of-order cache writes cannot resurrect revoked access.
+ *
+ * Returns `null` when the user genuinely has no access (removed/revoked). On a transient
+ * DB failure it reuses the last recorded decision for this (user, workflow) — including a
+ * previously recorded revocation (`null`) — and only falls back to `fallbackRole` when no
+ * decision has been recorded yet, so a blip neither blocks legitimate editors nor
+ * resurrects already-revoked access.
+ */
+export async function resolveCurrentWorkflowRole(
+  userId: string,
+  workflowId: string,
+  fallbackRole: string
+): Promise<string | null> {
+  const key = `${userId}:${workflowId}`
+  const cached = roleCache.get(key)
+  if (cached && cached.expiresAt > Date.now()) {
+    return cached.role
+  }
+
+  const inFlight = inFlightRoleResolutions.get(key)
+  if (inFlight) {
+    return inFlight
+  }
+
+  const resolution = resolveRoleUncached(key, userId, workflowId, fallbackRole).finally(() => {
+    inFlightRoleResolutions.delete(key)
+  })
+  inFlightRoleResolutions.set(key, resolution)
+  return resolution
+}
+
 /**
  * Live permission gate for mutating socket operations. Re-validates the user's workspace
  * role against the database (cached per pod for {@link ROLE_REVALIDATION_TTL_MS}) so that