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

* 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.

* fix(realtime): end the reader's failure streak on an idle read, not a busy one

Review findings, both accurate.

The streak reset sat after the entries were applied, so a blocking read that
timed out with nothing new — the idle steady state — skipped it via `continue`.
A healed outage's count therefore survived through normal polling, and the next
unrelated blip opened at the backoff cap: minutes of avoidable split-brain, and
a log line claiming a failure count it never earned. The streak now ends on the
read RETURNING, which is what proves the connection works.

Also: the new test built a raw `setTimeout` promise instead of the shared
`sleep`, which CLAUDE.md calls out by name.

* test(realtime): assert the retry delay, not a count inside a window

The streak-reset guard could pass on the very regression it exists to catch. It
counted read attempts inside a 1900ms window, and the jittered delay for a
carried streak is 1600–2400ms — so whenever jitter landed below about 0.95, a
second read fell inside the window and the assertion held even though the idle
reads had never cleared `failures`. A single falsification run happened to draw
a long delay, which is exactly how a guard like this goes quiet.

Assert the delay itself instead. The first retry after a reset is 500ms ±20%
(400–600ms); carried over it is the third, 2000ms ±20% (1600–2400ms). Those
ranges are disjoint, so the check no longer depends on which jitter is drawn:
against the old placement it now fails every time (measured 2160ms, 1925ms,
2046ms against the 1000ms bound).

Author: icecrasher321

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