fix(realtime): narrow join commit-guard to post-success; skip empty presence broadcast

Two Cursor findings on the prior audit-fix commit:

- The 'committed' guard in join-workflow/join-table was too broad: a failure
  BETWEEN the membership commit and the success ack (e.g. getWorkflowState) hit
  'if (committed) return' and emitted neither success nor error, hanging the
  client while it sat in the room. Replaced with the narrower shape: only the
  purely-decorative post-success steps (peer broadcast + log-metric) are wrapped
  best-effort; anything before the success ack still rolls back and surfaces a
  retryable error, so the client retries instead of hanging — while the original
  goal (a benign broadcast/metric blip never kicking a live, acked user) holds.
- broadcastPresenceUpdate read the roster via getRoomUsers, which swallows a Redis
  transport error to []. On a disconnect broadcast that emitted an empty roster and
  cleared every remaining collaborator's presence until the next healthy update.
  Split out a throwing readRoomUsers; broadcastPresenceUpdate now skips the
  broadcast on a read failure (getRoomUsers keeps its swallow contract).
- Tests: pre-success failure rolls back + retryable error (no hang); post-success
  failure keeps the user joined.

Gates: realtime tsc 0, 206 realtime tests, biome, boundaries, prune.

Author: waleedlatif1

apps/realtime/src/handlers/tables.ts

