fix(realtime): bound a client-requested flush to edits it has not written

A requested flush deliberately bypasses the cross-task dedup window, because a
deduped no-op acked as success would reintroduce the staleness the flush exists
to prevent. But room.edited is set on the first edit and never cleared, so every
repeat still performed a full projection: a Yjs-to-markdown conversion, a fresh
blob upload, and a delete of the previous key. A client emitting flush in a loop
could drive that unbounded.

Pairs a monotonic edit counter with the sequence the last successful persist
covered, so a flush with nothing new to write acks unchanged instead. The
sequence is captured before the projection and stored only on success, so an
edit arriving mid-write stays pending and a conflict is never mistaken for a
completed write.

Author: mzxchandra

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

@@ -461,6 +461,63 @@ describe('setupWorkspaceFileDocHandlers', () => {
       }
     })
 
+    it('does not re-project edits the durable file already has', async () => {
+      mockFetchFileDocSeed.mockResolvedValue(seedResult('# From server'))
+      mockFetchFileDocPersist.mockResolvedValue({ status: 'persisted', version: 9 })
+      const { io } = createIo()
+      const { handlers, socket } = setup('socket-1', io)
+      await joinAndEdit(handlers)
+
+      await handlers[FILE_DOC_EVENTS.FLUSH]({ fileId: 'file-1' })
+      await handlers[FILE_DOC_EVENTS.FLUSH]({ fileId: 'file-1' })
+      await handlers[FILE_DOC_EVENTS.FLUSH]({ fileId: 'file-1' })
+
+      // `edited` never clears, so without an edit-sequence check each repeat would mint another blob
+      // version. Only the first has anything to write; the rest are honest no-ops.
+      expect(mockFetchFileDocPersist).toHaveBeenCalledTimes(1)
+      expect(flushAcks(socket)).toEqual([
+        { fileId: 'file-1', status: 'persisted', version: 9 },
+        { fileId: 'file-1', status: 'unchanged' },
+        { fileId: 'file-1', status: 'unchanged' },
+      ])
+    })
+
+    it('writes again once a new edit lands after a flush', async () => {
+      mockFetchFileDocSeed.mockResolvedValue(seedResult('# From server'))
+      mockFetchFileDocPersist.mockResolvedValue({ status: 'persisted', version: 9 })
+      const { io } = createIo()
+      const { handlers } = setup('socket-1', io)
+      await joinAndEdit(handlers)
+      await handlers[FILE_DOC_EVENTS.FLUSH]({ fileId: 'file-1' })
+
+      const more = new Y.Doc()
+      more.getText(FILE_DOC_FIELD).insert(0, 'and more typing')
+      handlers[FILE_DOC_EVENTS.MESSAGE](
+        frame(FILE_DOC_MESSAGE_TYPE.SYNC, (e) =>
+          syncProtocol.writeUpdate(e, Y.encodeStateAsUpdate(more))
+        )
+      )
+      await flushMicrotasks()
+      await handlers[FILE_DOC_EVENTS.FLUSH]({ fileId: 'file-1' })
+
+      // The dedup must bound redundant writes without ever swallowing real edits.
+      expect(mockFetchFileDocPersist).toHaveBeenCalledTimes(2)
+    })
+
+    it('leaves the edits pending when a persist did not land', async () => {
+      mockFetchFileDocSeed.mockResolvedValue(seedResult('# From server'))
+      mockFetchFileDocPersist.mockResolvedValue({ status: 'conflict' })
+      const { io } = createIo()
+      const { handlers } = setup('socket-1', io)
+      await joinAndEdit(handlers)
+
+      await handlers[FILE_DOC_EVENTS.FLUSH]({ fileId: 'file-1' })
+      await handlers[FILE_DOC_EVENTS.FLUSH]({ fileId: 'file-1' })
+
+      // A conflict wrote nothing, so the second attempt must NOT be deduped away as already-durable.
+      expect(mockFetchFileDocPersist).toHaveBeenCalledTimes(2)
+    })
+
     it('refuses a flush for a file this socket never joined', async () => {
       mockFetchFileDocSeed.mockResolvedValue(seedResult('# From server'))
       const { io } = createIo()

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

@@ -148,6 +148,19 @@ interface FileDocRoom {
    * is last to leave — even one that only tailed the edits.
    */
   edited: boolean
+  /**
+   * Monotonic count of edits applied here. Paired with {@link FileDocRoom.persistedEditSeq} to answer
+   * "is there anything new to write?" — `edited` alone only answers "was this doc EVER edited", which
+   * stays true forever and would let a client force an unbounded run of redundant blob versions by
+   * repeatedly asking for a flush.
+   */
+  editSeq: number
+  /**
+   * The {@link FileDocRoom.editSeq} value the last successful persist projected. Equal to `editSeq`
+   * means the durable file already reflects every edit this room has seen. Captured BEFORE the
+   * projection and stored only on success, so an edit arriving mid-persist is never marked durable.
+   */
+  persistedEditSeq: number
   /** Whether this room has observed its doc become seeded — so a post-seed update counts as an edit but
    * the seed transition itself does not. See the `doc.on('update')` edit-tracking below. */
   seededObserved: boolean
@@ -308,6 +321,12 @@ async function flushPersist(
   // Never project a doc no user actually edited back over the file (see {@link FileDocRoom.edited}).
   // Nothing to write is not a failure — the durable content is already current.
   if (!room.edited || !room.workspaceId || !room.lastEditorUserId) return { status: 'unchanged' }
+  // Nor re-project edits the durable file already has. `edited` never clears, so without this a
+  // caller that can ask for a flush could force an unbounded run of identical blob versions — each
+  // one a fresh upload plus a delete of the old key. Captured here, before any await, so an edit
+  // landing mid-persist is compared against the value this projection actually covers.
+  const projectedEditSeq = room.editSeq
+  if (projectedEditSeq === room.persistedEditSeq) return { status: 'unchanged' }
   const store = getFileDocStore()
   const workspaceId = room.workspaceId
   const userId = room.lastEditorUserId
@@ -395,6 +414,9 @@ async function flushPersist(
     }
     if (result.status === 'persisted') {
       room.syncedVersion = Math.max(room.syncedVersion ?? 0, result.version)
+      // Only the edits this projection actually carried. An edit that arrived while the write was in
+      // flight keeps `editSeq` ahead, so the next flush still has something to do.
+      room.persistedEditSeq = Math.max(room.persistedEditSeq, projectedEditSeq)
       void store.setSyncedVersion(name, result.version)
       return { status: 'persisted', version: result.version }
     }
@@ -771,6 +793,8 @@ function getOrCreateRoom(io: Server, ref: RoomRef): FileDocRoom {
     workspaceId: null,
     lastEditorUserId: null,
     edited: false,
+    editSeq: 0,
+    persistedEditSeq: 0,
     seededObserved: false,
     persistTimer: null,
     persistDeadline: null,
@@ -837,8 +861,10 @@ function getOrCreateRoom(io: Server, ref: RoomRef): FileDocRoom {
       originSocketId(origin) ||
       origin === REDIS_SNAPSHOT_ORIGIN ||
       (seededBefore && origin === REDIS_ORIGIN)
-    )
+    ) {
       room.edited = true
+      room.editSeq++
+    }
     // Debounce a persist for LOCAL user edits only (peers debounce their own).
     if (originSocketId(origin)) schedulePersist(name, room)
   })

apps/sim/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/file-doc-room-context.test.tsx

@@ -0,0 +1,128 @@
+/**
+ * @vitest-environment jsdom
+ *
+ * The flush hand-off between the editor (which owns the realtime provider) and the file-detail
+ * header (which acts on it). The header renders the provider, so it owns the ref rather than
+ * reading it through context — these cover that the published flush is actually reachable from
+ * there, and stays reachable.
+ */
+import { act, type ReactNode, useRef } from 'react'
+import type { FlushFileDocResult } from '@sim/realtime-protocol/file-doc'
+import { createRoot, type Root } from 'react-dom/client'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import {
+  type FileDocFlush,
+  FileDocRoomProvider,
+  flushFileDocRef,
+  useReportFileDocFlush,
+} from '@/app/workspace/[workspaceId]/files/components/file-viewer/rich-markdown-editor/collaboration/file-doc-room-context'
+
+beforeEach(() => {
+  ;(globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true
+})
+
+const PERSISTED: FlushFileDocResult = { fileId: 'file-1', status: 'persisted', version: 1 }
+
+/**
+ * Mounts an owner that passes its own ref into the provider — the shape `files.tsx` uses — with a
+ * publisher underneath standing in for the editor. `read()` is what the header's retype handler does.
+ */
+function renderOwner(initialFlush: FileDocFlush | null) {
+  const container = document.createElement('div')
+  const root: Root = createRoot(container)
+  let read: (() => Promise<FlushFileDocResult>) | null = null
+
+  function Publisher({ flush }: { flush: FileDocFlush | null }) {
+    useReportFileDocFlush(flush)
+    return null
+  }
+
+  function Owner({ flush, children }: { flush: FileDocFlush | null; children?: ReactNode }) {
+    const flushRef = useRef<FileDocFlush | null>(null)
+    read = () => flushFileDocRef(flushRef)
+    return (
+      <FileDocRoomProvider flushRef={flushRef}>
+        <Publisher flush={flush} />
+        {children}
+      </FileDocRoomProvider>
+    )
+  }
+
+  const render = (flush: FileDocFlush | null) => {
+    act(() => {
+      root.render(<Owner flush={flush} />)
+    })
+  }
+  render(initialFlush)
+
+  return {
+    render,
+    read: () => {
+      if (!read) throw new Error('owner did not render')
+      return read()
+    },
+    unmount: () => act(() => root.unmount()),
+  }
+}
+
+describe('file-doc flush hand-off', () => {
+  it('reaches the ancestor-owned ref, not just descendants', async () => {
+    const flush = vi.fn(async () => PERSISTED)
+    const owner = renderOwner(flush)
+
+    await expect(owner.read()).resolves.toEqual(PERSISTED)
+    expect(flush).toHaveBeenCalledTimes(1)
+    owner.unmount()
+  })
+
+  it('resolves skipped when nothing collaborative is mounted', async () => {
+    const owner = renderOwner(null)
+
+    // The caller needs no null check: a non-collaborative file must cost nothing and never throw.
+    await expect(owner.read()).resolves.toMatchObject({ status: 'skipped' })
+    owner.unmount()
+  })
+
+  /**
+   * The regression that motivated the stable-publish design. A flush bound to the provider's
+   * identity republished on every socket churn, and a churn ending on `null` silently left the
+   * header with nothing to call — degrading a retype back to reading pre-edit bytes.
+   */
+  it('survives publisher churn that ends on a live flush', async () => {
+    const first = vi.fn(async () => PERSISTED)
+    const second = vi.fn(async () => PERSISTED)
+    const owner = renderOwner(first)
+
+    owner.render(null)
+    owner.render(second)
+
+    await expect(owner.read()).resolves.toEqual(PERSISTED)
+    expect(second).toHaveBeenCalledTimes(1)
+    expect(first).not.toHaveBeenCalled()
+    owner.unmount()
+  })
+
+  it('goes quiet once the publisher reports no flush', async () => {
+    const flush = vi.fn(async () => PERSISTED)
+    const owner = renderOwner(flush)
+
+    owner.render(null)
+
+    await expect(owner.read()).resolves.toMatchObject({ status: 'skipped' })
+    expect(flush).not.toHaveBeenCalled()
+    owner.unmount()
+  })
+
+  it('stops resolving a torn-down publisher', async () => {
+    const flush = vi.fn(async () => PERSISTED)
+    const owner = renderOwner(flush)
+    owner.unmount()
+
+    // Nothing to assert against the ref after unmount, but the published function must not be
+    // retained by a later mount — a fresh owner starts empty.
+    const next = renderOwner(null)
+    await expect(next.read()).resolves.toMatchObject({ status: 'skipped' })
+    expect(flush).not.toHaveBeenCalled()
+    next.unmount()
+  })
+})