soulcraft/brainy
2
0
Fork 0
Code Issues 1 Pull requests Projects Releases 335 Packages Wiki Activity Actions

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

Browse source 
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.
This commit is contained in:
David Snelling 2026-07-13 09:31:25 -07:00
parent d8301f8d08
commit 54c183668c
4 changed files with 302 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

143
src/db/generationStore.ts
View file

@ -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) =>
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 }