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

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

@@ -1,6 +1,7 @@
 /**
  * @vitest-environment node
  */
+import { sleep } from '@sim/utils/helpers'
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import * as Y from 'yjs'
 
@@ -15,6 +16,14 @@ 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
+  /** When each read was attempted, so a test can assert the BACKOFF rather than a count in a window. */
+  readTimes: 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 +36,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,14 +65,20 @@ function makeClient(): any {
       )
     },
     xRead: async (streams: { key: string; id: string }[]) => {
+      b().reads++
+      b().readTimes.push(Date.now())
+      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) {
         const after = (b().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))
+      await sleep(5)
       return null
     },
     set: async (key: string, val: string, opts?: { NX?: boolean }) => {
@@ -129,14 +148,85 @@ 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,
+      readTimes: [],
+      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 sleep(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()
+  })
+
+  /**
+   * The streak has to end on a read that RETURNS, not on one that carries messages: a blocking read
+   * timing out with nothing new is the idle steady state. Counting only message-bearing reads would
+   * keep a healed outage's streak alive through normal polling, so the next unrelated blip would open
+   * at the backoff cap — minutes of unnecessary split-brain — and log a count it never earned.
+   */
+  it('ends the failure streak on an idle read, so a later blip starts over', async () => {
+    const store = await newStore()
+    const doc = new Y.Doc()
+    await store.attachRoom(NAME, doc)
+
+    // Build a streak of two failures (retries back off ~0.5s, then ~1s).
+    state.backing!.readerClosed = true
+    await sleep(800)
+    // Redis comes back. Wait past the pending backoff so a read actually lands — and it returns
+    // nothing new, which is the idle case this test is about.
+    state.backing!.readerClosed = false
+    await sleep(1000)
+
+    // A fresh blip must retry at the START of the backoff curve, not partway up it. Assert the DELAY
+    // itself: counting attempts inside a fixed window cannot tell the two apart, because the jittered
+    // delay for a carried streak (1.6–2.4s) overlaps any window wide enough to catch a reset one.
+    state.backing!.readerClosed = true
+    state.backing!.readTimes.length = 0
+    await vi.waitFor(() => expect(state.backing!.readTimes.length).toBeGreaterThanOrEqual(2), {
+      timeout: 5000,
+      interval: 50,
+    })
+    const [first, second] = state.backing!.readTimes
+
+    // Streak reset ⇒ the first delay is 500ms ±20% ⇒ 400–600ms. Streak carried over ⇒ it is the third
+    // delay, 2000ms ±20% ⇒ 1600–2400ms. Disjoint ranges, so this cannot pass on the wrong one without
+    // the machine stalling the shorter sleep by 65%.
+    expect(second - first).toBeLessThan(1000)
+    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) {
@@ -662,6 +676,12 @@ export class FileDocStore {
           [...snapshot].map(([name, room]) => ({ key: streamKey(name), id: room.lastId })),
           { BLOCK: READ_BLOCK_MS, COUNT: READ_COUNT }
         )
+        // The streak ends HERE, on the read returning at all — not further down once entries are
+        // applied. A blocking read that times out with nothing new is the idle steady state, and it
+        // proves the connection works just as well as one carrying messages; leaving the streak
+        // standing through it would keep an old outage's count alive indefinitely, so the next
+        // unrelated blip would open at the backoff cap and log a failure count it never earned.
+        failures = 0
         if (!res) continue
         for (const stream of res) {
           const name = stream.name.slice(STREAM_PREFIX.length)
@@ -674,12 +694,40 @@ export class FileDocStore {
         }
       } 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