fix: refuse writes when single-op history cannot be made durable

The async group-commit flush that persists single-op generation history
swallowed persist failures as a bare warn: writes kept succeeding while their
before-images piled up in memory, never durable and unbounded, with no signal.

The generation store now accounts for every failed flush at one place
(flushPendingSingleOps), tolerates a transient blip (retry with capped
exponential backoff), and after PENDING_FLUSH_FAILURE_THRESHOLD consecutive
failures LATCHES a durability failure and refuses further single-op and transact
writes with a typed, exported PendingFlushDurabilityError — rather than promise a
durability it cannot deliver. Live canonical data is untouched; only the
immutable history is stuck. The latch self-heals: the moment a flush succeeds
(a retry, or an explicit flush()/close()) it lifts and writes resume.

Loud errors, never quiet losses.

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
+  }
+}

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<number, number>()
 
@@ -592,6 +612,11 @@ export class GenerationStore {
     execute: () => Promise<void>
   }): 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<void> {
     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<void> {
     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) =>
+        .catch(() => {})
-          prodLog.warn(`[GenerationStore] pending single-op flush failed: ${(err as Error).message}`)
-        )
       return
     }
     if (this.pendingFlushTimer !== null) return
     this.pendingFlushTimer = setTimeout(() => {
       this.pendingFlushTimer = null
-      void this.flushPendingSingleOps().catch((err) =>
+      void this.flushPendingSingleOps().catch(() => {})
-        prodLog.warn(`[GenerationStore] pending single-op flush failed: ${(err as Error).message}`)
-      )
     }, 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 }

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 {

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<string, unknown> => ({
+  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()
+  })
+})