fix(files): require explicit owners in live documents

Author: mzxchandra

.github/workflows/test-build.yml

@@ -128,7 +128,7 @@ jobs:
   http-e2e:
     name: End-to-end over real HTTP
     runs-on: ${{ (vars.CI_PROVIDER == '' || vars.CI_PROVIDER == 'blacksmith') && 'blacksmith-8vcpu-ubuntu-2404' || 'ubuntu-latest' }}
-    timeout-minutes: 40
+    timeout-minutes: 20
     services:
       redis:
         image: redis:8.2-alpine

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

@@ -68,9 +68,9 @@ const OWNER_ADAPTERS: Record<FileDocOwner['entityType'], FileDocOwnerAdapter> =
   },
 }
 
-/** Only registered scopes can use callbacks; legacy workspace addresses resolve their owner at join. */
-export function fileDocOwnerAdapter(owner?: FileDocOwner): FileDocOwnerAdapter {
-  const type = owner?.entityType ?? 'workspace'
+/** Select callbacks only after the document's canonical owner has been resolved. */
+export function fileDocOwnerAdapter(owner: FileDocOwner): FileDocOwnerAdapter {
+  const type = owner.entityType
   if (!Object.hasOwn(OWNER_ADAPTERS, type)) throw new Error('Unsupported document owner')
   return OWNER_ADAPTERS[type]
 }

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

@@ -62,9 +62,11 @@ describe('applyMarkdownToLiveFileDoc — multi-replica (store-enabled) ordering'
 
   it('drops a stale durable write against the SHARED synced version', async () => {
     // A durable write (e.g. a concurrent human save on another process) records the shared synced version.
-    expect(await applyMarkdownToLiveFileDoc('file-1', '# durable', { version: 100 })).toBe(
-      'applied'
-    )
+    expect(
+      await applyMarkdownToLiveFileDoc({ type: 'workspace-file-doc', id: 'file-1' }, '# durable', {
+        version: 100,
+      })
+    ).toBe('applied')
     expect(fakeStore.setSyncedVersion).toHaveBeenCalledWith(ROOM_NAME, 100, 'shared-generation')
     expect(fakeStore.getStreamState).toHaveBeenCalledWith(ROOM_NAME, 'shared-generation')
     expect(fakeStore.publishAndWait).toHaveBeenCalledWith(
@@ -76,15 +78,23 @@ describe('applyMarkdownToLiveFileDoc — multi-replica (store-enabled) ordering'
 
     // A durable write with an OLDER version than the SHARED synced version is stale — rejected under the
     // lock before any diff is built, so it can't regress the doc across replicas.
-    expect(await applyMarkdownToLiveFileDoc('file-1', '# older durable', { version: 50 })).toBe(
-      'stale'
-    )
+    expect(
+      await applyMarkdownToLiveFileDoc(
+        { type: 'workspace-file-doc', id: 'file-1' },
+        '# older durable',
+        { version: 50 }
+      )
+    ).toBe('stale')
     expect(mockFetchFileDocMerge).not.toHaveBeenCalled()
 
     // A newer durable write applies and advances the shared synced version.
-    expect(await applyMarkdownToLiveFileDoc('file-1', '# durable again', { version: 150 })).toBe(
-      'applied'
-    )
+    expect(
+      await applyMarkdownToLiveFileDoc(
+        { type: 'workspace-file-doc', id: 'file-1' },
+        '# durable again',
+        { version: 150 }
+      )
+    ).toBe('applied')
     expect(fakeStore.setSyncedVersion).toHaveBeenCalledWith(ROOM_NAME, 150, 'shared-generation')
     // setSyncedVersion fired only for the two applied durable writes, never for the stale one.
     expect(fakeStore.setSyncedVersion).toHaveBeenCalledTimes(2)
@@ -98,17 +108,25 @@ describe('applyMarkdownToLiveFileDoc — multi-replica (store-enabled) ordering'
     fakeStore.isAgentStreaming.mockResolvedValue(true)
 
     expect(
-      await applyMarkdownToLiveFileDoc('file-1', '# streamed by a client', { version: 100 })
+      await applyMarkdownToLiveFileDoc(
+        { type: 'workspace-file-doc', id: 'file-1' },
+        '# streamed by a client',
+        { version: 100 }
+      )
     ).toBe('applied')
     expect(mockFetchFileDocMerge).not.toHaveBeenCalled() // content deferred to the client
     expect(fakeStore.publishAndWait).not.toHaveBeenCalled()
     expect(fakeStore.setSyncedVersion).toHaveBeenCalledWith(ROOM_NAME, 100, 'shared-generation') // version still recorded
 
     // Once streaming stops the flag clears and the (now near-noop) durable merge resumes normally.
     fakeStore.isAgentStreaming.mockResolvedValue(false)
-    expect(await applyMarkdownToLiveFileDoc('file-1', '# final durable', { version: 150 })).toBe(
-      'applied'
-    )
+    expect(
+      await applyMarkdownToLiveFileDoc(
+        { type: 'workspace-file-doc', id: 'file-1' },
+        '# final durable',
+        { version: 150 }
+      )
+    ).toBe('applied')
     expect(mockFetchFileDocMerge).toHaveBeenCalledTimes(1)
   })
 })

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

@@ -376,7 +376,8 @@ describe('setupWorkspaceFileDocHandlers', () => {
       source.getText(FILE_DOC_FIELD).insert(0, 'Stale text')
       const acknowledge = vi.fn()
       try {
-        if (timing === 'before append') await invalidateLiveFileDocument('file-1', 2)
+        if (timing === 'before append')
+          await invalidateLiveFileDocument({ type: 'workspace-file-doc', id: 'file-1' }, 2)
         sent.length = 0
         handlers[FILE_DOC_EVENTS.UPDATE](
           {
@@ -389,7 +390,7 @@ describe('setupWorkspaceFileDocHandlers', () => {
         )
         if (timing !== 'before append') {
           expect(publish).toHaveBeenCalledTimes(1)
-          await invalidateLiveFileDocument('file-1', 2)
+          await invalidateLiveFileDocument({ type: 'workspace-file-doc', id: 'file-1' }, 2)
           if (timing === 'during append and reseed') {
             mockFetchFileDocSeed.mockResolvedValue({
               ...seedResult('Replacement', 'doc-new'),
@@ -949,19 +950,31 @@ describe('setupWorkspaceFileDocHandlers', () => {
     mockFetchFileDocMerge.mockResolvedValue(Y.encodeStateAsUpdate(new Y.Doc()))
 
     // A newer durable version lands and is recorded as the synced version.
-    expect(await applyMarkdownToLiveFileDoc('file-1', '# newer', { version: 100 })).toBe('applied')
+    expect(
+      await applyMarkdownToLiveFileDoc({ type: 'workspace-file-doc', id: 'file-1' }, '# newer', {
+        version: 100,
+      })
+    ).toBe('applied')
     mockFetchFileDocMerge.mockClear()
 
     // An older durable version arriving out of order (e.g. a concurrent write on another process) is
     // stale: skipped before any diff is computed, so the live doc never regresses to older content and
     // no diff is published that a later persist could write back.
-    expect(await applyMarkdownToLiveFileDoc('file-1', '# older, stale', { version: 50 })).toBe(
-      'stale'
-    )
+    expect(
+      await applyMarkdownToLiveFileDoc(
+        { type: 'workspace-file-doc', id: 'file-1' },
+        '# older, stale',
+        { version: 50 }
+      )
+    ).toBe('stale')
     // The same version is idempotent — also skipped.
-    expect(await applyMarkdownToLiveFileDoc('file-1', '# same version', { version: 100 })).toBe(
-      'stale'
-    )
+    expect(
+      await applyMarkdownToLiveFileDoc(
+        { type: 'workspace-file-doc', id: 'file-1' },
+        '# same version',
+        { version: 100 }
+      )
+    ).toBe('stale')
     expect(mockFetchFileDocMerge).not.toHaveBeenCalled()
   })
 
@@ -979,8 +992,8 @@ describe('setupWorkspaceFileDocHandlers', () => {
       .mockReturnValueOnce(new Promise((resolve) => (resolveFirst = resolve)))
       .mockResolvedValueOnce(noOpUpdate)
 
-    const first = applyMarkdownToLiveFileDoc('file-1', '# One')
-    const second = applyMarkdownToLiveFileDoc('file-1', '# Two')
+    const first = applyMarkdownToLiveFileDoc({ type: 'workspace-file-doc', id: 'file-1' }, '# One')
+    const second = applyMarkdownToLiveFileDoc({ type: 'workspace-file-doc', id: 'file-1' }, '# Two')
     await flushMicrotasks(SEED_CHAIN_TICKS)
     expect(mockFetchFileDocMerge).toHaveBeenCalledTimes(1) // second is queued behind the first
 

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

@@ -144,7 +144,7 @@ interface FileDocPresenceOwner {
 interface FileDocRoom {
   /** The `workspace_files.id` this room edits. */
   fileId: string
-  owner?: FileDocOwner
+  owner: FileDocOwner
   lastEditorConnectionId: string | null
   doc: Y.Doc
   awareness: awarenessProtocol.Awareness
@@ -786,12 +786,14 @@ interface MergeOrder {
  * reconcile treats it distinctly (retry later) rather than as "nothing to reconcile into".
  */
 export function applyMarkdownToLiveFileDoc(
-  fileId: string,
+  ref: RoomRef,
   markdown: string,
-  order: MergeOrder = {},
-  owner?: FileDocOwner
+  order: MergeOrder = {}
 ): Promise<'applied' | 'no-live-room' | 'merge-unavailable' | 'stale'> {
-  const name = roomName(fileDocRoom({ fileId, owner }))
+  const address = fileDocTargetFromRoom(ref)
+  if (!address) throw new Error('Invalid file document room')
+  const { fileId } = address
+  const name = roomName(ref)
   return serializeFileDocMutation(name, () => mergeMarkdownIntoRoom(name, fileId, markdown, order))
 }
 
@@ -819,11 +821,11 @@ async function acquireFileDocMergeSlot(name: string): Promise<string | null> {
 
 /** Serializes and version-orders an unsupported durable replacement with live Markdown merges. */
 export function invalidateLiveFileDocument(
-  fileId: string,
-  version: number,
-  owner?: FileDocOwner
+  ref: RoomRef,
+  version: number
 ): Promise<{ status: 'applied'; docId?: string } | { status: 'stale' }> {
-  const name = roomName(fileDocRoom({ fileId, owner }))
+  if (!fileDocTargetFromRoom(ref)) throw new Error('Invalid file document room')
+  const name = roomName(ref)
   return serializeFileDocMutation(name, async () => {
     const store = getFileDocStore()
     const token = await acquireFileDocMergeSlot(name)
@@ -970,19 +972,13 @@ async function mergeMarkdownIntoRoom(
   return 'applied'
 }
 
-function requireFileDocTarget(ref: RoomRef): FileDocTarget {
-  const target = fileDocTargetFromRoom(ref)
-  if (!target) throw new Error('Invalid file document room')
-  return target
-}
-
 /**
  * Get (or lazily create) the authoritative document for a room, wiring the two
  * relay handlers exactly once: document updates and awareness changes are
  * broadcast to the room.
  */
-function getOrCreateRoom(io: Server, ref: RoomRef): FileDocRoom {
-  const name = roomName(ref)
+function getOrCreateRoom(io: Server, target: FileDocTarget): FileDocRoom {
+  const name = roomName(fileDocRoom(target))
   const existing = fileDocRooms.get(name)
   if (existing) return existing
 
@@ -994,7 +990,7 @@ function getOrCreateRoom(io: Server, ref: RoomRef): FileDocRoom {
   // Started BEFORE the room is registered so no join can observe a room without its hydration handle.
   const hydrated = getFileDocStore().attachRoom(name, doc)
   const room: FileDocRoom = {
-    ...requireFileDocTarget(ref),
+    ...target,
     lastEditorConnectionId: null,
     doc,
     awareness,
@@ -1129,10 +1125,10 @@ function isFileDocWriteAllowed(
   name: string
 ): boolean | Promise<boolean> {
   const userId = socket.userId
-  const fileId = fileDocRooms.get(name)?.fileId
-  if (!userId || !fileId) return false
+  const document = fileDocRooms.get(name)
+  if (!userId || !document) return false
 
-  const owner = fileDocRooms.get(name)?.owner
+  const { fileId, owner } = document
   const room = fileDocRoom({ fileId, owner })
   if (fileDocOwnerAdapter(owner).requiresCurrentActor)
     return (async () => {
@@ -1343,8 +1339,8 @@ async function handleClientUpdate(
   if (
     !room ||
     (target.owner &&
-      (target.owner.entityType !== room.owner?.entityType ||
-        target.owner.entityId !== room.owner?.entityId))
+      (target.owner.entityType !== room.owner.entityType ||
+        target.owner.entityId !== room.owner.entityId))
   ) {
     reject('NOT_JOINED', true, candidate.updateId)
     return
@@ -1485,7 +1481,6 @@ export function setupWorkspaceFileDocHandlers(
   socket.on(FILE_DOC_EVENTS.JOIN, async (payload: JoinFileDocPayload) => {
     const { fileId, clientId } = payload
     const target = parseFileDocTarget(payload)
-    const owner = target?.owner
     // Hoisted so the catch can tell whether this join was superseded (a switch to another file)
     // before surfacing a retryable error for the abandoned one.
     let generation: number | undefined
@@ -1547,9 +1542,9 @@ export function setupWorkspaceFileDocHandlers(
         generation = joinGeneration.get(socket.id) ?? 0
       }
 
-      const room = fileDocRoom({ fileId, owner })
+      const room = fileDocRoom(target)
       const name = roomName(room)
-      const admissionName = fileDocAdmissionRoom({ fileId, owner })
+      const admissionName = fileDocAdmissionRoom(target)
 
       const authorizeJoin = () =>
         resolveRoomJoinAuth({
@@ -1569,11 +1564,21 @@ export function setupWorkspaceFileDocHandlers(
         })
       const authorized = await authorizeJoin()
       if (!authorized) return
-      if (owner?.entityType === 'workspace' && owner.entityId !== authorized.workspaceId) {
+      if (
+        target.owner?.entityType === 'workspace' &&
+        target.owner.entityId !== authorized.workspaceId
+      ) {
         emitJoinError(socket, fileId, clientId, 'File not found', 'NOT_FOUND', false)
         return
       }
 
+      const owner =
+        target.owner ??
+        (authorized.workspaceId
+          ? { entityType: 'workspace' as const, entityId: authorized.workspaceId }
+          : null)
+      if (!owner) throw new Error('Document authorization did not resolve its owner')
+
       // Server-authenticated identity for the presence roster (never trusts the client-set
       // awareness). Resolved here so the generation guard below also covers this await.
       const avatarUrl = await resolveAvatarUrl(socket, userId)
@@ -1606,11 +1611,7 @@ export function setupWorkspaceFileDocHandlers(
         discardInvalidatedRoom(name, io)
       }
 
-      const entry = getOrCreateRoom(io, room)
-      // The workspace the server-side persist writes back to — and what the seed is built from, so it
-      // must be captured BEFORE the room is prepared below.
-      if (authorized.workspaceId)
-        entry.owner = { entityType: 'workspace', entityId: authorized.workspaceId }
+      const entry = getOrCreateRoom(io, { fileId, owner })
 
       // Hold the room open across the awaits below: it has no owner until this join commits, so a
       // concurrent last-leave would otherwise tear down the very document being prepared.
@@ -1795,12 +1796,12 @@ export function setupWorkspaceFileDocHandlers(
           docId: docIdOf(entry.doc),
           version: joinedVersion,
           schemaVersion: FILE_DOC_SCHEMA_VERSION,
+          ...fileDocOwnerWireFields(entry.owner),
           ...(store.enabled || fileDocOwnerAdapter(owner).requiresCurrentActor
             ? { acknowledgedUpdates: true as const }
             : {}),
           ...(fileDocOwnerAdapter(owner).requiresCurrentActor
             ? {
-                ...fileDocOwnerWireFields(entry.owner),
                 canWrite: finalPermission === 'write' || finalPermission === 'admin',
               }
             : {}),
@@ -1846,7 +1847,8 @@ export function setupWorkspaceFileDocHandlers(
     } catch (error) {
       logger.error('Error joining file-doc room:', error)
       try {
-        const name = roomName(fileDocRoom({ fileId, owner }))
+        const name = target ? roomName(fileDocRoom(target)) : null
+        if (!name) throw error
         /**
          * Roll back ownership only if this attempt committed it. A failed provisional admission must
          * preserve a previous file's binding and any co-mounted provider already in the target room.