fix(realtime): keep the file-doc store reconnecting instead of dying quietly

A relay that lost Redis for longer than its retry budget did not degrade — it
went silently split-brain and stayed that way. The reconnect strategy returned
an `Error` after ten attempts, which tells node-redis to give up and CLOSE the
client, and a closed client rejects every command with "The client is closed"
for the rest of the process's life. From that point the task kept serving
clients while its rooms stopped receiving other tasks' updates, its own edits
stopped reaching the shared stream (also the crash buffer between persists), and
seeds, locks and the persist If-Match token all failed.

The tail loop then treated that as a transient read error — `running` is only
false during shutdown, so it retried every 500ms forever, one warning per
attempt. A tab left open overnight produced thousands of identical lines, which
is how the actual failure stayed invisible.

- Never stop reconnecting. This process holds live documents whose only
  convergence path is that connection, so a connection it can rebuild is always
  worth rebuilding. Same capped backoff, now via the shared
  `backoffWithJitter`, and no error return.
- Back the reader off after a failed read (500ms → 10s) instead of retrying at
  the read cadence, re-open a client that was CLOSED — node-redis reconnects a
  dropped client, never a closed one — and log the first failure of a streak
  then one in twenty, carrying the streak length, so an outage stays visible
  without burying itself.

Pinned by a test that models a closed connection: six read attempts in three
seconds before, about three after, and proof the reader is re-opened rather
than abandoned.

Author: icecrasher321

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

@@ -15,6 +15,12 @@ interface Backing {
   seq: number
   /** Number of upcoming xAdd calls to fail with a transient error (to exercise publish retry). */
   failXAdd: number
+  /** Set to fail every xRead the way node-redis does once a client has been closed. */
+  readerClosed: boolean
+  /** Failed reads served, so a test can prove the loop is not spinning at the read cadence. */
+  reads: number
+  /** `connect()` calls, so a test can prove a closed reader is re-opened rather than abandoned. */
+  connects: number
 }
 
 const state = vi.hoisted(() => ({ backing: null as Backing | null }))
