refactor(collab-doc): tidy merge-order docs + relay order object; cover multi-replica streaming stale-check

Author: waleedlatif1

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

@@ -0,0 +1,91 @@
+/**
+ * @vitest-environment node
+ *
+ * Multi-replica (store-enabled) coverage for the copilot live-merge stale-check. The main
+ * `file-doc.test.ts` runs with the store DISABLED (single-replica fallback); this file mocks an ENABLED
+ * store so the cross-process branch of `mergeMarkdownIntoRoom` — staleness against the SHARED synced
+ * version under the merge lock, and `recordVersion` writing `setSyncedVersion` — is exercised directly.
+ * The enabled merge path reads its base from the shared store (not an in-memory room), so no JOIN/seed
+ * is needed: calling `applyMarkdownToLiveFileDoc` against the fake store drives the branch on its own.
+ */
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import * as Y from 'yjs'
+
+const { mockFetchFileDocMerge } = vi.hoisted(() => ({
+  mockFetchFileDocMerge: vi.fn(),
+}))
+
+/**
+ * A minimal ENABLED store: in-memory monotonic synced version (mirrors SET_VERSION_IF_NEWER_SCRIPT), a
+ * non-null stream state so the merge has a base, and no-op locks/publish. Only the surface the
+ * store-enabled merge path touches is implemented.
+ */
+const fakeStore = {
+  enabled: true,
+  versions: new Map<string, number>(),
+  acquireMergeSlot: vi.fn(async () => 'token'),
+  releaseMergeSlot: vi.fn(async () => {}),
+  getStreamState: vi.fn(async () => new Uint8Array([1])),
+  publishAndWait: vi.fn(async () => {}),
+  getSyncedVersion: vi.fn(async (name: string) => fakeStore.versions.get(name) ?? null),
+  setSyncedVersion: vi.fn(async (name: string, version: number) => {
+    fakeStore.versions.set(name, Math.max(fakeStore.versions.get(name) ?? 0, version))
+  }),
+}
+
+vi.mock('@sim/platform-authz/rooms', () => ({ authorizeRoom: vi.fn() }))
+
+vi.mock('@/handlers/file-doc-app', () => ({
+  fetchFileDocSeed: vi.fn(),
+  fetchFileDocMerge: mockFetchFileDocMerge,
+  fetchFileDocPersist: vi.fn(),
+}))
+
+vi.mock('@/handlers/file-doc-store', () => ({
+  getFileDocStore: () => fakeStore,
+  REDIS_ORIGIN: Symbol('redis'),
+  REDIS_SNAPSHOT_ORIGIN: Symbol('redis-snapshot'),
+}))
+
+import { applyMarkdownToLiveFileDoc } from '@/handlers/file-doc'
+
+const ROOM_NAME = 'workspace-file-doc:file-1'
+
+describe('applyMarkdownToLiveFileDoc — multi-replica (store-enabled) ordering', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    fakeStore.versions.clear()
+    fakeStore.acquireMergeSlot.mockResolvedValue('token')
+    fakeStore.getStreamState.mockResolvedValue(new Uint8Array([1]))
+    mockFetchFileDocMerge.mockResolvedValue(Y.encodeStateAsUpdate(new Y.Doc()))
+  })
+
+  it('stale-checks against the SHARED synced version and never records a streaming merge', async () => {
+    // A durable write records the shared synced version cluster-wide.
+    expect(await applyMarkdownToLiveFileDoc('file-1', '# durable', { version: 100 })).toBe(
+      'applied'
+    )
+    expect(fakeStore.setSyncedVersion).toHaveBeenCalledWith(ROOM_NAME, 100)
+    mockFetchFileDocMerge.mockClear()
+
+    // A delayed streaming snapshot older than the SHARED version (e.g. from another process) is stale —
+    // rejected under the lock before any diff is built, so the live doc never regresses.
+    expect(await applyMarkdownToLiveFileDoc('file-1', '# older stream', { streamedAt: 50 })).toBe(
+      'stale'
+    )
+    expect(mockFetchFileDocMerge).not.toHaveBeenCalled()
+
+    // A streaming snapshot newer than the shared version applies (advances the live view)...
+    expect(await applyMarkdownToLiveFileDoc('file-1', '# newer stream', { streamedAt: 200 })).toBe(
+      'applied'
+    )
+    // ...but it is never recorded: a durable write between the two (150) still applies. Had 200 been
+    // recorded to the shared store, 150 would be rejected as stale.
+    expect(await applyMarkdownToLiveFileDoc('file-1', '# durable again', { version: 150 })).toBe(
+      'applied'
+    )
+    expect(fakeStore.setSyncedVersion).toHaveBeenCalledWith(ROOM_NAME, 150)
+    // setSyncedVersion fired only for the two durable writes, never for a streaming snapshot.
+    expect(fakeStore.setSyncedVersion).toHaveBeenCalledTimes(2)
+  })
+})

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

