diff --git a/apps/realtime/src/handlers/file-doc-store.test.ts b/apps/realtime/src/handlers/file-doc-store.test.ts index 1662a8a3426..815b755981f 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' @@ -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,6 +65,12 @@ 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 }[] }[] = [] for (const { key, id } of streams) { @@ -59,7 +78,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 }) => { @@ -129,7 +148,16 @@ 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, + readTimes: [], + connects: 0, + } stores = [] }) @@ -137,6 +165,68 @@ 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 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() diff --git a/apps/realtime/src/handlers/file-doc-store.ts b/apps/realtime/src/handlers/file-doc-store.ts index c806da5e6be..537f7f4db12 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) { @@ -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 { + 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