@@ -95,9 +95,6 @@ export function setupTablesHandlers(socket: AuthenticatedSocket, roomManager: IR
     // check right before the membership commit.
     const superseded = () => joinGeneration !== joinAttempt || socket.disconnected
     if (superseded()) return
-    // Flips true once addUserToRoom lands: past that point the user is genuinely joined, so a
-    // failure in the trailing ack/broadcast steps must NOT roll them back (see the catch).
-    let committed = false
     try {
       const userId = socket.userId
       const userName = socket.userName
@@ -207,7 +204,6 @@ export function setupTablesHandlers(socket: AuthenticatedSocket, roomManager: IR
       // this write can land after it, leaving a stale presence entry. Benign and self-correcting:
       // filterVisiblePresence hides it and sweepStalePresence reclaims it (same as the siblings).
       await roomManager.addUserToRoom(room, socket.id, presence)
-      committed = true
 
       // 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.
@@ -222,19 +218,24 @@ export function setupTablesHandlers(socket: AuthenticatedSocket, roomManager: IR
         presenceUsers,
       })
 
-      await roomManager.broadcastPresenceUpdate(room)
+      // Post-success, purely decorative: notify peers. The user is already joined and acked, so a
+      // Redis blip here must not surface as a join failure — swallow it (the next healthy broadcast
+      // reconciles peers). Kept OUT of the rollback catch below, which is only for pre-success failures.
+      try {
+        await roomManager.broadcastPresenceUpdate(room)
+      } catch (error) {
+        logger.warn(`Post-join presence broadcast failed for table room ${tableId}`, error)
+      }
 
       logger.info(`User ${userId} (${userName}) joined table room ${tableId}`)
     } catch (error) {
       logger.error('Error joining table room:', error)
-      // Past the membership commit the user is genuinely joined; a failure in the trailing
-      // ack/broadcast steps must not tear them out of the room. Leave them joined.
-      if (committed) return
       // Roll back a partial join: cleanup keys off the socket→room map, so a `socket.join` that
       // landed without a matching `addUserToRoom` (a throw in between) would otherwise leave the
-      // socket stranded in the Socket.IO room, unreclaimable by any later op. Safe to run even when
-      // superseded — serialization means the newer op hasn't committed yet, so this touches only
-      // this join's own (this-table) state, never the newer op's room.
+      // socket stranded in the Socket.IO room, unreclaimable by any later op. A failure between the
+      // commit and the success ack rolls back too and surfaces a retryable error, so the client
+      // retries rather than hanging. Safe to run even when superseded — serialization means the
+      // newer op hasn't committed yet, so this touches only this join's own (this-table) state.
       try {
         const room = tableRoom(tableId)
         socket.leave(roomName(room))

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

@@ -374,6 +374,31 @@ describe('setupWorkflowHandlers', () => {
     expect(socket.emit).not.toHaveBeenCalledWith('join-workflow-error', expect.anything())
   })
 
+  it('rolls back and surfaces a retryable error when a pre-success step fails after commit', async () => {
+    // getWorkflowState runs after addUserToRoom but before the success ack — its failure must roll
+    // back and emit a retryable error so the client retries, never hanging committed-but-unacked.
+    mockGetWorkflowState.mockRejectedValue(new Error('db blip'))
+    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).not.toHaveBeenCalledWith('join-workflow-success', expect.anything())
+    expect(roomManager.removeUserFromRoom).toHaveBeenCalledWith(
+      { type: 'workflow', id: 'workflow-1' },
+      'socket-1'
+    )
+    expect(socket.emit).toHaveBeenCalledWith(
+      'join-workflow-error',
+      expect.objectContaining({ code: 'JOIN_WORKFLOW_FAILED', retryable: true })
+    )
+  })
+
   it('leaves the workflow room even when the session key has expired', async () => {
     const { socket, handlers } = createSocket()
     const roomManager = createRoomManager({

apps/realtime/src/handlers/workflow.ts

@@ -42,9 +42,6 @@ export function setupWorkflowHandlers(socket: AuthenticatedSocket, roomManager:
     // check right before the membership commit.
     const superseded = () => joinGeneration !== joinAttempt || socket.disconnected
     if (superseded()) return
-    // Flips true once addUserToRoom lands: past that point the user is genuinely joined, so a
-    // failure in the trailing ack/broadcast/metric steps must NOT roll them back (see the catch).
-    let committed = false
     try {
       const userId = socket.userId
       const userName = socket.userName
@@ -223,7 +220,6 @@ export function setupWorkflowHandlers(socket: AuthenticatedSocket, roomManager:
 
       // Add user to room — the membership commit.
       await roomManager.addUserToRoom(wf(workflowId), socket.id, userPresence)
-      committed = true
 
       // Get current presence list for the join acknowledgment, filtered to live members so a new
       // joiner never sees a ghost from an entry the stale sweep hasn't reclaimed yet.
@@ -246,22 +242,26 @@ export function setupWorkflowHandlers(socket: AuthenticatedSocket, roomManager:
       // Send workflow state
       socket.emit('workflow-state', workflowState)
 
-      // Broadcast presence update to all users in the room
-      await roomManager.broadcastPresenceUpdate(wf(workflowId))
-
-      const uniqueUserCount = await roomManager.getUniqueUserCount(wf(workflowId))
-      logger.info(
-        `User ${userId} (${userName}) joined workflow ${workflowId}. Room now has ${uniqueUserCount} unique users.`
-      )
+      // Post-success, purely decorative: notify peers and log the count. The user is already joined
+      // and acked, so a Redis blip here must not surface as a join failure — swallow it (the next
+      // healthy presence broadcast reconciles peers). It must stay OUT of the rollback catch below,
+      // which is only for pre-success failures.
+      try {
+        await roomManager.broadcastPresenceUpdate(wf(workflowId))
+        const uniqueUserCount = await roomManager.getUniqueUserCount(wf(workflowId))
+        logger.info(
+          `User ${userId} (${userName}) joined workflow ${workflowId}. Room now has ${uniqueUserCount} unique users.`
+        )
+      } catch (error) {
+        logger.warn(`Post-join presence broadcast failed for workflow ${workflowId}`, error)
+      }
     } catch (error) {
       logger.error('Error joining workflow:', error)
-      // Past the membership commit the user is genuinely joined; a failure in the trailing
-      // ack/broadcast/metric steps must not tear them out of the room (a benign Redis blip on a
-      // pure log-metric call would otherwise kick a live collaborator). Leave them joined.
-      if (committed) return
       // Roll back a partial join: cleanup keys off the socket→room map, so a `socket.join` that
       // landed without a matching `addUserToRoom` (a throw in between) would otherwise strand the
-      // socket in the Socket.IO room, unreclaimable by any later op. Safe even when superseded —
+      // socket in the Socket.IO room, unreclaimable by any later op. A failure between the commit
+      // and the success ack rolls back too and surfaces a retryable error — so the client retries
+      // rather than hanging (never left committed-but-unacked). Safe even when superseded —
       // serialization means the newer op hasn't committed yet, so this touches only this join's own
       // room state, never the newer op's.
       socket.leave(workflowId)

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

@@ -298,19 +298,28 @@ export class RedisRoomManager implements IRoomManager {
     }
   }
 
+  /**
+   * Reads and parses the room roster. Throws on a transport error (so a caller can
+   * distinguish "genuinely empty" from "read failed"); a single corrupted entry is
+   * skipped, not fatal.
+   */
+  private async readRoomUsers(room: RoomRef): Promise<UserPresence[]> {
+    const users = await this.redis.hGetAll(KEYS.roomUsers(room))
+    return Object.entries(users)
+      .map(([socketId, json]) => {
+        try {
+          return JSON.parse(json) as UserPresence
+        } catch {
+          logger.warn(`Corrupted user data for socket ${socketId}, skipping`)
+          return null
+        }
+      })
+      .filter((u): u is UserPresence => u !== null)
+  }
+
   async getRoomUsers(room: RoomRef): Promise<UserPresence[]> {
     try {
-      const users = await this.redis.hGetAll(KEYS.roomUsers(room))
-      return Object.entries(users)
-        .map(([socketId, json]) => {
-          try {
-            return JSON.parse(json) as UserPresence
-          } catch {
-            logger.warn(`Corrupted user data for socket ${socketId}, skipping`)
-            return null
-          }
-        })
-        .filter((u): u is UserPresence => u !== null)
+      return await this.readRoomUsers(room)
     } catch (error) {
       logger.error(`Failed to get room users for ${room.type}:${room.id}:`, error)
       return []
@@ -374,7 +383,20 @@ export class RedisRoomManager implements IRoomManager {
   }
 
   async broadcastPresenceUpdate(room: RoomRef, excludeSocketId?: string): Promise<void> {
-    const users = await this.getRoomUsers(room)
+    let users: UserPresence[]
+    try {
+      // Read via the throwing variant, NOT getRoomUsers: a transport error there returns `[]`, which
+      // would broadcast an empty roster and clear every remaining collaborator's presence until the
+      // next healthy update. Skip instead — the next successful join/activity broadcast (or the
+      // stale sweep) reconciles peers.
+      users = await this.readRoomUsers(room)
+    } catch (error) {
+      logger.error(
+        `Skipping presence broadcast for ${room.type}:${room.id} (roster read failed):`,
+        error
+      )
+      return
+    }
     const visible = await filterVisiblePresence(this._io, room, users, excludeSocketId)
     // io.to() with the Redis adapter broadcasts to all pods.
     this._io.to(roomName(room)).emit(presenceEventName(room.type), visible)