From eb406479be64206e53565e3fe944fe8e86691580 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 29 Jul 2026 19:30:39 -0700 Subject: [PATCH 01/18] feat(collab-doc): optimistic-concurrency guard so persist never clobbers an out-of-band edit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The relay projected the live Yjs doc back to durable markdown unconditionally (last-write-wins), so a persist already in flight when an external write landed could overwrite it. Add RFC 7232 If-Match optimistic concurrency end to end, reconciling through the CRDT (never rejecting user work): - updateWorkspaceFileContent gains an expectedUpdatedAt guard: the write commits only if the file is still at that version (checked against the SELECT ... FOR UPDATE-locked row, so it is atomic with the write), else it throws the new ContentVersionConflictError without clobbering. - persistFileDoc takes expectedVersion and returns a discriminated result (persisted | missing | conflict). On conflict it returns the current durable content + version instead of writing. - The relay tracks the durable version its live doc is synced to — set on seed, advanced when a durable write is merged in (apply-edit carries the version), and on each successful persist. It is held cluster-wide in Redis (filedoc:syncver:{name}) so whichever task persists reads the same version, with the per-room value as the single-pod fallback. - flushPersist sends that version as If-Match. On a conflict it merges the current durable content into the live doc (so the out-of-band edit AND the live edits converge) and retries (bounded), so even a last-leave flush racing an external write persists the reconciled result rather than losing the session's edits. Threads the version through the seed + persist contracts and the apply-edit payload. No schema change (reuses workspace_files.updatedAt as the version token). Tests: app-side CAS (match writes, mismatch throws + cleans up the orphan upload), relay conflict handled gracefully without clobber/loop; 236 realtime + 76 sim collab/uploads tests, tsc x2, lint, api-validation, boundaries, prune all green. --- apps/realtime/src/handlers/file-doc-app.ts | 52 +++++--- apps/realtime/src/handlers/file-doc-store.ts | 32 +++++ apps/realtime/src/handlers/file-doc.test.ts | 63 +++++++--- apps/realtime/src/handlers/file-doc.ts | 114 +++++++++++++++--- apps/realtime/src/routes/http.ts | 10 +- .../api/internal/file-doc/persist/route.ts | 9 +- .../app/api/internal/file-doc/seed/route.ts | 1 + apps/sim/lib/api/contracts/file-doc.ts | 40 ++++-- apps/sim/lib/collab-doc/persist.ts | 107 +++++++++++----- apps/sim/lib/collab-doc/seed.test.ts | 1 + apps/sim/lib/collab-doc/seed.ts | 10 +- apps/sim/lib/realtime/notify.test.ts | 8 +- apps/sim/lib/realtime/notify.ts | 10 +- .../workspace/workspace-file-manager.ts | 44 ++++++- .../workspace-file-storage-accounting.test.ts | 47 +++++++- 15 files changed, 447 insertions(+), 101 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc-app.ts b/apps/realtime/src/handlers/file-doc-app.ts index 8223a5eef74..7c56b1593c5 100644 --- a/apps/realtime/src/handlers/file-doc-app.ts +++ b/apps/realtime/src/handlers/file-doc-app.ts @@ -26,7 +26,7 @@ function postToApp(path: string, payload: unknown, timeoutMs: number): Promise { +): 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,49 @@ 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. `markdown` + + * `version` are the current durable content + version so the relay reconciles and re-persists. + */ +export type PersistResult = + | { status: 'persisted'; version: number } + | { status: 'missing' } + | { status: 'conflict'; markdown: string; version: number } + /** * 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 { + docState: Uint8Array, + expectedVersion?: number +): Promise { 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') { + throw new Error(`Persist for file ${fileId} returned a malformed body`) + } + return body } diff --git a/apps/realtime/src/handlers/file-doc-store.ts b/apps/realtime/src/handlers/file-doc-store.ts index e4bd138612c..652eee8174a 100644 --- a/apps/realtime/src/handlers/file-doc-store.ts +++ b/apps/realtime/src/handlers/file-doc-store.ts @@ -79,6 +79,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 +446,36 @@ 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 { + 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. 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 { + if (!this.enabled || !this.write) return + await this.write + .set(`${SYNC_VERSION_PREFIX}${name}`, String(version), { EX: STREAM_TTL_SEC }) + .catch(() => {}) + } + async releaseMergeSlot(name: string, token: string): Promise { await this.releaseLock(`${MERGE_LOCK_PREFIX}${name}`, token) } diff --git a/apps/realtime/src/handlers/file-doc.test.ts b/apps/realtime/src/handlers/file-doc.test.ts index 83c4c0e144a..aa026e68172 100644 --- a/apps/realtime/src/handlers/file-doc.test.ts +++ b/apps/realtime/src/handlers/file-doc.test.ts @@ -130,11 +130,11 @@ async function flushMicrotasks(): Promise { * 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('retries a persist that keeps conflicting, then gives up without clobbering', async () => { + mockFetchFileDocSeed.mockResolvedValue(seedResult('# From server')) + // Every persist reports an out-of-band change (If-Match conflict) — the durable file must be left + // authoritative rather than overwritten, after a bounded reconcile-retry. + mockFetchFileDocPersist.mockResolvedValue({ + status: 'conflict', + markdown: '# external edit', + 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() + + // A persistent conflict is handled gracefully: the persist is attempted (never silently skipped) + // but the durable file is left authoritative — no clobber, no throw, no unbounded retry loop. + cleanupFileDocForSocket('socket-1', io, true) + await flushMicrotasks() + expect(mockFetchFileDocPersist).toHaveBeenCalled() + expect(mockFetchFileDocPersist.mock.calls.length).toBeLessThanOrEqual(3) + }) + 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 }) diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index 509d312c338..0867ee329dd 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -65,6 +65,10 @@ const PERSIST_DEBOUNCE_MS = 5_000 * 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 +/** How many times a persist re-tries after reconciling an out-of-band edit into the live doc before it + * gives up (leaving the durable content authoritative). One reconcile normally converges; the small + * budget covers a second racing write. */ +const PERSIST_CONFLICT_RETRIES = 2 /** 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 @@ -125,6 +129,13 @@ interface FileDocRoom { /** 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 + /** + * The durable file version (its `updatedAt`, epoch ms) this live doc is known to be synced to — set + * on seed, advanced when a durable write is merged in (apply-edit) and on each successful persist. It + * is the `If-Match` token {@link flushPersist} sends so a persist can never clobber an out-of-band + * edit the live doc hasn't incorporated. `null` before the first seed (persist writes unconditionally). + */ + syncedVersion: number | null } /** Live documents keyed by Socket.IO room name. Module-global: one Y.Doc per file. */ @@ -226,27 +237,70 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr // Never project a doc no user actually edited back over the file (see {@link FileDocRoom.edited}). if (!room.edited || !room.workspaceId || !room.lastEditorUserId) return const store = getFileDocStore() + const workspaceId = room.workspaceId + const userId = room.lastEditorUserId // Synchronous fallback capture — before any await, since the caller may destroy `room.doc` the moment // 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 + + // Capture the AUTHORITATIVE doc state: the shared stream when enabled (a copilot merge or a peer's + // edit published by another task may not be integrated into THIS task's `room.doc` yet), else the + // local snapshot. Re-read each attempt so a post-reconcile retry projects the converged state. + const captureState = async (): Promise => { + if (!store.enabled) return localState + try { + return (await store.getStreamState(name)) ?? localState + } catch (streamError) { + // A transient Redis read must NOT drop the write when we already hold a valid local snapshot — + // else the last-disconnect flush loses the session's edits as the room is torn down. + if (!localState) throw streamError + logger.warn(`Stream state unavailable for file ${room.fileId}; persisting local snapshot`, { + error: getErrorMessage(streamError), + }) + return localState + } + } + // The If-Match token: the cluster-wide synced version when enabled, else this task's room value. + const currentVersion = async (): Promise => { + const shared = store.enabled ? await store.getSyncedVersion(name) : null + return shared ?? room.syncedVersion ?? undefined + } + try { if (!final && !(await store.tryClaimPersistWindow(name, FILE_DOC_TIMEOUTS.persistRequestMs))) return - let docState = localState - if (store.enabled) { - try { - docState = (await store.getStreamState(name)) ?? localState - } catch (streamError) { - // A transient Redis read must NOT drop the write when we already hold a valid local snapshot — - // else the last-disconnect flush loses the session's edits as the room is torn down. - if (!localState) throw streamError - logger.warn(`Stream state unavailable for file ${room.fileId}; persisting local snapshot`, { - error: getErrorMessage(streamError), - }) + + // Persist under optimistic concurrency (RFC 7232 If-Match): the write commits only if the file is + // still at the version the live doc synced from, so a projection can never silently clobber an + // out-of-band edit. On a conflict, merge the current durable content into the live doc — so the + // out-of-band edit AND the live edits converge — and retry (bounded), so even a last-leave flush + // racing an external write persists the reconciled result rather than losing the session's edits. + for (let attempt = 0; attempt <= PERSIST_CONFLICT_RETRIES; attempt++) { + const docState = await captureState() + if (!docState) return // nothing seeded/authoritative to persist yet + const result = await fetchFileDocPersist( + workspaceId, + room.fileId, + userId, + docState, + await currentVersion() + ) + if (result.status === 'missing') return + if (result.status === 'persisted') { + room.syncedVersion = result.version + void store.setSyncedVersion(name, result.version) + return + } + if (attempt === PERSIST_CONFLICT_RETRIES) { + logger.warn( + `Persist for file ${room.fileId} still conflicting after reconcile; durable content left authoritative` + ) + return } + // Reconcile the out-of-band durable content into the live doc (advances the synced version), then + // loop to re-project and re-persist the converged state. + await applyMarkdownToLiveFileDoc(room.fileId, result.markdown, result.version) } - if (!docState) return // nothing seeded/authoritative to persist yet - await fetchFileDocPersist(room.workspaceId, room.fileId, room.lastEditorUserId, docState) } catch (error) { logger.warn(`Persist failed for file ${room.fileId}`, { error: getErrorMessage(error) }) } @@ -371,8 +425,15 @@ async function ensureServerSeed( } // We hold the seed lock — release it on EVERY exit from here (one `finally`, impossible to leak). try { - const update = await fetchFileDocSeed(workspaceId, room.fileId) + const seed = await fetchFileDocSeed(workspaceId, room.fileId) if (fileDocRooms.get(name) !== room || isDocSeeded(room.doc)) return + // Record the durable version this seed was built from: the live doc is now synced to it, so the + // first persist's If-Match guard compares against it. Cluster-wide (Redis) so any task's persist + // reads it, plus this task's room for the single-pod fallback. (Left null for an empty/missing file.) + if (seed) { + room.syncedVersion = seed.version + void store.setSyncedVersion(name, seed.version) + } // Build the seed (file content + seed flag, or just the flag for an empty/missing file) and write it // to the shared stream ATOMICALLY, iff the stream is still empty. This — NOT the seed lock — is the // split-brain guard: two tasks racing (even both past an expired lock) can never both seed, because @@ -380,7 +441,7 @@ async function ensureServerSeed( // seeded (via the local apply) only once the seed is durably in the stream, so a failed write leaves // the doc unseeded and the stream empty for a clean retry. SEED_ORIGIN keeps `doc.on('update')` from // re-publishing it. - const seedUpdate = update ?? emptySeedUpdate() + const seedUpdate = seed?.update ?? emptySeedUpdate() const didSeed = await store.seedIfEmpty(name, seedUpdate) if (fileDocRooms.get(name) !== room || isDocSeeded(room.doc)) return if (didSeed) { @@ -436,12 +497,15 @@ const fileDocMergeChains = new Map>() */ export function applyMarkdownToLiveFileDoc( fileId: string, - markdown: string + markdown: string, + version?: number ): Promise<'applied' | 'no-live-room'> { 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)) + const run = prior + .catch(() => {}) + .then(() => mergeMarkdownIntoRoom(name, fileId, markdown, version)) fileDocMergeChains.set( name, run.finally(() => { @@ -454,10 +518,21 @@ export function applyMarkdownToLiveFileDoc( async function mergeMarkdownIntoRoom( name: string, fileId: string, - markdown: string + markdown: string, + version?: number ): Promise<'applied' | 'no-live-room'> { const store = getFileDocStore() + // The durable version this merge carries is now incorporated in the live doc — record it (cluster-wide + // in Redis for multi-task, plus this task's room) so the persist If-Match guard treats this write as + // synced rather than an out-of-band conflict. Set on success below. + const recordVersion = () => { + if (version === undefined) return + const room = fileDocRooms.get(name) + if (room) room.syncedVersion = version + void store.setSyncedVersion(name, version) + } + if (store.enabled) { // Serialize merges to this file ACROSS tasks — the per-file chain above only covers this process. // Two copilot edits to the same file landing on different tasks must not diff the SAME shared base @@ -484,6 +559,7 @@ async function mergeMarkdownIntoRoom( if (!base) return 'no-live-room' const diff = await fetchFileDocMerge(fileId, base, markdown) await store.publishAndWait(name, diff) + recordVersion() return 'applied' } finally { await store.releaseMergeSlot(name, token) @@ -498,6 +574,7 @@ async function mergeMarkdownIntoRoom( if (fileDocRooms.get(name) !== room) return 'no-live-room' // No transaction origin → `doc.on('update')` relays to the WHOLE room (every editor sees copilot). Y.applyUpdate(room.doc, update) + recordVersion() return 'applied' } @@ -528,6 +605,7 @@ function getOrCreateRoom(io: Server, ref: RoomRef): FileDocRoom { seededObserved: false, persistTimer: null, persistDeadline: null, + syncedVersion: null, } // Register synchronously BEFORE the async catch-up so a concurrent join sees this room, not a second. fileDocRooms.set(name, room) diff --git a/apps/realtime/src/routes/http.ts b/apps/realtime/src/routes/http.ts index 3b184eb9f15..fe824a70b86 100644 --- a/apps/realtime/src/routes/http.ts +++ b/apps/realtime/src/routes/http.ts @@ -209,11 +209,17 @@ export function createHttpHandler(roomManager: IRoomManager, logger: Logger) { if (req.method === 'POST' && req.url === '/api/file-doc/apply-edit') { try { const body = await readRequestBody(req) - const { fileId, markdown } = JSON.parse(body) + const { fileId, markdown, version } = JSON.parse(body) if (!isNonEmptyString(fileId) || typeof markdown !== 'string') { return sendError(res, 'Invalid fileId or markdown', 400) } - const result = await applyMarkdownToLiveFileDoc(fileId, markdown) + // `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. + const result = await applyMarkdownToLiveFileDoc( + fileId, + markdown, + typeof version === 'number' ? version : undefined + ) res.writeHead(200, { 'Content-Type': 'application/json' }) res.end(JSON.stringify({ applied: result === 'applied' })) } catch (error) { diff --git a/apps/sim/app/api/internal/file-doc/persist/route.ts b/apps/sim/app/api/internal/file-doc/persist/route.ts index 9ef87415991..a85923deea7 100644 --- a/apps/sim/app/api/internal/file-doc/persist/route.ts +++ b/apps/sim/app/api/internal/file-doc/persist/route.ts @@ -23,16 +23,17 @@ export const POST = withRouteHandler(async (request: NextRequest) => { const parsed = await parseRequest(persistFileDocContract, request, {}) if (!parsed.success) return parsed.response - const { workspaceId, fileId, userId, docState } = parsed.data.body + const { workspaceId, fileId, userId, docState, expectedVersion } = parsed.data.body try { - const persisted = await persistFileDoc( + const result = await persistFileDoc( workspaceId, fileId, userId, - new Uint8Array(Buffer.from(docState, 'base64')) + new Uint8Array(Buffer.from(docState, 'base64')), + expectedVersion ) - return NextResponse.json({ persisted }) + return NextResponse.json(result) } catch (error) { logger.error('Failed to persist file-doc', { workspaceId, fileId, error }) return NextResponse.json( diff --git a/apps/sim/app/api/internal/file-doc/seed/route.ts b/apps/sim/app/api/internal/file-doc/seed/route.ts index 058e7a0003a..aaf051da094 100644 --- a/apps/sim/app/api/internal/file-doc/seed/route.ts +++ b/apps/sim/app/api/internal/file-doc/seed/route.ts @@ -28,6 +28,7 @@ export const POST = withRouteHandler(async (request: NextRequest) => { const seed = await buildFileDocSeed(workspaceId, fileId) return NextResponse.json({ update: seed ? Buffer.from(seed.update).toString('base64') : null, + version: seed ? seed.version : null, }) } catch (error) { logger.error('Failed to build file-doc seed', { workspaceId, fileId, error }) diff --git a/apps/sim/lib/api/contracts/file-doc.ts b/apps/sim/lib/api/contracts/file-doc.ts index a46bce7f5d3..127cdbf141d 100644 --- a/apps/sim/lib/api/contracts/file-doc.ts +++ b/apps/sim/lib/api/contracts/file-doc.ts @@ -13,6 +13,12 @@ export const buildFileDocSeedResponseSchema = z.object({ * file is missing/deleted (the caller treats that as an empty document). */ update: z.string().nullable(), + /** + * The file's durable `updatedAt` (epoch ms) this seed was built from — the version the relay records + * as what its freshly-seeded live doc is synced to (for the persist optimistic-concurrency guard). + * `null` when the file is missing. + */ + version: z.number().int().nullable(), }) export type BuildFileDocSeedResponse = z.output @@ -89,16 +95,36 @@ export const persistFileDocBodySchema = z.object({ .string() .min(1, 'docState is required') .max(16 * 1024 * 1024, 'docState is too large'), -}) -export type PersistFileDocBody = z.input - -export const persistFileDocResponseSchema = z.object({ /** - * `true` when the document was projected to markdown and written durably to the file; `false` when - * the file was missing/deleted (nothing to write). A transport/conversion failure is a non-2xx. + * Optimistic-concurrency guard (RFC 7232 `If-Match`): the durable `updatedAt` (epoch ms) the relay's + * live doc last synced from. The write commits only if the file is still at this version; otherwise + * the response is `status: 'conflict'` and nothing is written. Omit for an unconditional first write. */ - persisted: z.boolean(), + expectedVersion: z.number().int().optional(), }) +export type PersistFileDocBody = z.input + +export const persistFileDocResponseSchema = z.discriminatedUnion('status', [ + z.object({ + /** Projected to markdown and written durably. `version` is the new durable `updatedAt` (epoch ms). */ + status: z.literal('persisted'), + version: z.number().int(), + }), + z.object({ + /** The file was missing/deleted — nothing to write. */ + status: z.literal('missing'), + }), + z.object({ + /** + * The file changed out-of-band since `expectedVersion` — the write was refused to avoid clobbering + * it. `markdown` + `version` are the current durable content + version, so the relay can merge the + * change into its live doc and re-persist the reconciled result. + */ + status: z.literal('conflict'), + markdown: z.string(), + version: z.number().int(), + }), +]) export type PersistFileDocResponse = z.output /** diff --git a/apps/sim/lib/collab-doc/persist.ts b/apps/sim/lib/collab-doc/persist.ts index c83c0c6ee60..9082db83f6a 100644 --- a/apps/sim/lib/collab-doc/persist.ts +++ b/apps/sim/lib/collab-doc/persist.ts @@ -1,34 +1,56 @@ import { createLogger } from '@sim/logger' import { getErrorMessage } from '@sim/utils/errors' import * as Y from 'yjs' -import { getWorkspaceFile, updateWorkspaceFileContent } from '@/lib/uploads/contexts/workspace' +import { + ContentVersionConflictError, + fetchWorkspaceFileBuffer, + getWorkspaceFile, + updateWorkspaceFileContent, +} from '@/lib/uploads/contexts/workspace' import { hashMarkdown, saveCollabDocState } from './collab-state' import { yDocToFileMarkdown } from './converter' const logger = createLogger('FileDocPersist') /** - * Project a live collaborative document back to durable markdown and write it to the file. This is the - * server-authoritative durable path — the realtime relay owns the live Yjs doc but not the conversion - * engine or blob/DB access, so it ships the doc state here and the app persists it. Called debounced - * while the doc is being edited and when the last collaborator leaves; it replaces the editor's - * client-side autosave, so a copilot (or any server) edit can never be clobbered by a stale keystroke - * saving over it. + * Outcome of a persist attempt: + * - `persisted` — the live doc was projected to markdown and written; `version` is the new durable + * `updatedAt` (epoch ms) the relay records as the version its live doc is now synced to. + * - `missing` — the file is gone (deleted); nothing to write. + * - `conflict` — the file changed out-of-band since the relay's live doc last synced, so writing the + * projection would clobber that change (RFC 7232 `If-Match` failure). NOT written. `markdown` + + * `version` are the current durable content + version so the relay can merge them into its live doc + * and re-persist the reconciled result. + */ +export type PersistFileDocResult = + | { status: 'persisted'; version: number } + | { status: 'missing' } + | { status: 'conflict'; markdown: string; version: number } + +/** + * Project a live collaborative document back to durable markdown and write it to the file. The realtime + * relay owns the live Yjs doc but not the conversion engine or blob/DB access, so it ships the doc state + * here and the app persists it — the server-authoritative durable path that replaces the editor's + * client-side autosave. * - * Returns `false` when the file is genuinely absent (deleted) — nothing to write. Any other failure - * (conversion / blob / DB) THROWS so the caller surfaces a non-2xx and can retry on the next debounce. + * `expectedVersion` (the durable `updatedAt`, epoch ms, that the relay's live doc last synced from) is + * the optimistic-concurrency guard: the write commits only if the file is still at that version, so a + * projection built from a stale live doc can never silently overwrite an out-of-band edit. On a version + * mismatch this returns `conflict` (with the current durable content) instead of writing — the relay + * reconciles and retries. Omit `expectedVersion` to write unconditionally (e.g. the first persist, + * before any synced version exists). * - * `userId` is attribution only (blob metadata) — the last collaborator to touch the doc; the caller is - * already trusted via the internal `x-api-key` gate, so this does not re-authorize. + * `userId` is attribution only (blob metadata); the caller is already trusted via the `x-api-key` gate. */ export async function persistFileDoc( workspaceId: string, fileId: string, userId: string, - docState: Uint8Array -): Promise { + docState: Uint8Array, + expectedVersion?: number +): Promise { const record = await getWorkspaceFile(workspaceId, fileId, { throwOnError: true }) - if (!record) return false + if (!record) return { status: 'missing' } const ydoc = new Y.Doc() let markdownBuffer: Buffer @@ -39,24 +61,47 @@ export async function persistFileDoc( ydoc.destroy() } - // `syncLiveDoc: false`: this write IS the projection of the live doc back to markdown, so merging it - // back into that same doc would be a persist → merge → persist self-loop. The doc already holds this - // exact content; peers are already converged through the relay. - await updateWorkspaceFileContent(workspaceId, fileId, userId, markdownBuffer, undefined, { - syncLiveDoc: false, - }) - - // Cache the Yjs binary (tagged with the exact markdown just written) so a later cold room open loads - // it directly instead of re-converting markdown → Yjs. Best-effort: the markdown IS the durable file, - // so a cache failure only means the next cold open re-converts — never lost data. try { - await saveCollabDocState(fileId, docState, hashMarkdown(markdownBuffer)) + const updated = await updateWorkspaceFileContent( + workspaceId, + fileId, + userId, + markdownBuffer, + undefined, + { + // This write IS the projection of the live doc, so re-merging it into that same doc would loop. + syncLiveDoc: false, + // If-Match: only if the durable file is still at the version the live doc synced from. + expectedUpdatedAt: expectedVersion !== undefined ? new Date(expectedVersion) : undefined, + } + ) + + // Cache the Yjs binary (tagged with the exact markdown just written) so a later cold room open loads + // it directly instead of re-converting. Best-effort — the markdown is the durable source of truth. + try { + await saveCollabDocState(fileId, docState, hashMarkdown(markdownBuffer)) + } catch (error) { + logger.warn(`Failed to cache collab doc state for file ${fileId}`, { + error: getErrorMessage(error), + }) + } + + logger.info( + `Persisted live collaborative document to file ${fileId} (workspace ${workspaceId})` + ) + return { status: 'persisted', version: updated.updatedAt.getTime() } } catch (error) { - logger.warn(`Failed to cache collab doc state for file ${fileId}`, { - error: getErrorMessage(error), - }) + if (!(error instanceof ContentVersionConflictError)) throw error + // Out-of-band edit since the live doc last synced — DON'T clobber. Return the current durable + // content + version so the relay merges it into the live doc and re-persists the reconciled result. + const current = await getWorkspaceFile(workspaceId, fileId, { throwOnError: true }) + if (!current) return { status: 'missing' } + const currentBuffer = await fetchWorkspaceFileBuffer(current) + logger.warn(`Persist conflict for file ${fileId}; returning current durable state to reconcile`) + return { + status: 'conflict', + markdown: currentBuffer.toString('utf-8'), + version: current.updatedAt.getTime(), + } } - - logger.info(`Persisted live collaborative document to file ${fileId} (workspace ${workspaceId})`) - return true } diff --git a/apps/sim/lib/collab-doc/seed.test.ts b/apps/sim/lib/collab-doc/seed.test.ts index a3785a1c5ec..28e8562464a 100644 --- a/apps/sim/lib/collab-doc/seed.test.ts +++ b/apps/sim/lib/collab-doc/seed.test.ts @@ -35,6 +35,7 @@ describe('buildFileDocSeed', () => { name: 'note.md', key: 'k', context: 'workspace', + updatedAt: new Date('2026-01-01T00:00:00.000Z'), }) // Default: no cached binary → the conversion path runs (the case these tests cover). mockLoadFresh.mockResolvedValue(null) diff --git a/apps/sim/lib/collab-doc/seed.ts b/apps/sim/lib/collab-doc/seed.ts index 4fbfaa544c3..fdf4f0d578e 100644 --- a/apps/sim/lib/collab-doc/seed.ts +++ b/apps/sim/lib/collab-doc/seed.ts @@ -19,6 +19,11 @@ const MAX_SEED_BYTES = 5 * 1024 * 1024 export interface FileDocSeed { /** `Y.encodeStateAsUpdate` of the seeded document — apply with `Y.applyUpdate`. */ update: Uint8Array + /** + * The file's durable `updatedAt` (epoch ms) this seed was built from — the version the relay records + * as what its freshly-seeded live doc is synced to, for the persist optimistic-concurrency guard. + */ + version: number } /** @@ -41,6 +46,7 @@ export async function buildFileDocSeed( const record = await getWorkspaceFile(workspaceId, fileId, { throwOnError: true }) if (!record) return null + const version = record.updatedAt.getTime() const buffer = await fetchWorkspaceFileBuffer(record, { maxBytes: MAX_SEED_BYTES }) // Cold-start fast path: if we hold a cached Yjs binary derived from THIS exact markdown, apply it @@ -54,7 +60,7 @@ export async function buildFileDocSeed( // block the cold open — symmetric with persist's best-effort cache write. try { const cached = await loadFreshCollabDocState(fileId, hashMarkdown(buffer)) - if (cached) return { update: cached } + if (cached) return { update: cached, version } } catch (error) { logger.warn(`Failed to read cached collab doc state for file ${fileId}`, { error: getErrorMessage(error), @@ -74,7 +80,7 @@ export async function buildFileDocSeed( // Carry the frontmatter in the doc (not the body) so it merges across clients and a later // server-side edit can update it — the editor re-attaches this on autosave. config.set(FILE_DOC_SEED.frontmatterKey, frontmatter) - return { update: Y.encodeStateAsUpdate(ydoc) } + return { update: Y.encodeStateAsUpdate(ydoc), version } } finally { ydoc.destroy() } diff --git a/apps/sim/lib/realtime/notify.test.ts b/apps/sim/lib/realtime/notify.test.ts index 7e780e76130..005637ebe44 100644 --- a/apps/sim/lib/realtime/notify.test.ts +++ b/apps/sim/lib/realtime/notify.test.ts @@ -17,25 +17,25 @@ describe('mergeEditIntoLiveFileDoc', () => { const fetchMock = vi.fn().mockResolvedValue({ ok: true }) vi.stubGlobal('fetch', fetchMock) - await mergeEditIntoLiveFileDoc('file-1', '# hello') + await mergeEditIntoLiveFileDoc('file-1', '# hello', 42) expect(fetchMock).toHaveBeenCalledWith( 'http://realtime/api/file-doc/apply-edit', expect.objectContaining({ method: 'POST', headers: expect.objectContaining({ 'x-api-key': 'secret' }), - body: JSON.stringify({ fileId: 'file-1', markdown: '# hello' }), + body: JSON.stringify({ fileId: 'file-1', markdown: '# hello', version: 42 }), }) ) }) it('never throws when the realtime call fails (best-effort)', async () => { vi.stubGlobal('fetch', vi.fn().mockRejectedValue(new Error('socket pod down'))) - await expect(mergeEditIntoLiveFileDoc('file-1', '# hello')).resolves.toBeUndefined() + await expect(mergeEditIntoLiveFileDoc('file-1', '# hello', 42)).resolves.toBeUndefined() }) it('never throws on a non-2xx response', async () => { vi.stubGlobal('fetch', vi.fn().mockResolvedValue({ ok: false, status: 503 })) - await expect(mergeEditIntoLiveFileDoc('file-1', '# hello')).resolves.toBeUndefined() + await expect(mergeEditIntoLiveFileDoc('file-1', '# hello', 42)).resolves.toBeUndefined() }) }) diff --git a/apps/sim/lib/realtime/notify.ts b/apps/sim/lib/realtime/notify.ts index f8aedd6f011..92a4eb2f920 100644 --- a/apps/sim/lib/realtime/notify.ts +++ b/apps/sim/lib/realtime/notify.ts @@ -124,12 +124,18 @@ export async function notifyFolderResourceChanged( * Awaited (not fire-and-forget) so the fetch dispatches before the route handler returns; bounded to * {@link APPLY_EDIT_TIMEOUT_MS}, so it adds latency only when the socket pod is unreachable. */ -export async function mergeEditIntoLiveFileDoc(fileId: string, markdown: string): Promise { +export async function mergeEditIntoLiveFileDoc( + fileId: string, + markdown: string, + version: number +): Promise { try { const response = await fetch(`${getSocketServerUrl()}/api/file-doc/apply-edit`, { method: 'POST', headers: { 'Content-Type': 'application/json', 'x-api-key': env.INTERNAL_API_SECRET }, - body: JSON.stringify({ fileId, markdown }), + // `version` is the durable `updatedAt` (epoch ms) this markdown was written with — the relay + // records it as the version its live doc now incorporates (see the persist If-Match guard). + body: JSON.stringify({ fileId, markdown, version }), signal: AbortSignal.timeout(APPLY_EDIT_TIMEOUT_MS), }) if (!response.ok) { diff --git a/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts b/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts index 0eac611b3f8..4776949d152 100644 --- a/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts +++ b/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts @@ -1016,6 +1016,18 @@ export async function fetchWorkspaceFileBuffer( } } +/** + * Thrown by {@link updateWorkspaceFileContent} when its `expectedUpdatedAt` optimistic-concurrency + * guard fails — the file changed out-of-band since the caller read it. Callers catch this to reconcile + * (re-read + merge) rather than overwrite. Not a failure: the durable file is left untouched. + */ +export class ContentVersionConflictError extends Error { + constructor(readonly fileId: string) { + super(`Workspace file ${fileId} changed since it was read (optimistic-concurrency conflict)`) + this.name = 'ContentVersionConflictError' + } +} + /** * Updates a workspace file through a versioned object swap. Blob I/O completes * before the short metadata-and-ledger transaction. @@ -1036,6 +1048,14 @@ export async function updateWorkspaceFileContent( * content arrives via a subsequent write. */ syncLiveDoc?: boolean + /** + * Optimistic-concurrency guard (RFC 7232 `If-Match` semantics). When set, the write commits only + * if the file's `updatedAt` still equals this value — nothing else wrote in between; otherwise it + * throws {@link ContentVersionConflictError} without clobbering. Checked against the + * `SELECT … FOR UPDATE`-locked row, so it is atomic with the write. Used by the collab persist so + * projecting the live doc back to markdown can never silently overwrite an out-of-band edit. + */ + expectedUpdatedAt?: Date } ): Promise { logger.info(`Updating workspace file content: ${fileId} for workspace ${workspaceId}`) @@ -1095,6 +1115,17 @@ export async function updateWorkspaceFileContent( throw new Error('File not found') } + // Optimistic-concurrency guard: the row is `FOR UPDATE`-locked, so comparing its committed + // `updatedAt` to the caller's expected value and then writing is atomic — a racing writer blocks + // here until this transaction resolves. A mismatch means the file changed out-of-band since the + // caller last synced; abort rather than clobber it. + if ( + options?.expectedUpdatedAt && + currentFile.updatedAt.getTime() !== options.expectedUpdatedAt.getTime() + ) { + throw new ContentVersionConflictError(fileId) + } + const sizeDiff = content.length - currentFile.size const [updatedFile] = await tx .update(workspaceFiles) @@ -1164,7 +1195,14 @@ export async function updateWorkspaceFileContent( options?.syncLiveDoc !== false && isMarkdownFile({ type: nextContentType, name: finalized.file.originalName }) ) { - await mergeEditIntoLiveFileDoc(fileId, content.toString('utf-8')) + // Pass the new `updatedAt` as the version this write produced, so the relay records that its live + // doc now incorporates this durable version — the collab persist's optimistic-concurrency guard + // then won't treat this (already-merged) write as an out-of-band conflict. + await mergeEditIntoLiveFileDoc( + fileId, + content.toString('utf-8'), + finalized.file.updatedAt.getTime() + ) } const pathPrefix = getServePathPrefix() @@ -1189,6 +1227,10 @@ export async function updateWorkspaceFileContent( updatedAt: finalized.file.updatedAt, } } catch (error) { + // Preserve the typed conflict so callers can catch it and reconcile — it's an expected outcome of + // the optimistic-concurrency guard, not a failure to wrap. The orphan upload was already cleaned up + // by the inner finalization catch before it propagated here. + if (error instanceof ContentVersionConflictError) throw error logger.error(`Failed to update workspace file content ${fileId}:`, error) throw new Error(`Failed to update file content: ${getErrorMessage(error, 'Unknown error')}`) } diff --git a/apps/sim/lib/uploads/contexts/workspace/workspace-file-storage-accounting.test.ts b/apps/sim/lib/uploads/contexts/workspace/workspace-file-storage-accounting.test.ts index 138bb8e1293..7a7f1aa6cc1 100644 --- a/apps/sim/lib/uploads/contexts/workspace/workspace-file-storage-accounting.test.ts +++ b/apps/sim/lib/uploads/contexts/workspace/workspace-file-storage-accounting.test.ts @@ -69,6 +69,7 @@ vi.mock('@/lib/workspaces/permissions/utils', () => ({ })) import { + ContentVersionConflictError, deleteWorkspaceFile, registerUploadedWorkspaceFile, restoreWorkspaceFile, @@ -355,7 +356,11 @@ describe('workspace file metadata and storage accounting', () => { Buffer.from('# new content', 'utf-8') ) - expect(mockMergeEditIntoLiveFileDoc).toHaveBeenCalledWith(MD_ROW.id, '# new content') + expect(mockMergeEditIntoLiveFileDoc).toHaveBeenCalledWith( + MD_ROW.id, + '# new content', + updatedFile.updatedAt.getTime() + ) }) it('does NOT merge when syncLiveDoc is false (the relay persist / empty-shell opt-out)', async () => { @@ -392,4 +397,44 @@ describe('workspace file metadata and storage accounting', () => { expect(mockMergeEditIntoLiveFileDoc).not.toHaveBeenCalled() }) + + it('writes when the expectedUpdatedAt optimistic-concurrency guard matches', async () => { + const updatedFile = { ...FILE_ROW, size: 12 } + dbChainMockFns.limit.mockResolvedValueOnce([FILE_ROW]).mockResolvedValueOnce([FILE_ROW]) + dbChainMockFns.returning.mockResolvedValueOnce([updatedFile]) + mockUploadFile.mockResolvedValueOnce({ key: FILE_ROW.key }) + + const updated = await updateWorkspaceFileContent( + FILE_ROW.workspaceId, + FILE_ROW.id, + FILE_ROW.userId, + Buffer.alloc(12), + undefined, + { expectedUpdatedAt: FILE_ROW.updatedAt } + ) + + expect(updated.size).toBe(12) + expect(dbChainMockFns.returning).toHaveBeenCalled() + }) + + it('throws ContentVersionConflictError and does not write when the guard mismatches', async () => { + // The locked row's updatedAt differs from the caller's expected value → out-of-band edit. + dbChainMockFns.limit.mockResolvedValueOnce([FILE_ROW]).mockResolvedValueOnce([FILE_ROW]) + mockUploadFile.mockResolvedValueOnce({ key: FILE_ROW.key }) + + await expect( + updateWorkspaceFileContent( + FILE_ROW.workspaceId, + FILE_ROW.id, + FILE_ROW.userId, + Buffer.alloc(12), + undefined, + { expectedUpdatedAt: new Date('2020-01-01T00:00:00.000Z') } + ) + ).rejects.toBeInstanceOf(ContentVersionConflictError) + + // Never advanced the row, and cleaned up the orphan upload it staged before the conflict. + expect(dbChainMockFns.returning).not.toHaveBeenCalled() + expect(mockDeleteFile).toHaveBeenCalledWith({ key: FILE_ROW.key, context: 'workspace' }) + }) }) From 48190bdae01b3979717b6d6549a20354d11be328 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 29 Jul 2026 19:34:18 -0700 Subject: [PATCH 02/18] chore(collab-doc): heartbeat-refresh the synced-version key TTL alongside its stream Keep filedoc:syncver:{name} alive as long as the room's stream (it was only re-set on seed/merge/persist), so an open-but-idle doc's persist If-Match token can't expire and force a needless reconcile. --- apps/realtime/src/handlers/file-doc-store.ts | 3 +++ 1 file changed, 3 insertions(+) diff --git a/apps/realtime/src/handlers/file-doc-store.ts b/apps/realtime/src/handlers/file-doc-store.ts index 652eee8174a..66d02765374 100644 --- a/apps/realtime/src/handlers/file-doc-store.ts +++ b/apps/realtime/src/handlers/file-doc-store.ts @@ -566,6 +566,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(() => {}) } } } From ef3f342894c9f099b0d93868f5a882d6c0cd1042 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 29 Jul 2026 19:42:38 -0700 Subject: [PATCH 03/18] fix(collab-doc): stop persist-conflict retries when there is no live doc to reconcile MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit On an If-Match conflict with no live doc to reconcile into (last collaborator gone, no shared stream), applyMarkdownToLiveFileDoc returns no-live-room; re-projecting the same pre-teardown snapshot would only re-conflict, so break the retry loop immediately and leave the out-of-band (durable) content authoritative — the intended conflict policy. Addresses Greptile review. --- apps/realtime/src/handlers/file-doc.ts | 17 +++++++++++++++-- 1 file changed, 15 insertions(+), 2 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index 0867ee329dd..67ce2a432a0 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -298,8 +298,21 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr return } // Reconcile the out-of-band durable content into the live doc (advances the synced version), then - // loop to re-project and re-persist the converged state. - await applyMarkdownToLiveFileDoc(room.fileId, result.markdown, result.version) + // loop to re-project and re-persist the converged state. If there is no live doc to reconcile into + // — e.g. the last collaborator has left and the room is being torn down with no shared stream + // (single-pod) — stop: re-projecting the same pre-teardown snapshot would only re-conflict. The + // durable (out-of-band) content is left authoritative, which is the intended conflict policy. + const reconciled = await applyMarkdownToLiveFileDoc( + room.fileId, + result.markdown, + result.version + ) + if (reconciled === 'no-live-room') { + logger.warn( + `Persist conflict for file ${room.fileId} with no live doc to reconcile; durable content left authoritative` + ) + return + } } } catch (error) { logger.warn(`Persist failed for file ${room.fileId}`, { error: getErrorMessage(error) }) From 4a872ec83c99969e79f5d224ab13f75de70b9e9d Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 29 Jul 2026 19:55:24 -0700 Subject: [PATCH 04/18] fix(collab-doc): close three optimistic-concurrency edge cases from review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Single-pod persist retry projected the pre-reconcile snapshot (captureState always returned the initial localState), while the synced version had been advanced by the reconcile — so the If-Match could pass and clobber the reconciled edit. captureState now re-reads the live doc on each attempt (falling back to the pre-teardown snapshot only once the room is gone). - The synced version was recorded from this task's own seed FETCH before knowing whether this task's seed actually won; a peer winning with a different version could leave a newer token than the stream content. Record it only inside the didSeed branch (the task whose seed won); peer-seeded tasks read the winner's cluster value. - Persist wrote UNCONDITIONALLY when no version was available (relay version momentarily missing), which could clobber non-empty durable content. It now returns conflict for a non-empty file with no version (reconcile/retry once the version is re-established); an empty file's first write stays unconditional. --- apps/realtime/src/handlers/file-doc.ts | 26 ++++++++++++++++++-------- apps/sim/lib/collab-doc/persist.ts | 14 ++++++++++++++ 2 files changed, 32 insertions(+), 8 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index 67ce2a432a0..5cd8a77b57a 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -247,7 +247,15 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr // edit published by another task may not be integrated into THIS task's `room.doc` yet), else the // local snapshot. Re-read each attempt so a post-reconcile retry projects the converged state. const captureState = async (): Promise => { - if (!store.enabled) return localState + if (!store.enabled) { + // Single-pod: re-read the live doc so a post-reconcile retry projects the CONVERGED state, not the + // snapshot captured before the reconcile mutated it (which — with the version already advanced — + // would let the If-Match pass and clobber the reconciled edit). Fall back to that snapshot once the + // room is torn down (last-leave), where the doc is gone and there is nothing left to reconcile. + return fileDocRooms.get(name) === room && isDocSeeded(room.doc) + ? Y.encodeStateAsUpdate(room.doc) + : localState + } try { return (await store.getStreamState(name)) ?? localState } catch (streamError) { @@ -440,13 +448,6 @@ async function ensureServerSeed( try { const seed = await fetchFileDocSeed(workspaceId, room.fileId) if (fileDocRooms.get(name) !== room || isDocSeeded(room.doc)) return - // Record the durable version this seed was built from: the live doc is now synced to it, so the - // first persist's If-Match guard compares against it. Cluster-wide (Redis) so any task's persist - // reads it, plus this task's room for the single-pod fallback. (Left null for an empty/missing file.) - if (seed) { - room.syncedVersion = seed.version - void store.setSyncedVersion(name, seed.version) - } // Build the seed (file content + seed flag, or just the flag for an empty/missing file) and write it // to the shared stream ATOMICALLY, iff the stream is still empty. This — NOT the seed lock — is the // split-brain guard: two tasks racing (even both past an expired lock) can never both seed, because @@ -459,6 +460,15 @@ async function ensureServerSeed( if (fileDocRooms.get(name) !== room || isDocSeeded(room.doc)) return if (didSeed) { Y.applyUpdate(room.doc, seedUpdate, SEED_ORIGIN) + // Record the durable version ONLY now that THIS task's seed won — so the synced version reflects + // the stream's actual content. Setting it from our own fetch before knowing who won could record a + // newer version than a peer's winning seed and let a later persist clobber an out-of-band edit. + // Cluster-wide (Redis) so any task's persist reads it; local room is the single-pod fallback. A + // peer-seeded task leaves it null and reads the winner's cluster value. (Null for an empty file.) + if (seed) { + room.syncedVersion = seed.version + void store.setSyncedVersion(name, seed.version) + } } else { // A peer seeded first: its seed arrives via the tailer, so we must NOT apply our own — a second, // different-client-id seed IS the split-brain. Clear the guard so a later join can retry if that diff --git a/apps/sim/lib/collab-doc/persist.ts b/apps/sim/lib/collab-doc/persist.ts index 9082db83f6a..7481108b4cd 100644 --- a/apps/sim/lib/collab-doc/persist.ts +++ b/apps/sim/lib/collab-doc/persist.ts @@ -52,6 +52,20 @@ export async function persistFileDoc( const record = await getWorkspaceFile(workspaceId, fileId, { throwOnError: true }) if (!record) return { status: 'missing' } + // Optimistic concurrency needs a version for any file that already has content. If none was supplied + // (the relay's synced version was momentarily unavailable — e.g. a Redis blip on a peer-seeded task), + // refuse to overwrite non-empty durable content unconditionally: return `conflict` so the relay + // reconciles/retries once the version is re-established, rather than silently clobbering. An empty + // file has nothing to clobber, so its first unconditional write stays allowed. + if (expectedVersion === undefined && record.size > 0) { + const currentBuffer = await fetchWorkspaceFileBuffer(record) + return { + status: 'conflict', + markdown: currentBuffer.toString('utf-8'), + version: record.updatedAt.getTime(), + } + } + const ydoc = new Y.Doc() let markdownBuffer: Buffer try { From ee7433027332adeba20101bfb4f6b4232b9c1ba6 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 29 Jul 2026 20:09:08 -0700 Subject: [PATCH 05/18] fix(collab-doc): defer (not reconcile) on missing version, and use the freshest version token MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Missing-version persist now returns 'deferred' instead of 'conflict'. A missing version token (a Redis blip on a peer-seeded task) is NOT a genuine out-of-band change, so triggering a reconcile would wipe live edits (incoming-wins) even though nothing changed durably. Deferred means: don't write, don't reconcile — leave the edits in the stream and let a later persist write them once the version is re-established. - currentVersion now takes the MAX of the cluster (Redis) and local room versions rather than always preferring Redis, so a lagged/failed fire-and-forget Redis set can't shadow a newer local value and cause spurious If-Match conflicts. Versions are monotonic epoch-ms, so the larger is the later sync. --- apps/realtime/src/handlers/file-doc-app.ts | 8 +++++++- apps/realtime/src/handlers/file-doc.ts | 15 +++++++++++++-- apps/sim/lib/api/contracts/file-doc.ts | 9 +++++++++ apps/sim/lib/collab-doc/persist.ts | 17 +++++++---------- 4 files changed, 36 insertions(+), 13 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc-app.ts b/apps/realtime/src/handlers/file-doc-app.ts index 7c56b1593c5..834cb7f705b 100644 --- a/apps/realtime/src/handlers/file-doc-app.ts +++ b/apps/realtime/src/handlers/file-doc-app.ts @@ -84,6 +84,7 @@ export type PersistResult = | { status: 'persisted'; version: number } | { status: 'missing' } | { status: 'conflict'; markdown: string; version: number } + | { status: 'deferred' } /** * Ask the app to project a live collaborative document back to durable markdown and write it to the @@ -114,7 +115,12 @@ export async function fetchFileDocPersist( 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') { + 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 diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index 5cd8a77b57a..4b30ee1ee71 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -268,10 +268,14 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr return localState } } - // The If-Match token: the cluster-wide synced version when enabled, else this task's room value. + // The If-Match token: the FRESHEST synced version known — the max of the cluster-wide value (Redis) + // and this task's room value. Taking the max (not Redis-preferred) avoids a stale read shadowing a + // newer local value when a fire-and-forget `setSyncedVersion` to Redis lagged or failed; versions are + // monotonic epoch-ms, so the larger is the later sync point. `undefined` when neither is known. const currentVersion = async (): Promise => { const shared = store.enabled ? await store.getSyncedVersion(name) : null - return shared ?? room.syncedVersion ?? undefined + const best = Math.max(shared ?? 0, room.syncedVersion ?? 0) + return best > 0 ? best : undefined } try { @@ -294,6 +298,13 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr await currentVersion() ) if (result.status === 'missing') return + if (result.status === 'deferred') { + // The app couldn't persist safely without a version token (momentarily unavailable). Do NOT + // reconcile — nothing changed out-of-band, so merging would needlessly wipe live edits. Leave + // the edits in the stream; a later persist writes them once the version is re-established. + logger.warn(`Persist deferred for file ${room.fileId} (no synced version available yet)`) + return + } if (result.status === 'persisted') { room.syncedVersion = result.version void store.setSyncedVersion(name, result.version) diff --git a/apps/sim/lib/api/contracts/file-doc.ts b/apps/sim/lib/api/contracts/file-doc.ts index 127cdbf141d..97c9de47e70 100644 --- a/apps/sim/lib/api/contracts/file-doc.ts +++ b/apps/sim/lib/api/contracts/file-doc.ts @@ -114,6 +114,15 @@ export const persistFileDocResponseSchema = z.discriminatedUnion('status', [ /** The file was missing/deleted — nothing to write. */ status: z.literal('missing'), }), + z.object({ + /** + * Couldn't persist safely right now: no `expectedVersion` was supplied for a non-empty file (the + * relay's synced-version token was momentarily unavailable). Nothing was written and NO reconcile + * should run — the live edits stay in the stream and a later persist retries once the version is + * re-established. Distinct from `conflict`, which signals a genuine out-of-band change to merge. + */ + status: z.literal('deferred'), + }), z.object({ /** * The file changed out-of-band since `expectedVersion` — the write was refused to avoid clobbering diff --git a/apps/sim/lib/collab-doc/persist.ts b/apps/sim/lib/collab-doc/persist.ts index 7481108b4cd..ef3ecb9d991 100644 --- a/apps/sim/lib/collab-doc/persist.ts +++ b/apps/sim/lib/collab-doc/persist.ts @@ -26,6 +26,7 @@ export type PersistFileDocResult = | { status: 'persisted'; version: number } | { status: 'missing' } | { status: 'conflict'; markdown: string; version: number } + | { status: 'deferred' } /** * Project a live collaborative document back to durable markdown and write it to the file. The realtime @@ -53,17 +54,13 @@ export async function persistFileDoc( if (!record) return { status: 'missing' } // Optimistic concurrency needs a version for any file that already has content. If none was supplied - // (the relay's synced version was momentarily unavailable — e.g. a Redis blip on a peer-seeded task), - // refuse to overwrite non-empty durable content unconditionally: return `conflict` so the relay - // reconciles/retries once the version is re-established, rather than silently clobbering. An empty - // file has nothing to clobber, so its first unconditional write stays allowed. + // (the relay's synced-version token was momentarily unavailable — e.g. a Redis blip on a peer-seeded + // task), DEFER rather than write: an unconditional write could clobber an out-of-band edit, and a + // reconcile would wipe live edits even when nothing actually changed out-of-band (the version was + // merely missing). The live edits stay in the stream; a later persist writes them once the version is + // re-established. An empty file has nothing to clobber, so its first unconditional write stays allowed. if (expectedVersion === undefined && record.size > 0) { - const currentBuffer = await fetchWorkspaceFileBuffer(record) - return { - status: 'conflict', - markdown: currentBuffer.toString('utf-8'), - version: record.updatedAt.getTime(), - } + return { status: 'deferred' } } const ydoc = new Y.Doc() From 952393ef3af8e3da3ab2c618c701df2ade53fd70 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 29 Jul 2026 20:27:25 -0700 Subject: [PATCH 06/18] fix(collab-doc): make persist If-Match teardown-race-immune and recover missing version on final flush MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Close two last-leave concurrency holes Cursor flagged: - Thread the reconciled version LOCALLY through the persist retry loop. After a conflict+reconcile the correct next If-Match is exactly result.version, so carry it in a local var instead of re-deriving from room.syncedVersion/Redis. On a last-leave flush destroyRoomIfIdle removes the room from the map before the async flush finishes, so mergeMarkdownIntoRoom's recordVersion can no longer update room.syncedVersion — threading makes each retry's precondition correct by construction, immune to that dropped mutation and to a best-effort Redis re-read. - Cache the resolved version back into room.syncedVersion in currentVersion() so a peer-seeded/tail-only task (which never sets it locally) or a later transient Redis read failure still resolves it from the last value seen (monotonic max, never regresses). - On a FINAL flush, briefly retry resolving the If-Match when the version read momentarily fails, rather than deferring and stranding the session's edits in the TTL'd stream — the version is cluster-wide and heartbeat-refreshed. --- apps/realtime/src/handlers/file-doc.ts | 47 +++++++++++++++++++++----- 1 file changed, 39 insertions(+), 8 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index 4b30ee1ee71..97f73e8ede7 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -69,6 +69,13 @@ const PERSIST_MAX_WAIT_MS = 20_000 * gives up (leaving the durable content authoritative). One reconcile normally converges; the small * budget covers a second racing write. */ const PERSIST_CONFLICT_RETRIES = 2 +/** On a FINAL (last-leave/shutdown) flush the version read is the last chance to resolve the If-Match + * before the room is torn down; a transient Redis blip yielding no version would otherwise defer and + * strand the session's edits in the TTL'd stream. The version is cluster-wide and heartbeat-refreshed, so + * a brief bounded retry recovers a blip without stalling teardown (a genuinely-unset version never + * appears, so a long wait buys nothing). */ +const FINAL_VERSION_RETRIES = 2 +const FINAL_VERSION_RETRY_MS = 100 /** 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 @@ -271,10 +278,14 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr // The If-Match token: the FRESHEST synced version known — the max of the cluster-wide value (Redis) // and this task's room value. Taking the max (not Redis-preferred) avoids a stale read shadowing a // newer local value when a fire-and-forget `setSyncedVersion` to Redis lagged or failed; versions are - // monotonic epoch-ms, so the larger is the later sync point. `undefined` when neither is known. + // monotonic epoch-ms, so the larger is the later sync point. The resolved value is CACHED back into the + // room so a later transient Redis read failure — or a peer-seeded/tail-only task that never set the + // version locally — still resolves it from the last value we saw (caching the max never regresses). + // `undefined` when neither source has ever yielded a version. const currentVersion = async (): Promise => { const shared = store.enabled ? await store.getSyncedVersion(name) : null const best = Math.max(shared ?? 0, room.syncedVersion ?? 0) + if (best > 0) room.syncedVersion = best return best > 0 ? best : undefined } @@ -282,6 +293,28 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr if (!final && !(await store.tryClaimPersistWindow(name, FILE_DOC_TIMEOUTS.persistRequestMs))) return + // The If-Match token for the write, resolved ONCE up front and then advanced LOCALLY as we reconcile + // — never re-derived from `room.syncedVersion`/Redis mid-loop. On a last-leave flush the room is + // removed from `fileDocRooms` before this async flush finishes, so `mergeMarkdownIntoRoom`'s + // `recordVersion` can no longer write `room.syncedVersion`; carrying the reconciled version here makes + // each retry's precondition correct by construction rather than depending on that (now-dropped) + // mutation or a best-effort Redis re-read. + let ifMatch = await currentVersion() + // On a FINAL flush this is the last chance to persist before teardown; if the shared version read + // momentarily fails (a Redis blip) for a peer-seeded/tail-only task that never cached it locally, + // retry briefly rather than deferring and stranding the session's edits in the (TTL'd) stream — the + // version is cluster-wide and heartbeat-refreshed, so it is there to be read. Bounded small: a + // genuinely-unset version won't appear no matter how long we wait, and the flush must not stall + // teardown/shutdown. + for ( + let i = 0; + ifMatch === undefined && final && store.enabled && i < FINAL_VERSION_RETRIES; + i++ + ) { + await sleep(FINAL_VERSION_RETRY_MS) + ifMatch = await currentVersion() + } + // Persist under optimistic concurrency (RFC 7232 If-Match): the write commits only if the file is // still at the version the live doc synced from, so a projection can never silently clobber an // out-of-band edit. On a conflict, merge the current durable content into the live doc — so the @@ -290,13 +323,7 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr for (let attempt = 0; attempt <= PERSIST_CONFLICT_RETRIES; attempt++) { const docState = await captureState() if (!docState) return // nothing seeded/authoritative to persist yet - const result = await fetchFileDocPersist( - workspaceId, - room.fileId, - userId, - docState, - await currentVersion() - ) + const result = await fetchFileDocPersist(workspaceId, room.fileId, userId, docState, ifMatch) if (result.status === 'missing') return if (result.status === 'deferred') { // The app couldn't persist safely without a version token (momentarily unavailable). Do NOT @@ -332,6 +359,10 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr ) return } + // The out-of-band content is now reconciled into the live doc AT `result.version`, so that is the + // correct If-Match for the re-projection — advance to it directly rather than re-reading a version + // a concurrent teardown may have left stale in `room.syncedVersion`/Redis. + ifMatch = result.version } } catch (error) { logger.warn(`Persist failed for file ${room.fileId}`, { error: getErrorMessage(error) }) From d7a41ed9c56322b7aff94d92de841969904df482 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 29 Jul 2026 20:35:27 -0700 Subject: [PATCH 07/18] fix(collab-doc): stamp cluster sync version the moment the seed wins, before the liveness guard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The winning seeder set the If-Match token (room + Redis filedoc:syncver) only after the liveness/seeded guard that follows seedIfEmpty. But the tailer can integrate the just-appended seed DURING the seedIfEmpty await, so isDocSeeded(room.doc) is already true when the guard runs and it returns early — leaving the stream holding seed content with no cluster version. Later persists then send no If-Match, the app returns `deferred`, and session edits stay only in the TTL'd stream (the exact stranding this PR prevents elsewhere). Move the version stamp to immediately after seedIfEmpty wins, before the guard. Recording it only once our seed won (not from the fetch) is preserved, so it still can't shadow a peer's winning seed. --- apps/realtime/src/handlers/file-doc.ts | 23 ++++++++++++++--------- 1 file changed, 14 insertions(+), 9 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index 97f73e8ede7..7d6edae725a 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -499,18 +499,23 @@ async function ensureServerSeed( // re-publishing it. const seedUpdate = seed?.update ?? emptySeedUpdate() const didSeed = await store.seedIfEmpty(name, seedUpdate) + // Record the durable version the moment THIS task's seed is in the stream — BEFORE the liveness/ + // seeded guard below. Recording it only now that our seed WON (not from the fetch, before knowing who + // won) keeps it in step with the stream's actual content: a newer own-fetch version could otherwise + // shadow a peer's winning seed and let a later persist clobber an out-of-band edit. But it must not + // sit AFTER the guard: the tailer can integrate our just-appended seed during the await above, so + // `isDocSeeded` may already be true here — an early return would then leave the stream holding seed + // content with NO cluster If-Match token, and later persists would defer and strand session edits. + // Cluster-wide (Redis) so any task's persist reads it; the live room is the single-pod fallback / the + // read-through-cache seed. (No version for an empty/missing file — nothing durable to guard.) + if (didSeed && seed) { + const live = fileDocRooms.get(name) + if (live) live.syncedVersion = seed.version + void store.setSyncedVersion(name, seed.version) + } if (fileDocRooms.get(name) !== room || isDocSeeded(room.doc)) return if (didSeed) { Y.applyUpdate(room.doc, seedUpdate, SEED_ORIGIN) - // Record the durable version ONLY now that THIS task's seed won — so the synced version reflects - // the stream's actual content. Setting it from our own fetch before knowing who won could record a - // newer version than a peer's winning seed and let a later persist clobber an out-of-band edit. - // Cluster-wide (Redis) so any task's persist reads it; local room is the single-pod fallback. A - // peer-seeded task leaves it null and reads the winner's cluster value. (Null for an empty file.) - if (seed) { - room.syncedVersion = seed.version - void store.setSyncedVersion(name, seed.version) - } } else { // A peer seeded first: its seed arrives via the tailer, so we must NOT apply our own — a second, // different-client-id seed IS the split-brain. Clear the guard so a later join can retry if that From e9ced7e65d8e056706ad9b5c4502063ca76e9539 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 29 Jul 2026 20:40:13 -0700 Subject: [PATCH 08/18] fix(collab-doc): make the synced-version token monotonic at every write site MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The If-Match token is written fire-and-forget from the seed stamp, merges, and persists, both locally and to Redis. An out-of-order write (e.g. a seed's lagged setSyncedVersion landing after a later merge's) could regress it below the version the live doc already incorporates, causing spurious If-Match conflicts — and on a last-leave flush with no live room to reconcile into, a spurious conflict leaves durable authoritative and drops the session's edits. - setSyncedVersion now writes via SET_VERSION_IF_NEWER_SCRIPT (Redis-side compare-and-set): it overwrites only when the new value is greater, refreshing the TTL either way. - recordVersion / the persisted branch / the seed stamp all take Math.max instead of assigning room.syncedVersion directly. Versions are monotonic epoch-ms, so "newer" is a plain numeric compare, exact within a Lua double. --- apps/realtime/src/handlers/file-doc-store.ts | 23 +++++++++++++++++--- apps/realtime/src/handlers/file-doc.ts | 9 +++++--- 2 files changed, 26 insertions(+), 6 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc-store.ts b/apps/realtime/src/handlers/file-doc-store.ts index 66d02765374..9947db7f494 100644 --- a/apps/realtime/src/handlers/file-doc-store.ts +++ b/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 @@ -467,12 +479,17 @@ export class FileDocStore { } } - /** Record the durable version the shared live doc is now synced to. 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). */ + /** 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 { if (!this.enabled || !this.write) return await this.write - .set(`${SYNC_VERSION_PREFIX}${name}`, String(version), { EX: STREAM_TTL_SEC }) + .eval(SET_VERSION_IF_NEWER_SCRIPT, { + keys: [`${SYNC_VERSION_PREFIX}${name}`], + arguments: [String(version), String(STREAM_TTL_SEC)], + }) .catch(() => {}) } diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index 7d6edae725a..fa7c8e5777c 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -333,7 +333,7 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr return } if (result.status === 'persisted') { - room.syncedVersion = result.version + room.syncedVersion = Math.max(room.syncedVersion ?? 0, result.version) void store.setSyncedVersion(name, result.version) return } @@ -510,7 +510,7 @@ async function ensureServerSeed( // read-through-cache seed. (No version for an empty/missing file — nothing durable to guard.) if (didSeed && seed) { const live = fileDocRooms.get(name) - if (live) live.syncedVersion = seed.version + if (live) live.syncedVersion = Math.max(live.syncedVersion ?? 0, seed.version) void store.setSyncedVersion(name, seed.version) } if (fileDocRooms.get(name) !== room || isDocSeeded(room.doc)) return @@ -599,7 +599,10 @@ async function mergeMarkdownIntoRoom( const recordVersion = () => { if (version === undefined) return const room = fileDocRooms.get(name) - if (room) room.syncedVersion = version + // Never regress the token: merges/seeds/persists all write it (locally and via fire-and-forget + // Redis), so a lower value arriving out of order must not shadow a higher one the doc already + // incorporates (the Redis side is guarded identically by SET_VERSION_IF_NEWER_SCRIPT). + if (room) room.syncedVersion = Math.max(room.syncedVersion ?? 0, version) void store.setSyncedVersion(name, version) } From 0f7a0815ad4edccaf7c11a8e95217c0481b4d89f Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 29 Jul 2026 20:50:09 -0700 Subject: [PATCH 09/18] fix(collab-doc): close three last-leave persist edge cases from review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Stale snapshot after reconcile (High): the multi-task captureState fell back to the pre-await localState snapshot even after a reconcile advanced ifMatch, so a failed stream re-read could persist the pre-reconcile state against the new version and clobber the out-of-band edit the reconcile just incorporated. NULL localState after a reconcile so a failed read aborts instead. - Lock miss aborts reconcile (Medium): a merge-lock acquisition failure returned 'no-live-room', indistinguishable from an absent stream, so flushPersist treated transient contention as terminal. Return a distinct 'merge-unavailable' and handle it as retry-later (edits stay in the stream), never as "nothing to reconcile into". - Peer syncver never recovers (Medium): the winner's setSyncedVersion was fire-and-forget with swallowed errors — the only way a peer-seeded task learns the durable version — so a dropped write left that peer deferring forever. Make it retry (bounded) like appendUpdate/seedIfEmpty; the monotonic script keeps a racing retry a no-op. --- apps/realtime/src/handlers/file-doc-store.ts | 27 +++++++++++++---- apps/realtime/src/handlers/file-doc.ts | 32 ++++++++++++++++---- 2 files changed, 47 insertions(+), 12 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc-store.ts b/apps/realtime/src/handlers/file-doc-store.ts index 9947db7f494..b6cf65ce9e5 100644 --- a/apps/realtime/src/handlers/file-doc-store.ts +++ b/apps/realtime/src/handlers/file-doc-store.ts @@ -485,12 +485,27 @@ export class FileDocStore { * file's key can't outlive its room. No-op when disabled (single-pod fallback). */ async setSyncedVersion(name: string, version: number): Promise { if (!this.enabled || !this.write) return - await this.write - .eval(SET_VERSION_IF_NEWER_SCRIPT, { - keys: [`${SYNC_VERSION_PREFIX}${name}`], - arguments: [String(version), String(STREAM_TTL_SEC)], - }) - .catch(() => {}) + // 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 { diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index fa7c8e5777c..415b3db3631 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -248,7 +248,9 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr const userId = room.lastEditorUserId // Synchronous fallback capture — before any await, since the caller may destroy `room.doc` the moment // 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 + // NULLED after a reconcile (below): once we've merged an out-of-band edit in, this pre-reconcile + // snapshot predates it and must never be persisted as a fallback, or it would re-drop that edit. + let localState = isDocSeeded(room.doc) ? Y.encodeStateAsUpdate(room.doc) : null // Capture the AUTHORITATIVE doc state: the shared stream when enabled (a copilot merge or a peer's // edit published by another task may not be integrated into THIS task's `room.doc` yet), else the @@ -267,7 +269,9 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr return (await store.getStreamState(name)) ?? localState } catch (streamError) { // A transient Redis read must NOT drop the write when we already hold a valid local snapshot — - // else the last-disconnect flush loses the session's edits as the room is torn down. + // else the last-disconnect flush loses the session's edits as the room is torn down. But once a + // reconcile has run, `localState` is NULLED (it predates the merged-in out-of-band edit), so a + // failed read then correctly THROWS and aborts rather than clobbering with the stale snapshot. if (!localState) throw streamError logger.warn(`Stream state unavailable for file ${room.fileId}; persisting local snapshot`, { error: getErrorMessage(streamError), @@ -359,9 +363,22 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr ) return } + if (reconciled === 'merge-unavailable') { + // Merge-lock contention (NOT an absent stream): the reconcile could not run, so we have not + // incorporated the out-of-band edit and must not re-persist against it. Stop without clobbering; + // the session's edits stay in the stream and a later flush (or the next opener) retries. Distinct + // from `no-live-room` so a transient lock miss is never mistaken for "nothing to reconcile". + logger.warn( + `Persist conflict for file ${room.fileId}: merge lock unavailable, reconcile deferred (edits remain in stream)` + ) + return + } // The out-of-band content is now reconciled into the live doc AT `result.version`, so that is the // correct If-Match for the re-projection — advance to it directly rather than re-reading a version - // a concurrent teardown may have left stale in `room.syncedVersion`/Redis. + // a concurrent teardown may have left stale in `room.syncedVersion`/Redis. The pre-await + // `localState` snapshot now predates this reconcile, so drop it: a subsequent failed stream read + // must abort rather than fall back to a snapshot that would re-drop the just-merged out-of-band edit. + localState = null ifMatch = result.version } } catch (error) { @@ -564,12 +581,15 @@ const fileDocMergeChains = new Map>() * * Returns `'no-live-room'` when there is no shared state to merge against (no doc is or was recently * live): the caller (copilot) writes the file directly and the next open seeds from that markdown. + * Returns `'merge-unavailable'` when the cross-task merge lock could not be acquired (transient + * contention, not an absent stream) — the caller likewise falls back to a direct write, but a persist + * reconcile treats it distinctly (retry later) rather than as "nothing to reconcile into". */ export function applyMarkdownToLiveFileDoc( fileId: string, markdown: string, version?: number -): Promise<'applied' | 'no-live-room'> { +): Promise<'applied' | 'no-live-room' | 'merge-unavailable'> { 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. @@ -590,7 +610,7 @@ async function mergeMarkdownIntoRoom( fileId: string, markdown: string, version?: number -): Promise<'applied' | 'no-live-room'> { +): Promise<'applied' | 'no-live-room' | 'merge-unavailable'> { const store = getFileDocStore() // The durable version this merge carries is now incorporated in the live doc — record it (cluster-wide @@ -620,7 +640,7 @@ async function mergeMarkdownIntoRoom( } if (!token) { logger.warn(`Merge lock unavailable for file ${fileId}; skipping live merge`) - return 'no-live-room' + return 'merge-unavailable' } try { // Compute the diff against the committed SHARED state and PUBLISH it — every task with the doc From cf8802fe0f66312cc428aef7b2e0837a3ec396fc Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 29 Jul 2026 21:39:10 -0700 Subject: [PATCH 10/18] fix(collab-doc): scope the persist If-Match to a content version so metadata bumps can't clobber edits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The optimistic-concurrency validator was `updatedAt`, which rename/move/delete/restore also bump with no content change. A racing live-doc persist then saw a stale token, got `conflict`, reconciled the pre-edit durable body via updateYFragment (incoming-wins on overlap), and wiped the user's in-flight edits. Scope the validator to content (RFC 7232 semantics — validate the representation, not the row): - New `workspace_files.content_updated_at` (NOT NULL, `now()` fast-default — no table rewrite). Advances ONLY on content writes (upload / overwrite / create); metadata writes never touch it. - The FOR UPDATE CAS, the merge-notify version, and the seed version all use `content_updated_at`. A rename now leaves it unchanged, so the persist If-Match still matches -> no spurious conflict, no reconcile, no lost edits. Genuine out-of-band content writes still conflict and reconcile. - Consolidated the collab schema into one migration (the collab-state table + the new column) per request, rather than a separate follow-up migration. Relay/store/contracts unchanged (still a numeric monotonic version). --- apps/sim/lib/collab-doc/seed.ts | 5 ++- .../workspace/workspace-file-manager.ts | 37 ++++++++++++++----- .../workspace-file-storage-accounting.test.ts | 1 + ..._collab_doc_state_and_content_version.sql} | 1 + .../db/migrations/meta/0277_snapshot.json | 9 ++++- packages/db/migrations/meta/_journal.json | 4 +- packages/db/schema.ts | 11 ++++++ 7 files changed, 55 insertions(+), 13 deletions(-) rename packages/db/migrations/{0277_workspace_file_collab_state.sql => 0277_collab_doc_state_and_content_version.sql} (78%) diff --git a/apps/sim/lib/collab-doc/seed.ts b/apps/sim/lib/collab-doc/seed.ts index fdf4f0d578e..ece65549bf0 100644 --- a/apps/sim/lib/collab-doc/seed.ts +++ b/apps/sim/lib/collab-doc/seed.ts @@ -46,7 +46,10 @@ export async function buildFileDocSeed( const record = await getWorkspaceFile(workspaceId, fileId, { throwOnError: true }) if (!record) return null - const version = record.updatedAt.getTime() + // The content-scoped version (advances only on content writes, never on rename/move) is the persist + // If-Match token — so a metadata bump can't make a racing persist reconcile stale content and clobber + // live edits. `getWorkspaceFile` always maps it from the NOT NULL column; coalesce is a type guard only. + const version = (record.contentUpdatedAt ?? record.updatedAt).getTime() const buffer = await fetchWorkspaceFileBuffer(record, { maxBytes: MAX_SEED_BYTES }) // Cold-start fast path: if we hold a cached Yjs binary derived from THIS exact markdown, apply it diff --git a/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts b/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts index 4776949d152..3e08358bc9c 100644 --- a/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts +++ b/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts @@ -73,6 +73,13 @@ export interface WorkspaceFileRecord { deletedAt?: Date | null uploadedAt: Date updatedAt: Date + /** + * Content-scoped version (see the `content_updated_at` column): advances only on content writes, never + * on metadata. The collab persist's optimistic-concurrency validator. Optional on the DTO because + * display/mothership constructors don't carry it — the real DB mapper (the only source seed/persist + * read) always populates it from the NOT NULL column. + */ + contentUpdatedAt?: Date | null /** Pass-through to `downloadFile` when not default `workspace` (e.g. chat mothership uploads). */ storageContext?: 'workspace' | 'mothership' /** Public share state, attached at the API boundary. `null` when never shared. */ @@ -165,6 +172,8 @@ async function insertWorkspaceFileMetadataInTx( deletedAt: null, uploadedAt: new Date(), updatedAt: new Date(), + // Creation IS the first content write, so stamp the content version too (metadata writes never do). + contentUpdatedAt: new Date(), }) .onConflictDoNothing() .returning() @@ -701,6 +710,7 @@ function mapWorkspaceFileRecord( deletedAt: file.deletedAt, uploadedAt: file.uploadedAt, updatedAt: file.updatedAt, + contentUpdatedAt: file.contentUpdatedAt, } } @@ -1116,24 +1126,31 @@ export async function updateWorkspaceFileContent( } // Optimistic-concurrency guard: the row is `FOR UPDATE`-locked, so comparing its committed - // `updatedAt` to the caller's expected value and then writing is atomic — a racing writer blocks - // here until this transaction resolves. A mismatch means the file changed out-of-band since the - // caller last synced; abort rather than clobber it. + // CONTENT version to the caller's expected value and then writing is atomic — a racing writer + // blocks here until this transaction resolves. Compare `contentUpdatedAt` (which advances only on + // content writes), NOT `updatedAt` (which a rename/move also bumps): guarding on `updatedAt` let a + // metadata bump masquerade as an out-of-band CONTENT change, so a racing live-doc persist would + // reconcile stale durable content and clobber in-flight edits. Coalesce to `updatedAt` for rows + // predating the column. A mismatch means the CONTENT changed out-of-band; abort rather than clobber. if ( options?.expectedUpdatedAt && - currentFile.updatedAt.getTime() !== options.expectedUpdatedAt.getTime() + currentFile.contentUpdatedAt.getTime() !== options.expectedUpdatedAt.getTime() ) { throw new ContentVersionConflictError(fileId) } const sizeDiff = content.length - currentFile.size + // This IS a content write, so advance `contentUpdatedAt` in lockstep with `updatedAt` (same + // instant) — the new content version the collab relay records as its If-Match token. + const writeTimestamp = new Date() const [updatedFile] = await tx .update(workspaceFiles) .set({ key: uploadResult.key, size: content.length, contentType: nextContentType, - updatedAt: new Date(), + updatedAt: writeTimestamp, + contentUpdatedAt: writeTimestamp, }) .where( and( @@ -1195,13 +1212,14 @@ export async function updateWorkspaceFileContent( options?.syncLiveDoc !== false && isMarkdownFile({ type: nextContentType, name: finalized.file.originalName }) ) { - // Pass the new `updatedAt` as the version this write produced, so the relay records that its live - // doc now incorporates this durable version — the collab persist's optimistic-concurrency guard - // then won't treat this (already-merged) write as an out-of-band conflict. + // Pass the new CONTENT version this write produced, so the relay records that its live doc now + // incorporates this durable version — the collab persist's optimistic-concurrency guard then won't + // treat this (already-merged) write as an out-of-band conflict. Must be the SAME field the CAS + // guards on (`contentUpdatedAt`), not `updatedAt`, or the relay's token wouldn't match the CAS. await mergeEditIntoLiveFileDoc( fileId, content.toString('utf-8'), - finalized.file.updatedAt.getTime() + finalized.file.contentUpdatedAt.getTime() ) } @@ -1225,6 +1243,7 @@ export async function updateWorkspaceFileContent( deletedAt: finalized.file.deletedAt, uploadedAt: finalized.file.uploadedAt, updatedAt: finalized.file.updatedAt, + contentUpdatedAt: finalized.file.contentUpdatedAt, } } catch (error) { // Preserve the typed conflict so callers can catch it and reconcile — it's an expected outcome of diff --git a/apps/sim/lib/uploads/contexts/workspace/workspace-file-storage-accounting.test.ts b/apps/sim/lib/uploads/contexts/workspace/workspace-file-storage-accounting.test.ts index 7a7f1aa6cc1..a68da9ce5f5 100644 --- a/apps/sim/lib/uploads/contexts/workspace/workspace-file-storage-accounting.test.ts +++ b/apps/sim/lib/uploads/contexts/workspace/workspace-file-storage-accounting.test.ts @@ -100,6 +100,7 @@ const FILE_ROW = { deletedAt: null, uploadedAt: new Date('2026-07-01T00:00:00.000Z'), updatedAt: new Date('2026-07-01T00:00:00.000Z'), + contentUpdatedAt: new Date('2026-07-01T00:00:00.000Z'), } describe('workspace file metadata and storage accounting', () => { diff --git a/packages/db/migrations/0277_workspace_file_collab_state.sql b/packages/db/migrations/0277_collab_doc_state_and_content_version.sql similarity index 78% rename from packages/db/migrations/0277_workspace_file_collab_state.sql rename to packages/db/migrations/0277_collab_doc_state_and_content_version.sql index a40748242cc..0ff597bcc9c 100644 --- a/packages/db/migrations/0277_workspace_file_collab_state.sql +++ b/packages/db/migrations/0277_collab_doc_state_and_content_version.sql @@ -5,4 +5,5 @@ CREATE TABLE "workspace_file_collab_state" ( "updated_at" timestamp DEFAULT now() NOT NULL ); --> statement-breakpoint +ALTER TABLE "workspace_files" ADD COLUMN "content_updated_at" timestamp DEFAULT now() NOT NULL;--> statement-breakpoint ALTER TABLE "workspace_file_collab_state" ADD CONSTRAINT "workspace_file_collab_state_file_id_workspace_files_id_fk" FOREIGN KEY ("file_id") REFERENCES "public"."workspace_files"("id") ON DELETE cascade ON UPDATE no action; \ No newline at end of file diff --git a/packages/db/migrations/meta/0277_snapshot.json b/packages/db/migrations/meta/0277_snapshot.json index 66b90a36714..68f19300ddf 100644 --- a/packages/db/migrations/meta/0277_snapshot.json +++ b/packages/db/migrations/meta/0277_snapshot.json @@ -1,5 +1,5 @@ { - "id": "27607581-9ffb-4b7d-89a5-eaebe9f7d0d1", + "id": "970d87c2-7d8e-429a-ae8f-e91155ff6eb0", "prevId": "100afbc7-1a7f-4f05-9950-b2dbd6996c15", "version": "7", "dialect": "postgresql", @@ -17056,6 +17056,13 @@ "primaryKey": false, "notNull": true, "default": "now()" + }, + "content_updated_at": { + "name": "content_updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" } }, "indexes": { diff --git a/packages/db/migrations/meta/_journal.json b/packages/db/migrations/meta/_journal.json index a0b2463aa2e..0f5621c48e1 100644 --- a/packages/db/migrations/meta/_journal.json +++ b/packages/db/migrations/meta/_journal.json @@ -1937,8 +1937,8 @@ { "idx": 277, "version": "7", - "when": 1785366320079, - "tag": "0277_workspace_file_collab_state", + "when": 1785385489164, + "tag": "0277_collab_doc_state_and_content_version", "breakpoints": true } ] diff --git a/packages/db/schema.ts b/packages/db/schema.ts index 39eb0c2e7e6..8dc613492cf 100644 --- a/packages/db/schema.ts +++ b/packages/db/schema.ts @@ -1906,6 +1906,17 @@ export const workspaceFiles = pgTable( deletedAt: timestamp('deleted_at'), uploadedAt: timestamp('uploaded_at').notNull().defaultNow(), updatedAt: timestamp('updated_at').notNull().defaultNow(), + /** + * Content-scoped version: advances ONLY when the file's CONTENT changes (upload / content + * overwrite), never on metadata writes (rename, move, soft-delete, restore). It is the + * optimistic-concurrency validator the collaborative-document persist guards on (RFC 7232 `If-Match` + * semantics: validate the representation, not the row) — so a rename can't make a racing live-doc + * persist see a stale token, reconcile stale durable content, and clobber in-flight edits. NOT NULL + * with a `now()` default: Postgres applies this as a fast-default (no table rewrite), existing rows + * get a stable timestamp that — like every metadata write — never advances it, and every insert path + * is covered without per-call plumbing. Only a content write (upload / overwrite) advances it. + */ + contentUpdatedAt: timestamp('content_updated_at').notNull().defaultNow(), }, (table) => ({ keyActiveUniqueIdx: uniqueIndex('workspace_files_key_active_unique') From ffb9ec2d29b090b6785dbba6982f5a9b8712e2f8 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 29 Jul 2026 21:43:50 -0700 Subject: [PATCH 11/18] chore(collab-doc): condense the densest persist comments (no behavior change) Cleanup pass: tighten the three longest comment blocks added while hardening the persist path (currentVersion cache, ifMatch threading, final-flush version retry) without dropping any invariant. No dead code found (biome lint clean; all new symbols referenced). --- apps/realtime/src/handlers/file-doc.ts | 31 ++++++++++---------------- 1 file changed, 12 insertions(+), 19 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index 415b3db3631..54a4a4303dc 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -279,13 +279,10 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr return localState } } - // The If-Match token: the FRESHEST synced version known — the max of the cluster-wide value (Redis) - // and this task's room value. Taking the max (not Redis-preferred) avoids a stale read shadowing a - // newer local value when a fire-and-forget `setSyncedVersion` to Redis lagged or failed; versions are - // monotonic epoch-ms, so the larger is the later sync point. The resolved value is CACHED back into the - // room so a later transient Redis read failure — or a peer-seeded/tail-only task that never set the - // version locally — still resolves it from the last value we saw (caching the max never regresses). - // `undefined` when neither source has ever yielded a version. + // The If-Match token: the freshest synced version known — max of the cluster value (Redis) and the + // room's, so a lagging fire-and-forget `setSyncedVersion` can't let a stale read shadow a newer local + // value (versions are monotonic epoch-ms). Cached back into the room so a later transient Redis failure + // — or a peer-seeded/tail-only task that never set it locally — still resolves it. `undefined` if unknown. const currentVersion = async (): Promise => { const shared = store.enabled ? await store.getSyncedVersion(name) : null const best = Math.max(shared ?? 0, room.syncedVersion ?? 0) @@ -297,19 +294,15 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr if (!final && !(await store.tryClaimPersistWindow(name, FILE_DOC_TIMEOUTS.persistRequestMs))) return - // The If-Match token for the write, resolved ONCE up front and then advanced LOCALLY as we reconcile - // — never re-derived from `room.syncedVersion`/Redis mid-loop. On a last-leave flush the room is - // removed from `fileDocRooms` before this async flush finishes, so `mergeMarkdownIntoRoom`'s - // `recordVersion` can no longer write `room.syncedVersion`; carrying the reconciled version here makes - // each retry's precondition correct by construction rather than depending on that (now-dropped) - // mutation or a best-effort Redis re-read. + // The If-Match token, resolved ONCE and then advanced LOCALLY as we reconcile — never re-derived + // mid-loop. On a last-leave flush the room is already removed from `fileDocRooms`, so + // `mergeMarkdownIntoRoom`'s `recordVersion` can't write `room.syncedVersion`; carrying the reconciled + // version here keeps each retry's precondition correct without depending on that mutation or a Redis re-read. let ifMatch = await currentVersion() - // On a FINAL flush this is the last chance to persist before teardown; if the shared version read - // momentarily fails (a Redis blip) for a peer-seeded/tail-only task that never cached it locally, - // retry briefly rather than deferring and stranding the session's edits in the (TTL'd) stream — the - // version is cluster-wide and heartbeat-refreshed, so it is there to be read. Bounded small: a - // genuinely-unset version won't appear no matter how long we wait, and the flush must not stall - // teardown/shutdown. + // FINAL flush = last chance before teardown: if the version read momentarily fails (Redis blip) for a + // peer-seeded/tail-only task that never cached it, retry briefly rather than defer and strand the + // edits in the TTL'd stream (the version is cluster-wide + heartbeat-refreshed). Bounded — a genuinely + // unset version never appears, and the flush must not stall teardown. for ( let i = 0; ifMatch === undefined && final && store.enabled && i < FINAL_VERSION_RETRIES; From 5de2f033865509c215c9383f45b607c8730051ab Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 29 Jul 2026 21:53:25 -0700 Subject: [PATCH 12/18] fix(collab-doc): persist must return the content version, not updatedAt MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to the content-scoped If-Match: persistFileDoc still returned `updatedAt` as the version in both the persisted and conflict results, while the CAS/seed/merge all guard on `content_updated_at`. A content write sets both to the same instant, so it was coincidentally correct — until they diverge: if a metadata write bumps `updatedAt` past `content_updated_at`, the conflict path returned the larger `updatedAt`, so the relay's re-persist sent an If-Match the CAS (which checks `content_updated_at`) could never match → perpetual conflict → dropped reconciled edits. Return `contentUpdatedAt` in both paths so the relay's token always matches what it's checked against. --- apps/sim/lib/collab-doc/persist.ts | 21 ++++++++++++++++----- 1 file changed, 16 insertions(+), 5 deletions(-) diff --git a/apps/sim/lib/collab-doc/persist.ts b/apps/sim/lib/collab-doc/persist.ts index ef3ecb9d991..d38ee0e3f1b 100644 --- a/apps/sim/lib/collab-doc/persist.ts +++ b/apps/sim/lib/collab-doc/persist.ts @@ -15,7 +15,7 @@ const logger = createLogger('FileDocPersist') /** * Outcome of a persist attempt: * - `persisted` — the live doc was projected to markdown and written; `version` is the new durable - * `updatedAt` (epoch ms) the relay records as the version its live doc is now synced to. + * CONTENT version (`content_updated_at`, epoch ms) the relay records as what its live doc is synced to. * - `missing` — the file is gone (deleted); nothing to write. * - `conflict` — the file changed out-of-band since the relay's live doc last synced, so writing the * projection would clobber that change (RFC 7232 `If-Match` failure). NOT written. `markdown` + @@ -34,8 +34,9 @@ export type PersistFileDocResult = * here and the app persists it — the server-authoritative durable path that replaces the editor's * client-side autosave. * - * `expectedVersion` (the durable `updatedAt`, epoch ms, that the relay's live doc last synced from) is - * the optimistic-concurrency guard: the write commits only if the file is still at that version, so a + * `expectedVersion` (the durable CONTENT version, `content_updated_at` epoch ms, the relay's live doc last + * synced from) is the optimistic-concurrency guard: the write commits only if the file is still at that + * content version — a rename/move that only bumps `updatedAt` won't trip it, so a * projection built from a stale live doc can never silently overwrite an out-of-band edit. On a version * mismatch this returns `conflict` (with the current durable content) instead of writing — the relay * reconciles and retries. Omit `expectedVersion` to write unconditionally (e.g. the first persist, @@ -100,7 +101,14 @@ export async function persistFileDoc( logger.info( `Persisted live collaborative document to file ${fileId} (workspace ${workspaceId})` ) - return { status: 'persisted', version: updated.updatedAt.getTime() } + // Return the CONTENT version (what the CAS/seed/merge all guard on), not `updatedAt` — the relay + // records this as its new If-Match token, so it must be the same field a later persist is checked + // against. (A content write sets both to the same instant; using the wrong one only bites once they + // diverge — e.g. a metadata write bumping `updatedAt` afterward.) + return { + status: 'persisted', + version: (updated.contentUpdatedAt ?? updated.updatedAt).getTime(), + } } catch (error) { if (!(error instanceof ContentVersionConflictError)) throw error // Out-of-band edit since the live doc last synced — DON'T clobber. Return the current durable @@ -112,7 +120,10 @@ export async function persistFileDoc( return { status: 'conflict', markdown: currentBuffer.toString('utf-8'), - version: current.updatedAt.getTime(), + // The CONTENT version, not `updatedAt`: the relay reconciles to this and re-persists with it as the + // If-Match. If a metadata write bumped `updatedAt` past `contentUpdatedAt`, returning `updatedAt` + // would make the re-persist's CAS (which checks `contentUpdatedAt`) never match → perpetual conflict. + version: (current.contentUpdatedAt ?? current.updatedAt).getTime(), } } } From 26525738cdbfc17c0b77f0c61028c2adece07fe6 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 29 Jul 2026 22:06:26 -0700 Subject: [PATCH 13/18] fix(collab-doc): defer persist whenever the version is missing; guard the content-version test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Empty-file CAS race (Medium): the unconditional-write carve-out for size===0 read `record.size` outside the write transaction, so a concurrent first content write could land after the check and be clobbered. With content_updated_at NOT NULL every existing file always has a real version, so a missing expectedVersion is always transient — always defer, never write unconditionally. Removes the TOCTOU hole. - Content-version test (Low): the merge-chokepoint test kept updatedAt == contentUpdatedAt, so it passed even if wired to the wrong field. Mock distinct values and assert contentUpdatedAt, so a regression to updatedAt now fails the test. --- apps/sim/lib/collab-doc/persist.ts | 17 ++++++++++------- .../workspace-file-storage-accounting.test.ts | 11 +++++++++-- 2 files changed, 19 insertions(+), 9 deletions(-) diff --git a/apps/sim/lib/collab-doc/persist.ts b/apps/sim/lib/collab-doc/persist.ts index d38ee0e3f1b..0541d45106b 100644 --- a/apps/sim/lib/collab-doc/persist.ts +++ b/apps/sim/lib/collab-doc/persist.ts @@ -54,13 +54,16 @@ export async function persistFileDoc( const record = await getWorkspaceFile(workspaceId, fileId, { throwOnError: true }) if (!record) return { status: 'missing' } - // Optimistic concurrency needs a version for any file that already has content. If none was supplied - // (the relay's synced-version token was momentarily unavailable — e.g. a Redis blip on a peer-seeded - // task), DEFER rather than write: an unconditional write could clobber an out-of-band edit, and a - // reconcile would wipe live edits even when nothing actually changed out-of-band (the version was - // merely missing). The live edits stay in the stream; a later persist writes them once the version is - // re-established. An empty file has nothing to clobber, so its first unconditional write stays allowed. - if (expectedVersion === undefined && record.size > 0) { + // Optimistic concurrency needs a version. If none was supplied — the relay's synced-version token was + // momentarily unavailable (a Redis blip on a peer-seeded task) — DEFER rather than write: an + // unconditional write could clobber an out-of-band edit, and a reconcile would wipe live edits even + // when nothing changed out-of-band (the version was merely missing). The edits stay in the stream; a + // later persist writes them once the version is re-established. There is deliberately NO empty-file + // unconditional-write carve-out: every existing file has a `content_updated_at`, so the relay always + // has a real version to send and a missing one is always transient — and `record.size` is read outside + // the write transaction, so trusting it (an empty file "has nothing to clobber") is a TOCTOU race a + // concurrent first content write would lose. + if (expectedVersion === undefined) { return { status: 'deferred' } } diff --git a/apps/sim/lib/uploads/contexts/workspace/workspace-file-storage-accounting.test.ts b/apps/sim/lib/uploads/contexts/workspace/workspace-file-storage-accounting.test.ts index a68da9ce5f5..4bf6627c74c 100644 --- a/apps/sim/lib/uploads/contexts/workspace/workspace-file-storage-accounting.test.ts +++ b/apps/sim/lib/uploads/contexts/workspace/workspace-file-storage-accounting.test.ts @@ -345,7 +345,14 @@ describe('workspace file metadata and storage accounting', () => { const MD_ROW = { ...FILE_ROW, originalName: 'note.md', contentType: 'text/markdown' } it('streams a markdown overwrite into any open collaborative editor (the shared merge chokepoint)', async () => { - const updatedFile = { ...MD_ROW, size: 12 } + // Distinct updatedAt vs contentUpdatedAt so the assertion proves the merge carries the CONTENT + // version (the persist If-Match token), not `updatedAt` — reverting that wiring would fail here. + const updatedFile = { + ...MD_ROW, + size: 12, + updatedAt: new Date('2026-07-05T00:00:00.000Z'), + contentUpdatedAt: new Date('2026-07-04T00:00:00.000Z'), + } dbChainMockFns.limit.mockResolvedValueOnce([MD_ROW]).mockResolvedValueOnce([MD_ROW]) dbChainMockFns.returning.mockResolvedValueOnce([updatedFile]) mockUploadFile.mockResolvedValueOnce({ key: MD_ROW.key }) @@ -360,7 +367,7 @@ describe('workspace file metadata and storage accounting', () => { expect(mockMergeEditIntoLiveFileDoc).toHaveBeenCalledWith( MD_ROW.id, '# new content', - updatedFile.updatedAt.getTime() + updatedFile.contentUpdatedAt.getTime() ) }) From 255d28d344406a415520bc1b6cd5c5c224e2cdb3 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 29 Jul 2026 22:20:22 -0700 Subject: [PATCH 14/18] fix(collab-doc): don't reconcile a conflict the live doc already reflects (would wipe newer edits) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit flushPersist reconciled the durable body into the live doc on every conflict. But when the conflict comes from a racing self-persist (or an apply-edit the chokepoint already merged), the durable body is a STALE SUBSET of the live stream, and the incoming-wins updateYFragment merge moves the doc backward — wiping newer in-flight edits, which the retry then persists. Before reconciling, re-check the freshest synced version. If it already covers the conflict version, the live doc has already incorporated that content (or is ahead), so skip the reconcile and just retry with the freshest version as If-Match — the re-projection captures the current live stream, preserving every edit. Only a genuine out-of-band change the live doc hasn't incorporated (freshest < conflict version) is reconciled in. freshest never exceeds the durable version, so this can't loop. --- apps/realtime/src/handlers/file-doc.ts | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index 54a4a4303dc..b5421033b0b 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -340,6 +340,18 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr ) return } + // Only reconcile a GENUINE out-of-band change the live doc hasn't incorporated. If the freshest + // synced version already covers the conflict version, the durable body is a STALE SUBSET of the live + // stream — a racing persist wrote it FROM this stream, or an apply-edit already merged it (the write + // chokepoint advances the synced version). Reconciling it (incoming-wins) would move the doc BACKWARD + // and wipe newer in-flight edits. Skip it and just retry with the freshest version as If-Match; the + // re-projection captures the current live stream, preserving every edit. (`freshest` never exceeds + // the durable version — you can't sync from a version that doesn't exist — so this can't loop.) + const freshest = await currentVersion() + if (freshest !== undefined && freshest >= result.version) { + ifMatch = freshest + continue + } // Reconcile the out-of-band durable content into the live doc (advances the synced version), then // loop to re-project and re-persist the converged state. If there is no live doc to reconcile into // — e.g. the last collaborator has left and the room is being torn down with no shared stream From d001aa3906e6ac08ecb0805edd3c6949ce881436 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 29 Jul 2026 22:34:20 -0700 Subject: [PATCH 15/18] fix(collab-doc): make content_updated_at monotonic per file; skip-reconcile can't loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The If-Match token was stamped with app-local new Date() on each content write, so cross-instance clock skew could stamp a later write with an EARLIER content_updated_at — breaking the version ordering the whole optimistic-concurrency scheme (and the skip-reconcile branch's freshest>=version assumption) depends on. Under skew the relay's monotonic syncedVersion could exceed the durable version, sticking the If-Match: persist conflicts forever, exhausts retries, drops the session's edits. - Stamp content_updated_at strictly after the current committed value (we hold the row's FOR UPDATE lock): new Date(max(now, currentFile.contentUpdatedAt + 1ms)). Monotonic per file regardless of clocks; also removes same-millisecond collisions. updatedAt stays plain wall-clock (display/sort). - Skip-reconcile branch retries with result.version (the durable value the CAS will match), never freshest (which could exceed it and loop). Belt-and-suspenders now that the version is monotonic. --- apps/realtime/src/handlers/file-doc.ts | 9 +++++---- .../workspace/workspace-file-manager.ts | 17 ++++++++++++----- 2 files changed, 17 insertions(+), 9 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index b5421033b0b..b2d8a18e888 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -344,12 +344,13 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr // synced version already covers the conflict version, the durable body is a STALE SUBSET of the live // stream — a racing persist wrote it FROM this stream, or an apply-edit already merged it (the write // chokepoint advances the synced version). Reconciling it (incoming-wins) would move the doc BACKWARD - // and wipe newer in-flight edits. Skip it and just retry with the freshest version as If-Match; the - // re-projection captures the current live stream, preserving every edit. (`freshest` never exceeds - // the durable version — you can't sync from a version that doesn't exist — so this can't loop.) + // and wipe newer in-flight edits. Skip it and retry with the DURABLE conflict version as If-Match + // (`result.version`, what the CAS will actually match — never `freshest`, which could exceed it and + // loop): the re-projection captures the current live stream, preserving every edit. (`content_updated_at` + // is monotonic per file, so `freshest` normally can't exceed it — using `result.version` is belt-and-suspenders.) const freshest = await currentVersion() if (freshest !== undefined && freshest >= result.version) { - ifMatch = freshest + ifMatch = result.version continue } // Reconcile the out-of-band durable content into the live doc (advances the synced version), then diff --git a/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts b/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts index 3e08358bc9c..becee921859 100644 --- a/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts +++ b/apps/sim/lib/uploads/contexts/workspace/workspace-file-manager.ts @@ -1140,17 +1140,24 @@ export async function updateWorkspaceFileContent( } const sizeDiff = content.length - currentFile.size - // This IS a content write, so advance `contentUpdatedAt` in lockstep with `updatedAt` (same - // instant) — the new content version the collab relay records as its If-Match token. - const writeTimestamp = new Date() + const now = new Date() + // `contentUpdatedAt` is the persist If-Match token, so it MUST be strictly monotonic per file — a + // bare `new Date()` is not: cross-instance clock skew can stamp a later write with an earlier time, + // breaking the version ordering the whole optimistic-concurrency scheme relies on (stuck If-Match, + // wrong reconcile). We hold this row's FOR UPDATE lock, so `currentFile.contentUpdatedAt` is the + // latest committed value; stamp strictly after it. (Also removes same-millisecond collisions.) + // `updatedAt` stays plain wall-clock — it is display/sort only, never the concurrency token. + const contentUpdatedAt = new Date( + Math.max(now.getTime(), currentFile.contentUpdatedAt.getTime() + 1) + ) const [updatedFile] = await tx .update(workspaceFiles) .set({ key: uploadResult.key, size: content.length, contentType: nextContentType, - updatedAt: writeTimestamp, - contentUpdatedAt: writeTimestamp, + updatedAt: now, + contentUpdatedAt, }) .where( and( From f965bc171d1cfffe01cd2c6d6e55b1789be8ab93 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 30 Jul 2026 00:20:19 -0700 Subject: [PATCH 16/18] refactor(collab-doc): drop the destructive in-persist reconcile; adopt-version-and-retry on conflict MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The in-persist reconcile projected the durable body back over the live doc via updateYFragment ("make the doc match"). That is destructive: when the live stream is already ahead — the common case, because the write chokepoint (mergeEditIntoLiveFileDoc) already merged the out-of-band change into the stream — it moved the doc backward and wiped newer in-flight edits. This produced a run of races (stale snapshot, wipe-newer-edits, version-lag skip miss) that a full-document reconcile fundamentally can't avoid, since deciding when it's safe relies on a laggy cross-task version token. Remove it. On conflict, adopt the durable version as the new If-Match and retry: captureState re-reads the current stream (which holds the out-of-band change AND the live edits), so the re-projection persists the converged result. The durable change reaches the live doc via the chokepoint, never here. Trade-off: the only unmerged out-of-band write is one whose chokepoint merge itself failed (rare, logged), which we accept over the frequent reconcile-wipes-edits race. - flushPersist: conflict -> ifMatch = result.version, retry (bounded). No applyMarkdownToLiveFileDoc. - conflict response drops `markdown` (contract + relay type + persist) — no body needed, saves a blob fetch. applyMarkdownToLiveFileDoc stays (still used by the apply-edit route / the chokepoint). --- apps/realtime/src/handlers/file-doc-app.ts | 6 +- apps/realtime/src/handlers/file-doc.test.ts | 5 +- apps/realtime/src/handlers/file-doc.ts | 65 +++++---------------- apps/sim/lib/api/contracts/file-doc.ts | 6 +- apps/sim/lib/collab-doc/persist.ts | 34 ++++++----- 5 files changed, 39 insertions(+), 77 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc-app.ts b/apps/realtime/src/handlers/file-doc-app.ts index 834cb7f705b..abee966dd46 100644 --- a/apps/realtime/src/handlers/file-doc-app.ts +++ b/apps/realtime/src/handlers/file-doc-app.ts @@ -77,13 +77,13 @@ export async function fetchFileDocMerge( * 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. `markdown` + - * `version` are the current durable content + version so the relay reconciles and re-persists. + * - `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'; markdown: string; version: number } + | { status: 'conflict'; version: number } | { status: 'deferred' } /** diff --git a/apps/realtime/src/handlers/file-doc.test.ts b/apps/realtime/src/handlers/file-doc.test.ts index aa026e68172..11f53e9ccbb 100644 --- a/apps/realtime/src/handlers/file-doc.test.ts +++ b/apps/realtime/src/handlers/file-doc.test.ts @@ -301,11 +301,10 @@ describe('setupWorkspaceFileDocHandlers', () => { it('retries a persist that keeps conflicting, then gives up without clobbering', async () => { mockFetchFileDocSeed.mockResolvedValue(seedResult('# From server')) - // Every persist reports an out-of-band change (If-Match conflict) — the durable file must be left - // authoritative rather than overwritten, after a bounded reconcile-retry. + // Every persist reports an out-of-band change (If-Match conflict) — the relay adopts the durable + // version and retries (bounded); the durable file is left authoritative rather than overwritten. mockFetchFileDocPersist.mockResolvedValue({ status: 'conflict', - markdown: '# external edit', version: 999, }) const { io } = createIo() diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index b2d8a18e888..8ba6635b204 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -248,9 +248,7 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr const userId = room.lastEditorUserId // Synchronous fallback capture — before any await, since the caller may destroy `room.doc` the moment // this yields. Only meaningful once seeded; used only when the authoritative stream state is absent. - // NULLED after a reconcile (below): once we've merged an out-of-band edit in, this pre-reconcile - // snapshot predates it and must never be persisted as a fallback, or it would re-drop that edit. - let localState = isDocSeeded(room.doc) ? Y.encodeStateAsUpdate(room.doc) : null + const localState = isDocSeeded(room.doc) ? Y.encodeStateAsUpdate(room.doc) : null // Capture the AUTHORITATIVE doc state: the shared stream when enabled (a copilot merge or a peer's // edit published by another task may not be integrated into THIS task's `room.doc` yet), else the @@ -314,9 +312,7 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr // Persist under optimistic concurrency (RFC 7232 If-Match): the write commits only if the file is // still at the version the live doc synced from, so a projection can never silently clobber an - // out-of-band edit. On a conflict, merge the current durable content into the live doc — so the - // out-of-band edit AND the live edits converge — and retry (bounded), so even a last-leave flush - // racing an external write persists the reconciled result rather than losing the session's edits. + // out-of-band edit. On a conflict, adopt the durable version and retry (bounded) — see below. for (let attempt = 0; attempt <= PERSIST_CONFLICT_RETRIES; attempt++) { const docState = await captureState() if (!docState) return // nothing seeded/authoritative to persist yet @@ -336,55 +332,20 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr } if (attempt === PERSIST_CONFLICT_RETRIES) { logger.warn( - `Persist for file ${room.fileId} still conflicting after reconcile; durable content left authoritative` + `Persist for file ${room.fileId} still conflicting after ${PERSIST_CONFLICT_RETRIES} retries; durable content left authoritative` ) return } - // Only reconcile a GENUINE out-of-band change the live doc hasn't incorporated. If the freshest - // synced version already covers the conflict version, the durable body is a STALE SUBSET of the live - // stream — a racing persist wrote it FROM this stream, or an apply-edit already merged it (the write - // chokepoint advances the synced version). Reconciling it (incoming-wins) would move the doc BACKWARD - // and wipe newer in-flight edits. Skip it and retry with the DURABLE conflict version as If-Match - // (`result.version`, what the CAS will actually match — never `freshest`, which could exceed it and - // loop): the re-projection captures the current live stream, preserving every edit. (`content_updated_at` - // is monotonic per file, so `freshest` normally can't exceed it — using `result.version` is belt-and-suspenders.) - const freshest = await currentVersion() - if (freshest !== undefined && freshest >= result.version) { - ifMatch = result.version - continue - } - // Reconcile the out-of-band durable content into the live doc (advances the synced version), then - // loop to re-project and re-persist the converged state. If there is no live doc to reconcile into - // — e.g. the last collaborator has left and the room is being torn down with no shared stream - // (single-pod) — stop: re-projecting the same pre-teardown snapshot would only re-conflict. The - // durable (out-of-band) content is left authoritative, which is the intended conflict policy. - const reconciled = await applyMarkdownToLiveFileDoc( - room.fileId, - result.markdown, - result.version - ) - if (reconciled === 'no-live-room') { - logger.warn( - `Persist conflict for file ${room.fileId} with no live doc to reconcile; durable content left authoritative` - ) - return - } - if (reconciled === 'merge-unavailable') { - // Merge-lock contention (NOT an absent stream): the reconcile could not run, so we have not - // incorporated the out-of-band edit and must not re-persist against it. Stop without clobbering; - // the session's edits stay in the stream and a later flush (or the next opener) retries. Distinct - // from `no-live-room` so a transient lock miss is never mistaken for "nothing to reconcile". - logger.warn( - `Persist conflict for file ${room.fileId}: merge lock unavailable, reconcile deferred (edits remain in stream)` - ) - return - } - // The out-of-band content is now reconciled into the live doc AT `result.version`, so that is the - // correct If-Match for the re-projection — advance to it directly rather than re-reading a version - // a concurrent teardown may have left stale in `room.syncedVersion`/Redis. The pre-await - // `localState` snapshot now predates this reconcile, so drop it: a subsequent failed stream read - // must abort rather than fall back to a snapshot that would re-drop the just-merged out-of-band edit. - localState = null + // Conflict: the durable file advanced out-of-band since our If-Match token. We deliberately do NOT + // project the durable body back over the live doc — that is a destructive "make the doc match" + // operation, and when the live stream is already ahead (the common case, because the write chokepoint + // has ALREADY merged the out-of-band change into the stream) it would move the doc BACKWARD and wipe + // newer in-flight edits. Instead adopt the durable version as the new If-Match and retry: + // `captureState` re-reads the current stream — which holds the out-of-band change AND the live edits + // — so the re-projection persists the converged result. The durable change reaches the live doc via + // the chokepoint (`mergeEditIntoLiveFileDoc`, which every external markdown write goes through), never + // here — so the only unmerged out-of-band write is one whose chokepoint merge itself failed, a rare + // degraded case we accept over the far more frequent reconcile-wipes-live-edits race. ifMatch = result.version } } catch (error) { diff --git a/apps/sim/lib/api/contracts/file-doc.ts b/apps/sim/lib/api/contracts/file-doc.ts index 97c9de47e70..4284640570d 100644 --- a/apps/sim/lib/api/contracts/file-doc.ts +++ b/apps/sim/lib/api/contracts/file-doc.ts @@ -126,11 +126,11 @@ export const persistFileDocResponseSchema = z.discriminatedUnion('status', [ z.object({ /** * The file changed out-of-band since `expectedVersion` — the write was refused to avoid clobbering - * it. `markdown` + `version` are the current durable content + version, so the relay can merge the - * change into its live doc and re-persist the reconciled result. + * it. `version` is the current durable version the relay adopts as its new If-Match to re-persist the + * current live stream (which already holds the out-of-band change via the write chokepoint). No + * markdown body: the relay never projects the durable body back over the live doc. */ status: z.literal('conflict'), - markdown: z.string(), version: z.number().int(), }), ]) diff --git a/apps/sim/lib/collab-doc/persist.ts b/apps/sim/lib/collab-doc/persist.ts index 0541d45106b..24d4590ff50 100644 --- a/apps/sim/lib/collab-doc/persist.ts +++ b/apps/sim/lib/collab-doc/persist.ts @@ -3,7 +3,6 @@ import { getErrorMessage } from '@sim/utils/errors' import * as Y from 'yjs' import { ContentVersionConflictError, - fetchWorkspaceFileBuffer, getWorkspaceFile, updateWorkspaceFileContent, } from '@/lib/uploads/contexts/workspace' @@ -18,14 +17,14 @@ const logger = createLogger('FileDocPersist') * CONTENT version (`content_updated_at`, epoch ms) the relay records as what its live doc is synced to. * - `missing` — the file is gone (deleted); nothing to write. * - `conflict` — the file changed out-of-band since the relay's live doc last synced, so writing the - * projection would clobber that change (RFC 7232 `If-Match` failure). NOT written. `markdown` + - * `version` are the current durable content + version so the relay can merge them into its live doc - * and re-persist the reconciled result. + * projection would clobber that change (RFC 7232 `If-Match` failure). NOT written. `version` is the + * current durable version the relay adopts as its new If-Match to re-persist the current live stream + * (which already holds the out-of-band change via the write chokepoint). */ export type PersistFileDocResult = | { status: 'persisted'; version: number } | { status: 'missing' } - | { status: 'conflict'; markdown: string; version: number } + | { status: 'conflict'; version: number } | { status: 'deferred' } /** @@ -38,9 +37,9 @@ export type PersistFileDocResult = * synced from) is the optimistic-concurrency guard: the write commits only if the file is still at that * content version — a rename/move that only bumps `updatedAt` won't trip it, so a * projection built from a stale live doc can never silently overwrite an out-of-band edit. On a version - * mismatch this returns `conflict` (with the current durable content) instead of writing — the relay - * reconciles and retries. Omit `expectedVersion` to write unconditionally (e.g. the first persist, - * before any synced version exists). + * mismatch this returns `conflict` (the current durable version) instead of writing — the relay adopts + * it as its new If-Match and retries against the current live stream. Omit `expectedVersion` to write + * unconditionally (e.g. the first persist, before any synced version exists). * * `userId` is attribution only (blob metadata); the caller is already trusted via the `x-api-key` gate. */ @@ -114,18 +113,21 @@ export async function persistFileDoc( } } catch (error) { if (!(error instanceof ContentVersionConflictError)) throw error - // Out-of-band edit since the live doc last synced — DON'T clobber. Return the current durable - // content + version so the relay merges it into the live doc and re-persists the reconciled result. + // Out-of-band content change since the live doc last synced — DON'T clobber. Return the current + // durable version so the relay adopts it as its new If-Match and re-persists the current live stream + // (which already holds the out-of-band change, merged in via the write chokepoint). No markdown body + // is returned: the relay never projects the durable body back over the live doc (that would be a + // destructive "make it match" that could move the doc backward and wipe newer edits). const current = await getWorkspaceFile(workspaceId, fileId, { throwOnError: true }) if (!current) return { status: 'missing' } - const currentBuffer = await fetchWorkspaceFileBuffer(current) - logger.warn(`Persist conflict for file ${fileId}; returning current durable state to reconcile`) + logger.warn( + `Persist conflict for file ${fileId}; durable content changed out-of-band since sync` + ) return { status: 'conflict', - markdown: currentBuffer.toString('utf-8'), - // The CONTENT version, not `updatedAt`: the relay reconciles to this and re-persists with it as the - // If-Match. If a metadata write bumped `updatedAt` past `contentUpdatedAt`, returning `updatedAt` - // would make the re-persist's CAS (which checks `contentUpdatedAt`) never match → perpetual conflict. + // The CONTENT version, not `updatedAt`: the relay adopts this as its If-Match. If a metadata write + // bumped `updatedAt` past `contentUpdatedAt`, returning `updatedAt` would make the re-persist's CAS + // (which checks `contentUpdatedAt`) never match → perpetual conflict. version: (current.contentUpdatedAt ?? current.updatedAt).getTime(), } } From 22dfc9cef073a089ad809b83c96b643520637b31 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 30 Jul 2026 00:30:41 -0700 Subject: [PATCH 17/18] fix(collab-doc): don't let a last-leave conflict retry clobber via the stale local snapshot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Regression from dropping the reconcile: on conflict the retry adopts result.version and re-reads captureState. But after single-pod last-leave teardown the room is already destroyed, so captureState falls back to the pre-teardown localState (which lacks the out-of-band change); the retry then CAS-passes and overwrites the committed external write — undoing the external-wins last-leave policy. Null localState on the first conflict, so the retry can only use freshly-read authoritative state (stream / live doc). When none is available (single-pod room gone, or a transient stream-read failure) captureState returns null and the retry stops, leaving durable content authoritative. Covers both the single-pod and multi-task-stream-unavailable variants of the stale-snapshot clobber. --- apps/realtime/src/handlers/file-doc.ts | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index 8ba6635b204..b7534c1b575 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -248,7 +248,9 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr const userId = room.lastEditorUserId // Synchronous fallback capture — before any await, since the caller may destroy `room.doc` the moment // 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 + // NULLED on the first conflict (below): once the durable file has advanced out-of-band, this snapshot + // predates that change, so it must never be persisted as a fallback or it would re-clobber the write. + let localState = isDocSeeded(room.doc) ? Y.encodeStateAsUpdate(room.doc) : null // Capture the AUTHORITATIVE doc state: the shared stream when enabled (a copilot merge or a peer's // edit published by another task may not be integrated into THIS task's `room.doc` yet), else the @@ -346,6 +348,13 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr // the chokepoint (`mergeEditIntoLiveFileDoc`, which every external markdown write goes through), never // here — so the only unmerged out-of-band write is one whose chokepoint merge itself failed, a rare // degraded case we accept over the far more frequent reconcile-wipes-live-edits race. + // + // Drop `localState` first: the durable body has changed, so the pre-await session snapshot is now + // stale. Without this, a retry whose fresh read is unavailable — single-pod after last-leave teardown + // (room already destroyed), or a transient stream-read failure — would fall back to that snapshot and + // CAS-pass over the committed out-of-band write. Nulling it makes `captureState` return null there, so + // the retry stops and leaves the durable content authoritative (the external-wins last-leave policy). + localState = null ifMatch = result.version } } catch (error) { From df929697f8c8e7c63a760684ce13d6663cfa445c Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Thu, 30 Jul 2026 00:47:11 -0700 Subject: [PATCH 18/18] =?UTF-8?q?fix(collab-doc):=20stop=20(don't=20re-per?= =?UTF-8?q?sist)=20on=20a=20persist=20conflict=20=E2=80=94=20closes=20the?= =?UTF-8?q?=20commit-window=20clobber?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The conflict retry adopted the durable version and immediately re-persisted the current stream, assuming the stream already held the out-of-band change. But an external write commits durable BEFORE its chokepoint merge (mergeEditIntoLiveFileDoc) reaches the stream, so a persist landing in that window CAS-passed with a stream that still lacked the external content and clobbered the committed write — not just the rare merge-failed path, but a race on every external write, worst at last-leave flushes. Make persist a single attempt: on conflict, STOP and leave durable authoritative. The chokepoint merges the change into the stream and — only once it is actually there — advances the synced version via its own recordVersion; a later flush (debounced or final) then projects the converged stream with a matching token. The session's edits stay in the stream meanwhile. The conflict handler deliberately does NOT advance the synced version, or the next flush would clobber with a still-behind stream. Removes the retry loop and PERSIST_CONFLICT_RETRIES. --- apps/realtime/src/handlers/file-doc.test.ts | 15 ++-- apps/realtime/src/handlers/file-doc.ts | 82 ++++++++------------- 2 files changed, 37 insertions(+), 60 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc.test.ts b/apps/realtime/src/handlers/file-doc.test.ts index 11f53e9ccbb..4e0d9ffa2de 100644 --- a/apps/realtime/src/handlers/file-doc.test.ts +++ b/apps/realtime/src/handlers/file-doc.test.ts @@ -299,10 +299,12 @@ describe('setupWorkspaceFileDocHandlers', () => { expect(mockFetchFileDocPersist).toHaveBeenCalled() }) - it('retries a persist that keeps conflicting, then gives up without clobbering', async () => { + it('stops on a persist conflict without clobbering (single attempt, durable left authoritative)', async () => { mockFetchFileDocSeed.mockResolvedValue(seedResult('# From server')) - // Every persist reports an out-of-band change (If-Match conflict) — the relay adopts the durable - // version and retries (bounded); the durable file is left authoritative rather than overwritten. + // 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, @@ -321,12 +323,11 @@ describe('setupWorkspaceFileDocHandlers', () => { ) await flushMicrotasks() - // A persistent conflict is handled gracefully: the persist is attempted (never silently skipped) - // but the durable file is left authoritative — no clobber, no throw, no unbounded retry loop. + // 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).toHaveBeenCalled() - expect(mockFetchFileDocPersist.mock.calls.length).toBeLessThanOrEqual(3) + expect(mockFetchFileDocPersist).toHaveBeenCalledTimes(1) }) it('flushAllFileDocRooms persists open EDITED rooms (graceful shutdown), skips unedited', async () => { diff --git a/apps/realtime/src/handlers/file-doc.ts b/apps/realtime/src/handlers/file-doc.ts index b7534c1b575..194943153a6 100644 --- a/apps/realtime/src/handlers/file-doc.ts +++ b/apps/realtime/src/handlers/file-doc.ts @@ -65,10 +65,6 @@ const PERSIST_DEBOUNCE_MS = 5_000 * 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 -/** How many times a persist re-tries after reconciling an out-of-band edit into the live doc before it - * gives up (leaving the durable content authoritative). One reconcile normally converges; the small - * budget covers a second racing write. */ -const PERSIST_CONFLICT_RETRIES = 2 /** On a FINAL (last-leave/shutdown) flush the version read is the last chance to resolve the If-Match * before the room is torn down; a transient Redis blip yielding no version would otherwise defer and * strand the session's edits in the TTL'd stream. The version is cluster-wide and heartbeat-refreshed, so @@ -248,9 +244,7 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr const userId = room.lastEditorUserId // Synchronous fallback capture — before any await, since the caller may destroy `room.doc` the moment // this yields. Only meaningful once seeded; used only when the authoritative stream state is absent. - // NULLED on the first conflict (below): once the durable file has advanced out-of-band, this snapshot - // predates that change, so it must never be persisted as a fallback or it would re-clobber the write. - let localState = isDocSeeded(room.doc) ? Y.encodeStateAsUpdate(room.doc) : null + const localState = isDocSeeded(room.doc) ? Y.encodeStateAsUpdate(room.doc) : null // Capture the AUTHORITATIVE doc state: the shared stream when enabled (a copilot merge or a peer's // edit published by another task may not be integrated into THIS task's `room.doc` yet), else the @@ -294,10 +288,7 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr if (!final && !(await store.tryClaimPersistWindow(name, FILE_DOC_TIMEOUTS.persistRequestMs))) return - // The If-Match token, resolved ONCE and then advanced LOCALLY as we reconcile — never re-derived - // mid-loop. On a last-leave flush the room is already removed from `fileDocRooms`, so - // `mergeMarkdownIntoRoom`'s `recordVersion` can't write `room.syncedVersion`; carrying the reconciled - // version here keeps each retry's precondition correct without depending on that mutation or a Redis re-read. + // The If-Match token: the durable content version the live doc is synced to. let ifMatch = await currentVersion() // FINAL flush = last chance before teardown: if the version read momentarily fails (Redis blip) for a // peer-seeded/tail-only task that never cached it, retry briefly rather than defer and strand the @@ -314,49 +305,34 @@ async function flushPersist(name: string, room: FileDocRoom, final: boolean): Pr // Persist under optimistic concurrency (RFC 7232 If-Match): the write commits only if the file is // still at the version the live doc synced from, so a projection can never silently clobber an - // out-of-band edit. On a conflict, adopt the durable version and retry (bounded) — see below. - for (let attempt = 0; attempt <= PERSIST_CONFLICT_RETRIES; attempt++) { - const docState = await captureState() - if (!docState) return // nothing seeded/authoritative to persist yet - const result = await fetchFileDocPersist(workspaceId, room.fileId, userId, docState, ifMatch) - if (result.status === 'missing') return - if (result.status === 'deferred') { - // The app couldn't persist safely without a version token (momentarily unavailable). Do NOT - // reconcile — nothing changed out-of-band, so merging would needlessly wipe live edits. Leave - // the edits in the stream; a later persist writes them once the version is re-established. - logger.warn(`Persist deferred for file ${room.fileId} (no synced version available yet)`) - return - } - if (result.status === 'persisted') { - room.syncedVersion = Math.max(room.syncedVersion ?? 0, result.version) - void store.setSyncedVersion(name, result.version) - return - } - if (attempt === PERSIST_CONFLICT_RETRIES) { - logger.warn( - `Persist for file ${room.fileId} still conflicting after ${PERSIST_CONFLICT_RETRIES} retries; durable content left authoritative` - ) - return - } - // Conflict: the durable file advanced out-of-band since our If-Match token. We deliberately do NOT - // project the durable body back over the live doc — that is a destructive "make the doc match" - // operation, and when the live stream is already ahead (the common case, because the write chokepoint - // has ALREADY merged the out-of-band change into the stream) it would move the doc BACKWARD and wipe - // newer in-flight edits. Instead adopt the durable version as the new If-Match and retry: - // `captureState` re-reads the current stream — which holds the out-of-band change AND the live edits - // — so the re-projection persists the converged result. The durable change reaches the live doc via - // the chokepoint (`mergeEditIntoLiveFileDoc`, which every external markdown write goes through), never - // here — so the only unmerged out-of-band write is one whose chokepoint merge itself failed, a rare - // degraded case we accept over the far more frequent reconcile-wipes-live-edits race. - // - // Drop `localState` first: the durable body has changed, so the pre-await session snapshot is now - // stale. Without this, a retry whose fresh read is unavailable — single-pod after last-leave teardown - // (room already destroyed), or a transient stream-read failure — would fall back to that snapshot and - // CAS-pass over the committed out-of-band write. Nulling it makes `captureState` return null there, so - // the retry stops and leaves the durable content authoritative (the external-wins last-leave policy). - localState = null - ifMatch = result.version + // out-of-band edit. A single attempt — on conflict we STOP rather than retry (see below). + const docState = await captureState() + if (!docState) return // nothing seeded/authoritative to persist yet + const result = await fetchFileDocPersist(workspaceId, room.fileId, userId, docState, ifMatch) + if (result.status === 'missing') return // the file was deleted; nothing to write + if (result.status === 'deferred') { + // No version token available (momentarily — a Redis blip on a peer-seeded task). Leave the edits in + // the stream; a later persist writes them once the version is re-established. + logger.warn(`Persist deferred for file ${room.fileId} (no synced version available yet)`) + return } + if (result.status === 'persisted') { + room.syncedVersion = Math.max(room.syncedVersion ?? 0, result.version) + void store.setSyncedVersion(name, result.version) + return + } + // status === 'conflict': the durable file advanced out-of-band since our If-Match token. We do NOT + // re-persist against the current stream: an external write commits durable BEFORE its chokepoint merge + // (`mergeEditIntoLiveFileDoc`) reaches the stream, so a re-persist landing in that window would CAS-pass + // with a stream that still lacks the external content and clobber the committed write. Instead leave the + // durable content authoritative — the chokepoint merges the change into the stream and, ONLY once it is + // actually there, advances the synced version (via the merge's own `recordVersion`); a later flush + // (a subsequent debounced persist, or the final flush) then projects the converged stream with a token + // that matches. The session's edits stay in the stream meanwhile. Deliberately do NOT advance the synced + // version here: before the stream reflects the durable content, that would let the next flush clobber it. + logger.warn( + `Persist conflict for file ${room.fileId}; durable content advanced out-of-band, left authoritative` + ) } catch (error) { logger.warn(`Persist failed for file ${room.fileId}`, { error: getErrorMessage(error) }) }