fix(realtime): fence file joins by owner-qualified room

Author: mzxchandra

apps/realtime/src/handlers/file-doc-project.integration.ts

@@ -13,6 +13,7 @@ import {
   type JoinFileDocError,
   type JoinFileDocPayload,
   type JoinFileDocSuccess,
+  type LeaveFileDocPayload,
 } from '@sim/realtime-protocol/file-doc'
 import { createDeferred } from '@sim/testing/helpers/deferred'
 import { getErrorMessage } from '@sim/utils/errors'
@@ -296,6 +297,97 @@ describe('Project documents across the Socket.IO and internal HTTP boundary', ()
     })
   }
 
+  for (const action of ['join', 'leave'] as const) {
+    check(
+      action === 'join'
+        ? 'a join for another owner supersedes the pending admission for the same file'
+        : 'a leave for another owner preserves the pending admission for the same file',
+      async () => {
+        const actor = `owner-intent-${generateId()}`
+        actors.set(actor, 'read')
+        const gate = {
+          requests: 0,
+          entered: createDeferred<void>(),
+          release: createDeferred<void>(),
+          status: 200,
+        }
+        accessGates.set(actor, gate)
+        const socket = connect(socketUrl, { transports: ['websocket'], auth: { actor } })
+        clients.push(socket)
+        await new Promise<void>((resolve) => socket.once('connect', resolve))
+        const server = io.sockets.sockets.get(socket.id ?? '')
+        if (!server) throw new Error('Pending socket missing')
+        const originalJoin = server.listeners(FILE_DOC_EVENTS.JOIN)[0]
+        const originalLeave = server.listeners(FILE_DOC_EVENTS.LEAVE)[0]
+        if (!originalJoin || !originalLeave) throw new Error('Document lifecycle handlers missing')
+        const provider = new Y.Doc()
+        const nextProvider = new Y.Doc()
+        const done = createDeferred<void>()
+        const actionDone = createDeferred<void>()
+        const observeJoin = async (payload: JoinFileDocPayload) => {
+          try {
+            await originalJoin.call(server, payload)
+          } finally {
+            if (payload.clientId === provider.clientID) done.resolve()
+            else actionDone.resolve()
+          }
+        }
+        const observeLeave = async (payload: LeaveFileDocPayload) => {
+          try {
+            await originalLeave.call(server, payload)
+          } finally {
+            actionDone.resolve()
+          }
+        }
+        server.off(FILE_DOC_EVENTS.JOIN, originalJoin)
+        server.on(FILE_DOC_EVENTS.JOIN, observeJoin)
+        server.off(FILE_DOC_EVENTS.LEAVE, originalLeave)
+        server.on(FILE_DOC_EVENTS.LEAVE, observeLeave)
+        const admissions: JoinFileDocSuccess[] = []
+        const errors: JoinFileDocError[] = []
+        socket.on(FILE_DOC_EVENTS.JOIN_SUCCESS, (payload) => admissions.push(payload))
+        socket.on(FILE_DOC_EVENTS.JOIN_ERROR, (payload) => errors.push(payload))
+        try {
+          socket.emit(FILE_DOC_EVENTS.JOIN, {
+            fileId,
+            projectId,
+            clientId: provider.clientID,
+            schemaVersion: FILE_DOC_SCHEMA_VERSION,
+          })
+          await gate.entered.promise
+          const otherOwner = { fileId, projectId: generateId() }
+          if (action === 'join')
+            socket.emit(FILE_DOC_EVENTS.JOIN, {
+              ...otherOwner,
+              clientId: nextProvider.clientID,
+              schemaVersion: FILE_DOC_SCHEMA_VERSION,
+            })
+          else socket.emit(FILE_DOC_EVENTS.LEAVE, otherOwner)
+          await actionDone.promise
+          gate.release.resolve()
+          await done.promise
+          const drained = new Promise<void>((resolve) => socket.once('fixture-drained', resolve))
+          server.emit('fixture-drained')
+          await drained
+          const shouldAdmit = action === 'leave'
+          expect(server.rooms.has(`project-file-doc:${projectId}/${fileId}`)).toBe(shouldAdmit)
+          expect(admissions).toHaveLength(shouldAdmit ? 1 : 0)
+          expect(errors.map((error) => error.code)).toEqual(shouldAdmit ? [] : ['NOT_FOUND'])
+        } finally {
+          gate.release.resolve()
+          accessGates.delete(actor)
+          server.off(FILE_DOC_EVENTS.JOIN, observeJoin)
+          server.on(FILE_DOC_EVENTS.JOIN, originalJoin)
+          server.off(FILE_DOC_EVENTS.LEAVE, observeLeave)
+          server.on(FILE_DOC_EVENTS.LEAVE, originalLeave)
+          socket.disconnect()
+          provider.destroy()
+          nextProvider.destroy()
+        }
+      }
+    )
+  }
+
   for (const rejectLeave of [false, true]) {
     check(
       `a stale rejoin preserves the admitted document when delayed cleanup ${rejectLeave ? 'rejects' : 'resolves'}`,

apps/realtime/src/handlers/file-doc.ts

@@ -1515,11 +1515,8 @@ export function setupWorkspaceFileDocHandlers(
   roomManager: IRoomManager
 ) {
   const io = roomManager.io
-  // The file this socket currently intends to edit (set when a join starts). A leave targeting it
-  // — or an unscoped leave — advances the join generation to cancel an in-flight join, so a join
-  // awaiting authorization can't complete after the client left and register a ghost owner. A
-  // leave for a DIFFERENT file must NOT cancel it (a document switch), mirroring workspace-files.
-  let currentFileId: string | null = null
+  // Owner-qualified intent fences delayed admissions and leaves across document switches.
+  let currentFileRoom: string | null = null
   /** Co-mounted providers share invalidation membership until their last admission settles. */
   const pendingMemberships = new Map<string, number>()
 
@@ -1594,19 +1591,18 @@ export function setupWorkspaceFileDocHandlers(
         return
       }
 
-      // A generation represents the socket's intended FILE, not an individual provider. Co-mounted
-      // providers for the same file must be allowed to join concurrently; switching files advances the
-      // generation so every in-flight join for the old file is cancelled together.
-      if (currentFileId !== fileId) {
+      const room = fileDocRoom(fileId, projectId)
+      const name = roomName(room)
+
+      // Co-mounted providers share a generation only while their owner-qualified target matches.
+      if (currentFileRoom !== name) {
         generation = (joinGeneration.get(socket.id) ?? 0) + 1
         joinGeneration.set(socket.id, generation)
-        currentFileId = fileId
+        currentFileRoom = name
       } else {
         generation = joinGeneration.get(socket.id) ?? 0
       }
 
-      const room = fileDocRoom(fileId, projectId)
-      const name = roomName(room)
       const admissionName = fileDocAdmissionRoom(fileId, projectId)
 
       const authorizeJoin = () =>
@@ -1957,23 +1953,20 @@ export function setupWorkspaceFileDocHandlers(
 
   socket.on(FILE_DOC_EVENTS.LEAVE, async (payload?: LeaveFileDocPayload) => {
     try {
-      // Cancel an in-flight join whose file the client is now leaving (or an unscoped leave): a
-      // join still awaiting authorization would otherwise complete after the client left, register
-      // as an owner, and broadcast a ghost collaborator until disconnect. Guard on the current
-      // file intent so a stale/deferred leave for a DIFFERENT file can't abort the join the client
-      // has since switched to (bumping the generation blindly caused that regression in #5941).
-      if (!payload?.fileId || payload.fileId === currentFileId) {
+      const leavingRoom = payload?.fileId
+        ? roomName(fileDocRoom(payload.fileId, payload.projectId))
+        : undefined
+      if (!leavingRoom || leavingRoom === currentFileRoom) {
         joinGeneration.set(socket.id, (joinGeneration.get(socket.id) ?? 0) + 1)
-        currentFileId = null
+        currentFileRoom = null
       }
       // Tear down membership only for a REGISTERED room; a leave that raced ahead of an in-flight
       // join (nothing registered yet) has already cancelled it above.
       const name = socketToRoomName.get(socket.id)
       if (!name) return
       // Scope the leave to the named file when provided: a deferred leave from a
       // prior document must not evict the socket from one it has since opened.
-      if (payload?.fileId && roomName(fileDocRoom(payload.fileId, payload.projectId)) !== name)
-        return
+      if (leavingRoom && leavingRoom !== name) return
       cleanupFileDocForSocket(socket.id, io)
       await leaveUnclaimedRoom(name)
     } catch (error) {