From 8e2733c676d52e1e25e88737556d1b9d3f04d16e Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Thu, 13 Aug 2026 10:16:24 -0700 Subject: [PATCH 1/3] fix(realtime): keep the file-doc store reconnecting instead of dying quietly MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../src/handlers/file-doc-store.test.ts | 52 +++++++++++++++++- apps/realtime/src/handlers/file-doc-store.ts | 55 +++++++++++++++++-- 2 files changed, 99 insertions(+), 8 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc-store.test.ts b/apps/realtime/src/handlers/file-doc-store.test.ts index 1662a8a3426..40aa1cd1def 100644 --- a/apps/realtime/src/handlers/file-doc-store.test.ts +++ b/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 }[] }[] = [] for (const { key, id } of streams) { @@ -129,7 +144,15 @@ async function newStore(): Promise { 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 = [] }) @@ -137,6 +160,31 @@ describe('FileDocStore', () => { 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() diff --git a/apps/realtime/src/handlers/file-doc-store.ts b/apps/realtime/src/handlers/file-doc-store.ts index c806da5e6be..bc819666b1e 100644 --- a/apps/realtime/src/handlers/file-doc-store.ts +++ b/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 { + 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 { + 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 From 0b7e0adcaaafc5d6c3ec8e9bd0f17486360e9990 Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Thu, 13 Aug 2026 10:26:31 -0700 Subject: [PATCH 2/3] fix(realtime): end the reader's failure streak on an idle read, not a busy one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../src/handlers/file-doc-store.test.ts | 36 +++++++++++++++++-- apps/realtime/src/handlers/file-doc-store.ts | 7 +++- 2 files changed, 40 insertions(+), 3 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc-store.test.ts b/apps/realtime/src/handlers/file-doc-store.test.ts index 40aa1cd1def..7bbb5db475c 100644 --- a/apps/realtime/src/handlers/file-doc-store.test.ts +++ b/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' @@ -74,7 +75,7 @@ function makeClient(): any { 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 }) => { @@ -173,7 +174,7 @@ describe('FileDocStore', () => { 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)) + 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 @@ -185,6 +186,37 @@ describe('FileDocStore', () => { 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 at the cap. + state.backing!.readerClosed = true + const before = state.backing!.reads + await sleep(1900) + const attempts = state.backing!.reads - before + + // Streak reset ⇒ retries at ~0, ~0.5s, ~1.5s ⇒ 3 attempts (2 even if the machine is loaded and + // every sleep overshoots by half). Streak carried over ⇒ ~2s then ~4s ⇒ at most 1. + expect(attempts).toBeGreaterThanOrEqual(2) + doc.destroy() + }) + it('elects exactly one seeder across tasks (no split-brain seed)', async () => { const a = await newStore() const b = await newStore() diff --git a/apps/realtime/src/handlers/file-doc-store.ts b/apps/realtime/src/handlers/file-doc-store.ts index bc819666b1e..537f7f4db12 100644 --- a/apps/realtime/src/handlers/file-doc-store.ts +++ b/apps/realtime/src/handlers/file-doc-store.ts @@ -676,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) @@ -686,7 +692,6 @@ 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 await this.recoverReader(++failures, error) From 064b4aa1da8a0938cd2d70d31067255278ee90e3 Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Thu, 13 Aug 2026 10:34:35 -0700 Subject: [PATCH 3/3] test(realtime): assert the retry delay, not a count inside a window MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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). --- .../src/handlers/file-doc-store.test.ts | 24 +++++++++++++------ 1 file changed, 17 insertions(+), 7 deletions(-) diff --git a/apps/realtime/src/handlers/file-doc-store.test.ts b/apps/realtime/src/handlers/file-doc-store.test.ts index 7bbb5db475c..815b755981f 100644 --- a/apps/realtime/src/handlers/file-doc-store.test.ts +++ b/apps/realtime/src/handlers/file-doc-store.test.ts @@ -20,6 +20,8 @@ interface Backing { 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 } @@ -64,6 +66,7 @@ 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') @@ -152,6 +155,7 @@ describe('FileDocStore', () => { failXAdd: 0, readerClosed: false, reads: 0, + readTimes: [], connects: 0, } stores = [] @@ -205,15 +209,21 @@ describe('FileDocStore', () => { state.backing!.readerClosed = false await sleep(1000) - // A fresh blip must retry at the START of the backoff curve, not at the cap. + // 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 - const before = state.backing!.reads - await sleep(1900) - const attempts = state.backing!.reads - before + 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 ⇒ retries at ~0, ~0.5s, ~1.5s ⇒ 3 attempts (2 even if the machine is loaded and - // every sleep overshoots by half). Streak carried over ⇒ ~2s then ~4s ⇒ at most 1. - expect(attempts).toBeGreaterThanOrEqual(2) + // 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() })