fix(file-doc): reschedule an unattributed persist and pin admission and persist attribution with tests

Author: waleedlatif1

apps/realtime/src/handlers/file-doc.redis-store.test.ts

@@ -20,17 +20,18 @@ import * as syncProtocol from 'y-protocols/sync'
 import * as Y from 'yjs'
 import type { IRoomManager } from '@/rooms'
 
-const { mockAuthorizeRoom, mockFetchFileDocSeed } = vi.hoisted(() => ({
+const { mockAuthorizeRoom, mockFetchFileDocSeed, mockFetchFileDocPersist } = vi.hoisted(() => ({
   mockAuthorizeRoom: vi.fn(),
   mockFetchFileDocSeed: vi.fn(),
+  mockFetchFileDocPersist: vi.fn(),
 }))
 
 vi.mock('@sim/platform-authz/rooms', () => ({ authorizeRoom: mockAuthorizeRoom }))
 
 vi.mock('@/handlers/file-doc-app', () => ({
   fetchFileDocSeed: mockFetchFileDocSeed,
   fetchFileDocMerge: vi.fn(),
-  fetchFileDocPersist: vi.fn().mockResolvedValue({ status: 'persisted', version: 1 }),
+  fetchFileDocPersist: mockFetchFileDocPersist,
 }))
 
 /** One in-memory Redis backing per test — only the stream/lock ops the store actually uses. */
@@ -269,6 +270,7 @@ describe('file-doc join readiness (shared store enabled)', () => {
       workspacePermission: 'write',
     })
     mockFetchFileDocSeed.mockResolvedValue(null)
+    mockFetchFileDocPersist.mockResolvedValue({ status: 'persisted', version: 2 })
   })
 
   afterEach(() => {
@@ -326,4 +328,49 @@ describe('file-doc join readiness (shared store enabled)', () => {
     expect(mockFetchFileDocSeed).not.toHaveBeenCalled()
     expect(statesDeliveredTo(socket)).toEqual(['seeded by the writer that held the lock'])
   })
+
+  it('persists unattributed stream state under the local editor when the last collaborator leaves', async () => {
+    backing.kv.set(`filedoc:generation:${ROOM_NAME}`, 'doc-1')
+    backing.kv.set(`filedoc:syncver:${ROOM_NAME}`, '1')
+    const doc = new Y.Doc()
+    doc.getText(FIELD).insert(0, 'AAA')
+    doc.getMap(FILE_DOC_SEED.configMap).set(FILE_DOC_SEED.flag, true)
+    doc.getMap(FILE_DOC_SEED.configMap).set(FILE_DOC_SEED.docIdKey, 'doc-1')
+    backing.streams.set(STREAM_KEY, [
+      {
+        id: `${++backing.seq}-0`,
+        message: { u: Buffer.from(Y.encodeStateAsUpdate(doc)).toString('base64'), g: 'doc-1' },
+      },
+    ])
+    const afterSeed = Y.encodeStateVector(doc)
+    doc.getText(FIELD).insert(0, 'BBB')
+    const edit = Y.encodeStateAsUpdate(doc, afterSeed)
+    doc.destroy()
+
+    const { handlers } = setup('socket-1', sockets)
+    await handlers[FILE_DOC_EVENTS.JOIN]({
+      fileId: FILE_ID,
+      clientId: 1,
+      schemaVersion: FILE_DOC_SCHEMA_VERSION,
+    })
+    const acknowledge = vi.fn()
+    await (handlers[FILE_DOC_EVENTS.UPDATE] as (...args: unknown[]) => Promise<void>)(
+      { fileId: FILE_ID, docId: 'doc-1', updateId: 'update-1', update: edit },
+      acknowledge
+    )
+    await vi.waitFor(() =>
+      expect(acknowledge).toHaveBeenCalledWith({ status: 'accepted', updateId: 'update-1' })
+    )
+
+    cleanupFileDocForSocket('socket-1', createIo(sockets), true)
+    await vi.waitFor(() =>
+      expect(mockFetchFileDocPersist).toHaveBeenCalledWith(
+        'ws-1',
+        FILE_ID,
+        'user-1',
+        expect.any(Uint8Array),
+        1
+      )
+    )
+  })
 })

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

@@ -687,6 +687,28 @@ describe('setupWorkspaceFileDocHandlers', () => {
     }
   })
 
+  it('tells an admitted writer it can write even if its cached role lapses before the reply', async () => {
+    // A distinct user/file: role decisions recorded under fake timers outlive this test.
+    vi.useFakeTimers({ toFake: ['Date'] })
+    try {
+      const { io } = createIo()
+      const data: Record<string, unknown> = {}
+      // Registration records presence just before the reply; expire the role cache right there.
+      Object.defineProperty(data, 'fileDocPresence', {
+        set: () => vi.setSystemTime(Date.now() + ROLE_REVALIDATION_TTL_MS + 1),
+      })
+      const { socket, handlers } = setup('socket-lapse', io, { userId: 'user-lapse', data })
+      await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-lapse', clientId: 9 })
+
+      expect(socket.emit).toHaveBeenCalledWith(
+        FILE_DOC_EVENTS.JOIN_SUCCESS,
+        expect.objectContaining({ fileId: 'file-lapse', canWrite: true })
+      )
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
   it('publishes a re-validated permission only when it changes what the socket last heard', async () => {
     const { io } = createIo()
     const { socket, handlers } = setup('socket-sweep', io)

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

@@ -467,7 +467,11 @@ async function persistRoom(name: string, room: FileDocRoom, final: boolean): Pro
     const docState = await captureState()
     if (!docState) return // nothing seeded/authoritative to persist yet
     if (!(await store.isDocumentGenerationCurrent(name, generation))) return
-    if (!snapshotEditor) return
+    if (!snapshotEditor) {
+      // The claimed window must not strand the edits; a later flush may find an author.
+      if (!final && fileDocRooms.get(name) === room) schedulePersist(name, room)
+      return
+    }
     const result = await fileDocOwnerAdapter(room.owner).persist(
       room,
       snapshotEditor,

apps/realtime/src/handlers/room-permission.ts

@@ -1,5 +1,6 @@
 import { createLogger } from '@sim/logger'
 import type { RoomRef, RoomType } from '@sim/realtime-protocol/rooms'
+import { getErrorMessage } from '@sim/utils/errors'
 import type { AuthenticatedSocket } from '@/middleware/auth'
 
 const logger = createLogger('RoomPermission')
@@ -36,6 +37,8 @@ export function publishRoomPermission(
   try {
     publisher(socket, room, permission)
   } catch (error) {
-    logger.warn(`Room permission publish failed for socket ${socket.id} on ${room.type}`, error)
+    logger.warn(`Room permission publish failed for socket ${socket.id} on ${room.type}`, {
+      error: getErrorMessage(error),
+    })
   }
 }