@@ -27,7 +33,11 @@ function makeClient(): any {
     return state.backing
   }
   const client: any = {
-    connect: async () => {},
+    isOpen: true,
+    connect: async () => {
+      client.isOpen = true
+      b().connects++
+    },
     quit: async () => {},
     on: () => client,
     duplicate: () => makeClient(),
@@ -52,6 +62,11 @@ function makeClient(): any {
       )
     },
     xRead: async (streams: { key: string; id: string }[]) => {
+      b().reads++
+      if (b().readerClosed) {
+        client.isOpen = false
+        throw new Error('The client is closed')
+      }
       const res: { name: string; messages: { id: string; message: Record<string, string> }[] }[] =
         []
       for (const { key, id } of streams) {
@@ -129,14 +144,47 @@ async function newStore(): Promise<FileDocStore> {
 
 describe('FileDocStore', () => {
   beforeEach(() => {
-    state.backing = { streams: new Map(), kv: new Map(), seq: 0, failXAdd: 0 }
+    state.backing = {
+      streams: new Map(),
+      kv: new Map(),
+      seq: 0,
+      failXAdd: 0,
+      readerClosed: false,
+      reads: 0,
+      connects: 0,
+    }
     stores = []
   })
 
   afterEach(async () => {
     await Promise.all(stores.map((s) => s.shutdown()))
   })
 
+  /**
+   * A connection that stops serving reads used to spin the tailer at the read cadence — two attempts a
+   * second, one warning each, forever — while the task quietly stopped converging with every other one.
+   * The loop must back off instead, and re-open a client that was closed rather than reading a dead one.
+   */
+  it('backs off and re-opens the reader when its connection is closed, instead of spinning', async () => {
+    const store = await newStore()
+    const doc = new Y.Doc()
+    await store.attachRoom(NAME, doc)
+    state.backing!.readerClosed = true
+
+    state.backing!.connects = 0 // ignore the two `init` connects; count only recovery attempts
+    const before = state.backing!.reads
+    await new Promise((r) => setTimeout(r, 3000))
+    const attempts = state.backing!.reads - before
+
+    // A fixed 500ms retry manages 6–7 attempts in this window; backing off (500 → 1s → 2s → …) manages
+    // about 3. Exact counts are timing-dependent, so assert the property — it slowed down — not a number.
+    expect(attempts).toBeGreaterThan(0)
+    expect(attempts).toBeLessThanOrEqual(4)
+    // …and it tried to bring the connection back rather than leaving the tailer dead forever.
+    expect(state.backing!.connects).toBeGreaterThan(0)
+    doc.destroy()
+  })
+
   it('elects exactly one seeder across tasks (no split-brain seed)', async () => {
     const a = await newStore()
     const b = await newStore()

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

@@ -160,6 +160,12 @@ const SEED_LOCK_TTL_MS = FILE_DOC_TIMEOUTS.seedRequestMs + 4_000
 const STREAM_TTL_SEC = 600
 /** Refresh every occupied stream's TTL on this cadence, so a live doc's stream never expires. */
 const HEARTBEAT_MS = 60_000
+/** Cap on the delay between reconnection attempts — the strategy retries indefinitely (see `init`). */
+const RECONNECT_MAX_DELAY_MS = 3_000
+/** Cap on the reader's own retry backoff after a failed read. */
+const READER_RETRY_MAX_MS = 10_000
+/** After the first failure of a streak, log one reader failure in this many. */
+const READER_ERROR_LOG_EVERY = 20
 
 const streamKey = (name: string) => `${STREAM_PREFIX}${name}`
 
@@ -246,10 +252,17 @@ export class FileDocStore {
     const options = {
       url: this.redisUrl,
       socket: {
-        reconnectStrategy: (retries: number) => {
-          if (retries > 10) return new Error('FileDocStore Redis reconnection failed')
-          return Math.min(retries * 100, 3000)
-        },
+        /**
+         * Never stop reconnecting. Returning an `Error` here tells node-redis to give up and CLOSE the
+         * client — and a closed client rejects every command with "The client is closed" for the rest of
+         * the process's life. So an outage longer than the retry budget does not degrade this task, it
+         * takes it out silently: its rooms stop receiving other tasks' updates, its own edits stop
+         * reaching the shared stream, seeds and locks fail, and the only symptom is a warning per retry.
+         * This process holds live documents whose sole convergence path is this connection, so a
+         * connection it can rebuild is always worth rebuilding.
+         */
+        reconnectStrategy: (retries: number) =>
+          backoffWithJitter(retries + 1, null, { baseMs: 100, maxMs: RECONNECT_MAX_DELAY_MS }),
       },
     }
     this.write = createClient(options)
@@ -651,6 +664,7 @@ export class FileDocStore {
    * apply new entries. One blocking connection for the whole process regardless of open-file count.
    */
   private async runReader(): Promise<void> {
+    let failures = 0
     while (this.running && this.read) {
       const snapshot = new Map(this.rooms)
       if (snapshot.size === 0) {
@@ -672,14 +686,43 @@ export class FileDocStore {
           if (!room || room !== snapshot.get(name)) continue
           for (const entry of stream.messages) this.applyEntry(room, entry.id, entry.message)
         }
+        failures = 0
       } catch (error) {
         if (!this.running) break
-        logger.warn('FileDocStore reader error; retrying', { error: getErrorMessage(error) })
-        await sleep(500)
+        await this.recoverReader(++failures, error)
       }
     }
   }
 
+  /**
+   * A failed read is either a transient blip or a connection that is gone, and this loop cannot tell
+   * them apart — so it backs off instead of retrying at the read cadence. Without that, a connection
+   * that cannot serve reads spins this loop forever at two attempts a second, one warning each, which
+   * is how an outage turns into thousands of identical log lines that bury the reason for it.
+   *
+   * It also re-opens a CLOSED client. node-redis reconnects a client that merely dropped, but never one
+   * it has closed; the strategy above no longer closes one, so this covers a client closed some other
+   * way (an explicit disconnect, a shutdown that raced a read) rather than leaving the tailer dead.
+   *
+   * Logs the first failure of a streak and then one in every {@link READER_ERROR_LOG_EVERY}, carrying
+   * the streak length, so a real outage stays visible without filling the log.
+   */
+  private async recoverReader(failures: number, error: unknown): Promise<void> {
+    if (failures === 1 || failures % READER_ERROR_LOG_EVERY === 0) {
+      logger.warn(`FileDocStore reader failed ${failures}x in a row; retrying`, {
+        error: getErrorMessage(error),
+      })
+    }
+    await sleep(backoffWithJitter(failures, null, { baseMs: 500, maxMs: READER_RETRY_MAX_MS }))
+    if (this.running && this.read && !this.read.isOpen) {
+      await this.read.connect().catch((reconnectError) => {
+        logger.warn('FileDocStore could not re-open the reader connection', {
+          error: getErrorMessage(reconnectError),
+        })
+      })
+    }
+  }
+
   /**
    * Snapshot-then-trim compaction: append a full-state snapshot and drop the older deltas it subsumes,
    * so the stream stays bounded while a fresh task can still catch up from the head. Lock-guarded so