From a859d6ecf8807e435542cff01a9ddc15d540b3de Mon Sep 17 00:00:00 2001 From: David Snelling Date: Tue, 30 Jun 2026 10:30:17 -0700 Subject: [PATCH] perf(8.0): bound per-id generation history chains (O(W+L) resident, was O(N)) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The MVCC time-travel layer kept nounChains/verbChains as unbounded Map — one resident chain per id ever touched across retained history (O(N) RAM, defeating billion-scale time travel). Replace with a hot-tail window + bounded cold LRU + a mutex-free bulk resolver, keeping resolveAt exactly correct. The most-recent W generations stay resident as full chains (the common recent-pin read is O(log), zero scan); deeper pins reconstruct one id's chain on demand into an L-bounded LRU. Reconstruction is LOCK-LIGHT — it holds no commit mutex, so a historical read never stalls writers; correctness holds because a live pin's answer is always > minPinnedGeneration, which compaction can never reclaim, and a concurrently-reclaimed gen below that is skipped. materializeAtGeneration routes through a new mutex-free resolveManyAt (one forward pass, O(R + |ids|)) — without it a deep-pin materialize both regresses to O(N*R) and deadlocks on snapshotWith's mutex. The cold cache is invalidated on re-touch to stay coherent. Resident RAM is O(W*d + L), independent of N. 11-case test (oracle-vs-bruteforce, held-Db across eviction + concurrent compaction, no-deadlock, write-path I/O-free); unit 1735 + integration 613 (db-mvcc/db-temporal/db-asof green). --- src/brainy.ts | 74 +++- src/db/generationStore.ts | 410 ++++++++++++++++--- tests/unit/db/bounded-chains.test.ts | 542 +++++++++++++++++++++++++ tests/unit/db/generation-chain.test.ts | 2 +- 4 files changed, 956 insertions(+), 72 deletions(-) create mode 100644 tests/unit/db/bounded-chains.test.ts diff --git a/src/brainy.ts b/src/brainy.ts index ad05e702..82c05d21 100644 --- a/src/brainy.ts +++ b/src/brainy.ts @@ -7475,23 +7475,42 @@ export class Brainy implements BrainyInterface { const snapshotStorage = new MemoryStorage() await snapshotStorage.init() - /** Copy one id's at-`generation` state into the snapshot (or ensure absence). */ - const copyAt = async (kind: 'noun' | 'verb', id: string): Promise => { - const resolved = await this.generationStore.resolveAt(kind, id, generation) + // Copy one id's at-`generation` state into the snapshot (or ensure absence) + // as a PURE LOOKUP against a precomputed `firstAfter` map (id → the first + // generation after the pin that touched it). An id absent from the map was + // untouched since the pin → its live storage state IS its at-`generation` + // state; an id present resolves from that generation's immutable + // before-image. Both `resolveManyAt` (builds the map) and + // `readGenerationRecord` (reads the before-image) are MUTEX-FREE, so this is + // safe to call inside the reconciliation pass below (which holds the commit + // mutex via `snapshotWith`) — a per-id `resolveAt` there would re-enter the + // mutex and DEADLOCK, and would regress the O(R) bulk copy to O(N·R). + const copyAt = async ( + kind: 'noun' | 'verb', + id: string, + firstAfter: Map + ): Promise => { let record: { metadata: any; vector: any | null } | null = null - if (resolved.source === 'record') { - if (resolved.metadata !== null) { - record = { metadata: resolved.metadata, vector: resolved.vector } - } - } else if (resolved.source === 'current') { + const candidate = firstAfter.get(id) + if (candidate === undefined) { + // Untouched since the pin → live storage IS the at-`generation` state. const raw = kind === 'noun' ? await this.storage.readNounRaw(id) : await this.storage.readVerbRaw(id) - if (raw.metadata !== null) { - record = raw + if (raw.metadata !== null) record = raw + } else { + const before = await this.generationStore.readGenerationRecord(kind, candidate, id) + if (before === null) { + throw new Error( + `Generation record missing: _generations/${candidate}/prev/${id}.json ` + + `(store corrupted or records removed outside compactHistory())` + ) } + // A non-null metadata before-image is the at-`generation` live state; a + // null-metadata before-image (the create sentinel, or a metadata-less + // stored state) is absent at this generation. + if (before.metadata !== null) record = { metadata: before.metadata, vector: before.vector } } - // `record === null` — absent at this generation (or a metadata-less - // stored state, which is not a live entity): writing nulls ensures the + // `record === null` — absent at this generation: writing nulls ensures the // id is absent in the snapshot, which also un-copies ids created by // transactions that commit mid-build. if (kind === 'noun') { @@ -7516,25 +7535,42 @@ export class Brainy implements BrainyInterface { const id = entityIdFromCanonicalPath(path) if (id) verbIds.add(id) } - for (const id of nounIds) await copyAt('noun', id) - for (const id of verbIds) await copyAt('verb', id) + // ONE ascending pass per kind resolves the whole universe's first-after + // generations (O(R) getDelta reads, NOT O(N·R)), then each copy is a lookup. + const nounFirstAfter = await this.generationStore.resolveManyAt('noun', nounIds, generation) + const verbFirstAfter = await this.generationStore.resolveManyAt('verb', verbIds, generation) + for (const id of nounIds) await copyAt('noun', id, nounFirstAfter) + for (const id of verbIds) await copyAt('verb', id, verbFirstAfter) // Reconciliation pass under the commit mutex: nothing can commit during // this section, so afterwards the snapshot is exactly the at-`generation` // record set even when transactions committed while the bulk copy ran. // Reconciled ids join the enumeration sets so the vector-index build - // below sees them too. + // below sees them too. `resolveManyAt`/`readGenerationRecord` are mutex-free, + // so this section never re-enters the commit mutex (no deadlock). await this.generationStore.snapshotWith(async () => { const committedNow = this.generationStore.committedGeneration() if (committedNow === watermark) return const delta = await this.generationStore.changedBetween(watermark, committedNow) - for (const id of delta.nouns) { + const reconNouns = new Set(delta.nouns) + const reconVerbs = new Set(delta.verbs) + const reconNounFirstAfter = await this.generationStore.resolveManyAt( + 'noun', + reconNouns, + generation + ) + const reconVerbFirstAfter = await this.generationStore.resolveManyAt( + 'verb', + reconVerbs, + generation + ) + for (const id of reconNouns) { nounIds.add(id) - await copyAt('noun', id) + await copyAt('noun', id, reconNounFirstAfter) } - for (const id of delta.verbs) { + for (const id of reconVerbs) { verbIds.add(id) - await copyAt('verb', id) + await copyAt('verb', id, reconVerbFirstAfter) } watermark = committedNow }) diff --git a/src/db/generationStore.ts b/src/db/generationStore.ts index a94a333e..bd83684d 100644 --- a/src/db/generationStore.ts +++ b/src/db/generationStore.ts @@ -147,19 +147,80 @@ export class GenerationStore { * ({@link committedRanges}) * (which is O(database-age) — confirmed by the scalability spike: a read of an * unchanged entity at an old pin scaled 11.9x for 10x history depth). This - * mirrors cor's `delta_history` chain. Built lazily once (under the mutex), - * maintained on commit, invalidated on compaction. + * mirrors cor's `delta_history` chain. + * + * BOUNDED-RAM design (Approach C — hot-tail window + cold LRU). The old + * unbounded `Map` held ONE resident chain per id ever touched + * across retained history → O(distinct-ids-ever-touched) RAM, which defeats + * billion-scale time travel (the chains, not the data, become the heap floor). + * These three structures replace it with RAM bounded by the knobs W/L, + * INDEPENDENT of the corpus size: + * + * 1. {@link recentNounChains}/{@link recentVerbChains} — FULL per-id chains, + * but only over the HOT TAIL window `[windowLo, counter]` (the last + * {@link recentWindowGenerations} generations). Pins at-or-above + * `windowLo` (the overwhelming common case — recent history) resolve here + * with the same O(log chain) binary search as before. + * 2. {@link coldNounChains}/{@link coldVerbChains} — a bounded LRU of FULL + * per-id chains for DEEP pins (`gen < windowLo`). Reconstructed on demand + * from the persisted deltas ({@link reconstructColdChain}), cached so a + * re-read is O(log chain), evicted oldest-first past + * {@link coldChainLruMax}. Empty chains are cached too (bounded negative + * cache for untouched-in-cold-region ids). + * 3. {@link windowNounDeltas}/{@link windowVerbDeltas} — the window's INVERSE + * index (`gen → ids touched at that gen`), fed from the touched sets + * already handed to {@link extendChains}. It makes sliding `windowLo` + * forward an O(d̄) front-trim of exactly the ids leaving the window, with + * ZERO disk I/O — the write path never reads `tx.json` on a slide. + * + * INVARIANT — purely derived, NEVER persisted. All three structures (and the + * cold LRU) are reconstructable from the on-disk deltas at any time; eviction + * (cold LRU) and window slide-out drop entries that are then rebuilt on the + * next read that needs them. Correctness across eviction/reconstruction is + * guaranteed by pin-gating: a live historical `Db` pins at `g_old`, and the + * before-image a read needs lives at the FIRST generation after `g_old` + * (`> g_old ≥ minPinned`), which compaction can never reclaim while the pin is + * held — so a reconstructed chain yields the identical answer as the original. */ - private readonly nounChains = new Map() - private readonly verbChains = new Map() - /** Whether {@link nounChains}/{@link verbChains} reflect every committed generation. */ - private chainsReady = false - /** De-dupes concurrent first-callers of {@link ensureChains}. */ - private chainsBuilding: Promise | null = null + private readonly recentNounChains = new Map() + private readonly recentVerbChains = new Map() + /** Window inverse index (`gen → ids touched`), one map per kind — the O(d̄), + * I/O-free slide-trim driver. Holds exactly the gens in `[windowLo, counter]`. */ + private readonly windowNounDeltas = new Map() + private readonly windowVerbDeltas = new Map() + /** Bounded LRU of reconstructed deep-pin chains (`gen < windowLo`). JS Map + * insertion order is the LRU order: a hit re-inserts (delete+set) to mark + * MRU; eviction removes `keys().next().value` (the oldest) past the cap. */ + private readonly coldNounChains = new Map() + private readonly coldVerbChains = new Map() + /** Inclusive lower bound of the resident hot-tail window (`0` until built). */ + private windowLo = 0 + /** Whether the hot-tail window + its inverse index are built and current. */ + private windowReady = false + /** De-dupes concurrent first-callers of {@link ensureWindow}. */ + private windowBuilding: Promise | null = null + + /** + * Hot-tail window width W — how many of the newest generations keep a resident + * per-id chain ({@link recentNounChains}). Bounds recent-chain + window-delta + * RAM to O(W·d̄), independent of corpus size. Deep pins below the window fall to + * the cold LRU. Not `readonly` so tests can shrink it to exercise the + * cold/slide paths without building thousands of generations. + */ + private recentWindowGenerations = 4096 + + /** + * Cold-chain LRU capacity L — how many reconstructed deep-pin chains stay + * resident. Bounds cold-chain RAM to O(L·avg-chain-length), independent of how + * many distinct ids are deep-read. Not `readonly` so tests can shrink it to + * exercise eviction + reconstruction. + */ + private coldChainLruMax = 4096 /** * Cap on resident {@link deltaCache} entries (Model-B RAM bound). The per-id - * {@link nounChains}/{@link verbChains} are the hot read structure; the raw + * {@link recentNounChains}/{@link recentVerbChains} window is the hot read + * structure; the raw * deltas are only needed for range queries ({@link changedBetween}, `diff`, * `since`) and chain (re)builds, and {@link getDelta} transparently re-reads an * evicted delta from storage. Bounding this keeps a long-lived high-write @@ -290,8 +351,8 @@ export class GenerationStore { // committedRanges only includes generations ≤ the manifest watermark. this.counter = Math.max(this.counter, gen) } - // Chains are rebuilt lazily from the (possibly changed) generation set on - // the next historical read — covers reopen and reopen-after-restore. + // The history window is rebuilt lazily from the (possibly changed) generation + // set on the next historical read — covers reopen and reopen-after-restore. this.invalidateChains() if (rolledBack > 0) { @@ -1016,16 +1077,39 @@ export class GenerationStore { | { source: 'absent' } | { source: 'record'; metadata: any; vector: any | null } > { - await this.ensureChains() - const chain = (kind === 'noun' ? this.nounChains : this.verbChains).get(id) - if (chain === undefined) { - // The id was never touched by any committed generation → the live - // storage state IS its state at `gen`. - return { source: 'current' } + await this.ensureWindow() + // Snapshot windowLo, then read the chain synchronously (no await between): + // a concurrent commit can only slide the window at the `await` above, so the + // routing decision and the chain it reads are a consistent pair. + const windowLo = this.windowLo + let candidate: number | undefined + if (gen >= windowLo) { + // HOT path — every generation that could hold the before-image + // (`> gen ≥ windowLo`) is inside the resident window, so the recent chain + // is complete for this pin. O(log chain). + const chain = (kind === 'noun' ? this.recentNounChains : this.recentVerbChains).get(id) + if (chain === undefined) return { source: 'current' } + candidate = firstGenerationAfter(chain, gen) + } else { + // COLD path — a deep pin below the window. Serve from the bounded LRU, + // reconstructing (lock-light) on a miss. Caching empty chains too keeps + // repeated deep reads of untouched-in-cold-region ids O(1). + const cold = kind === 'noun' ? this.coldNounChains : this.coldVerbChains + let chain = cold.get(id) + if (chain !== undefined) { + cold.delete(id) // re-insert to mark most-recently-used + cold.set(id, chain) + } else { + chain = await this.reconstructColdChain(kind, id) + cold.set(id, chain) + while (cold.size > this.coldChainLruMax) { + const oldest = cold.keys().next().value + if (oldest === undefined) break + cold.delete(oldest) + } + } + candidate = firstGenerationAfter(chain, gen) } - // The first committed generation strictly greater than `gen` that touched - // `id` holds the before-image of its state AS OF `gen` — O(log chain). - const candidate = firstGenerationAfter(chain, gen) if (candidate === undefined) { // Nothing after `gen` touched `id`; the live state is its state at `gen`. return { source: 'current' } @@ -1045,6 +1129,87 @@ export class GenerationStore { return { source: 'record', metadata: record.metadata, vector: record.vector } } + /** + * @description Resolve the FIRST reserved generation after `gen` that touched + * each of `ids`, by kind — the bulk, MUTEX-FREE counterpart of + * {@link resolveAt}'s chain lookup. A SINGLE ascending pass over the reserved + * (committed ∪ pending) generations records `firstAfter[id] = g` for the first + * `g > gen` whose delta touched `id`, stopping early once every id is resolved. + * + * Ids absent from the returned map were never touched after `gen` → their live + * storage state IS their state at `gen` (the {@link resolveAt} `'current'` + * answer). Ids present map to the generation whose before-image + * ({@link readGenerationRecord}) holds their state as of `gen`. + * + * WHY MUTEX-FREE (load-bearing): {@link Brainy.materializeAtGeneration} runs its + * final reconciliation inside {@link snapshotWith} (which HOLDS the commit + * mutex). Resolving ids one-by-one through {@link resolveAt} there would (a) + * re-enter the mutex via {@link ensureWindow} → DEADLOCK, and (b) regress an + * O(R) bulk copy to O(N·R). This method takes no lock and makes one pass, so a + * deep-pin materialize over N ids stays O(R·d̄ + |ids|) with O(R) `getDelta` + * calls. It iterates SNAPSHOTS of the small committed-range list (O(gaps)) and + * the pending list (O(pending)) — NOT a materialized O(R) gen array — so it is + * safe against concurrent commits and bounded in memory. + * + * @param kind - `'noun'` for entities, `'verb'` for relationships. + * @param ids - The ids to resolve. + * @param gen - The pinned generation. + * @returns `id → first reserved generation after `gen` that touched it` for the + * subset of `ids` touched after `gen`. + */ + async resolveManyAt( + kind: 'noun' | 'verb', + ids: Set, + gen: number + ): Promise> { + const firstAfter = new Map() + if (ids.size === 0) return firstAfter + // Snapshot the small range list + pending list (NOT the expanded gens) so the + // pass is immune to a concurrent compaction splice / commit append and stays + // O(gaps + pending) in memory. + const ranges = this.committedRanges.map(([s, e]) => [s, e] as [number, number]) + const pending = [...this.pendingGens] + const record = async (g: number): Promise => { + const delta = await this.getDelta(g) + const touched = kind === 'noun' ? delta.nouns : delta.verbs + for (const id of touched) { + if (ids.has(id) && !firstAfter.has(id)) firstAfter.set(id, g) + } + return firstAfter.size === ids.size + } + for (const [s, e] of ranges) { + if (e <= gen) continue + for (let g = Math.max(s, gen + 1); g <= e; g++) { + if (await record(g)) return firstAfter + } + } + for (const g of pending) { + if (g <= gen) continue + if (await record(g)) return firstAfter + } + return firstAfter + } + + /** + * @description Read the before-image of `id` recorded by generation `gen` — the + * public, MUTEX-FREE accessor {@link Brainy.materializeAtGeneration}'s `copyAt` + * uses after {@link resolveManyAt} hands it the generation. Pairs with + * `resolveManyAt` so the bulk historical copy never touches the commit mutex + * (no deadlock inside {@link snapshotWith}, no per-id chain rebuild). + * @param kind - `'noun'` for entities, `'verb'` for relationships. + * @param gen - The generation whose before-image to read. + * @param id - The id to read. + * @returns The stored {@link GenerationRecord}, or `null` when the record-set + * has no before-image for `id`. + */ + async readGenerationRecord( + kind: 'noun' | 'verb', + gen: number, + id: string + ): Promise { + return this.readBeforeImage(kind, gen, id) + } + /** * @description Read the before-image of `id` stored by generation `gen` from * the right tier: the in-memory pending buffer while the generation is @@ -1066,50 +1231,176 @@ export class GenerationStore { } /** - * @description Ensure the per-id history chains ({@link nounChains} / - * {@link verbChains}) reflect every committed generation. Built once, lazily, - * under the commit mutex (so no concurrent commit mutates {@link committedRanges} - * mid-build); thereafter maintained incrementally by {@link commit} and - * invalidated by {@link compact}. Concurrent first-callers share one build. - * O(committed generations) the first time; O(1) afterwards. + * @description Build the resident hot-tail window — the per-id chains + * ({@link recentNounChains}/{@link recentVerbChains}) and the inverse index + * ({@link windowNounDeltas}/{@link windowVerbDeltas}) over the last + * {@link recentWindowGenerations} reserved generations, `[windowLo, counter]`. + * Built once, lazily, under the commit mutex (so no concurrent commit mutates + * {@link committedRanges}/pending mid-build); thereafter maintained + * incrementally by {@link extendChains} and invalidated by {@link compact} / + * reopen. Concurrent first-callers share one build. O(W) `getDelta` reads the + * first time (only the window, NOT all of history); O(1) afterwards. Deep pins + * below `windowLo` are served by {@link reconstructColdChain} on demand. */ - private async ensureChains(): Promise { - if (this.chainsReady) return - if (!this.chainsBuilding) { - this.chainsBuilding = this.withMutex(async () => { - if (this.chainsReady) return - this.nounChains.clear() - this.verbChains.clear() - // reservedGensAsc (committed ∪ pending) is sorted ascending, so chains - // accrue in ascending order and un-flushed single-ops are indexed too. + private async ensureWindow(): Promise { + if (this.windowReady) return + if (!this.windowBuilding) { + this.windowBuilding = this.withMutex(async () => { + if (this.windowReady) return + this.recentNounChains.clear() + this.recentVerbChains.clear() + this.windowNounDeltas.clear() + this.windowVerbDeltas.clear() + // The window is the newest W reserved generations, clamped above the + // compaction horizon (gens ≤ horizon are reclaimed — never resident). + this.windowLo = Math.max(this.horizonGen + 1, this.counter - this.recentWindowGenerations + 1) + // reservedGensAsc (committed ∪ pending) is ascending, so chains accrue in + // order and un-flushed single-ops in-window are indexed too. for (const g of this.reservedGensAsc()) { + if (g < this.windowLo) continue const delta = await this.getDelta(g) - for (const nounId of delta.nouns) appendToChain(this.nounChains, nounId, g) - for (const verbId of delta.verbs) appendToChain(this.verbChains, verbId, g) + const nouns = [...delta.nouns] + const verbs = [...delta.verbs] + for (const id of nouns) appendToChain(this.recentNounChains, id, g) + for (const id of verbs) appendToChain(this.recentVerbChains, id, g) + this.windowNounDeltas.set(g, nouns) + this.windowVerbDeltas.set(g, verbs) } - this.chainsReady = true - this.chainsBuilding = null + this.windowReady = true + this.windowBuilding = null }) } - await this.chainsBuilding + await this.windowBuilding } - /** Record that generation `gen` (the newest) touched these ids — keeps chains - * current after a commit without a full rebuild. No-op until chains are built - * (the eventual build reads `gen` from {@link committedRanges}). */ + /** + * @description Reconstruct the FULL per-id chain for a deep pin (`gen < + * windowLo`) — every generation that touched `id`, ascending. Built from the + * persisted deltas BELOW the window plus the resident in-window tail, then + * cached in the cold LRU by the caller. + * + * LOCK-LIGHT — deliberately does NOT hold the commit mutex (a deep read must + * never stall writers). Correctness without the lock: + * - The recent tail and `windowLo` are snapshotted SYNCHRONOUSLY up front, so + * even if the window slides during the (awaited) cold scan, the chain is the + * one consistent with the snapshot. New writes during the scan land ABOVE the + * snapshot and cannot change any deep pin's first-after answer; the cold + * entry is also dropped by {@link extendChains} the moment `id` is next + * touched, so it is never served stale. + * - The cold scan iterates a SNAPSHOT of the small committed-range list + * (O(gaps) memory, immune to a concurrent compaction splice). + * - A generation concurrently reclaimed by compaction surfaces as a missing + * delta. Such a gen is always `≤ minPinned` (compaction never reclaims above + * the lowest live pin) and therefore NEVER a live pin's answer (`first-after + * > pin ≥ minPinned`), so it is skipped, not raised. A missing delta ABOVE + * `minPinned` is genuine corruption and still throws. + * + * @param kind - `'noun'` for entities, `'verb'` for relationships. + * @param id - The id whose chain to reconstruct. + * @returns The id's full ascending generation chain (`[]` when never touched). + */ + private async reconstructColdChain(kind: 'noun' | 'verb', id: string): Promise { + // Synchronous, consistent snapshot of the window state. + const windowLo = this.windowLo + const recentTail = (kind === 'noun' ? this.recentNounChains : this.recentVerbChains).get(id) + const recentSnapshot = recentTail ? [...recentTail] : [] + const ranges = this.committedRanges.map(([s, e]) => [s, e] as [number, number]) + const pendingCold = this.pendingGens.filter((g) => g < windowLo) + + const cold: number[] = [] + const scan = async (g: number): Promise => { + let delta: { nouns: Set; verbs: Set } + try { + delta = await this.getDelta(g) + } catch (err) { + // Concurrent-compaction tolerance: a reclaimed gen (≤ minPinned, never a + // live pin's answer) is skipped; a missing gen above the lowest pin is + // real corruption and rethrows. + if (g <= this.minPinnedGeneration()) return + throw err + } + if ((kind === 'noun' ? delta.nouns : delta.verbs).has(id)) cold.push(g) + } + + for (const [s, e] of ranges) { + if (s >= windowLo) break // ascending → nothing newer is below the window + const hi = Math.min(e, windowLo - 1) + for (let g = s; g <= hi; g++) await scan(g) + } + // Cold pending gens (only when more than W writes are buffered, i.e. tests + // with a small window) sit between the committed gens and the window. + for (const g of pendingCold) await scan(g) + + // Cold gens (< windowLo) then the resident in-window tail (≥ windowLo) — both + // ascending and disjoint, so the concatenation is the full ascending chain. + return recentSnapshot.length ? cold.concat(recentSnapshot) : cold + } + + /** + * Record that generation `gen` (the newest) touched these ids and advance the + * hot-tail window. Keeps the resident window current after a commit without a + * rebuild. No-op until the window is built (the eventual build reads `gen` from + * {@link reservedGensAsc}). + * + * Three steps, all in-memory and I/O-FREE: + * 1. Append `gen` to each touched id's recent chain, and DROP each touched id + * from the cold LRU — a freshly-touched id's cached deep chain is now + * incomplete, so it must be reconstructed on the next deep read (else a + * deep pin whose first-after just became `gen` would wrongly read `current`). + * 2. Record the window inverse index entry `gen → touched ids` (the slide-trim + * driver). + * 3. Slide `windowLo` forward to `max(horizon+1, gen-W+1)`, front-trimming the + * chains of exactly the ids leaving the window using the inverse index — NO + * `getDelta`, so the write path never reads `tx.json` on a slide regardless + * of W vs the delta-cache size. + */ private extendChains(gen: number, nouns: Iterable, verbs: Iterable): void { - if (!this.chainsReady) return - for (const nounId of nouns) appendToChain(this.nounChains, nounId, gen) - for (const verbId of verbs) appendToChain(this.verbChains, verbId, gen) + if (!this.windowReady) return + const nounArr = [...nouns] + const verbArr = [...verbs] + for (const id of nounArr) { + appendToChain(this.recentNounChains, id, gen) + this.coldNounChains.delete(id) + } + for (const id of verbArr) { + appendToChain(this.recentVerbChains, id, gen) + this.coldVerbChains.delete(id) + } + this.windowNounDeltas.set(gen, nounArr) + this.windowVerbDeltas.set(gen, verbArr) + + const newLo = Math.max(this.horizonGen + 1, gen - this.recentWindowGenerations + 1) + if (newLo <= this.windowLo) return + // Collect the ids leaving the window from the inverse index (O(slid·d̄)), then + // front-trim their chains to the new low watermark (each id once). + const trimNouns = new Set() + const trimVerbs = new Set() + for (let g = this.windowLo; g < newLo; g++) { + const n = this.windowNounDeltas.get(g) + if (n) for (const id of n) trimNouns.add(id) + this.windowNounDeltas.delete(g) + const v = this.windowVerbDeltas.get(g) + if (v) for (const id of v) trimVerbs.add(id) + this.windowVerbDeltas.delete(g) + } + for (const id of trimNouns) dropChainFront(this.recentNounChains, id, newLo) + for (const id of trimVerbs) dropChainFront(this.recentVerbChains, id, newLo) + this.windowLo = newLo } - /** Drop the chains so the next read rebuilds them from the surviving - * generations (called after compaction removes record-sets). */ + /** Drop the resident window + cold LRU so the next read rebuilds from the + * surviving generations (called after compaction removes record-sets / on + * reopen). */ private invalidateChains(): void { - this.chainsReady = false - this.chainsBuilding = null - this.nounChains.clear() - this.verbChains.clear() + this.windowReady = false + this.windowBuilding = null + this.recentNounChains.clear() + this.recentVerbChains.clear() + this.windowNounDeltas.clear() + this.windowVerbDeltas.clear() + this.coldNounChains.clear() + this.coldVerbChains.clear() + this.windowLo = 0 } /** @@ -1603,6 +1894,21 @@ function appendToChain(chains: Map, id: string, gen: number): } } +/** + * @description Drop the leading entries of an id's ascending chain that fall + * below `lo` (generations that slid out of the hot-tail window), deleting the + * chain entirely when it empties. Used by the O(d̄) window slide in + * {@link GenerationStore.extendChains}. Tolerates an already-trimmed front. + */ +function dropChainFront(chains: Map, id: string, lo: number): void { + const chain = chains.get(id) + if (chain === undefined) return + let i = 0 + while (i < chain.length && chain[i] < lo) i++ + if (i === chain.length) chains.delete(id) + else if (i > 0) chain.splice(0, i) +} + /** * @description Binary-search an ascending generation chain for the FIRST entry * strictly greater than `gen` (the generation whose before-image holds the diff --git a/tests/unit/db/bounded-chains.test.ts b/tests/unit/db/bounded-chains.test.ts new file mode 100644 index 00000000..034bc663 --- /dev/null +++ b/tests/unit/db/bounded-chains.test.ts @@ -0,0 +1,542 @@ +/** + * @module tests/unit/db/bounded-chains + * @description Correctness + RAM-boundedness oracle for the bounded per-id + * generation history chains in `src/db/generationStore.ts` (GA #33, Approach C: + * hot-tail window + bounded cold LRU + a mutex-free bulk `resolveManyAt`). + * + * The old design kept ONE resident chain per id ever touched across retained + * history (`Map`) → O(distinct-ids) RAM, which defeats + * billion-scale time travel. The replacement bounds RAM to O(W·d̄)/O(L) — a + * resident window over the newest `W` generations plus a bounded LRU of + * reconstructed deep-pin chains — while staying EXACTLY correct. + * + * These tests are the correctness oracle for that change: + * + * 1. Oracle vs brute-force: for every id × every pin, `resolveAt` must equal a + * brute-force scan of every reserved generation's persisted `tx.json`. + * 2. Bounded RAM independent of N: the resident structures stay O(W)/O(L) no + * matter how many distinct ids the corpus touched. + * 3. Held-Db across cold eviction + a concurrent compaction (lock-light). + * 4. Un-flushed pending single-ops resolve identically before and after flush. + * 5. A deep-pin materialize is O(R) `getDelta` (not O(N·R)) and never deadlocks + * inside the mutex-held reconciliation pass. + * 6. `resolveManyAt` ≡ per-id `resolveAt` for random id-sets × random pins. + * 7. The write path is I/O-free across window slides (the inverse-index ring + * supplies slid-out ids — zero `tx.json` reads on commit). + * 8. Compaction invalidates + rebuilds the window over survivors; a pin below + * the new horizon throws `GenerationCompactedError`. + */ + +import { describe, it, expect, beforeEach, afterEach } from 'vitest' +import { MemoryStorage } from '../../../src/storage/adapters/memoryStorage.js' +import { GenerationStore } from '../../../src/db/generationStore.js' +import { GenerationCompactedError } from '../../../src/db/errors.js' +import { Brainy } from '../../../src/index.js' +import { NounType } from '../../../src/types/graphTypes.js' +import { createTestConfig, generateTestVector } from '../../helpers/test-factory.js' + +/** A precomputed embedding so adds skip the (slow) embedding model — these tests + * exercise the generation layer, not semantics. */ +const VEC = generateTestVector() + +/** Deterministic UUID-shaped id (the sharded storage layout derives the shard + * from the UUID hex), unique per `n`. */ +function uid(n: number): string { + return `00000000-0000-4000-8000-${n.toString(16).padStart(12, '0')}` +} + +/** Stored-metadata fixture carrying a `version` the oracle compares on. */ +function fixture(version: number): Record { + return { + noun: NounType.Document, + subtype: 'note', + data: `payload-v${version}`, + version, + createdAt: 1000, + updatedAt: 1000 + version, + _rev: version + } +} + +/** A small deterministic PRNG so the random oracle runs are reproducible. */ +function mulberry32(seed: number): () => number { + let a = seed >>> 0 + return () => { + a |= 0 + a = (a + 0x6d2b79f5) | 0 + let t = Math.imul(a ^ (a >>> 15), 1 | a) + t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t + return ((t ^ (t >>> 14)) >>> 0) / 4294967296 + } +} + +/** Reject after `ms` so a deadlock surfaces as a test failure, not a hang. */ +function withTimeout(p: Promise, ms: number, label: string): Promise { + return Promise.race([ + p, + new Promise((_, reject) => + setTimeout(() => reject(new Error(`timed out after ${ms}ms (likely deadlock): ${label}`)), ms) + ) + ]) +} + +/** Normalize a `resolveAt` result to a comparable scalar. */ +function norm(r: { source: string; metadata?: any }): string { + if (r.source === 'current') return 'current' + if (r.source === 'absent') return 'absent' + return `rec:${r.metadata?.version}` +} + +describe('db/GenerationStore — bounded per-id chains (GA #33)', () => { + let storage: MemoryStorage + let store: GenerationStore + + beforeEach(async () => { + storage = new MemoryStorage() + await storage.init() + store = new GenerationStore(storage) + await store.open() + }) + + /** Commit one single-op noun write (its own generation); returns the gen. */ + async function writeNoun(id: string, version: number): Promise { + const { generation } = await store.commitSingleOp({ + touched: { nouns: [id] }, + execute: async () => { + await storage.saveNounMetadata(id, fixture(version)) + } + }) + return generation + } + + /** Commit one single-op verb write (raw, so before-images are captured). */ + async function writeVerb(id: string, version: number): Promise { + const { generation } = await store.commitSingleOp({ + touched: { verbs: [id] }, + execute: async () => { + await storage.writeVerbRaw(id, { + metadata: { ...fixture(version), verb: 'RelatedTo' }, + vector: { sourceId: uid(900001), targetId: uid(900002), verb: 'RelatedTo' } + }) + } + }) + return generation + } + + /** Brute-force `resolveAt`: scan EVERY reserved generation's persisted tx.json + * (the gens must be flushed) for the first one after `pin` that touched `id`, + * then read its before-image. Fully independent of the resident chains. */ + async function bruteForce( + kind: 'noun' | 'verb', + id: string, + pin: number, + maxGen: number + ): Promise { + for (let g = pin + 1; g <= maxGen; g++) { + const delta = (await storage.readRawObject(`_generations/${g}/tx.json`)) as + | { nouns: string[]; verbs: string[] } + | null + if (!delta) continue + const touched = kind === 'noun' ? delta.nouns : delta.verbs + if (Array.isArray(touched) && touched.includes(id)) { + const rec = (await storage.readRawObject(`_generations/${g}/prev/${id}.json`)) as + | { metadata: any; vector: any } + | null + if (!rec || (rec.metadata === null && rec.vector === null)) return 'absent' + return `rec:${rec.metadata?.version}` + } + } + return 'current' + } + + // ========================================================================== + // 1. Oracle vs brute-force — every id × every pin, nouns + verbs. + // ========================================================================== + it('resolveAt equals a brute-force tx.json scan for every id × every pin (recent, deep, boundary)', async () => { + const s = store as any + s.recentWindowGenerations = 8 // W + s.coldChainLruMax = 4 // L + + const nounIds = Array.from({ length: 5 }, (_, i) => uid(100 + i)) + const verbIds = Array.from({ length: 5 }, (_, i) => uid(200 + i)) + const rand = mulberry32(1234) + let v = 0 + + // 60 overlapping single-op generations. Trigger a historical read at the + // halfway point so the SECOND half exercises the live slide path + // (extendChains + window advance), not just the lazy build. + for (let i = 0; i < 60; i++) { + if (i === 30) { + // Force the window to build so the rest of the writes drive slides. + await store.resolveAt('noun', nounIds[0], 1) + } + if (rand() < 0.5) { + await writeNoun(nounIds[Math.floor(rand() * nounIds.length)], ++v) + } else { + await writeVerb(verbIds[Math.floor(rand() * verbIds.length)], ++v) + } + } + // Flush so every reserved generation has a persisted tx.json for the oracle. + await store.flushPendingSingleOps() + const maxGen = store.generation() + + // Window must have actually slid below the head (proves the slide path ran). + expect(s.windowLo).toBeGreaterThan(1) + expect(s.windowLo).toBe(Math.max(1, maxGen - 8 + 1)) + + // Every id × every pin from 0..maxGen — covers recent (≥windowLo), deep + // ( { + const s = store as any + s.recentWindowGenerations = 4 + s.coldChainLruMax = 16 + + const X = uid(300) + const filler = (i: number) => uid(400 + i) + + const g1 = await writeNoun(X, 1) // X = v1 + // Push X out of the window with filler writes (X untouched). + for (let i = 0; i < 10; i++) await writeNoun(filler(i), 100 + i) + + // Deep read at g1 BEFORE any later touch: X untouched after g1 → 'current'. + expect(norm(await store.resolveAt('noun', X, g1))).toBe('current') + expect((await store.resolveAt('noun', X, g1)).source).toBe('current') // now cached cold + + // Now touch X again (v2). The stale cached cold chain would still answer + // 'current' → reading the NEW live value (v1 corrupted to v2). extendChains + // must invalidate the cold entry so the re-read resolves the g1 before-image. + await writeNoun(X, 2) + await store.flushPendingSingleOps() + + const resolved = await store.resolveAt('noun', X, g1) + expect(resolved.source).toBe('record') + if (resolved.source === 'record') expect(resolved.metadata.version).toBe(1) + }) + + // ========================================================================== + // 2. Bounded RAM independent of N. + // ========================================================================== + it('resident window/cold structures stay O(W)/O(L), independent of distinct-id count', async () => { + const s = store as any + const W = 8 + const L = 4 + s.recentWindowGenerations = W + s.coldChainLruMax = L + + const N = 1200 // distinct ids, one single-op each + for (let i = 0; i < N; i++) await writeNoun(uid(1000 + i), i) + await store.flushPendingSingleOps() + const maxGen = store.generation() + + // Build the window (first historical read). + await store.resolveAt('noun', uid(1000), maxGen) + + // The hot-tail window holds at most W generations' worth of ids; one id per + // gen here ⇒ exactly W resident recent chains + W inverse-index entries. + expect(s.recentNounChains.size).toBeLessThanOrEqual(W) + expect(s.windowNounDeltas.size).toBeLessThanOrEqual(W) + expect(s.recentNounChains.size).toBeLessThan(N) + + // Deep-read many distinct ids → the cold LRU is capped at L regardless. + for (let i = 0; i < 40; i++) await store.resolveAt('noun', uid(1000 + i), 1) + expect(s.coldNounChains.size).toBeLessThanOrEqual(L) + }) + + // ========================================================================== + // 3. Held-Db across cold eviction + a concurrent compaction (lock-light). + // ========================================================================== + it('a held pin reads correctly across cold eviction AND a concurrent compact()', async () => { + const s = store as any + s.recentWindowGenerations = 4 + s.coldChainLruMax = 4 + + const heldIds = [uid(500), uid(501), uid(502)] + const filler = (i: number) => uid(600 + i) + + // Each held id gets value 10/11/12 at g_old, then a later mutation so a + // before-image exists to resolve. + const gOldVals = new Map() + let pOld = 0 + for (let k = 0; k < heldIds.length; k++) { + const val = 10 + k + pOld = await writeNoun(heldIds[k], val) + gOldVals.set(heldIds[k], val) + } + const gOld = pOld // pin: at gOld every held id holds its value above + store.pin(gOld) + + // Mutate the held ids AFTER the pin (so resolveAt resolves a before-image), + // then slide the window well past gOld with lots of filler writes. + for (const id of heldIds) await writeNoun(id, 99) + for (let i = 0; i < 30; i++) await writeNoun(filler(i), 700 + i) + await store.flushPendingSingleOps() + + // Populate then evict the held ids' cold chains by cycling other deep ids. + for (const id of heldIds) await store.resolveAt('noun', id, gOld) + for (let i = 0; i < 12; i++) await store.resolveAt('noun', filler(i), 1) + + // Re-read the held ids WHILE a compaction (which reclaims everything ≤ the + // pin) runs concurrently. Lock-light reconstruction must skip the reclaimed + // gens (all ≤ minPinned, never an answer) and still match the at-gOld oracle. + const reads = heldIds.map((id) => + withTimeout(store.resolveAt('noun', id, gOld), 5000, `held read ${id}`) + ) + const compaction = withTimeout(store.compact(), 5000, 'concurrent compact') + const [r0, r1, r2] = await Promise.all(reads) + await compaction + + for (const [id, res] of [ + [heldIds[0], r0], + [heldIds[1], r1], + [heldIds[2], r2] + ] as const) { + expect(res.source).toBe('record') + if (res.source === 'record') expect(res.metadata.version).toBe(gOldVals.get(id)) + } + + // And a fresh read after compaction still matches (reconstruct over survivors). + for (const id of heldIds) { + const res = await store.resolveAt('noun', id, gOld) + expect(res.source).toBe('record') + if (res.source === 'record') expect(res.metadata.version).toBe(gOldVals.get(id)) + } + store.release(gOld) + }) + + // ========================================================================== + // 4. Un-flushed pending single-ops — flush-agnostic before-images. + // ========================================================================== + it('a now()-style pin reads pre-mutation before-images of un-flushed single-ops, identical after flush', async () => { + const X = uid(800) + const g1 = await writeNoun(X, 1) // flushed below to make g1 durable history + + // More UN-FLUSHED single-ops mutating the same id. + await writeNoun(X, 2) + await writeNoun(X, 3) + expect(store.committedGeneration()).toBe(0) // nothing on disk yet + + const before = await store.resolveAt('noun', X, g1) + expect(before.source).toBe('record') + if (before.source === 'record') expect(before.metadata.version).toBe(1) + + // Flush → the answer is identical (resolution is flush-agnostic). + await store.flushPendingSingleOps() + const after = await store.resolveAt('noun', X, g1) + expect(norm(after)).toBe(norm(before)) + }) + + // ========================================================================== + // 6. resolveManyAt ≡ per-id resolveAt (random id-sets × random pins). + // ========================================================================== + it('resolveManyAt equals per-id resolveAt for random id-sets across recent and deep pins', async () => { + const s = store as any + s.recentWindowGenerations = 6 + s.coldChainLruMax = 8 + + const ids = Array.from({ length: 8 }, (_, i) => uid(2000 + i)) + const rand = mulberry32(99) + let v = 0 + for (let i = 0; i < 50; i++) await writeNoun(ids[Math.floor(rand() * ids.length)], ++v) + await store.flushPendingSingleOps() + const maxGen = store.generation() + + /** Map a resolveManyAt firstAfter entry to the same scalar resolveAt yields. */ + const derived = async (id: string, firstAfter: Map): Promise => { + const g = firstAfter.get(id) + if (g === undefined) return 'current' + const rec = await store.readGenerationRecord('noun', g, id) + if (!rec || (rec.metadata === null && rec.vector === null)) return 'absent' + return `rec:${(rec.metadata as any)?.version}` + } + + for (let pin = 0; pin <= maxGen; pin++) { + // A random subset of ids for this pin. + const subset = new Set(ids.filter(() => rand() < 0.6)) + if (subset.size === 0) subset.add(ids[0]) + const many = await store.resolveManyAt('noun', subset, pin) + for (const id of subset) { + expect(await derived(id, many)).toBe(norm(await store.resolveAt('noun', id, pin))) + } + } + }) + + // ========================================================================== + // 7. Write path is I/O-free across window slides (the ring supplies slid-out + // ids — no tx.json reads on commit, even with W > deltaCacheMax). + // ========================================================================== + it('committing across window slides reads ZERO tx.json from disk (W > deltaCacheMax)', async () => { + const s = store as any + s.recentWindowGenerations = 6 // W + s.deltaCacheMax = 2 // W > deltaCacheMax: a naive slide would re-read tx.json + + // Seed + build the window so subsequent commits drive slides. + for (let i = 0; i < 8; i++) await writeNoun(uid(3000 + i), i) + await store.flushPendingSingleOps() + await store.resolveAt('noun', uid(3000), store.generation()) + + // Count tx.json reads from this point on. + let txReads = 0 + const realRead = storage.readRawObject.bind(storage) + ;(storage as any).readRawObject = async (path: string) => { + if (typeof path === 'string' && path.endsWith('/tx.json')) txReads++ + return realRead(path) + } + try { + // Many more commits, each sliding the window past evicted-from-cache gens. + for (let i = 0; i < 20; i++) await writeNoun(uid(3100 + i), 1000 + i) + } finally { + ;(storage as any).readRawObject = realRead + } + expect(txReads).toBe(0) + }) + + // ========================================================================== + // 8. Compaction invalidates + rebuilds the window over survivors; a pin below + // the new horizon throws GenerationCompactedError. + // ========================================================================== + it('rebuilds the window over survivors after compaction and rejects pins below the horizon', async () => { + const s = store as any + s.recentWindowGenerations = 4 + + const X = uid(900) + const g0 = await writeNoun(X, 0) + for (let v = 1; v <= 6; v++) await writeNoun(X, v) + await store.flushPendingSingleOps() + const maxGen = store.generation() + + // Build the window, then compact to keep only the 2 most recent gens. + await store.resolveAt('noun', X, maxGen) + const res = await store.compact({ maxGenerations: 2 }) + expect(res.removedGenerations).toBeGreaterThan(0) + expect(s.windowReady).toBe(false) // invalidated + + // A pin ABOVE the horizon rebuilds the window from survivors and resolves. + const aboveHorizon = res.horizon + 1 + const resolved = await store.resolveAt('noun', X, aboveHorizon) + expect(['record', 'current']).toContain(resolved.source) + + // A pin below the horizon is unreachable. + expect(() => store.assertReachable(g0)).toThrow(GenerationCompactedError) + }) +}) + +// ============================================================================ +// 5. Deep-pin materialize — O(R) getDelta (not O(N·R)) + no mutex re-entrancy. +// ============================================================================ +describe('materializeAtGeneration — bounded & deadlock-free (GA #33)', () => { + let brain: Brainy + + beforeEach(async () => { + brain = new Brainy(createTestConfig()) + await brain.init() + }) + afterEach(async () => { + await brain.close() + }) + + it('resolveManyAt + readGenerationRecord do NOT re-enter the commit mutex (safe inside snapshotWith)', async () => { + const store = (brain as any).generationStore + const id = await brain.add({ data: 'a', type: NounType.Document, subtype: 'note', vector: VEC }) + await brain.add({ data: 'b', type: NounType.Document, subtype: 'note', vector: VEC }) + const pin = 1 + + // snapshotWith HOLDS the commit mutex. If resolveManyAt/readGenerationRecord + // took the mutex, this section would deadlock — assert it completes promptly. + const out = await withTimeout( + store.snapshotWith(async () => { + const many = await store.resolveManyAt('noun', new Set([id]), pin) + const rec = many.has(id) ? await store.readGenerationRecord('noun', many.get(id), id) : null + return { many, rec } + }), + 3000, + 'snapshotWith + resolveManyAt/readGenerationRecord' + ) + expect(out.many).toBeInstanceOf(Map) + }) + + it('a deep-pin materialize over many ids invokes getDelta O(R) (not O(N·R)) and matches a per-id oracle', async () => { + const store = (brain as any).generationStore + + const N = 400 + for (let i = 0; i < N; i++) { + await brain.add({ data: `doc ${i}`, type: NounType.Document, subtype: 'note', metadata: { i }, vector: VEC }) + } + const R = brain.generation() // ≈ N (each add is its own generation) + expect(R).toBeGreaterThanOrEqual(N) + + const deepGen = 1 + + // Count getDelta invocations during the materialize. + const realGetDelta = store.getDelta.bind(store) + let getDeltaCalls = 0 + store.getDelta = async (g: number) => { + getDeltaCalls++ + return realGetDelta(g) + } + + let handle: any + try { + handle = await withTimeout( + (brain as any).materializeAtGeneration(deepGen), + 15000, + 'deep materialize' + ) + } finally { + store.getDelta = realGetDelta + } + + // O(R): a small constant number of ascending passes (changedBetween + + // resolveManyAt per kind), NOT a per-id rescan (which would be ~N·R). + expect(getDeltaCalls).toBeLessThan(R * 5) + expect(getDeltaCalls).toBeLessThan(N * N) // the regression guard + + // The materialized at-gen-1 brain holds exactly the one entity that existed. + const atGen1 = await handle.find({ limit: N + 10 }) + expect(atGen1.length).toBe(1) + await handle.close() + }) + + it('a materialize completes (no deadlock) under a forced concurrent commit', async () => { + const N = 120 + for (let i = 0; i < N; i++) { + await brain.add({ data: `x ${i}`, type: NounType.Document, subtype: 'note', vector: VEC }) + } + const deepGen = 1 + + // Fire the materialize and several commits together: the reconciliation pass + // (under snapshotWith's mutex) must resolve the raced ids via the mutex-free + // bulk path without deadlocking. + const matPromise = withTimeout( + (brain as any).materializeAtGeneration(deepGen), + 15000, + 'materialize under concurrent commits' + ) + const commits = Promise.all( + Array.from({ length: 5 }, (_, i) => + brain.add({ data: `concurrent ${i}`, type: NounType.Document, subtype: 'note', vector: VEC }) + ) + ) + const [handle] = await Promise.all([matPromise, commits]) + expect(handle).toBeTruthy() + await handle.close() + }) +}) diff --git a/tests/unit/db/generation-chain.test.ts b/tests/unit/db/generation-chain.test.ts index 1435e6a9..d00bbe65 100644 --- a/tests/unit/db/generation-chain.test.ts +++ b/tests/unit/db/generation-chain.test.ts @@ -82,7 +82,7 @@ describe('generation history chains (Model-B scalability)', () => { const g0 = await seedX() for (let v = 1; v <= 12; v++) await bumpX(v) // 12 gens > cap 4 → evictions - // asOf still resolves correctly (ensureChains re-read every evicted delta to build the chain)… + // asOf still resolves correctly (the window build re-read every evicted delta)… expect(await vAt(g0)).toBe(0) // …and a range op (since scans committedGens via getDelta, re-reading evicted deltas) still works. const now = await brain.now()