fix(files): serve a whole document on join, and stop persistence deadlocking

Opening a file right after an edit replayed it — the block you had just moved
moved again, in front of you. A room rebuilds itself from the file's Redis
stream one entry at a time, into the same Y.Doc that fans every update out to
its room, and the join attached the socket before that finished and before the
server seed. So a client was never sent the document; it was sent the
document's history, and it watched that replay. The new join-readiness test
reproduces it exactly — ['AAA', 'BBBAAA'] where one state was due — and fails
without the fix.

Underneath it, persistence had deadlocked. The If-Match token is a REMEMBERED
timestamp: held in the room, which dies with it, and in a cluster key written
fire-and-forget. A relay that exits in the moments after a successful write
comes back holding a version older than the file's, every persist then fails
the CAS, and because a conflict neither writes nor advances the token, that
document can never be persisted again. Measured on a live file: token
1786594348911 against a content version of 1786594418409, with 103 unpersisted
entries still in the stream. The durable markdown froze there — and the
editor's placeholder is built from it, so every reload painted the pre-edit
document and the live one corrected it on screen.

- Assemble a room before attaching a client to it. The join awaits the stream
  catch-up and the seed, so the first sync IS the finished document, in one
  message. The seed is memoized on the room, so concurrent joins wait for the
  same one instead of the second being served an empty doc; a task that loses
  the seed lock PULLS the winner's seed from the stream rather than waiting for
  the tailer to push it, which is what left a freshly uploaded file read-only
  until its readiness deadline lapsed.
- Drop the client-side quiet-frame gate that was standing in for this. It was
  unsound in both directions: it delayed a document that was already correct,
  and it opened mid-flight anyway whenever updates arrived more than a frame
  apart, which is what a cross-region Redis and a long room history produce.
- Await the cluster version write after a successful persist. It is the only
  record that survives a teardown, and one round trip after a blob write is not
  a cost worth a wedged document.
- On a version conflict, ask the CONTENT, not the clock. If the file still
  holds the bytes this document last projected — the tag written with every
  persist — then nothing out-of-band exists to protect, so re-sync the token and
  write. Any other bytes and the conflict stands exactly as before.
- Actually run the stale-storage-key recovery. It was requested from inside the
  failing read's own queryFn, where react-query drops it, so nothing re-resolved
  the record and the reader sat on a dead key showing "Failed to load file
  content" until something unrelated refetched. It now runs off that cycle and
  cancels a read already in flight, which could only hand back the dead key.
  While the record is re-resolving the surface reports loading, not failure.
- One document per file, for its whole life. A document rebuilt from markdown is
  a DIFFERENT document to Yjs — its items carry new client ids — so a client
  still holding the old one merges the file into itself, twice, on both sides.
  The seed now stores what it builds (until that row existed, a file opened but
  never edited was rebuilt on every open) and resumes it with a CRDT diff when
  the markdown moved on out-of-band. A document also carries an identity the
  join ack names, so a tab that outlived its room is refused rather than merged.
- Let a browser keep an embedded image. The inline route sent no-cache with no
  validator, so every open re-downloaded the whole image — measured at ~1 MB per
  open of a real document, with the image area blank until it landed. A `key`
  names one storage object and a content write never rewrites one, so those
  bytes are immutable; a `fileId` names the file, whose bytes move, so that form
  still revalidates.

Author: icecrasher321

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

@@ -18,10 +18,12 @@
  *   client receives each update exactly once, from its own task's local broadcast — no adapter
  *   amplification, and every task's doc stays converged. (Awareness/presence stay on the adapter: they
  *   are ephemeral and need no convergence or replay.)
- * - {@link attachRoom} does a synchronous catch-up read from the head of the stream when a task first
- *   opens a file, so a late-joining task (the normal case under autoscaling) loads the current shared
- *   state before its first client syncs. Catch-up + tail are seamless: the tailer resumes from the
- *   exact id catch-up stopped at.
+ * - {@link attachRoom} reads the stream from the head when a task first opens a file, and the relay
+ *   AWAITS it before attaching a client, so a late-joining task (the normal case under autoscaling)
+ *   holds the current shared state before its first client syncs — a client must never watch the
+ *   catch-up land entry by entry, which is the document's edit history replaying on screen. Catch-up +
+ *   tail are seamless: the tailer resumes from the exact id catch-up stopped at, and {@link catchUp}
+ *   can re-run at any time for a caller that must converge without waiting on the tailer.
  * - The one-time seed is written via the atomic {@link seedIfEmpty} (append-iff-empty in one Redis
  *   step), so exactly one task ever writes the seed cluster-wide (the fix for split-brain) — even if two
  *   tasks race. {@link shouldSeed} is a Redis lock + empty-stream check layered on top ONLY as an
@@ -185,6 +187,18 @@ function applyEntryToDoc(
   }
 }
 
+/**
+ * Whether stream id `id` sorts after `than`. A Redis stream id is `<ms>-<seq>`, so a lexicographic
+ * compare is wrong the moment the millisecond part changes digit length (`'9999-0' > '10000-0'`);
+ * compare the two parts numerically instead. The initial `'0'` (nothing applied) has no `-seq` part,
+ * which reads as sequence 0 — before every real entry.
+ */
+function isAfterStreamId(id: string, than: string): boolean {
+  const [ms, seq = '0'] = id.split('-')
+  const [thanMs, thanSeq = '0'] = than.split('-')
+  return Number(ms) === Number(thanMs) ? Number(seq) > Number(thanSeq) : Number(ms) > Number(thanMs)
+}
+
 /** Whether a doc carries the seed flag (mirrors the relay's `isDocSeeded`), so the store can tell the
  * one-time seed transition from a real post-seed edit without re-implementing the check divergently. */
 function isDocSeeded(doc: Y.Doc): boolean {
@@ -260,10 +274,9 @@ export class FileDocStore {
   }
 
   /**
-   * Register a locally-opened room and load the shared state into its doc: read the whole stream from
-   * the head, apply every entry (origin {@link REDIS_ORIGIN}), and remember the last id so the tailer
-   * resumes exactly after it. A brand-new file has an empty stream and loads nothing (it is seeded
-   * shortly after, via {@link shouldSeed}). No-op when disabled.
+   * Register a locally-opened room and load the shared state into its doc ({@link catchUp}). A
+   * brand-new file has an empty stream and loads nothing (it is seeded shortly after, via
+   * {@link shouldSeed}). No-op when disabled.
    */
   async attachRoom(name: string, doc: Y.Doc): Promise<void> {
     if (!this.enabled || !this.write) return
@@ -277,12 +290,31 @@ export class FileDocStore {
       realEdited: false,
     }
     this.rooms.set(name, room)
+    await this.catchUp(name)
+  }
+
+  /**
+   * PULL the shared state into a registered room: read the stream and apply every entry the doc has
+   * not integrated yet (origin {@link REDIS_ORIGIN}), advancing `lastId` so the tailer resumes exactly
+   * after it. This is the ONLY way a room loads shared state, so a caller that must not depend on the
+   * tailer's asynchronous push — the join, which may not serve a client a half-assembled document —
+   * can converge on demand. Idempotent and safe to call repeatedly; no-op when disabled or the room is
+   * not registered (a fast open→close detached it). Never throws.
+   */
+  async catchUp(name: string): Promise<void> {
+    if (!this.enabled || !this.write) return
+    const room = this.rooms.get(name)
+    if (!room) return
     try {
       const entries = await this.write.xRange(streamKey(name), '-', '+')
       for (const entry of entries) {
-        // The room can be detached + its doc destroyed while catch-up is in flight (a fast open→close);
-        // stop touching it the moment that happens.
+        // The room can be detached + its doc destroyed while the read is in flight (a fast
+        // open→close); stop touching it the moment that happens.
         if (this.rooms.get(name) !== room) return
+        // Applying a Yjs update twice is a no-op, but `applyEntry`'s bookkeeping is not: re-applying
+        // the SEED after `seededObserved` latched would count it as a post-seed edit and let a
+        // compaction snapshot claim content no user ever typed. Skip what this room already holds.
+        if (!isAfterStreamId(entry.id, room.lastId)) continue
         this.applyEntry(room, entry.id, entry.message)
       }
       await this.write.expire(streamKey(name), STREAM_TTL_SEC)

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

@@ -0,0 +1,305 @@
+/**
+ * @vitest-environment node
+ *
+ * The join's readiness contract, with the shared store ENABLED (`file-doc.test.ts` runs it disabled).
+ *
+ * A room loads its document from the file's Redis stream one entry at a time, into the same `Y.Doc`
+ * that fans every update out to the room. So a client attached while that is happening is not sent the
+ * document — it is sent the document's history, and it watches the history replay on screen (reload
+ * right after moving a block and the block moves again in front of you). These tests pin the fix: the
+ * join waits for the room to hold the whole document, so the client's first sync is authoritative.
+ */
+import {
+  FILE_DOC_EVENTS,
+  FILE_DOC_MESSAGE_TYPE,
+  FILE_DOC_SEED,
+} from '@sim/realtime-protocol/file-doc'
+import * as decoding from 'lib0/decoding'
+import * as encoding from 'lib0/encoding'
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'
+import * as syncProtocol from 'y-protocols/sync'
+import * as Y from 'yjs'
+import type { IRoomManager } from '@/rooms'
+
+const { mockAuthorizeRoom, mockFetchFileDocSeed } = vi.hoisted(() => ({
+  mockAuthorizeRoom: vi.fn(),
+  mockFetchFileDocSeed: vi.fn(),
+}))
+
+vi.mock('@sim/platform-authz/rooms', () => ({ authorizeRoom: mockAuthorizeRoom }))
+
+vi.mock('@/handlers/file-doc-app', () => ({
+  fetchFileDocSeed: mockFetchFileDocSeed,
+  fetchFileDocMerge: vi.fn(),
+  fetchFileDocPersist: vi.fn().mockResolvedValue({ status: 'persisted', version: 1 }),
+}))
+
+/** One in-memory Redis backing per test — only the stream/lock ops the store actually uses. */
+const backing = vi.hoisted(() => ({
+  streams: new Map<string, { id: string; message: Record<string, string> }[]>(),
+  kv: new Map<string, string>(),
+  seq: 0,
+  /** Ticks of event-loop delay each xRange takes, modelling a remote (cross-region) Redis. */
+  readDelayTicks: 0,
+}))
+
+const seqOf = (id: string) => Number(id.split('-')[0])
+
+vi.mock('redis', () => {
+  const makeClient = (): Record<string, unknown> => {
+    const client: Record<string, unknown> = {
+      connect: async () => {},
+      quit: async () => {},
+      on: () => client,
+      duplicate: () => makeClient(),
+      xAdd: async (key: string, _star: string, fields: Record<string, string>) => {
+        const id = `${++backing.seq}-0`
+        const arr = backing.streams.get(key) ?? []
+        arr.push({ id, message: { ...fields } })
+        backing.streams.set(key, arr)
+        return id
+      },
+      xRange: async (key: string) => {
+        for (let i = 0; i < backing.readDelayTicks; i++) await Promise.resolve()
+        return (backing.streams.get(key) ?? []).map((e) => ({ ...e }))
+      },
+      xLen: async (key: string) => (backing.streams.get(key) ?? []).length,
+      xRead: async (streams: { key: string; id: string }[]) => {
+        const res: { name: string; messages: { id: string; message: Record<string, string> }[] }[] =
+          []
+        for (const { key, id } of streams) {
+          const after = (backing.streams.get(key) ?? []).filter((e) => seqOf(e.id) > seqOf(id))
+          if (after.length) res.push({ name: key, messages: after.map((e) => ({ ...e })) })
+        }
+        if (res.length) return res
+        await new Promise((r) => setTimeout(r, 5))
+        return null
+      },
+      set: async (key: string, val: string, opts?: { NX?: boolean }) => {
+        if (opts?.NX && backing.kv.has(key)) return null
+        backing.kv.set(key, val)
+        return 'OK'
+      },
+      eval: async (script: string, opts: { keys: string[]; arguments: string[] }) => {
+        const [key] = opts.keys
+        if (script.includes('xlen')) {
+          const [field, value] = opts.arguments
+          const arr = backing.streams.get(key) ?? []
+          if (arr.length > 0) return 0
+          arr.push({ id: `${++backing.seq}-0`, message: { [field]: value } })
+          backing.streams.set(key, arr)
+          return 1
+        }
+        const [token] = opts.arguments
+        if (backing.kv.get(key) === token) {
+          backing.kv.delete(key)
+          return 1
+        }
+        return 0
+      },
+      expire: async () => 1,
+      get: async (key: string) => backing.kv.get(key) ?? null,
+      exists: async (key: string) => (backing.kv.has(key) ? 1 : 0),
+    }
+    return client
+  }
+  return { createClient: () => makeClient() }
+})
+
+import { cleanupFileDocForSocket, setupWorkspaceFileDocHandlers } from '@/handlers/file-doc'
+import { getFileDocStore, initFileDocStore } from '@/handlers/file-doc-store'
+
+const FILE_ID = 'file-1'
+const ROOM_NAME = `workspace-file-doc:${FILE_ID}`
+const STREAM_KEY = `filedoc:stream:${ROOM_NAME}`
+const FIELD = 'default'
+
+type Handler = (payload?: unknown) => Promise<void> | void
+
+interface FakeSocket {
+  id: string
+  emit: (event: string, payload: unknown) => void
+  rooms: Set<string>
+}
+
+/**
+ * An `io` that actually DELIVERS: a room emit reaches every socket that joined that room, so a frame
+ * the relay fans out mid-assembly lands on the joiner's `emit` exactly as it would in the browser.
+ * Recording the emits without routing them would hide the very thing these tests are about.
+ */
+function createIo(sockets: FakeSocket[]) {
+  const emitTo = (target: string, except: string | null, event: string, payload: unknown) => {
+    for (const socket of sockets) {
+      if (socket.id === except || !socket.rooms.has(target)) continue
+      socket.emit(event, payload)
+    }
+  }
+  const to = vi.fn((target: string) => ({
+    except: (exclude: string) => ({
+      emit: (event: string, payload: unknown) => emitTo(target, exclude, event, payload),
+    }),
+    emit: (event: string, payload: unknown) => emitTo(target, null, event, payload),
+  }))
+  return {
+    to,
+    in: vi.fn(() => ({ socketsLeave: () => {} })),
+    local: { to },
+  } as unknown as IRoomManager['io']
+}
+
+function setup(id: string, sockets: FakeSocket[]) {
+  const handlers: Record<string, Handler> = {}
+  const rooms = new Set<string>()
+  const socket = {
+    id,
+    userId: 'user-1',
+    userName: 'Test User',
+    userImage: 'avatar.png',
+    disconnected: false,
+    rooms,
+    on: vi.fn((event: string, handler: Handler) => {
+      handlers[event] = handler
+    }),
+    emit: vi.fn(),
+    join: vi.fn((name: string) => rooms.add(name)),
+    leave: vi.fn((name: string) => rooms.delete(name)),
+  }
+  sockets.push(socket as unknown as FakeSocket)
+  setupWorkspaceFileDocHandlers(
+    socket as unknown as Parameters<typeof setupWorkspaceFileDocHandlers>[0],
+    { isReady: () => true, io: createIo(sockets) } as unknown as IRoomManager
+  )
+  return { socket, handlers }
+}
+
+/** Append a Yjs update to the file's stream, exactly as `publish`/`seedIfEmpty` would. */
+function appendToStream(update: Uint8Array): void {
+  const arr = backing.streams.get(STREAM_KEY) ?? []
+  arr.push({ id: `${++backing.seq}-0`, message: { u: Buffer.from(update).toString('base64') } })
+  backing.streams.set(STREAM_KEY, arr)
+}
+
+/**
+ * A warm room's history: the seed, then a later edit — the "I moved a block, then reloaded" case.
+ * Returns the markdown-equivalent text of each state.
+ */
+function seedWarmStreamHistory(): { intermediate: string; final: string } {
+  const doc = new Y.Doc()
+  doc.getText(FIELD).insert(0, 'AAA')
+  doc.getMap(FILE_DOC_SEED.configMap).set(FILE_DOC_SEED.flag, true)
+  appendToStream(Y.encodeStateAsUpdate(doc))
+  const afterSeed = Y.encodeStateVector(doc)
+  doc.getText(FIELD).insert(0, 'BBB')
+  appendToStream(Y.encodeStateAsUpdate(doc, afterSeed))
+  doc.destroy()
+  return { intermediate: 'AAA', final: 'BBBAAA' }
+}
+
+/** Every document state this socket was ever shown, in order. */
+function statesDeliveredTo(socket: { emit: ReturnType<typeof vi.fn> }): string[] {
+  const clientDoc = new Y.Doc()
+  const states: string[] = []
+  for (const [event, payload] of socket.emit.mock.calls) {
+    if (event !== FILE_DOC_EVENTS.MESSAGE || !(payload instanceof Uint8Array)) continue
+    const decoder = decoding.createDecoder(payload)
+    if (decoding.readVarUint(decoder) !== FILE_DOC_MESSAGE_TYPE.SYNC) continue
+    syncProtocol.readSyncMessage(decoder, encoding.createEncoder(), clientDoc, null)
+    const text = clientDoc.getText(FIELD).toString()
+    if (text !== (states.at(-1) ?? '')) states.push(text)
+  }
+  clientDoc.destroy()
+  return states
+}
+
+/** Let anything the join left running (a catch-up, a seed) settle, so a frame it fans out afterwards
+ *  is counted — that late delivery IS the replay these tests exist to rule out. */
+async function flushPendingWork(): Promise<void> {
+  for (let i = 0; i < 20; i++) await Promise.resolve()
+}
+
+/** Ask the server for its state the way a client does after the join ack. */
+function requestSyncStep2(handlers: Record<string, Handler>): void {
+  const encoder = encoding.createEncoder()
+  encoding.writeVarUint(encoder, FILE_DOC_MESSAGE_TYPE.SYNC)
+  syncProtocol.writeSyncStep1(encoder, new Y.Doc())
+  handlers[FILE_DOC_EVENTS.MESSAGE](encoding.toUint8Array(encoder))
+}
+
+describe('file-doc join readiness (shared store enabled)', () => {
+  /** Every socket the test created, so a room emit can be routed to its members. */
+  const sockets: FakeSocket[] = []
+
+  // One store for the whole file: `initFileDocStore` is idempotent once enabled, so a per-test
+  // shutdown would leave every later test running against a store with closed clients.
+  beforeAll(async () => {
+    await initFileDocStore('redis://fake')
+  })
+
+  afterAll(async () => {
+    await getFileDocStore().shutdown()
+  })
+
+  beforeEach(() => {
+    vi.clearAllMocks()
+    backing.streams.clear()
+    backing.kv.clear()
+    backing.seq = 0
+    backing.readDelayTicks = 0
+    mockAuthorizeRoom.mockResolvedValue({
+      allowed: true,
+      status: 200,
+      workspaceId: 'ws-1',
+      workspacePermission: 'write',
+    })
+    mockFetchFileDocSeed.mockResolvedValue(null)
+  })
+
+  afterEach(() => {
+    cleanupFileDocForSocket('socket-1', createIo(sockets), true)
+    sockets.length = 0
+  })
+
+  it('hands a joiner the final document, never the room history it was rebuilt from', async () => {
+    const { intermediate, final } = seedWarmStreamHistory()
+    // The catch-up read is not instantaneous — the case that made this visible is a cross-region Redis.
+    backing.readDelayTicks = 6
+    const { socket, handlers } = setup('socket-1', sockets)
+
+    await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: FILE_ID, clientId: 1 })
+    requestSyncStep2(handlers)
+    await flushPendingWork()
+
+    // One state, and it is the final one: the client never saw the pre-move document.
+    expect(statesDeliveredTo(socket)).toEqual([final])
+    expect(statesDeliveredTo(socket)).not.toContain(intermediate)
+  })
+
+  it('does not fetch a seed for a room the stream can already reconstruct', async () => {
+    seedWarmStreamHistory()
+    const { handlers } = setup('socket-1', sockets)
+
+    await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: FILE_ID, clientId: 1 })
+
+    expect(mockFetchFileDocSeed).not.toHaveBeenCalled()
+  })
+
+  it('pulls a seed another writer put in the stream instead of waiting for the tailer to push it', async () => {
+    // The seed lock is held by a writer whose room has since been dropped (a fast open→close on a
+    // freshly created file), and its seed lands in the stream. Waiting to be told about it is what
+    // left a new file un-editable until the client's readiness deadline lapsed; the join reads it.
+    backing.kv.set(`filedoc:seedlock:${ROOM_NAME}`, 'held-by-a-writer-that-is-gone')
+    const doc = new Y.Doc()
+    doc.getText(FIELD).insert(0, 'seeded by the writer that held the lock')
+    doc.getMap(FILE_DOC_SEED.configMap).set(FILE_DOC_SEED.flag, true)
+    appendToStream(Y.encodeStateAsUpdate(doc))
+    doc.destroy()
+
+    const { socket, handlers } = setup('socket-1', sockets)
+    await handlers[FILE_DOC_EVENTS.JOIN]({ fileId: FILE_ID, clientId: 1 })
+    requestSyncStep2(handlers)
+    await flushPendingWork()
+
+    expect(mockFetchFileDocSeed).not.toHaveBeenCalled()
+    expect(statesDeliveredTo(socket)).toEqual(['seeded by the writer that held the lock'])
+  })
+})