@@ -553,15 +553,19 @@ 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', 100)).toBe('applied')
+    expect(await applyMarkdownToLiveFileDoc('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', 50)).toBe('stale')
+    expect(await applyMarkdownToLiveFileDoc('file-1', '# older, stale', { version: 50 })).toBe(
+      'stale'
+    )
     // The same version is idempotent — also skipped.
-    expect(await applyMarkdownToLiveFileDoc('file-1', '# same version', 100)).toBe('stale')
+    expect(await applyMarkdownToLiveFileDoc('file-1', '# same version', { version: 100 })).toBe(
+      'stale'
+    )
     expect(mockFetchFileDocMerge).not.toHaveBeenCalled()
   })
 
@@ -575,22 +579,25 @@ 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', '# durable', 100)).toBe('applied')
+    expect(await applyMarkdownToLiveFileDoc('file-1', '# durable', { version: 100 })).toBe(
+      'applied'
+    )
 
     // A delayed streaming snapshot OLDER than that durable version — e.g. a throttled copilot snapshot
     // from another process arriving late — is stale, so it can't regress the doc back toward its content.
-    // (`version` omitted, `streamedAt` passed as the 4th arg.)
-    expect(await applyMarkdownToLiveFileDoc('file-1', '# older stream', undefined, 50)).toBe(
+    expect(await applyMarkdownToLiveFileDoc('file-1', '# older stream', { streamedAt: 50 })).toBe(
       'stale'
     )
 
     // A streaming snapshot NEWER than the durable version applies (it advances the live view)...
-    expect(await applyMarkdownToLiveFileDoc('file-1', '# newer stream', undefined, 200)).toBe(
+    expect(await applyMarkdownToLiveFileDoc('file-1', '# newer stream', { streamedAt: 200 })).toBe(
       'applied'
     )
     // ...but it is never RECORDED as the synced version: a durable write between the two (150) still
     // applies. Had the streaming 200 been recorded, 150 would have been rejected as stale.
-    expect(await applyMarkdownToLiveFileDoc('file-1', '# durable again', 150)).toBe('applied')
+    expect(await applyMarkdownToLiveFileDoc('file-1', '# durable again', { version: 150 })).toBe(
+      'applied'
+    )
   })
 
   it('serializes concurrent merges for the same file (second waits for the first)', async () => {

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

@@ -515,6 +515,15 @@ function emptySeedUpdate(): Uint8Array {
 /** Serializes live merges per file so overlapping calls never race the same doc (see below). */
 const fileDocMergeChains = new Map<string, Promise<unknown>>()
 
+/**
+ * How a merge is positioned on the file's version line — mirrors the sim-side `LiveFileDocMergeOrder`
+ * wire fields. A durable `version` is checked AND recorded; a streaming `streamedAt` is checked only.
+ */
+interface MergeOrder {
+  version?: number
+  streamedAt?: number
+}
+
 /**
  * Apply new markdown into a file's LIVE collaborative document (Stage C — copilot writing into an open
  * doc). Ships the document's current state to the app to build a minimal Yjs diff, applies it — which
@@ -540,15 +549,12 @@ const fileDocMergeChains = new Map<string, Promise<unknown>>()
 export function applyMarkdownToLiveFileDoc(
   fileId: string,
   markdown: string,
-  version?: number,
-  streamedAt?: number
+  order: MergeOrder = {}
 ): Promise<'applied' | 'no-live-room' | 'merge-unavailable' | 'stale'> {
   const name = roomName(fileDocRoom(fileId))
   const prior = fileDocMergeChains.get(name) ?? Promise.resolve()
   // `.catch` so a failed prior merge doesn't reject this one — each merge is independent.
-  const run = prior
-    .catch(() => {})
-    .then(() => mergeMarkdownIntoRoom(name, fileId, markdown, version, streamedAt))
+  const run = prior.catch(() => {}).then(() => mergeMarkdownIntoRoom(name, fileId, markdown, order))
   fileDocMergeChains.set(
     name,
     run.finally(() => {
@@ -562,8 +568,7 @@ async function mergeMarkdownIntoRoom(
   name: string,
   fileId: string,
   markdown: string,
-  version?: number,
-  streamedAt?: number
+  { version, streamedAt }: MergeOrder
 ): Promise<'applied' | 'no-live-room' | 'merge-unavailable' | 'stale'> {
   const store = getFileDocStore()
 
@@ -590,6 +595,11 @@ async function mergeMarkdownIntoRoom(
   // monotonic token stays high, so a later persist could write that stale content back over the durable
   // file. Skip it. This is what stops a delayed streaming snapshot from clobbering a newer durable merge
   // across processes (the per-process caller chain cannot). A merge with neither key is never stale.
+  //
+  // Only durable `version` (DB-monotonic `contentUpdatedAt`) is ever recorded, so the durable line is
+  // immune to clock skew. `streamedAt` is best-effort wall-clock used ONLY to order transient streaming
+  // snapshots; correctness never depends on it — a skewed clock at worst drops or briefly regresses the
+  // live preview, which the next durable write reconciles.
   const orderingVersion = version ?? streamedAt
   const isStale = (current: number): boolean =>
     orderingVersion !== undefined && orderingVersion <= current

apps/realtime/src/routes/http.ts

@@ -216,12 +216,10 @@ export function createHttpHandler(roomManager: IRoomManager, logger: Logger) {
         // `version` (the durable updatedAt this markdown was written with) records that the live doc now
         // incorporates that durable version, so the persist If-Match guard won't flag it as a conflict.
         // `streamedAt` orders a streaming snapshot on the same version line without recording it.
-        const result = await applyMarkdownToLiveFileDoc(
-          fileId,
-          markdown,
-          typeof version === 'number' ? version : undefined,
-          typeof streamedAt === 'number' ? streamedAt : undefined
-        )
+        const result = await applyMarkdownToLiveFileDoc(fileId, markdown, {
+          version: typeof version === 'number' ? version : undefined,
+          streamedAt: typeof streamedAt === 'number' ? streamedAt : undefined,
+        })
         res.writeHead(200, { 'Content-Type': 'application/json' })
         res.end(JSON.stringify({ applied: result === 'applied' }))
       } catch (error) {

apps/sim/lib/realtime/notify.ts

@@ -109,6 +109,17 @@ export async function notifyFolderResourceChanged(
   await FOLDER_RESOURCE_NOTIFIERS[resourceType]?.(workspaceId)
 }
 
+/**
+ * How a live-doc merge is positioned on the file's monotonic version line. Pass one key or the other,
+ * never both; passing neither applies the merge without ordering it (legacy).
+ */
+export interface LiveFileDocMergeOrder {
+  /** A durable write's `contentUpdatedAt` (epoch ms): orders the merge AND is recorded as the synced version. */
+  version?: number
+  /** A streaming snapshot's production time (epoch ms): orders the merge only — never recorded as a checkpoint. */
+  streamedAt?: number
+}
+
 /**
  * Best-effort: ask the realtime relay to merge a copilot edit into a file's LIVE collaborative
  * document, so open editors see it stream in as a CRDT merge (Stage C) rather than the file changing
@@ -125,33 +136,19 @@ export async function notifyFolderResourceChanged(
  * streaming caller fires and forgets it. Bounded to {@link APPLY_EDIT_TIMEOUT_MS}, so it adds latency
  * only when the socket pod is unreachable.
  *
- * `order` positions this merge on the file's monotonic version line so a stale write never regresses
- * the doc — the relay drops any merge not NEWER than the version the doc already incorporates:
- * - `version` — a DURABLE write's `contentUpdatedAt` (epoch ms). Both orders the merge AND is recorded
- *   as the doc's synced version (the persist If-Match guard), so a later persist treats this write as
- *   synced rather than an out-of-band conflict.
- * - `streamedAt` — the wall-clock time (epoch ms) a STREAMING snapshot was produced (the copilot stream
- *   mid-flight). Orders the merge so a delayed snapshot older than a newer durable write — possibly from
- *   another app process — is dropped, but is NEVER recorded: the synced version stays pinned to the last
- *   durable write, which is exactly the copilot tool's final `edit_content` write that reconciles the file.
- *
- * Pass one or the other, never both. Passing neither applies the merge without ordering it (legacy).
+ * `order` ({@link LiveFileDocMergeOrder}) positions this merge so a stale write never regresses the doc:
+ * the relay drops any merge not NEWER than the version the doc already incorporates. A durable `version`
+ * is recorded as the synced version (the persist If-Match guard); a streaming `streamedAt` is checked but
+ * never recorded, so the synced version stays pinned to the last durable write — which the copilot tool's
+ * final `edit_content` write carries, reconciling the durable file.
  *
  * Ordering is enforced at two scales. Within this process, merges for a file run on a single serialized
  * chain — each chained after the current tail — so a durable write applies after any in-flight streaming
  * merge and after every earlier durable write, never concurrently. Across processes, the per-process
- * chain does not apply, so the relay orders merges by the monotonic version above (durable version /
- * streaming `streamedAt`) under a cluster-wide lock. The copilot streaming caller uses
- * {@link isLiveDocMergeInFlight} to skip redundant snapshots while one is in flight, so a slow relay
- * can't backlog stale snapshots.
+ * chain does not apply, so the relay orders merges by that monotonic version under a cluster-wide lock.
+ * The copilot streaming caller uses {@link isLiveDocMergeInFlight} to skip redundant snapshots while one
+ * is in flight, so a slow relay can't backlog stale snapshots.
  */
-export interface LiveFileDocMergeOrder {
-  /** A durable write's `contentUpdatedAt` (epoch ms): orders the merge AND is recorded as the synced version. */
-  version?: number
-  /** A streaming snapshot's production time (epoch ms): orders the merge only — never recorded as a checkpoint. */
-  streamedAt?: number
-}
-
 export async function mergeEditIntoLiveFileDoc(
   fileId: string,
   markdown: string,