fix(collab-doc): close durability gaps at deploy boundaries + audit polish

From a comprehensive from-scratch audit (correctness, SOTA, cleanliness,
feature-completeness):

- Persist max-wait: a continuous edit burst kept resetting the 5s debounce and
  never persisted; cap it so a burst flushes at least every 20s, bounding
  unpersisted edits.
- Graceful-shutdown flush: flushAllFileDocRooms awaited in shutdown so a rolling
  deploy / scale-in secures open edited rooms to durable markdown before exit,
  instead of relying on the stream + a surviving task.
- Compacted-snapshot catch-up now marks the doc edited (REDIS_SNAPSHOT_ORIGIN): a
  snapshot folds seed+edits into one frame, so a task catching up purely from it
  no longer treats real edits as an unedited seed and skips persisting.
- Polish: delete dead __setFileDocStoreForTest; bounded retry loop; rename
  acquirePersistSlot -> tryClaimPersistWindow with accurate docs; tailer
  object-identity guard; fix stale comments (seed route, edit-content autosave).

Author: waleedlatif1

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

@@ -220,6 +220,11 @@ describe('FileDocStore', () => {
     Y.applyUpdate(doc, (await a.getStreamState(NAME))!)
     expect(doc.getText('body').toString()).toBe('PEER1PEER2')
     doc.destroy()
+
+    // The appended snapshot entry must carry the snapshot marker, so a fresh catch-up task treats it as
+    // edited content (not a bare seed) and persists on last-disconnect.
+    const stream = state.backing!.streams.get(streamKey)!
+    expect(stream[stream.length - 1].message.s).toBe('1')
   })
 
   it('retries a transient append failure so the edit is not lost from the shared log', async () => {

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

@@ -56,14 +56,26 @@ const RELEASE_LOCK_SCRIPT =
  */
 export const REDIS_ORIGIN = Symbol('file-doc-redis')
 
+/**
+ * Origin for a COMPACTED SNAPSHOT applied from the stream. A snapshot folds the seed + all prior edits
+ * into one entry, so a fresh task catching up from it would otherwise never see a separate post-seed
+ * edit frame and would treat the doc as unedited. The relay's edit-tracker uses this origin to mark the
+ * doc edited (a snapshot only exists after the stream crossed the compaction threshold, i.e. real edits
+ * happened). Behaves like {@link REDIS_ORIGIN} otherwise (already in the stream — never re-published).
+ */
+export const REDIS_SNAPSHOT_ORIGIN = Symbol('file-doc-redis-snapshot')
+
 const STREAM_PREFIX = 'filedoc:stream:'
 const SEED_LOCK_PREFIX = 'filedoc:seedlock:'
 const COMPACT_LOCK_PREFIX = 'filedoc:compactlock:'
 const PERSIST_LOCK_PREFIX = 'filedoc:persistlock:'
 const MERGE_LOCK_PREFIX = 'filedoc:mergelock:'
 
-/** The single field each stream entry carries — a base64 Yjs update. */
+/** The field each stream entry carries — a base64 Yjs update. */
 const UPDATE_FIELD = 'u'
+/** Marks a stream entry as a compaction SNAPSHOT (folds seed + edits), so the tailer applies it with
+ * {@link REDIS_SNAPSHOT_ORIGIN}. Present only on snapshot entries. */
+const SNAPSHOT_FIELD = 's'
 
 /** Sentinel token a DISABLED store returns from a lock acquire, so single-replica callers proceed
  * without special-casing; {@link FileDocStore.releaseLock} treats it as a no-op. Not a real UUID, so it
@@ -228,12 +240,12 @@ export class FileDocStore {
   private async appendUpdate(name: string, update: Uint8Array): Promise<void> {
     if (!this.write) return
     const encoded = Buffer.from(update).toString('base64')
-    for (let attempt = 0; ; attempt++) {
+    for (let attempt = 0; attempt <= PUBLISH_MAX_RETRIES; attempt++) {
       try {
         await this.write.xAdd(streamKey(name), '*', { [UPDATE_FIELD]: encoded })
         break
       } catch (error) {
-        if (attempt >= PUBLISH_MAX_RETRIES) {
+        if (attempt === PUBLISH_MAX_RETRIES) {
           logger.error(`FileDocStore append failed for ${name}`, { error: getErrorMessage(error) })
           throw error
         }
@@ -352,12 +364,14 @@ export class FileDocStore {
   }
 
   /**
-   * Try to claim the right to persist this file for the current debounce window, so concurrent tasks
-   * editing the same file don't each write a redundant blob version. Returns true (proceed) when
-   * disabled, or when this task wins a short lock. The final last-collaborator flush does NOT gate on
-   * this — it must always write.
+   * A best-effort TTL dedup WINDOW (NOT a lock): claim the right to run a debounced persist for the next
+   * `ttlMs`, so concurrent tasks editing the same file don't each write a redundant blob version. It is
+   * never released — it simply expires after `ttlMs`, gating the debounced persist to ~once per window
+   * cluster-wide. Fails OPEN (returns true on a Redis error): a redundant persist is a harmless
+   * idempotent write, so it must never block a real one. The final last-collaborator flush does NOT gate
+   * on this — it must always write.
    */
-  async acquirePersistSlot(name: string, ttlMs: number): Promise<boolean> {
+  async tryClaimPersistWindow(name: string, ttlMs: number): Promise<boolean> {
     if (!this.enabled || !this.write) return true
     try {
       const won = await this.write.set(`${PERSIST_LOCK_PREFIX}${name}`, '1', {
@@ -388,7 +402,10 @@ export class FileDocStore {
 
   private applyEntry(room: StoreRoom, id: string, message: Record<string, string>): void {
     room.lastId = id
-    applyEntryToDoc(room.doc, id, message, REDIS_ORIGIN)
+    // A compaction snapshot folds seed + edits into one frame; stamp it so the relay's edit-tracker
+    // treats a fresh catch-up from it as edited (a snapshot only exists once real edits accumulated).
+    const origin = message[SNAPSHOT_FIELD] ? REDIS_SNAPSHOT_ORIGIN : REDIS_ORIGIN
+    applyEntryToDoc(room.doc, id, message, origin)
   }
 
   /**
@@ -397,21 +414,24 @@ export class FileDocStore {
    */
   private async runReader(): Promise<void> {
     while (this.running && this.read) {
-      const snapshot = [...this.rooms.entries()]
-      if (snapshot.length === 0) {
+      const snapshot = new Map(this.rooms)
+      if (snapshot.size === 0) {
         await sleep(IDLE_POLL_MS)
         continue
       }
       try {
         const res = await this.read.xRead(
-          snapshot.map(([name, room]) => ({ key: streamKey(name), id: room.lastId })),
+          [...snapshot].map(([name, room]) => ({ key: streamKey(name), id: room.lastId })),
           { BLOCK: READ_BLOCK_MS, COUNT: READ_COUNT }
         )
         if (!res) continue
         for (const stream of res) {
           const name = stream.name.slice(STREAM_PREFIX.length)
           const room = this.rooms.get(name)
-          if (!room) continue // detached mid-read; its doc is being destroyed
+          // Skip if detached mid-read, OR replaced by a close→reopen (a DIFFERENT StoreRoom): applying
+          // entries read against the OLD room's lastId to the new one could regress its lastId (harmless
+          // but wasteful re-delivery). The new room caught itself up via xRange already.
+          if (!room || room !== snapshot.get(name)) continue
           for (const entry of stream.messages) this.applyEntry(room, entry.id, entry.message)
         }
       } catch (error) {
@@ -446,7 +466,11 @@ export class FileDocStore {
         // appended snapshot id instead would silently drop those un-integrated peer entries.
         const upTo = room.lastId
         const snapshot = Buffer.from(Y.encodeStateAsUpdate(room.doc)).toString('base64')
-        await this.write.xAdd(streamKey(name), '*', { [UPDATE_FIELD]: snapshot })
+        // Mark it a snapshot so a fresh catch-up task treats it as edited content, not a bare seed.
+        await this.write.xAdd(streamKey(name), '*', {
+          [UPDATE_FIELD]: snapshot,
+          [SNAPSHOT_FIELD]: '1',
+        })
         // MINID keeps entries with id >= upTo: the snapshot, any un-integrated peer entries, and
         // `upTo` itself (redundant with the snapshot, harmless); it drops only the folded older deltas.
         await this.write.xTrim(streamKey(name), 'MINID', upTo)
@@ -486,8 +510,3 @@ export function getFileDocStore(): FileDocStore {
   if (!store) store = new FileDocStore(undefined)
   return store
 }
-
-/** Test-only: reset the singleton so a test can install its own instance. */
-export function __setFileDocStoreForTest(next: FileDocStore | null): void {
-  store = next
-}

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

@@ -35,6 +35,7 @@ vi.mock('@/handlers/file-doc-app', () => ({
 import {
   applyMarkdownToLiveFileDoc,
   cleanupFileDocForSocket,
+  flushAllFileDocRooms,
   setupWorkspaceFileDocHandlers,
 } from '@/handlers/file-doc'
 
@@ -296,6 +297,32 @@ describe('setupWorkspaceFileDocHandlers', () => {
     expect(mockFetchFileDocPersist).toHaveBeenCalled()
   })
 
+  it('flushAllFileDocRooms persists open EDITED rooms (graceful shutdown), skips unedited', async () => {
+    mockFetchFileDocSeed.mockResolvedValue(encodedSeedUpdate('# From server'))
+    const { io } = createIo()
+    const { handlers } = setup('socket-1', io)
+    await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
+    await flushMicrotasks()
+
+    // Seed-only room: a graceful-shutdown flush must NOT persist it.
+    mockFetchFileDocPersist.mockClear()
+    await flushAllFileDocRooms()
+    expect(mockFetchFileDocPersist).not.toHaveBeenCalled()
+
+    // After a real user edit, the same flush persists (edits would otherwise be lost on deploy).
+    const edit = new Y.Doc()
+    edit.getText(FILE_DOC_FIELD).insert(0, 'typed')
+    handlers[FILE_DOC_EVENTS.MESSAGE](
+      frame(FILE_DOC_MESSAGE_TYPE.SYNC, (e) =>
+        syncProtocol.writeUpdate(e, Y.encodeStateAsUpdate(edit))
+      )
+    )
+    await flushMicrotasks()
+    mockFetchFileDocPersist.mockClear()
+    await flushAllFileDocRooms()
+    expect(mockFetchFileDocPersist).toHaveBeenCalled()
+  })
+
   it('joins the room, sends sync step 1, and seeds the document from the server', async () => {
     mockFetchFileDocSeed.mockResolvedValue(encodedSeedUpdate('# From server'))
     const { io } = createIo()

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

@@ -45,7 +45,7 @@ import * as syncProtocol from 'y-protocols/sync'
 import * as Y from 'yjs'
 import { resolveAvatarUrl } from '@/handlers/avatar'
 import { fetchFileDocMerge, fetchFileDocPersist, fetchFileDocSeed } from '@/handlers/file-doc-app'
-import { getFileDocStore, REDIS_ORIGIN } from '@/handlers/file-doc-store'
+import { getFileDocStore, REDIS_ORIGIN, REDIS_SNAPSHOT_ORIGIN } from '@/handlers/file-doc-store'
 import { resolveRoomJoinAuth } from '@/handlers/room-join-auth'
 import type { AuthenticatedSocket } from '@/middleware/auth'
 import type { IRoomManager } from '@/rooms'
@@ -61,6 +61,10 @@ const SEED_ORIGIN = Symbol('file-doc-seed')
 
 /** Debounce window for the server-side project-to-markdown persist while a doc is actively edited. */
 const PERSIST_DEBOUNCE_MS = 5_000
+/** Max-wait cap on the persist debounce: a CONTINUOUS edit burst keeps resetting the 5s debounce and
+ * would otherwise never persist until an idle pause, so force a flush at least this often — bounding how
+ * many edits are unpersisted (in the stream only) if the task dies mid-burst. */
+const PERSIST_MAX_WAIT_MS = 20_000
 
 /** Cross-task merge lock. The TTL must exceed the whole critical section it guards — stream fold +
  * `fetchFileDocMerge` (bounded at `mergeRequestMs`) + the awaited publish — so the lock never expires
@@ -118,6 +122,9 @@ interface FileDocRoom {
   seededObserved: boolean
   /** The pending debounced persist timer, if any. */
   persistTimer: ReturnType<typeof setTimeout> | null
+  /** Absolute time (ms) by which a debounced persist must fire even under continuous editing (the
+   * max-wait cap); null when no persist is pending. */
+  persistDeadline: number | null
 }
 
 /** Live documents keyed by Socket.IO room name. Module-global: one Y.Doc per file. */
@@ -184,24 +191,29 @@ function broadcastLocal(
 
 /**
  * Schedule a debounced server-side persist of the live doc back to durable markdown. Coalesces rapid
- * edits; a no-op until the room knows its workspace (set at join). The final flush on last-disconnect
- * is separate ({@link flushPersist} with `final`).
+ * edits; a no-op until the room knows its workspace (set at join). A {@link PERSIST_MAX_WAIT_MS}
+ * max-wait caps the debounce so a continuous burst still persists periodically. The final flush on
+ * last-disconnect is separate ({@link flushPersist} with `final`).
  */
 function schedulePersist(name: string, room: FileDocRoom): void {
   if (!room.workspaceId || !room.lastEditorUserId) return
+  const now = Date.now()
+  if (room.persistDeadline === null) room.persistDeadline = now + PERSIST_MAX_WAIT_MS
   if (room.persistTimer) clearTimeout(room.persistTimer)
+  const delay = Math.max(0, Math.min(PERSIST_DEBOUNCE_MS, room.persistDeadline - now))
   room.persistTimer = setTimeout(() => {
     room.persistTimer = null
+    room.persistDeadline = null
     void flushPersist(name, room, false)
-  }, PERSIST_DEBOUNCE_MS)
+  }, delay)
 }
 
 /**
  * Project the live doc to markdown and write it durably via the app. `final` (last collaborator
- * leaving) always writes; a debounced mid-edit flush first claims a cross-task slot (held for the whole
- * write, so a concurrent task can't issue an overlapping blob write) so tasks don't each write a
- * redundant version. Best-effort: never throws (a failure is retried on the next debounce; the stream
- * holds the state meanwhile).
+ * leaving) always writes; a debounced mid-edit flush first claims a best-effort cross-task dedup WINDOW
+ * (a TTL key that just expires, so at most ~one persist per window cluster-wide) so concurrent tasks
+ * editing the same file don't each write a redundant blob version. Best-effort: never throws (a failure
+ * is retried on the next debounce; the stream holds the state meanwhile).
  *
  * Persists the AUTHORITATIVE shared state (the stream), not this task's local doc: a copilot merge — or
  * a peer's edit — published by another task may not be integrated into `room.doc` yet (and the stream
@@ -218,7 +230,7 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr
   // this yields. Only meaningful once seeded; used only when the authoritative stream state is absent.
   const localState = isDocSeeded(room.doc) ? Y.encodeStateAsUpdate(room.doc) : null
   try {
-    if (!final && !(await store.acquirePersistSlot(name, FILE_DOC_TIMEOUTS.persistRequestMs)))
+    if (!final && !(await store.tryClaimPersistWindow(name, FILE_DOC_TIMEOUTS.persistRequestMs)))
       return
     let docState = localState
     if (store.enabled) {
@@ -294,6 +306,7 @@ function awarenessUpdateClientIds(update: Uint8Array): number[] {
 function destroyRoomIfIdle(name: string) {
   const room = fileDocRooms.get(name)
   if (!room || room.owners.size > 0) return
+  room.persistDeadline = null
   if (room.persistTimer) {
     clearTimeout(room.persistTimer)
     room.persistTimer = null
@@ -307,6 +320,21 @@ function destroyRoomIfIdle(name: string) {
   fileDocRooms.delete(name)
 }
 
+/**
+ * Flush every open, edited room's converged doc to durable markdown, AWAITING the writes. Called on
+ * graceful shutdown (rolling deploy / scale-in) so edits since the last debounce aren't left only in the
+ * ephemeral stream — the per-socket disconnect flush is fire-and-forget and would race `process.exit`.
+ * Best-effort and bounded by each persist's own timeout; never throws. Rooms are NOT torn down here (the
+ * process is exiting); only their durable state is secured.
+ */
+export async function flushAllFileDocRooms(): Promise<void> {
+  const flushes: Promise<void>[] = []
+  for (const [name, room] of fileDocRooms) {
+    if (room.edited) flushes.push(flushPersist(name, room, true))
+  }
+  await Promise.all(flushes)
+}
+
 /**
  * Seed a room's document server-side, once, on the first join: ask the app to build the seed (the
  * file's current markdown → Yjs, through the exact editor engine) and apply it, which relays the
@@ -498,6 +526,7 @@ function getOrCreateRoom(io: Server, ref: RoomRef): FileDocRoom {
     edited: false,
     seededObserved: false,
     persistTimer: null,
+    persistDeadline: null,
   }
   // Register synchronously BEFORE the async catch-up so a concurrent join sees this room, not a second.
   fileDocRooms.set(name, room)
@@ -509,18 +538,27 @@ function getOrCreateRoom(io: Server, ref: RoomRef): FileDocRoom {
     // Fan out to THIS task's clients only (excluding the origin socket if local). Cross-task delivery
     // rides the shared stream — every task's tailer applies + runs its own local fan-out.
     broadcastLocal(io, name, encoding.toUint8Array(encoder), originSocketId(origin))
-    // Share every locally-originated update to the stream so peers converge. Skip REDIS_ORIGIN (already
-    // in the stream) and SEED_ORIGIN — the seed is published EXPLICITLY and AWAITED under the seed lock
-    // (so it lands before the lock releases), which a fire-and-forget publish here couldn't guarantee.
-    if (origin !== REDIS_ORIGIN && origin !== SEED_ORIGIN) getFileDocStore().publish(name, update)
+    // Share every locally-originated update to the stream so peers converge. Skip updates that already
+    // came FROM the stream (REDIS_ORIGIN / REDIS_SNAPSHOT_ORIGIN) and SEED_ORIGIN — the seed is published
+    // EXPLICITLY and AWAITED under the seed lock (so it lands before the lock releases), which a
+    // fire-and-forget publish here couldn't guarantee.
+    if (origin !== REDIS_ORIGIN && origin !== REDIS_SNAPSHOT_ORIGIN && origin !== SEED_ORIGIN)
+      getFileDocStore().publish(name, update)
     // Edit tracking for persistence. Mark the doc dirty on any update applied AFTER it was seeded — a
     // local user edit (socket origin) OR a peer's edit relayed via the tailer (REDIS_ORIGIN) — so
-    // whichever task is last to leave persists real edits, even one that only tailed them. The seed
-    // transition itself is never counted (nor a purely-local copilot merge, already durable via
-    // copilot's direct file write), so a seeded-but-unedited doc is never projected back over the file.
+    // whichever task is last to leave persists real edits, even one that only tailed them. A compaction
+    // snapshot on catch-up (REDIS_SNAPSHOT_ORIGIN) also counts: it folds real edits into one frame, so a
+    // fresh task catching up purely from it must not treat the doc as unedited. The seed transition
+    // itself is never counted (nor a purely-local copilot merge, already durable via copilot's direct
+    // file write), so a seeded-but-unedited doc is never projected back over the file.
     const seededBefore = room.seededObserved
     if (isDocSeeded(room.doc)) room.seededObserved = true
-    if (originSocketId(origin) || (seededBefore && origin === REDIS_ORIGIN)) room.edited = true
+    if (
+      originSocketId(origin) ||
+      origin === REDIS_SNAPSHOT_ORIGIN ||
+      (seededBefore && origin === REDIS_ORIGIN)
+    )
+      room.edited = true
     // Debounce a persist for LOCAL user edits only (peers debounce their own).
     if (originSocketId(origin)) schedulePersist(name, room)
   })

apps/realtime/src/index.ts

@@ -6,6 +6,7 @@ import { createSocketIOServer, shutdownSocketIOAdapter } from '@/config/socket'
 import { assertSchemaCompatibility } from '@/database/preflight'
 import { env } from '@/env'
 import { setupAllHandlers } from '@/handlers'
+import { flushAllFileDocRooms } from '@/handlers/file-doc'
 import { getFileDocStore, initFileDocStore } from '@/handlers/file-doc-store'
 import { type AuthenticatedSocket, authenticateSocket } from '@/middleware/auth'
 import { type IRoomManager, MemoryRoomManager, RedisRoomManager } from '@/rooms'
@@ -116,6 +117,15 @@ async function main() {
 
     accessRevalidation.stop()
 
+    // Flush open collaborative docs to durable markdown BEFORE tearing down Redis/the store — the
+    // per-socket disconnect flush is fire-and-forget and would race process exit.
+    try {
+      await flushAllFileDocRooms()
+      logger.info('Flushed open collaborative documents')
+    } catch (error) {
+      logger.error('Error flushing collaborative documents on shutdown:', error)
+    }
+
     try {
       await roomManager.shutdown()
       logger.info('RoomManager shutdown complete')

apps/sim/app/api/internal/file-doc/seed/route.ts

@@ -14,7 +14,7 @@ const logger = createLogger('FileDocSeedAPI')
  * POST /api/internal/file-doc/seed — build a server-authoritative collaborative-document seed
  * (markdown → Yjs) for the realtime relay to apply on room creation. Internal only: gated on the
  * shared `x-api-key: INTERNAL_API_SECRET` secret, matching the header the realtime relay sends
- * (`apps/realtime/src/handlers/file-doc-seed.ts`) and the realtime server's own inbound validator.
+ * (`apps/realtime/src/handlers/file-doc-app.ts`) and the realtime server's own inbound validator.
  */
 export const POST = withRouteHandler(async (request: NextRequest) => {
   const auth = checkInternalApiKey(request)