diff --git a/src/brainy.ts b/src/brainy.ts index 2e1d2de0..6d3a7927 100644 --- a/src/brainy.ts +++ b/src/brainy.ts @@ -272,6 +272,7 @@ type ResolvedBrainyConfig = Required< | 'eagerEmbeddings' | 'migrationWaitTimeoutMs' | 'transactionBudgetFloorMs' + | 'persistence' > > & Pick< @@ -285,6 +286,7 @@ type ResolvedBrainyConfig = Required< | 'eagerEmbeddings' | 'migrationWaitTimeoutMs' | 'transactionBudgetFloorMs' + | 'persistence' > /** @@ -684,6 +686,14 @@ export class Brainy implements BrainyInterface { private _aggregationIndex?: AggregationIndex // Incremental aggregation engine private _aggregationBackfillFlight: Promise | null = null // Single-flight backfill walk private _aggregationCatchUpFlight: Promise | 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 | null = null + private _persistBackgroundFlight: Promise | null = null // 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 // one loud error per query, never a full store walk per query. @@ -1829,6 +1839,64 @@ export class Brainy implements BrainyInterface { * @param run - The single-op's existing operation batch builder (the * `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( touched: { nouns?: string[]; verbs?: string[] }, run: TransactionFunction, @@ -1921,6 +1989,7 @@ export class Brainy implements BrainyInterface { ) } } + this.noteWriteForPersistence() return receipt } @@ -7714,6 +7783,7 @@ export class Brainy implements BrainyInterface { // A rejected batch throws at commitTransaction and never reaches here. this.emitCommitted(plan.changeEvents, undefined, generation, timestamp) + this.noteWriteForPersistence() const receipt: TransactReceipt = { generation, timestamp, ids: plan.ids } return this.createPinnedDb({ generation, timestamp, receipt }) } @@ -14857,7 +14927,10 @@ export class Brainy implements BrainyInterface { requireSubtype: config?.requireSubtype ?? true, // Multi-process safety 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 implements BrainyInterface { * This ensures deferred persistence mode data is saved */ async close(): Promise { + // 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 // writer (merge-deletes), and no delete pass may start mid- or post-close. this._backgroundDedup?.cancelPending() diff --git a/src/types/brainy.types.ts b/src/types/brainy.types.ts index 6c1f0ffd..cfd23d9f 100644 --- a/src/types/brainy.types.ts +++ b/src/types/brainy.types.ts @@ -2028,6 +2028,39 @@ export interface BrainyConfig { */ 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 ============= diff --git a/tests/unit/brainy/persistence-policy.test.ts b/tests/unit/brainy/persistence-policy.test.ts new file mode 100644 index 00000000..98a0afc2 --- /dev/null +++ b/tests/unit/brainy/persistence-policy.test.ts @@ -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 { + 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(() => {})) + + 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 | null })._persistBackgroundFlight = + null + }) +})