feat(persistence): the engine owns its flush cadence — callers never call flush() in hot paths again
A4 of the service-class pair (SELF-ENGINE-LIFECYCLE-SPRINT, David-directed: 'why do we need manual flushes at all?'). The production disease: 829 caller-scheduled per-write flushes convoying into 45-66s write walls — cadence hand-rolled a layer above the only layer that can see dirty-node counts and IO pressure. - BrainyConfig.persistence: policy 'auto' (DEFAULT) | 'manual', with flushEveryWrites (512) / flushIntervalMs (30s) / flushOnIdleMs (2s) triggers. Auto = the engine kicks ONE single-flight BACKGROUND flush at a threshold or when the store goes quiet; write acks NEVER await it (a hung flush cannot block a write — pinned); a failed background flush is LOUD and re-arms the trigger. 'manual' restores caller-owned cadence. - Triggers wired at both write chokepoints (single-op post-commit + transact post-commit); idle timer unref'd; close() tears the timer down and drains the flight before its own final flush. - RECOVERY SEMANTICS documented on the config: canonical records are durable per-write regardless of policy — a crash between background flushes loses derived state only, which converges at next open (epoch machinery + the new incremental aggregation catch-up), bounded by the un-flushed window. Never data loss. Pins: write-count trigger fires one background flush with zero caller calls · idle trigger · manual never self-flushes · THE ACK LAW (writes acknowledge under a never-resolving flush). Gates: unit 1917/1917 · integration 760 · conformance 27/27 — green WITH auto as the default.
This commit is contained in:
parent
1dc861d299
commit
3236a01bef
3 changed files with 214 additions and 1 deletions
|
|
@ -272,6 +272,7 @@ type ResolvedBrainyConfig = Required<
|
||||||
| 'eagerEmbeddings'
|
| 'eagerEmbeddings'
|
||||||
| 'migrationWaitTimeoutMs'
|
| 'migrationWaitTimeoutMs'
|
||||||
| 'transactionBudgetFloorMs'
|
| 'transactionBudgetFloorMs'
|
||||||
|
| 'persistence'
|
||||||
>
|
>
|
||||||
> &
|
> &
|
||||||
Pick<
|
Pick<
|
||||||
|
|
@ -285,6 +286,7 @@ type ResolvedBrainyConfig = Required<
|
||||||
| 'eagerEmbeddings'
|
| 'eagerEmbeddings'
|
||||||
| 'migrationWaitTimeoutMs'
|
| 'migrationWaitTimeoutMs'
|
||||||
| 'transactionBudgetFloorMs'
|
| 'transactionBudgetFloorMs'
|
||||||
|
| 'persistence'
|
||||||
>
|
>
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
@ -684,6 +686,14 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
private _aggregationIndex?: AggregationIndex // Incremental aggregation engine
|
private _aggregationIndex?: AggregationIndex // Incremental aggregation engine
|
||||||
private _aggregationBackfillFlight: Promise<void> | null = null // Single-flight backfill walk
|
private _aggregationBackfillFlight: Promise<void> | null = null // Single-flight backfill walk
|
||||||
private _aggregationCatchUpFlight: Promise<void> | null = null // Single-flight behind-stamp catch-up
|
private _aggregationCatchUpFlight: Promise<void> | null = null // Single-flight behind-stamp catch-up
|
||||||
|
|
||||||
|
// ENGINE-OWNED PERSISTENCE CADENCE (SELF-ENGINE-LIFECYCLE-SPRINT):
|
||||||
|
// write-count / interval / idle triggers → ONE background flush at a time.
|
||||||
|
// Write acks NEVER await it; a failed background flush is LOUD and re-armed.
|
||||||
|
private _persistDirtyWrites = 0
|
||||||
|
private _persistLastFlushAt = Date.now()
|
||||||
|
private _persistIdleTimer: ReturnType<typeof setTimeout> | null = null
|
||||||
|
private _persistBackgroundFlight: Promise<void> | null = null
|
||||||
// A failed walk latches its error: retries within the cooldown rethrow it
|
// A failed walk latches its error: retries within the cooldown rethrow it
|
||||||
// instantly instead of re-walking, so a tight caller-side retry loop costs
|
// instantly instead of re-walking, so a tight caller-side retry loop costs
|
||||||
// one loud error per query, never a full store walk per query.
|
// one loud error per query, never a full store walk per query.
|
||||||
|
|
@ -1829,6 +1839,64 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
* @param run - The single-op's existing operation batch builder (the
|
* @param run - The single-op's existing operation batch builder (the
|
||||||
* `tx => {…}` body previously passed straight to `executeTransaction`).
|
* `tx => {…}` body previously passed straight to `executeTransaction`).
|
||||||
*/
|
*/
|
||||||
|
/**
|
||||||
|
* @description The write-side persistence trigger (policy `'auto'`): count
|
||||||
|
* the committed write, kick a single-flight BACKGROUND flush when the
|
||||||
|
* write-count or interval threshold is crossed, and (re)arm the idle
|
||||||
|
* timer. Never awaited by the write path — the ack is already durable at
|
||||||
|
* the canonical layer; this schedules DERIVED-state persistence on the
|
||||||
|
* engine's own cadence (callers never call flush() in hot paths).
|
||||||
|
*/
|
||||||
|
private noteWriteForPersistence(): void {
|
||||||
|
const cfg = this.config.persistence
|
||||||
|
if (this.isReadOnly || cfg?.policy === 'manual') return
|
||||||
|
this._persistDirtyWrites++
|
||||||
|
const every = cfg?.flushEveryWrites ?? 512
|
||||||
|
const intervalMs = cfg?.flushIntervalMs ?? 30_000
|
||||||
|
const idleMs = cfg?.flushOnIdleMs ?? 2_000
|
||||||
|
|
||||||
|
if (
|
||||||
|
this._persistDirtyWrites >= every ||
|
||||||
|
Date.now() - this._persistLastFlushAt >= intervalMs
|
||||||
|
) {
|
||||||
|
this.kickBackgroundFlush('threshold')
|
||||||
|
}
|
||||||
|
|
||||||
|
if (this._persistIdleTimer) clearTimeout(this._persistIdleTimer)
|
||||||
|
const timer = setTimeout(() => {
|
||||||
|
this._persistIdleTimer = null
|
||||||
|
if (this._persistDirtyWrites > 0) this.kickBackgroundFlush('idle')
|
||||||
|
}, idleMs)
|
||||||
|
// Never hold the process open for a cadence timer.
|
||||||
|
;(timer as { unref?: () => void }).unref?.()
|
||||||
|
this._persistIdleTimer = timer
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @description Start (or join) the ONE background flush. The dirty counter
|
||||||
|
* resets at kick time so writes landing during the flush re-accumulate
|
||||||
|
* toward the next trigger. A failure is LOUD and leaves the writes counted
|
||||||
|
* again — silence is not an option, and neither is a retry storm (the next
|
||||||
|
* trigger re-attempts).
|
||||||
|
*/
|
||||||
|
private kickBackgroundFlush(reason: 'threshold' | 'idle'): void {
|
||||||
|
if (this._persistBackgroundFlight) return
|
||||||
|
const counted = this._persistDirtyWrites
|
||||||
|
this._persistDirtyWrites = 0
|
||||||
|
this._persistLastFlushAt = Date.now()
|
||||||
|
this._persistBackgroundFlight = this.flush()
|
||||||
|
.catch((err) => {
|
||||||
|
this._persistDirtyWrites += counted // re-arm the trigger honestly
|
||||||
|
prodLog.error(
|
||||||
|
`[Brainy] background flush (${reason}) FAILED: ${(err as Error).message} — ` +
|
||||||
|
`derived-state persistence retries at the next trigger; canonical data is unaffected`
|
||||||
|
)
|
||||||
|
})
|
||||||
|
.finally(() => {
|
||||||
|
this._persistBackgroundFlight = null
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
private async persistSingleOp(
|
private async persistSingleOp(
|
||||||
touched: { nouns?: string[]; verbs?: string[] },
|
touched: { nouns?: string[]; verbs?: string[] },
|
||||||
run: TransactionFunction<void>,
|
run: TransactionFunction<void>,
|
||||||
|
|
@ -1921,6 +1989,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
this.noteWriteForPersistence()
|
||||||
return receipt
|
return receipt
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -7714,6 +7783,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
// A rejected batch throws at commitTransaction and never reaches here.
|
// A rejected batch throws at commitTransaction and never reaches here.
|
||||||
this.emitCommitted(plan.changeEvents, undefined, generation, timestamp)
|
this.emitCommitted(plan.changeEvents, undefined, generation, timestamp)
|
||||||
|
|
||||||
|
this.noteWriteForPersistence()
|
||||||
const receipt: TransactReceipt = { generation, timestamp, ids: plan.ids }
|
const receipt: TransactReceipt = { generation, timestamp, ids: plan.ids }
|
||||||
return this.createPinnedDb({ generation, timestamp, receipt })
|
return this.createPinnedDb({ generation, timestamp, receipt })
|
||||||
}
|
}
|
||||||
|
|
@ -14857,7 +14927,10 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
requireSubtype: config?.requireSubtype ?? true,
|
requireSubtype: config?.requireSubtype ?? true,
|
||||||
// Multi-process safety
|
// Multi-process safety
|
||||||
mode: config?.mode ?? 'writer',
|
mode: config?.mode ?? 'writer',
|
||||||
force: config?.force ?? false
|
force: config?.force ?? false,
|
||||||
|
// Engine-owned persistence cadence — defaults resolve at the trigger
|
||||||
|
// site (policy 'auto': 512 writes / 30s interval / 2s idle).
|
||||||
|
persistence: config?.persistence
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -16401,6 +16474,16 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
* This ensures deferred persistence mode data is saved
|
* This ensures deferred persistence mode data is saved
|
||||||
*/
|
*/
|
||||||
async close(): Promise<void> {
|
async close(): Promise<void> {
|
||||||
|
// Persistence cadence teardown: no background flush may fire after close
|
||||||
|
// begins (close() runs its own final flush).
|
||||||
|
if (this._persistIdleTimer) {
|
||||||
|
clearTimeout(this._persistIdleTimer)
|
||||||
|
this._persistIdleTimer = null
|
||||||
|
}
|
||||||
|
if (this._persistBackgroundFlight) {
|
||||||
|
await this._persistBackgroundFlight.catch(() => {})
|
||||||
|
}
|
||||||
|
|
||||||
// Cancel any pending post-import background deduplication FIRST — it is a
|
// Cancel any pending post-import background deduplication FIRST — it is a
|
||||||
// writer (merge-deletes), and no delete pass may start mid- or post-close.
|
// writer (merge-deletes), and no delete pass may start mid- or post-close.
|
||||||
this._backgroundDedup?.cancelPending()
|
this._backgroundDedup?.cancelPending()
|
||||||
|
|
|
||||||
|
|
@ -2028,6 +2028,39 @@ export interface BrainyConfig {
|
||||||
*/
|
*/
|
||||||
force?: boolean
|
force?: boolean
|
||||||
|
|
||||||
|
/**
|
||||||
|
* THE ENGINE OWNS ITS FLUSH CADENCE (the persistence policy —
|
||||||
|
* SELF-ENGINE-LIFECYCLE-SPRINT, David-directed: "why do we need manual
|
||||||
|
* flushes at all?"). Under `'auto'` (the DEFAULT) the engine schedules
|
||||||
|
* single-flight background flushes itself — triggered by write count,
|
||||||
|
* elapsed time, and idle — so callers NEVER call `flush()` in a hot path
|
||||||
|
* (a production consumer's 829 per-write flushes convoyed into 45–66s
|
||||||
|
* write walls; the cadence belongs to the layer that can see dirty-node
|
||||||
|
* counts and IO pressure). `flush()` remains public as an awaitable
|
||||||
|
* durability BARRIER for the rare "must be on disk before I proceed"
|
||||||
|
* moment — calling it is never wrong, just no longer necessary.
|
||||||
|
*
|
||||||
|
* RECOVERY SEMANTICS (the documented promise): canonical records are
|
||||||
|
* durable per-write, independent of this policy — a crash between
|
||||||
|
* background flushes loses NO data. What a flush persists is DERIVED
|
||||||
|
* state (index postings, deferred HNSW nodes, counters, aggregation
|
||||||
|
* stamps); after a crash, derived state converges at the next open from
|
||||||
|
* canonical records (epoch machinery + incremental aggregation catch-up),
|
||||||
|
* paying a bounded catch-up cost proportional to the un-flushed window —
|
||||||
|
* never data loss.
|
||||||
|
*
|
||||||
|
* `'manual'` restores the pre-9.1 behavior: the engine never flushes on
|
||||||
|
* its own (except at `close()`); the caller owns the cadence.
|
||||||
|
*/
|
||||||
|
persistence?: {
|
||||||
|
policy?: 'auto' | 'manual'
|
||||||
|
/** Background flush after this many committed writes (default 512). */
|
||||||
|
flushEveryWrites?: number
|
||||||
|
/** Background flush when this much time has passed since the last flush, checked at write time (default 30_000). */
|
||||||
|
flushIntervalMs?: number
|
||||||
|
/** Background flush after the store goes quiet for this long with dirty state (default 2_000). */
|
||||||
|
flushOnIdleMs?: number
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// ============= Neural API Types =============
|
// ============= Neural API Types =============
|
||||||
|
|
|
||||||
97
tests/unit/brainy/persistence-policy.test.ts
Normal file
97
tests/unit/brainy/persistence-policy.test.ts
Normal file
|
|
@ -0,0 +1,97 @@
|
||||||
|
/**
|
||||||
|
* @module tests/unit/brainy/persistence-policy
|
||||||
|
* @description THE ENGINE-OWNED FLUSH CADENCE pins (A4,
|
||||||
|
* SELF-ENGINE-LIFECYCLE-SPRINT, David-directed: callers NEVER call flush()
|
||||||
|
* in hot paths). The production disease: 829 caller-scheduled per-write
|
||||||
|
* flushes convoying into 45–66 second write walls — cadence hand-rolled a
|
||||||
|
* layer above the only layer that can see dirty state and IO pressure.
|
||||||
|
*
|
||||||
|
* Pinned here: (1) the write-count trigger fires a BACKGROUND flush without
|
||||||
|
* any caller flush(); (2) the idle trigger; (3) `'manual'` restores
|
||||||
|
* caller-owned cadence exactly; (4) THE ACK LAW — a write acknowledges
|
||||||
|
* without awaiting any background flush, even one that never resolves.
|
||||||
|
*/
|
||||||
|
import { describe, it, expect, afterEach, vi } from 'vitest'
|
||||||
|
import { Brainy } from '../../../src/index.js'
|
||||||
|
import { NounType } from '../../../src/types/graphTypes.js'
|
||||||
|
|
||||||
|
const brains: Brainy[] = []
|
||||||
|
|
||||||
|
async function mk(persistence?: {
|
||||||
|
policy?: 'auto' | 'manual'
|
||||||
|
flushEveryWrites?: number
|
||||||
|
flushIntervalMs?: number
|
||||||
|
flushOnIdleMs?: number
|
||||||
|
}): Promise<Brainy> {
|
||||||
|
const b = new Brainy({
|
||||||
|
storage: { type: 'memory' },
|
||||||
|
requireSubtype: false,
|
||||||
|
...(persistence && { persistence })
|
||||||
|
})
|
||||||
|
await b.init()
|
||||||
|
brains.push(b)
|
||||||
|
return b
|
||||||
|
}
|
||||||
|
|
||||||
|
afterEach(async () => {
|
||||||
|
for (const b of brains.splice(0)) await b.close().catch(() => {})
|
||||||
|
vi.restoreAllMocks()
|
||||||
|
})
|
||||||
|
|
||||||
|
describe('persistence policy — the engine owns its flush cadence', () => {
|
||||||
|
it('write-count trigger: N committed writes fire ONE background flush, no caller flush()', async () => {
|
||||||
|
const brain = await mk({ flushEveryWrites: 5, flushOnIdleMs: 60_000, flushIntervalMs: 600_000 })
|
||||||
|
const flushSpy = vi.spyOn(brain, 'flush')
|
||||||
|
|
||||||
|
for (let i = 0; i < 5; i++) {
|
||||||
|
await brain.add({ data: `w${i}`, type: NounType.Document, metadata: { i } })
|
||||||
|
}
|
||||||
|
|
||||||
|
await vi.waitFor(() => expect(flushSpy).toHaveBeenCalled(), { timeout: 5000 })
|
||||||
|
// Single-flight: the threshold crossing kicks exactly one.
|
||||||
|
expect(flushSpy.mock.calls.length).toBe(1)
|
||||||
|
})
|
||||||
|
|
||||||
|
it('idle trigger: a quiet store with dirty writes flushes itself', async () => {
|
||||||
|
const brain = await mk({ flushEveryWrites: 10_000, flushIntervalMs: 600_000, flushOnIdleMs: 60 })
|
||||||
|
const flushSpy = vi.spyOn(brain, 'flush')
|
||||||
|
|
||||||
|
await brain.add({ data: 'lone write', type: NounType.Document, metadata: {} })
|
||||||
|
|
||||||
|
await vi.waitFor(() => expect(flushSpy).toHaveBeenCalled(), { timeout: 5000 })
|
||||||
|
})
|
||||||
|
|
||||||
|
it("'manual' policy: the engine NEVER flushes on its own", async () => {
|
||||||
|
const brain = await mk({ policy: 'manual', flushEveryWrites: 2, flushOnIdleMs: 30 })
|
||||||
|
const flushSpy = vi.spyOn(brain, 'flush')
|
||||||
|
|
||||||
|
for (let i = 0; i < 6; i++) {
|
||||||
|
await brain.add({ data: `m${i}`, type: NounType.Document, metadata: { i } })
|
||||||
|
}
|
||||||
|
await new Promise((r) => setTimeout(r, 150))
|
||||||
|
|
||||||
|
expect(flushSpy).not.toHaveBeenCalled()
|
||||||
|
})
|
||||||
|
|
||||||
|
it('THE ACK LAW: writes acknowledge without awaiting the background flush — even a hung one', async () => {
|
||||||
|
const brain = await mk({ flushEveryWrites: 2, flushOnIdleMs: 60_000, flushIntervalMs: 600_000 })
|
||||||
|
// A flush that NEVER resolves: if any write ack awaited it, the test
|
||||||
|
// would time out. (The engine's background flight must be fire-and-log.)
|
||||||
|
vi.spyOn(brain, 'flush').mockImplementation(() => new Promise<void>(() => {}))
|
||||||
|
|
||||||
|
for (let i = 0; i < 6; i++) {
|
||||||
|
const id = await brain.add({ data: `a${i}`, type: NounType.Document, metadata: { i } })
|
||||||
|
expect(id).toBeTruthy()
|
||||||
|
}
|
||||||
|
// All six writes acked while the "flush" hangs forever.
|
||||||
|
const rows = await brain.find({ type: NounType.Document, limit: 10 })
|
||||||
|
expect(rows.length).toBe(6)
|
||||||
|
|
||||||
|
// Un-hang before afterEach close(): restore the method AND drop the
|
||||||
|
// never-resolving in-flight promise (close() awaits the flight — with a
|
||||||
|
// real flush that is correct; here it is the test's own artifact).
|
||||||
|
vi.restoreAllMocks()
|
||||||
|
;(brain as unknown as { _persistBackgroundFlight: Promise<void> | null })._persistBackgroundFlight =
|
||||||
|
null
|
||||||
|
})
|
||||||
|
})
|
||||||
Loading…
Add table
Add a link
Reference in a new issue