diff --git a/src/db/errors.ts b/src/db/errors.ts index 7cf9c3d0..22f405be 100644 --- a/src/db/errors.ts +++ b/src/db/errors.ts @@ -230,3 +230,62 @@ export class StoreInconsistentError extends Error { this.cause = cause } } + +/** + * @description Thrown by a write when the store cannot make single-op + * generation **history** durable: the asynchronous group-commit flush that + * persists buffered before-images to disk has failed repeatedly (a real + * storage fault — a full disk, an EIO, a permission loss — not a transient + * blip). Rather than keep accepting writes whose history silently piles up in + * memory and is never persisted — an invisible durability loss, and an + * unbounded leak — the store LATCHES this failure and refuses further writes + * until the pending tier drains. + * + * This is the loud, honest response to a broken history-durability path: + * - Live canonical data is unaffected (single-op live bytes are written before + * the history flush; only the immutable before-image history is stuck). + * - The background flush keeps retrying with backoff; when the underlying fault + * clears and a flush succeeds, the latch lifts and writes resume + * automatically. An explicit `flush()` also clears it on success. + * - {@link cause} is the underlying storage error from the last failed flush. + * + * A caller seeing this should treat it exactly like a full disk: stop writing, + * resolve the storage fault, and retry. It is NOT a data-corruption error — no + * committed generation is lost — it is a refusal to *promise* durability the + * store currently cannot deliver. + * + * @example + * try { + * await brain.add({ ... }) + * } catch (err) { + * if (err instanceof PendingFlushDurabilityError) { + * // History can't be persisted right now (disk fault). Resolve storage, + * // then retry — the store self-heals once a flush succeeds. + * console.error('history durability stalled:', err.cause) + * } + * } + */ +export class PendingFlushDurabilityError extends Error { + /** The underlying storage error from the most recent failed flush. */ + public override readonly cause: Error + /** How many consecutive flush attempts had failed when the latch tripped. */ + public readonly failedAttempts: number + + /** + * @param cause - The storage error from the last failed pending-tier flush. + * @param failedAttempts - Consecutive failed flush attempts at latch time. + */ + constructor(cause: Error, failedAttempts: number) { + super( + `Write refused: single-op generation history could not be made durable ` + + `after ${failedAttempts} consecutive flush attempts (${cause.message}). ` + + `Live data is intact, but buffered history is not yet persisted — the ` + + `store refuses further writes rather than silently accumulate ` + + `un-durable history. Resolve the underlying storage fault; the store ` + + `self-heals and resumes writes once a flush succeeds. Cause: ${cause.message}` + ) + this.name = 'PendingFlushDurabilityError' + this.cause = cause + this.failedAttempts = failedAttempts + } +} diff --git a/src/db/generationStore.ts b/src/db/generationStore.ts index 5492ed70..d7a302c9 100644 --- a/src/db/generationStore.ts +++ b/src/db/generationStore.ts @@ -32,7 +32,7 @@ */ import { prodLog } from '../utils/logger.js' -import { GenerationCompactedError, GenerationConflictError, StoreInconsistentError } from './errors.js' +import { GenerationCompactedError, GenerationConflictError, PendingFlushDurabilityError, StoreInconsistentError } from './errors.js' import type { UnreconciledRecord } from './errors.js' import { TransactionRollbackError } from '../transaction/errors.js' import type { @@ -284,6 +284,26 @@ export class GenerationStore { /** Coalescing window (ms) before an idle pending tier is flushed. */ private static readonly PENDING_FLUSH_DELAY_MS = 50 + /** + * Latched pending-flush durability failure. Set once background flushes have + * failed {@link PENDING_FLUSH_FAILURE_THRESHOLD} consecutive times (a real, + * persistent storage fault — not a blip); while set, writes are REFUSED with + * a {@link PendingFlushDurabilityError} rather than silently accumulate + * un-durable history in memory. Cleared the moment a flush finally succeeds + * (the store self-heals). `null` = history-durability path is healthy. + */ + private pendingFlushError: Error | null = null + /** Consecutive failed background pending-flush attempts (resets on success). */ + private pendingFlushFailures = 0 + /** + * Consecutive failed flush attempts before the durability latch trips and + * writes start refusing. A small tolerance so a single transient fault does + * not trip the alarm, while a genuinely stuck disk stops the bleed fast. + */ + private static readonly PENDING_FLUSH_FAILURE_THRESHOLD = 3 + /** Upper bound (ms) on the exponential retry backoff between failed flushes. */ + private static readonly PENDING_FLUSH_RETRY_CAP_MS = 30_000 + /** Live pin refcounts, keyed by pinned generation. */ private readonly pins = new Map() @@ -592,6 +612,11 @@ export class GenerationStore { execute: () => Promise }): Promise<{ generation: number; timestamp: number }> { return this.withMutex(async () => { + // A latched history-durability failure compromises the whole generation + // spine — refuse a transact too (advancing the manifest past stuck, + // un-durable single-op generations would be inconsistent). Same loud + // error; self-clears when the pending tier drains. + this.assertHistoryDurable() if (args.ifAtGeneration !== undefined && args.ifAtGeneration !== this.counter) { throw new GenerationConflictError(args.ifAtGeneration, this.counter) } @@ -904,6 +929,11 @@ export class GenerationStore { precommit?: (before: CommitBeforeImages) => void }): Promise<{ generation: number; timestamp: number; degraded?: string[] }> { return this.withMutex(async () => { + // Refuse to accept a write whose history we cannot make durable: if the + // pending-tier flush has latched a persistent failure, accepting more + // writes would silently pile un-durable before-images into memory. Fail + // loud instead (the latch self-clears when a flush finally succeeds). + this.assertHistoryDurable() const nouns = args.touched.nouns ? [...new Set(args.touched.nouns)] : [] const verbs = args.touched.verbs ? [...new Set(args.touched.verbs)] : [] const gen = ++this.counter @@ -1030,6 +1060,22 @@ export class GenerationStore { */ async flushPendingSingleOps(): Promise { if (this.pendingGens.length === 0) return + // Centralize durability accounting here so EVERY failure path — the + // background scheduler AND an explicit flush()/close() — latches + // consistently, and success (from any caller) clears the latch. The + // scheduler owns only the retry cadence. + try { + await this.flushPendingSingleOpsUnlocked() + this.onPendingFlushSuccess() + } catch (err) { + this.recordPendingFlushFailure(err as Error) + throw err + } + } + + /** The actual pending-tier flush under the mutex (see {@link flushPendingSingleOps}, + * which wraps this with durability accounting). */ + private async flushPendingSingleOpsUnlocked(): Promise { return this.withMutex(async () => { if (this.pendingGens.length === 0) return this.clearPendingFlushTimer() @@ -1140,24 +1186,105 @@ export class GenerationStore { }) } + /** + * @description Reset the durability-failure state after a successful flush. + * If a failure was latched (writes were being refused), log the recovery so + * the transition out of the loud state is as visible as the transition in. + */ + private onPendingFlushSuccess(): void { + if (this.pendingFlushError) { + prodLog.info( + `[GenerationStore] pending single-op history flush recovered after ` + + `${this.pendingFlushFailures} failed attempt(s); writes resume.` + ) + } + this.pendingFlushFailures = 0 + this.pendingFlushError = null + } + + /** + * @description Account for a failed pending-tier flush (called from + * {@link flushPendingSingleOps}' catch, so it fires on EVERY failure path — + * background scheduler and explicit flush()/close() alike). Escalates a + * transient blip (a warn, keep retrying) into a latched durability failure + * once {@link PENDING_FLUSH_FAILURE_THRESHOLD} consecutive attempts fail — + * from that point writes are refused with a {@link PendingFlushDurabilityError} + * until a flush finally succeeds. Always ensures a backoff retry is scheduled + * so a recovering disk self-heals without operator action, regardless of who + * triggered the failing flush. The pending before-images are retained (the + * failed flush never cleared them), so no history is lost — only not-yet-durable. + */ + private recordPendingFlushFailure(err: Error): void { + this.pendingFlushFailures++ + const attempts = this.pendingFlushFailures + if (attempts >= GenerationStore.PENDING_FLUSH_FAILURE_THRESHOLD) { + // Latch: refuse further writes rather than pile un-durable history. + this.pendingFlushError = err + prodLog.error( + `[GenerationStore] pending single-op history flush has FAILED ${attempts} ` + + `consecutive times (${err.message}). Generation history is not durable; ` + + `writes are now REFUSED until it drains. Live canonical data is intact. ` + + `Retrying with backoff; the store resumes writes when a flush succeeds.` + ) + } else { + prodLog.warn( + `[GenerationStore] pending single-op flush failed (attempt ${attempts}/` + + `${GenerationStore.PENDING_FLUSH_FAILURE_THRESHOLD}): ${err.message}; retrying with backoff.` + ) + } + this.scheduleFlushRetry(attempts) + } + + /** + * @description (Re)arm the pending-flush timer at a capped exponential + * backoff after a failure, so a persistent fault is retried at a decaying + * cadence instead of hot-looping. Shares the single {@link pendingFlushTimer} + * slot with {@link schedulePendingFlush} (only one flush is ever scheduled). + * The retry's own failure is re-accounted inside {@link flushPendingSingleOps}. + */ + private scheduleFlushRetry(attempt: number): void { + if (this.pendingFlushTimer !== null) return + const backoff = Math.min( + GenerationStore.PENDING_FLUSH_DELAY_MS * 2 ** Math.min(attempt, 6), + GenerationStore.PENDING_FLUSH_RETRY_CAP_MS + ) + this.pendingFlushTimer = setTimeout(() => { + this.pendingFlushTimer = null + // flushPendingSingleOps records the failure + reschedules internally. + void this.flushPendingSingleOps().catch(() => {}) + }, backoff) + const t = this.pendingFlushTimer as unknown as { unref?: () => void } + if (t && typeof t.unref === 'function') t.unref() + } + + /** + * @description Throw if the store has a latched history-durability failure. + * Called at the top of every write commit so a broken durability path fails + * loud and immediately instead of silently accumulating un-durable history. + */ + private assertHistoryDurable(): void { + if (this.pendingFlushError) { + throw new PendingFlushDurabilityError(this.pendingFlushError, this.pendingFlushFailures) + } + } + /** Schedule a coalesced pending-tier flush (size trigger fires immediately on * the next microtask; otherwise a {@link PENDING_FLUSH_DELAY_MS} timer). Both - * defer outside the current mutex section so the flush can re-acquire it. */ + * defer outside the current mutex section so the flush can re-acquire it. A + * failed flush is accounted + rescheduled inside {@link flushPendingSingleOps} + * (latch-on-persistent-failure), never swallowed as a bare log line. */ private schedulePendingFlush(): void { if (this.pendingGens.length >= this.pendingFlushThreshold) { + // Failure is recorded + retried inside flushPendingSingleOps. void Promise.resolve() .then(() => this.flushPendingSingleOps()) - .catch((err) => - prodLog.warn(`[GenerationStore] pending single-op flush failed: ${(err as Error).message}`) - ) + .catch(() => {}) return } if (this.pendingFlushTimer !== null) return this.pendingFlushTimer = setTimeout(() => { this.pendingFlushTimer = null - void this.flushPendingSingleOps().catch((err) => - prodLog.warn(`[GenerationStore] pending single-op flush failed: ${(err as Error).message}`) - ) + void this.flushPendingSingleOps().catch(() => {}) }, GenerationStore.PENDING_FLUSH_DELAY_MS) // A background flush must not keep the process alive (Node only). const t = this.pendingFlushTimer as unknown as { unref?: () => void } diff --git a/src/index.ts b/src/index.ts index 3915025d..63a07690 100644 --- a/src/index.ts +++ b/src/index.ts @@ -175,7 +175,8 @@ export { GenerationConflictError, SpeculativeOverlayError, GenerationCompactedError, - StoreInconsistentError + StoreInconsistentError, + PendingFlushDurabilityError } from './db/errors.js' export type { UnreconciledRecord } from './db/errors.js' export type { diff --git a/tests/unit/db/pending-flush-durability.test.ts b/tests/unit/db/pending-flush-durability.test.ts new file mode 100644 index 00000000..f8501c46 --- /dev/null +++ b/tests/unit/db/pending-flush-durability.test.ts @@ -0,0 +1,106 @@ +/** + * @module tests/unit/db/pending-flush-durability + * @description Finding 8: the async group-commit flush that persists single-op + * generation HISTORY must not swallow persist failures. Before the fix a failed + * background flush was a bare `prodLog.warn` — writes kept succeeding while their + * before-images silently piled up in memory, never durable and unbounded. + * + * The contract now: + * - a single transient flush failure does NOT trip the alarm (tolerance), and + * - after PENDING_FLUSH_FAILURE_THRESHOLD consecutive failures the store LATCHES + * and REFUSES further writes with a typed PendingFlushDurabilityError rather + * than accumulate un-durable history, and + * - it self-heals: once a flush finally succeeds the latch lifts and writes resume. + * Live canonical data is never touched — only the immutable history is stuck. + * + * Fake timers keep the background retry/coalesce timers from firing, so the exact + * number of flush attempts (and thus the latch point) is deterministic — the test + * drives every flush explicitly. + */ +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest' +import { MemoryStorage } from '../../../src/storage/adapters/memoryStorage.js' +import { GenerationStore } from '../../../src/db/generationStore.js' +import { PendingFlushDurabilityError } from '../../../src/db/errors.js' +import { NounType } from '../../../src/types/graphTypes.js' + +const ID = (suffix: string): string => `00000000-0000-4000-8000-0000000000${suffix}` +const meta = (v: number): Record => ({ + noun: NounType.Document, + subtype: 'note', + data: `payload-v${v}`, + version: v, + _rev: v +}) + +describe('db/GenerationStore pending-flush durability (finding 8)', () => { + let storage: MemoryStorage + let store: GenerationStore + + beforeEach(async () => { + vi.useFakeTimers() + storage = new MemoryStorage() + await storage.init() + store = new GenerationStore(storage) + await store.open() + // High size threshold so single-ops never auto-flush — the test owns every flush. + ;(store as any).pendingFlushThreshold = 1000 + }) + + afterEach(() => { + vi.useRealTimers() + }) + + /** One single-op write that buffers a pending generation. */ + const singleOp = (id: string, v: number) => + store.commitSingleOp({ + touched: { nouns: [id], verbs: [] }, + execute: async () => { + await storage.saveNounMetadata(id, meta(v)) + } + }) + + it('latches after repeated flush failures, refuses writes, then self-heals', async () => { + // Buffer a pending generation while storage is healthy. + await singleOp(ID('aa'), 1) + + // Fault the HISTORY flush (raw generation-record writes) — NOT the live + // canonical write, which uses a different path (saveNounMetadata). + const fault = Object.assign(new Error('EIO simulated'), { code: 'EIO' }) + const spy = vi.spyOn(storage as any, 'writeRawObject').mockRejectedValue(fault) + + // Each explicit flush fails and is accounted; the third trips the latch. + for (let i = 0; i < 3; i++) { + await expect(store.flushPendingSingleOps()).rejects.toThrow('EIO simulated') + } + + // Writes are now refused LOUDLY rather than piling up un-durable history. + await expect(singleOp(ID('bb'), 1)).rejects.toBeInstanceOf(PendingFlushDurabilityError) + // The live counter did not advance for the refused write (no generation consumed). + expect(store.generation()).toBe(1) + + // Storage recovers → an explicit flush drains the retained tier and lifts the latch. + spy.mockRestore() + await expect(store.flushPendingSingleOps()).resolves.toBeUndefined() + + // Writes resume; the next single-op commits normally. + const receipt = await singleOp(ID('cc'), 1) + expect(receipt.generation).toBe(2) + }) + + it('a single transient flush failure does NOT latch — writes continue', async () => { + await singleOp(ID('aa'), 1) + + const spy = vi.spyOn(storage as any, 'writeRawObject').mockRejectedValue(new Error('blip')) + await expect(store.flushPendingSingleOps()).rejects.toThrow('blip') // failure #1, below threshold + spy.mockRestore() + + // Below the latch threshold → the store still accepts writes. + const receipt = await singleOp(ID('bb'), 1) + expect(receipt.generation).toBe(2) + + // A subsequent healthy flush drains the tier and resets the failure counter. + await expect(store.flushPendingSingleOps()).resolves.toBeUndefined() + expect((store as any).pendingFlushFailures).toBe(0) + expect((store as any).pendingFlushError).toBeNull() + }) +})