perf(8.0): bound per-id generation history chains (O(W+L) resident, was O(N))
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:
parent
8f4787bb10
commit
a859d6ecf8
4 changed files with 956 additions and 72 deletions
|
|
@ -7475,23 +7475,42 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
const snapshotStorage = new MemoryStorage()
|
const snapshotStorage = new MemoryStorage()
|
||||||
await snapshotStorage.init()
|
await snapshotStorage.init()
|
||||||
|
|
||||||
/** Copy one id's at-`generation` state into the snapshot (or ensure absence). */
|
// Copy one id's at-`generation` state into the snapshot (or ensure absence)
|
||||||
const copyAt = async (kind: 'noun' | 'verb', id: string): Promise<void> => {
|
// as a PURE LOOKUP against a precomputed `firstAfter` map (id → the first
|
||||||
const resolved = await this.generationStore.resolveAt(kind, id, generation)
|
// 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
|
let record: { metadata: any; vector: any | null } | null = null
|
||||||
if (resolved.source === 'record') {
|
const candidate = firstAfter.get(id)
|
||||||
if (resolved.metadata !== null) {
|
if (candidate === undefined) {
|
||||||
record = { metadata: resolved.metadata, vector: resolved.vector }
|
// Untouched since the pin → live storage IS the at-`generation` state.
|
||||||
}
|
|
||||||
} else if (resolved.source === 'current') {
|
|
||||||
const raw =
|
const raw =
|
||||||
kind === 'noun' ? await this.storage.readNounRaw(id) : await this.storage.readVerbRaw(id)
|
kind === 'noun' ? await this.storage.readNounRaw(id) : await this.storage.readVerbRaw(id)
|
||||||
if (raw.metadata !== null) {
|
if (raw.metadata !== null) record = raw
|
||||||
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
|
// `record === null` — absent at this generation: writing nulls ensures the
|
||||||
// stored state, which is not a live entity): writing nulls ensures the
|
|
||||||
// id is absent in the snapshot, which also un-copies ids created by
|
// id is absent in the snapshot, which also un-copies ids created by
|
||||||
// transactions that commit mid-build.
|
// transactions that commit mid-build.
|
||||||
if (kind === 'noun') {
|
if (kind === 'noun') {
|
||||||
|
|
@ -7516,25 +7535,42 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
const id = entityIdFromCanonicalPath(path)
|
const id = entityIdFromCanonicalPath(path)
|
||||||
if (id) verbIds.add(id)
|
if (id) verbIds.add(id)
|
||||||
}
|
}
|
||||||
for (const id of nounIds) await copyAt('noun', id)
|
// ONE ascending pass per kind resolves the whole universe's first-after
|
||||||
for (const id of verbIds) await copyAt('verb', id)
|
// 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
|
// Reconciliation pass under the commit mutex: nothing can commit during
|
||||||
// this section, so afterwards the snapshot is exactly the at-`generation`
|
// this section, so afterwards the snapshot is exactly the at-`generation`
|
||||||
// record set even when transactions committed while the bulk copy ran.
|
// record set even when transactions committed while the bulk copy ran.
|
||||||
// Reconciled ids join the enumeration sets so the vector-index build
|
// 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 () => {
|
await this.generationStore.snapshotWith(async () => {
|
||||||
const committedNow = this.generationStore.committedGeneration()
|
const committedNow = this.generationStore.committedGeneration()
|
||||||
if (committedNow === watermark) return
|
if (committedNow === watermark) return
|
||||||
const delta = await this.generationStore.changedBetween(watermark, committedNow)
|
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)
|
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)
|
verbIds.add(id)
|
||||||
await copyAt('verb', id)
|
await copyAt('verb', id, reconVerbFirstAfter)
|
||||||
}
|
}
|
||||||
watermark = committedNow
|
watermark = committedNow
|
||||||
})
|
})
|
||||||
|
|
|
||||||
|
|
@ -147,19 +147,80 @@ export class GenerationStore {
|
||||||
* ({@link committedRanges})
|
* ({@link committedRanges})
|
||||||
* (which is O(database-age) — confirmed by the scalability spike: a read of an
|
* (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
|
* 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),
|
* mirrors cor's `delta_history` chain.
|
||||||
* maintained on commit, invalidated on compaction.
|
*
|
||||||
|
* 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 recentNounChains = new Map<string, number[]>()
|
||||||
private readonly verbChains = new Map<string, number[]>()
|
private readonly recentVerbChains = new Map<string, number[]>()
|
||||||
/** Whether {@link nounChains}/{@link verbChains} reflect every committed generation. */
|
/** Window inverse index (`gen → ids touched`), one map per kind — the O(d̄),
|
||||||
private chainsReady = false
|
* I/O-free slide-trim driver. Holds exactly the gens in `[windowLo, counter]`. */
|
||||||
/** De-dupes concurrent first-callers of {@link ensureChains}. */
|
private readonly windowNounDeltas = new Map<number, string[]>()
|
||||||
private chainsBuilding: Promise<void> | null = null
|
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
|
* 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`,
|
* deltas are only needed for range queries ({@link changedBetween}, `diff`,
|
||||||
* `since`) and chain (re)builds, and {@link getDelta} transparently re-reads an
|
* `since`) and chain (re)builds, and {@link getDelta} transparently re-reads an
|
||||||
* evicted delta from storage. Bounding this keeps a long-lived high-write
|
* 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.
|
// committedRanges only includes generations ≤ the manifest watermark.
|
||||||
this.counter = Math.max(this.counter, gen)
|
this.counter = Math.max(this.counter, gen)
|
||||||
}
|
}
|
||||||
// Chains are rebuilt lazily from the (possibly changed) generation set on
|
// The history window is rebuilt lazily from the (possibly changed) generation
|
||||||
// the next historical read — covers reopen and reopen-after-restore.
|
// set on the next historical read — covers reopen and reopen-after-restore.
|
||||||
this.invalidateChains()
|
this.invalidateChains()
|
||||||
|
|
||||||
if (rolledBack > 0) {
|
if (rolledBack > 0) {
|
||||||
|
|
@ -1016,16 +1077,39 @@ export class GenerationStore {
|
||||||
| { source: 'absent' }
|
| { source: 'absent' }
|
||||||
| { source: 'record'; metadata: any; vector: any | null }
|
| { source: 'record'; metadata: any; vector: any | null }
|
||||||
> {
|
> {
|
||||||
await this.ensureChains()
|
await this.ensureWindow()
|
||||||
const chain = (kind === 'noun' ? this.nounChains : this.verbChains).get(id)
|
// Snapshot windowLo, then read the chain synchronously (no await between):
|
||||||
if (chain === undefined) {
|
// a concurrent commit can only slide the window at the `await` above, so the
|
||||||
// The id was never touched by any committed generation → the live
|
// routing decision and the chain it reads are a consistent pair.
|
||||||
// storage state IS its state at `gen`.
|
const windowLo = this.windowLo
|
||||||
return { source: 'current' }
|
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) {
|
if (candidate === undefined) {
|
||||||
// Nothing after `gen` touched `id`; the live state is its state at `gen`.
|
// Nothing after `gen` touched `id`; the live state is its state at `gen`.
|
||||||
return { source: 'current' }
|
return { source: 'current' }
|
||||||
|
|
@ -1045,6 +1129,87 @@ export class GenerationStore {
|
||||||
return { source: 'record', metadata: record.metadata, vector: record.vector }
|
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
|
* @description Read the before-image of `id` stored by generation `gen` from
|
||||||
* the right tier: the in-memory pending buffer while the generation is
|
* 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} /
|
* @description Build the resident hot-tail window — the per-id chains
|
||||||
* {@link verbChains}) reflect every committed generation. Built once, lazily,
|
* ({@link recentNounChains}/{@link recentVerbChains}) and the inverse index
|
||||||
* under the commit mutex (so no concurrent commit mutates {@link committedRanges}
|
* ({@link windowNounDeltas}/{@link windowVerbDeltas}) over the last
|
||||||
* mid-build); thereafter maintained incrementally by {@link commit} and
|
* {@link recentWindowGenerations} reserved generations, `[windowLo, counter]`.
|
||||||
* invalidated by {@link compact}. Concurrent first-callers share one build.
|
* Built once, lazily, under the commit mutex (so no concurrent commit mutates
|
||||||
* O(committed generations) the first time; O(1) afterwards.
|
* {@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> {
|
private async ensureWindow(): Promise<void> {
|
||||||
if (this.chainsReady) return
|
if (this.windowReady) return
|
||||||
if (!this.chainsBuilding) {
|
if (!this.windowBuilding) {
|
||||||
this.chainsBuilding = this.withMutex(async () => {
|
this.windowBuilding = this.withMutex(async () => {
|
||||||
if (this.chainsReady) return
|
if (this.windowReady) return
|
||||||
this.nounChains.clear()
|
this.recentNounChains.clear()
|
||||||
this.verbChains.clear()
|
this.recentVerbChains.clear()
|
||||||
// reservedGensAsc (committed ∪ pending) is sorted ascending, so chains
|
this.windowNounDeltas.clear()
|
||||||
// accrue in ascending order and un-flushed single-ops are indexed too.
|
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()) {
|
for (const g of this.reservedGensAsc()) {
|
||||||
|
if (g < this.windowLo) continue
|
||||||
const delta = await this.getDelta(g)
|
const delta = await this.getDelta(g)
|
||||||
for (const nounId of delta.nouns) appendToChain(this.nounChains, nounId, g)
|
const nouns = [...delta.nouns]
|
||||||
for (const verbId of delta.verbs) appendToChain(this.verbChains, verbId, g)
|
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.windowReady = true
|
||||||
this.chainsBuilding = null
|
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
|
* @description Reconstruct the FULL per-id chain for a deep pin (`gen <
|
||||||
* (the eventual build reads `gen` from {@link committedRanges}). */
|
* 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 {
|
private extendChains(gen: number, nouns: Iterable<string>, verbs: Iterable<string>): void {
|
||||||
if (!this.chainsReady) return
|
if (!this.windowReady) return
|
||||||
for (const nounId of nouns) appendToChain(this.nounChains, nounId, gen)
|
const nounArr = [...nouns]
|
||||||
for (const verbId of verbs) appendToChain(this.verbChains, verbId, gen)
|
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
|
/** Drop the resident window + cold LRU so the next read rebuilds from the
|
||||||
* generations (called after compaction removes record-sets). */
|
* surviving generations (called after compaction removes record-sets / on
|
||||||
|
* reopen). */
|
||||||
private invalidateChains(): void {
|
private invalidateChains(): void {
|
||||||
this.chainsReady = false
|
this.windowReady = false
|
||||||
this.chainsBuilding = null
|
this.windowBuilding = null
|
||||||
this.nounChains.clear()
|
this.recentNounChains.clear()
|
||||||
this.verbChains.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
|
* @description Binary-search an ascending generation chain for the FIRST entry
|
||||||
* strictly greater than `gen` (the generation whose before-image holds the
|
* strictly greater than `gen` (the generation whose before-image holds the
|
||||||
|
|
|
||||||
542
tests/unit/db/bounded-chains.test.ts
Normal file
542
tests/unit/db/bounded-chains.test.ts
Normal 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()
|
||||||
|
})
|
||||||
|
})
|
||||||
|
|
@ -82,7 +82,7 @@ describe('generation history chains (Model-B scalability)', () => {
|
||||||
const g0 = await seedX()
|
const g0 = await seedX()
|
||||||
for (let v = 1; v <= 12; v++) await bumpX(v) // 12 gens > cap 4 → evictions
|
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)
|
expect(await vAt(g0)).toBe(0)
|
||||||
// …and a range op (since scans committedGens via getDelta, re-reading evicted deltas) still works.
|
// …and a range op (since scans committedGens via getDelta, re-reading evicted deltas) still works.
|
||||||
const now = await brain.now()
|
const now = await brain.now()
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue