feat(collab-doc): If-Match optimistic concurrency so persist never clobbers an out-of-band edit

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

@@ -26,7 +26,7 @@ function postToApp(path: string, payload: unknown, timeoutMs: number): Promise<R
 export async function fetchFileDocSeed(
   workspaceId: string,
   fileId: string
-): Promise<Uint8Array | null> {
+): Promise<{ update: Uint8Array; version: number } | null> {
   const response = await postToApp(
     '/api/internal/file-doc/seed',
     { workspaceId, fileId },
@@ -35,16 +35,16 @@ export async function fetchFileDocSeed(
   if (!response.ok) {
     throw new Error(`Seed fetch failed for file ${fileId}: ${response.status}`)
   }
-  const body = (await response.json()) as { update?: unknown }
+  const body = (await response.json()) as { update?: unknown; version?: unknown }
   const update = body?.update
-  // A well-formed response is `{ update: base64-string | null }`. Anything else is a contract
-  // violation, not a "genuinely empty file" — throw so the caller retries rather than silently
-  // treating a malformed body as empty and stranding the room unseeded.
+  // A well-formed response is `{ update: base64-string | null, version: number | null }`. Anything
+  // else is a contract violation, not a "genuinely empty file" — throw so the caller retries rather
+  // than silently treating a malformed body as empty and stranding the room unseeded.
   if (update === null) return null
-  if (typeof update !== 'string') {
+  if (typeof update !== 'string' || typeof body?.version !== 'number') {
     throw new Error(`Seed fetch for file ${fileId} returned a malformed body`)
   }
-  return new Uint8Array(Buffer.from(update, 'base64'))
+  return { update: new Uint8Array(Buffer.from(update, 'base64')), version: body.version }
 }
 
 /**
@@ -73,25 +73,55 @@ export async function fetchFileDocMerge(
   return new Uint8Array(Buffer.from(body.update, 'base64'))
 }
 
+/**
+ * Result of a persist attempt (mirrors the app's `persistFileDoc` contract):
+ * - `persisted` — written; `version` is the new durable version the relay records as synced.
+ * - `missing` — the file is gone.
+ * - `conflict` — the file changed out-of-band since `expectedVersion`; NOT written. `version` is the
+ *   current durable version the relay adopts as its new If-Match to re-persist the current live stream.
+ */
+export type PersistResult =
+  | { status: 'persisted'; version: number }
+  | { status: 'missing' }
+  | { status: 'conflict'; version: number }
+  | { status: 'deferred' }
+
 /**
  * Ask the app to project a live collaborative document back to durable markdown and write it to the
- * file (Yjs → markdown, through the exact editor engine). This is the server-authoritative durable
- * path — called debounced while the doc is edited and when the last collaborator leaves — that
- * replaces the editor's client autosave, so a server/copilot edit can't be clobbered by a stale
- * keystroke. THROWS on a transport failure so the caller can log/retry on the next debounce.
+ * file (Yjs → markdown, through the exact editor engine) — the server-authoritative durable path that
+ * replaces the editor's client autosave. `expectedVersion` (the durable version the live doc synced
+ * from) is the optimistic-concurrency guard: on a mismatch the app returns `conflict` (rather than
+ * clobbering) so the caller reconciles and retries. THROWS only on a transport/contract failure.
  */
 export async function fetchFileDocPersist(
   workspaceId: string,
   fileId: string,
   userId: string,
-  docState: Uint8Array
-): Promise<void> {
+  docState: Uint8Array,
+  expectedVersion?: number
+): Promise<PersistResult> {
   const response = await postToApp(
     '/api/internal/file-doc/persist',
-    { workspaceId, fileId, userId, docState: Buffer.from(docState).toString('base64') },
+    {
+      workspaceId,
+      fileId,
+      userId,
+      docState: Buffer.from(docState).toString('base64'),
+      ...(expectedVersion !== undefined ? { expectedVersion } : {}),
+    },
     FILE_DOC_TIMEOUTS.persistRequestMs
   )
   if (!response.ok) {
     throw new Error(`Persist failed for file ${fileId}: ${response.status}`)
   }
+  const body = (await response.json()) as PersistResult
+  if (
+    body?.status !== 'persisted' &&
+    body?.status !== 'missing' &&
+    body?.status !== 'conflict' &&
+    body?.status !== 'deferred'
+  ) {
+    throw new Error(`Persist for file ${fileId} returned a malformed body`)
+  }
+  return body
 }

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

@@ -61,6 +61,18 @@ const RELEASE_LOCK_SCRIPT =
 const SEED_IF_EMPTY_SCRIPT =
   "if redis.call('xlen', KEYS[1]) == 0 then redis.call('xadd', KEYS[1], '*', ARGV[1], ARGV[2]); return 1 else return 0 end"
 
+/**
+ * Monotonic set of the synced-version token: overwrite ONLY when the new value is greater than the
+ * stored one (or none is stored). The token is written fire-and-forget from multiple sites (seed stamp,
+ * merge, persist) and across tasks, so an out-of-order write must never REGRESS it to a version older
+ * than the live doc already incorporates — a regressed token causes spurious If-Match conflicts and, on a
+ * last-leave flush with no live room to reconcile into, a lost persist. Refreshes the TTL on both paths
+ * so a write skipped as older still keeps the (higher) value alive. Versions are monotonic epoch-ms,
+ * comfortably within a Lua double, so the numeric compare is exact.
+ */
+const SET_VERSION_IF_NEWER_SCRIPT =
+  "local c = redis.call('get', KEYS[1]); if c == false or tonumber(c) < tonumber(ARGV[1]) then redis.call('set', KEYS[1], ARGV[1], 'EX', ARGV[2]) else redis.call('expire', KEYS[1], ARGV[2]) end; return 1"
+
 /**
  * The transaction origin the store stamps on updates it applies from the stream. The relay's
  * `doc.on('update')` handler uses it to distinguish an update that ARRIVED from a peer (fan out to
@@ -79,6 +91,8 @@ export const REDIS_ORIGIN = Symbol('file-doc-redis')
 export const REDIS_SNAPSHOT_ORIGIN = Symbol('file-doc-redis-snapshot')
 
 const STREAM_PREFIX = 'filedoc:stream:'
+/** Cluster-wide "durable version the live doc is synced to" (the persist If-Match token). */
+const SYNC_VERSION_PREFIX = 'filedoc:syncver:'
 const SEED_LOCK_PREFIX = 'filedoc:seedlock:'
 const COMPACT_LOCK_PREFIX = 'filedoc:compactlock:'
 const PERSIST_LOCK_PREFIX = 'filedoc:persistlock:'
@@ -444,6 +458,56 @@ export class FileDocStore {
     return this.acquireLock(`${MERGE_LOCK_PREFIX}${name}`, ttlMs)
   }
 
+  /**
+   * The durable file version (its `updatedAt`, epoch ms) the shared live doc is synced to — the
+   * cluster-wide {@link https://www.rfc-editor.org/rfc/rfc7232 `If-Match`} token for persistence. Held
+   * in Redis (not per-task room state) so whichever task runs a debounced/last-leave persist reads the
+   * SAME version, even though the write that advanced it (a seed or a merged edit) may have run on
+   * another task. Returns `null` when unset/expired (persist then falls back to the local room's value).
+   */
+  async getSyncedVersion(name: string): Promise<number | null> {
+    if (!this.enabled || !this.write) return null
+    try {
+      const value = await this.write.get(`${SYNC_VERSION_PREFIX}${name}`)
+      const parsed = value === null ? Number.NaN : Number(value)
+      return Number.isFinite(parsed) ? parsed : null
+    } catch (error) {
+      logger.warn(`FileDocStore getSyncedVersion failed for ${name}`, {
+        error: getErrorMessage(error),
+      })
+      return null
+    }
+  }
+
+  /** Record the durable version the shared live doc is now synced to. MONOTONIC — writes only when the
+   * new value exceeds the stored one ({@link SET_VERSION_IF_NEWER_SCRIPT}), so an out-of-order
+   * fire-and-forget write can't regress the token. Best-effort; TTL-bounded like the stream so an idle
+   * file's key can't outlive its room. No-op when disabled (single-pod fallback). */
+  async setSyncedVersion(name: string, version: number): Promise<void> {
+    if (!this.enabled || !this.write) return
+    // Retry a transient failure (bounded) rather than swallow it: this token is the ONLY way a
+    // peer-seeded task learns the durable version, so a dropped write would leave that peer's persists
+    // deferring forever with the session's edits stranded in the TTL'd stream. The monotonic script makes
+    // a retry that races a newer value a no-op, never a regression.
+    for (let attempt = 0; attempt <= PUBLISH_MAX_RETRIES; attempt++) {
+      try {
+        await this.write.eval(SET_VERSION_IF_NEWER_SCRIPT, {
+          keys: [`${SYNC_VERSION_PREFIX}${name}`],
+          arguments: [String(version), String(STREAM_TTL_SEC)],
+        })
+        return
+      } catch (error) {
+        if (attempt === PUBLISH_MAX_RETRIES) {
+          logger.warn(`FileDocStore setSyncedVersion failed for ${name}`, {
+            error: getErrorMessage(error),
+          })
+          return
+        }
+        await sleep(backoffWithJitter(attempt + 1, null, { baseMs: 50, maxMs: 500 }))
+      }
+    }
+  }
+
   async releaseMergeSlot(name: string, token: string): Promise<void> {
     await this.releaseLock(`${MERGE_LOCK_PREFIX}${name}`, token)
   }
@@ -534,6 +598,9 @@ export class FileDocStore {
     if (!this.write) return
     for (const name of this.rooms.keys()) {
       await this.write.expire(streamKey(name), STREAM_TTL_SEC).catch(() => {})
+      // Keep the synced-version key alive as long as its stream, so an open-but-idle doc's persist
+      // If-Match token can't expire out from under it (which would force a needless reconcile).
+      await this.write.expire(`${SYNC_VERSION_PREFIX}${name}`, STREAM_TTL_SEC).catch(() => {})
     }
   }
 }

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

@@ -130,11 +130,11 @@ async function flushMicrotasks(): Promise<void> {
  * An encoded Yjs update shaped like the server seed builder's output: some content in the shared
  * `default` type plus the {@link FILE_DOC_SEED} flag, so applying it marks the doc seeded.
  */
-function encodedSeedUpdate(content: string): Uint8Array {
+function seedResult(content: string): { update: Uint8Array; version: number } {
   const doc = new Y.Doc()
   doc.getText(FILE_DOC_FIELD).insert(0, content)
   doc.getMap(FILE_DOC_SEED.configMap).set(FILE_DOC_SEED.flag, true)
-  return Y.encodeStateAsUpdate(doc)
+  return { update: Y.encodeStateAsUpdate(doc), version: 1 }
 }
 
 /** Apply a server sync reply frame (`[SYNC tag][sync message]`) into a fresh client doc. */
@@ -189,6 +189,8 @@ describe('setupWorkspaceFileDocHandlers', () => {
     // Default: the merge builder returns a valid no-op (empty-doc) update. Tests exercising copilot
     // merges override it.
     mockFetchFileDocMerge.mockResolvedValue(Y.encodeStateAsUpdate(new Y.Doc()))
+    // Default: persist succeeds. Tests asserting conflict/reconcile override this per-case.
+    mockFetchFileDocPersist.mockResolvedValue({ status: 'persisted', version: 1 })
   })
 
   afterEach(() => {
@@ -262,7 +264,7 @@ describe('setupWorkspaceFileDocHandlers', () => {
   })
 
   it('does NOT persist a seeded-but-unedited doc on last disconnect (no clobber of a concurrent write)', async () => {
-    mockFetchFileDocSeed.mockResolvedValue(encodedSeedUpdate('# From server'))
+    mockFetchFileDocSeed.mockResolvedValue(seedResult('# From server'))
     const { io } = createIo()
     const { handlers } = setup('socket-1', io)
     await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
@@ -276,7 +278,7 @@ describe('setupWorkspaceFileDocHandlers', () => {
   })
 
   it('persists on last disconnect once a genuine user edit has landed', async () => {
-    mockFetchFileDocSeed.mockResolvedValue(encodedSeedUpdate('# From server'))
+    mockFetchFileDocSeed.mockResolvedValue(seedResult('# From server'))
     const { io } = createIo()
     const { handlers } = setup('socket-1', io)
     await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
@@ -297,8 +299,39 @@ describe('setupWorkspaceFileDocHandlers', () => {
     expect(mockFetchFileDocPersist).toHaveBeenCalled()
   })
 
+  it('stops on a persist conflict without clobbering (single attempt, durable left authoritative)', async () => {
+    mockFetchFileDocSeed.mockResolvedValue(seedResult('# From server'))
+    // A persist reports an out-of-band change (If-Match conflict). The relay must NOT re-persist against
+    // the current stream (the external write commits durable before its chokepoint merge lands, so a
+    // re-persist could clobber it) — it stops after a single attempt and leaves the durable file
+    // authoritative; a later flush projects the converged stream once the merge lands.
+    mockFetchFileDocPersist.mockResolvedValue({
+      status: 'conflict',
+      version: 999,
+    })
+    const { io } = createIo()
+    const { handlers } = setup('socket-1', io)
+    await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
+    await flushMicrotasks()
+
+    const edit = new Y.Doc()
+    edit.getText(FILE_DOC_FIELD).insert(0, 'user typed this')
+    handlers[FILE_DOC_EVENTS.MESSAGE](
+      frame(FILE_DOC_MESSAGE_TYPE.SYNC, (e) =>
+        syncProtocol.writeUpdate(e, Y.encodeStateAsUpdate(edit))
+      )
+    )
+    await flushMicrotasks()
+
+    // The conflict is handled gracefully: the persist is attempted exactly once (never silently skipped,
+    // never retried against a possibly-behind stream) and the durable file is left authoritative.
+    cleanupFileDocForSocket('socket-1', io, true)
+    await flushMicrotasks()
+    expect(mockFetchFileDocPersist).toHaveBeenCalledTimes(1)
+  })
+
   it('flushAllFileDocRooms persists open EDITED rooms (graceful shutdown), skips unedited', async () => {
-    mockFetchFileDocSeed.mockResolvedValue(encodedSeedUpdate('# From server'))
+    mockFetchFileDocSeed.mockResolvedValue(seedResult('# From server'))
     const { io } = createIo()
     const { handlers } = setup('socket-1', io)
     await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
@@ -324,7 +357,7 @@ describe('setupWorkspaceFileDocHandlers', () => {
   })
 
   it('joins the room, sends sync step 1, and seeds the document from the server', async () => {
-    mockFetchFileDocSeed.mockResolvedValue(encodedSeedUpdate('# From server'))
+    mockFetchFileDocSeed.mockResolvedValue(seedResult('# From server'))
     const { io } = createIo()
     const { socket, handlers } = setup('socket-1', io)
 
@@ -360,7 +393,7 @@ describe('setupWorkspaceFileDocHandlers', () => {
   it('seeds the document only once from the server across concurrent joiners of the same file', async () => {
     // Keep the first seed fetch IN FLIGHT so the doc is still unseeded when the second socket joins:
     // that forces the dedup onto `serverSeedStarted` (the in-flight guard) rather than `isDocSeeded`.
-    let resolveSeed: (v: Uint8Array | null) => void = () => {}
+    let resolveSeed: (v: { update: Uint8Array; version: number } | null) => void = () => {}
     mockFetchFileDocSeed.mockReturnValueOnce(new Promise((resolve) => (resolveSeed = resolve)))
     const { io } = createIo()
     const a = setup('socket-a', io)
@@ -370,7 +403,7 @@ describe('setupWorkspaceFileDocHandlers', () => {
     await b.handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 2 })
     // Second join happened with the fetch still pending; only after this does the seed land.
     expect(mockFetchFileDocSeed).toHaveBeenCalledTimes(1)
-    resolveSeed(encodedSeedUpdate('# From server'))
+    resolveSeed(seedResult('# From server'))
     await flushMicrotasks()
     expect(mockFetchFileDocSeed).toHaveBeenCalledTimes(1)
   })
@@ -401,7 +434,7 @@ describe('setupWorkspaceFileDocHandlers', () => {
   it('makes one seed attempt and releases the guard on failure so a later join retries', async () => {
     mockFetchFileDocSeed
       .mockRejectedValueOnce(new Error('transport blip'))
-      .mockResolvedValueOnce(encodedSeedUpdate('# Recovered'))
+      .mockResolvedValueOnce(seedResult('# Recovered'))
     const { io } = createIo()
     const { socket, handlers } = setup('socket-1', io)
 
@@ -428,7 +461,7 @@ describe('setupWorkspaceFileDocHandlers', () => {
   })
 
   it('does not seed a room that was dropped while the seed fetch was in flight', async () => {
-    let resolveSeed: (v: Uint8Array | null) => void = () => {}
+    let resolveSeed: (v: { update: Uint8Array; version: number } | null) => void = () => {}
     mockFetchFileDocSeed.mockReturnValueOnce(new Promise((resolve) => (resolveSeed = resolve)))
     const { io } = createIo()
     const { handlers } = setup('socket-1', io)
@@ -437,7 +470,7 @@ describe('setupWorkspaceFileDocHandlers', () => {
     // The only owner leaves → the room (and its doc) is destroyed while the fetch is still pending.
     cleanupFileDocForSocket('socket-1', io, true)
     // Resolving now must not touch the destroyed doc or throw (liveness re-check after the await).
-    resolveSeed(encodedSeedUpdate('# Too late'))
+    resolveSeed(seedResult('# Too late'))
     await expect(flushMicrotasks()).resolves.toBeUndefined()
   })
 
@@ -447,7 +480,7 @@ describe('setupWorkspaceFileDocHandlers', () => {
     // edits are readiness-gated), but even if some update landed content in the doc before the seed
     // fetch resolved, the seed must still apply and set the flag — or the client's
     // `synced && initialContentLoaded` gate would never open.
-    let resolveSeed: (v: Uint8Array | null) => void = () => {}
+    let resolveSeed: (v: { update: Uint8Array; version: number } | null) => void = () => {}
     mockFetchFileDocSeed.mockReturnValueOnce(new Promise((resolve) => (resolveSeed = resolve)))
     const { io } = createIo()
     const { socket, handlers } = setup('socket-1', io)
@@ -461,7 +494,7 @@ describe('setupWorkspaceFileDocHandlers', () => {
         syncProtocol.writeUpdate(e, Y.encodeStateAsUpdate(placeholder))
       )
     )
-    resolveSeed(encodedSeedUpdate('# Seeded'))
+    resolveSeed(seedResult('# Seeded'))
     await flushMicrotasks()
 
     socket.emit.mockClear()
@@ -478,7 +511,7 @@ describe('setupWorkspaceFileDocHandlers', () => {
   })
 
   it('merges a copilot edit into a seeded live room and relays it to editors', async () => {
-    mockFetchFileDocSeed.mockResolvedValue(encodedSeedUpdate('# Original'))
+    mockFetchFileDocSeed.mockResolvedValue(seedResult('# Original'))
     const { io, sent } = createIo()
     const { handlers } = setup('socket-1', io)
     await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })
@@ -511,7 +544,7 @@ describe('setupWorkspaceFileDocHandlers', () => {
   })
 
   it('serializes concurrent merges for the same file (second waits for the first)', async () => {
-    mockFetchFileDocSeed.mockResolvedValue(encodedSeedUpdate('# Original'))
+    mockFetchFileDocSeed.mockResolvedValue(seedResult('# Original'))
     const { io } = createIo()
     const { handlers } = setup('socket-1', io)
     await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: 'file-1', clientId: 1 })