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

perf(8.0): bound per-id generation history chains (O(W+L) resident, was O(N))

Browse source 
The MVCC time-travel layer kept nounChains/verbChains as unbounded
Map<id, number[]> — 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).
This commit is contained in:
David Snelling 2026-06-30 10:30:17 -07:00
parent 8f4787bb10
commit a859d6ecf8
4 changed files with 956 additions and 72 deletions
Download patch file Download diff file Expand all files Collapse all files

74
src/brainy.ts
View file

@ -7475,23 +7475,42 @@ export class Brainy<T = any> implements BrainyInterface<T> {
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<void> => {
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<string, number>
): Promise<void> => {
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<T = any> implements BrainyInterface<T> {
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<string>(delta.nouns)
const reconVerbs = new Set<string>(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
})

410
src/db/generationStore.ts
View file

@ -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<id, number[]>` 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<string, number[]>()
private readonly verbChains = new Map<string, number[]>()
/** Whether {@link nounChains}/{@link verbChains} reflect every committed generation. */
private chainsReady = false
/** De-dupes concurrent first-callers of {@link ensureChains}. */
private chainsBuilding: Promise<void> | null = null
private readonly recentNounChains = new Map<string, number[]>()
private readonly recentVerbChains = new Map<string, number[]>()
/** 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<number, string[]>()
private readonly windowVerbDeltas = new Map<number, string[]>()
/** 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<string, number[]>()
private readonly coldVerbChains = new Map<string, number[]>()
/** 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<void> | 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<string>,
gen: number
): Promise<Map<string, number>> {
const firstAfter = new Map<string, number>()
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<boolean> => {
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<GenerationRecord | null> {
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<void> {
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<void> {
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<number[]> {
// 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<void> => {
let delta: { nouns: Set<string>; verbs: Set<string> }
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<string>, verbs: Iterable<string>): 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<string>()
const trimVerbs = new Set<string>()
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<string, number[]>, 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<string, number[]>, 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

542
tests/unit/db/bounded-chains.test.ts Normal file
View file

@ -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<id, number[]>`) → 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<string, unknown> {
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<T>(p: Promise<T>, ms: number, label: string): Promise<T> {
return Promise.race([
p,
new Promise<T>((_, 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<number> {
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<number> {
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<string> {
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
// (<windowLo), and the exact window/horizon boundaries.
for (let pin = 0; pin <= maxGen; pin++) {
for (const id of nounIds) {
expect(norm(await store.resolveAt('noun', id, pin))).toBe(
await bruteForce('noun', id, pin, maxGen)
)
}
for (const id of verbIds) {
expect(norm(await store.resolveAt('verb', id, pin))).toBe(
await bruteForce('verb', id, pin, maxGen)
)
}
}
})
// ==========================================================================
// 1b. Cold-cache coherence: an id touched AFTER it was cold-cached must NOT be
// served stale (the current↔record transition the unbounded design got
// for free).
// ==========================================================================
it('a deep pin stays correct when the id is touched again after its cold chain was cached', async () => {
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<string, number>()
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<string, number>): Promise<string> => {
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()
})
})

2
tests/unit/db/generation-chain.test.ts
View file

@ -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()