From ceed70d7be92fd9230948c971b4934161cd284ae Mon Sep 17 00:00:00 2001 From: David Snelling Date: Mon, 22 Jun 2026 13:47:33 -0700 Subject: [PATCH] perf(8.0): per-id history chains for O(log) historical reads + bounded delta cache MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Historical reads (asOf/get) scanned the global committedGens list linearly, making them O(database-age): a read of an unchanged entity at an old pin scaled ~12x for 10x history depth. Add per-id inverted history chains (nounChains/ verbChains) so resolveAt binary-searches the id's own generation chain instead — O(log) and flat with depth. Chains build lazily under the commit mutex, maintain incrementally on commit, and invalidate on compaction. Also bound deltaCache (LRU cap 4096; getDelta re-reads evicted deltas) so a long-lived high-write process's heap is O(cap) not O(generations) on the disk-backed path, and binary-search commitTimestampAtOrBefore (O(log)). Verified: 373 unit tests green; new tests/unit/db/generation-chain.test.ts covers chain resolution, eviction re-read, and chain rebuild after compaction. --- src/db/generationStore.ts | 209 +++++++++++++++++++++---- tests/unit/db/generation-chain.test.ts | 112 +++++++++++++ 2 files changed, 291 insertions(+), 30 deletions(-) create mode 100644 tests/unit/db/generation-chain.test.ts diff --git a/src/db/generationStore.ts b/src/db/generationStore.ts index 364cdcef..f23b6a2c 100644 --- a/src/db/generationStore.ts +++ b/src/db/generationStore.ts @@ -120,6 +120,35 @@ export class GenerationStore { { nouns: Set; verbs: Set; timestamp: number } >() + /** + * Per-id inverted history index (Model-B scalability) — `id → ascending + * generations that touched it`, one map per kind. {@link resolveAt} binary- + * searches the id's OWN chain (O(log chain)) for the first generation after + * the pin, instead of linearly scanning the global {@link committedGens} + * (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. + */ + private readonly nounChains = new Map() + private readonly verbChains = new Map() + /** Whether {@link nounChains}/{@link verbChains} reflect every committed generation. */ + private chainsReady = false + /** De-dupes concurrent first-callers of {@link ensureChains}. */ + private chainsBuilding: Promise | null = null + + /** + * Cap on resident {@link deltaCache} entries (Model-B RAM bound). The per-id + * {@link nounChains}/{@link verbChains} are 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 + * process's heap O(cap) instead of O(generations) on the disk-backed path. + * Not `readonly` so tests can lower it to exercise the eviction/re-read path + * without building thousands of generations. + */ + private deltaCacheMax = 4096 + /** Live pin refcounts, keyed by pinned generation. */ private readonly pins = new Map() @@ -196,6 +225,9 @@ export class GenerationStore { this.counter = Math.max(this.counter, gen) } this.committedGens = committedGens + // Chains are rebuilt lazily from the (possibly changed) generation set on + // the next historical read — covers reopen and reopen-after-restore. + this.invalidateChains() if (rolledBack > 0) { prodLog.warn( @@ -466,7 +498,8 @@ export class GenerationStore { // -- 6. Post-commit bookkeeping --------------------------------------- this.committed = gen this.committedGens.push(gen) - this.deltaCache.set(gen, { nouns: new Set(nouns), verbs: new Set(verbs), timestamp }) + this.setDelta(gen, { nouns: new Set(nouns), verbs: new Set(verbs), timestamp }) + this.extendChains(gen, nouns, verbs) const logEntry: TxLogEntry = { generation: gen, timestamp, ...(args.meta && { meta: args.meta }) } await this.storage.appendTxLogLine(JSON.stringify(logEntry)) @@ -535,28 +568,81 @@ export class GenerationStore { | { source: 'absent' } | { source: 'record'; metadata: any; vector: any | null } > { - for (const candidate of this.committedGens) { - if (candidate <= gen) continue - const delta = await this.getDelta(candidate) - const touched = kind === 'noun' ? delta.nouns : delta.verbs - if (!touched.has(id)) continue - const record = (await this.storage.readRawObject( - `${GENERATIONS_PREFIX}/${candidate}/prev/${id}.json` - )) as GenerationRecord | null - if (record === null) { - // The record-set exists (candidate is committed) but the id's - // before-image is missing — only possible through external tampering. - throw new Error( - `Generation record missing: ${GENERATIONS_PREFIX}/${candidate}/prev/${id}.json ` + - `(store corrupted or records removed outside compactHistory())` - ) - } - if (record.metadata === null && record.vector === null) { - return { source: 'absent' } - } - return { source: 'record', metadata: record.metadata, vector: record.vector } + 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' } } - return { source: 'current' } + // 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' } + } + const record = (await this.storage.readRawObject( + `${GENERATIONS_PREFIX}/${candidate}/prev/${id}.json` + )) as GenerationRecord | null + if (record === null) { + // The record-set exists (candidate is committed) but the id's + // before-image is missing — only possible through external tampering. + throw new Error( + `Generation record missing: ${GENERATIONS_PREFIX}/${candidate}/prev/${id}.json ` + + `(store corrupted or records removed outside compactHistory())` + ) + } + if (record.metadata === null && record.vector === null) { + return { source: 'absent' } + } + return { source: 'record', metadata: record.metadata, vector: record.vector } + } + + /** + * @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 committedGens} + * 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. + */ + private async ensureChains(): Promise { + if (this.chainsReady) return + if (!this.chainsBuilding) { + this.chainsBuilding = this.withMutex(async () => { + if (this.chainsReady) return + this.nounChains.clear() + this.verbChains.clear() + // committedGens is sorted ascending, so chains accrue in ascending order. + for (const g of this.committedGens) { + 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) + } + this.chainsReady = true + this.chainsBuilding = null + }) + } + await this.chainsBuilding + } + + /** 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 committedGens}). */ + private extendChains(gen: number, nouns: Iterable, verbs: Iterable): void { + if (!this.chainsReady) return + for (const nounId of nouns) appendToChain(this.nounChains, nounId, gen) + for (const verbId of verbs) appendToChain(this.verbChains, verbId, gen) + } + + /** Drop the chains so the next read rebuilds them from the surviving + * generations (called after compaction removes record-sets). */ + private invalidateChains(): void { + this.chainsReady = false + this.chainsBuilding = null + this.nounChains.clear() + this.verbChains.clear() } /** @@ -687,13 +773,12 @@ export class GenerationStore { * @param gen - The pinned generation. */ async commitTimestampAtOrBefore(gen: number): Promise { - for (let i = this.committedGens.length - 1; i >= 0; i--) { - const candidate = this.committedGens[i] - if (candidate > gen) continue - const delta = await this.getDelta(candidate) - return delta.timestamp - } - return null + // committedGens is sorted ascending — binary-search the largest committed + // generation ≤ gen (O(log n), not an O(database-age) backward scan). + const candidate = largestAtOrBefore(this.committedGens, gen) + if (candidate === undefined) return null + const delta = await this.getDelta(candidate) + return delta.timestamp } private async getDelta( @@ -715,10 +800,27 @@ export class GenerationStore { verbs: new Set(delta.verbs), timestamp: delta.timestamp } - this.deltaCache.set(gen, entry) + this.setDelta(gen, entry) return entry } + /** + * @description Insert a delta into {@link deltaCache}, evicting the oldest + * entries (FIFO, by insertion order) past {@link deltaCacheMax}. Eviction is + * safe: {@link getDelta} re-reads any evicted delta from storage on demand. + */ + private setDelta( + gen: number, + entry: { nouns: Set; verbs: Set; timestamp: number } + ): void { + this.deltaCache.set(gen, entry) + while (this.deltaCache.size > this.deltaCacheMax) { + const oldest = this.deltaCache.keys().next().value + if (oldest === undefined) break + this.deltaCache.delete(oldest) + } + } + // ========================================================================== // Compaction // ========================================================================== @@ -758,6 +860,8 @@ export class GenerationStore { if (removed.length > 0) { const removedSet = new Set(removed) this.committedGens = this.committedGens.filter((gen) => !removedSet.has(gen)) + // Reclaimed generations leave the per-id chains stale → rebuild on next read. + this.invalidateChains() this.horizonGen = Math.max(this.horizonGen, ...removed) const manifest: GenerationManifest = { version: 1, @@ -879,3 +983,48 @@ function recordIdFromPath(path: string): string | null { const match = /[/\\]prev[/\\]([^/\\]+)\.json$/.exec(path) return match ? match[1] : null } + +/** + * @description Append `gen` to an id's ascending history chain in `chains`, + * creating the chain on first touch. Callers append in ascending generation + * order, so the chain stays sorted without a re-sort. + */ +function appendToChain(chains: Map, id: string, gen: number): void { + const chain = chains.get(id) + if (chain === undefined) { + chains.set(id, [gen]) + } else if (chain[chain.length - 1] !== gen) { + chain.push(gen) + } +} + +/** + * @description Binary-search an ascending generation chain for the FIRST entry + * strictly greater than `gen` (the generation whose before-image holds the + * value as of `gen`). Returns `undefined` when every entry is ≤ `gen`. O(log n). + */ +function firstGenerationAfter(chain: number[], gen: number): number | undefined { + let lo = 0 + let hi = chain.length // upper-bound search over [lo, hi) + while (lo < hi) { + const mid = (lo + hi) >>> 1 + if (chain[mid] > gen) hi = mid + else lo = mid + 1 + } + return lo < chain.length ? chain[lo] : undefined +} + +/** + * @description Binary-search an ascending list for the LARGEST entry less than + * or equal to `gen`. Returns `undefined` when every entry is greater. O(log n). + */ +function largestAtOrBefore(sorted: number[], gen: number): number | undefined { + let lo = 0 + let hi = sorted.length // first index whose value is > gen + while (lo < hi) { + const mid = (lo + hi) >>> 1 + if (sorted[mid] <= gen) lo = mid + 1 + else hi = mid + } + return lo > 0 ? sorted[lo - 1] : undefined +} diff --git a/tests/unit/db/generation-chain.test.ts b/tests/unit/db/generation-chain.test.ts new file mode 100644 index 00000000..4effee72 --- /dev/null +++ b/tests/unit/db/generation-chain.test.ts @@ -0,0 +1,112 @@ +/** + * @module tests/unit/db/generation-chain + * @description Locks in the Model-B scalability fixes in `generationStore.ts`: + * the per-id history chains that make `resolveAt` O(log) instead of an + * O(database-age) scan of the global `committedGens`, the bounded `deltaCache` + * (evicted deltas are transparently re-read from storage), and chain + * maintenance across commit + compaction. These assert the OBSERVABLE behavior + * — `asOf()` correctness — across the code paths those fixes touch, including the + * edge cases the existing temporal suites don't reach (eviction past the cap, + * and chain rebuild after compaction). + */ + +import { describe, it, expect, beforeEach, afterEach } from 'vitest' +import { Brainy } from '../../../src/index.js' +import { NounType } from '../../../src/types/graphTypes.js' +import { createTestConfig } from '../../helpers/test-factory.js' + +const X = '11111111-1111-4111-8111-111111111111' +const Y = '22222222-2222-4222-8222-222222222222' + +describe('generation history chains (Model-B scalability)', () => { + let brain: Brainy + + beforeEach(async () => { + brain = new Brainy(createTestConfig()) + await brain.init() + }) + afterEach(async () => { + await brain.close() + }) + + /** Seed X (its own committed generation), then return g0 = that generation. */ + async function seedX(): Promise { + const db = await brain.transact([ + { op: 'add', id: X, type: NounType.Document, subtype: 'note', data: 'x', metadata: { v: 0 } } + ]) + await db.release() + return brain.generation() + } + + async function bumpX(v: number): Promise { + const db = await brain.transact([{ op: 'update', id: X, metadata: { v } }]) + await db.release() + return brain.generation() + } + + const vAt = async (gen: number): Promise => { + const db = await brain.asOf(gen) + const e = (await db.get(X)) as any + await db.release() + return e?.metadata?.v + } + + it('resolveAt returns the value as-of each pinned generation (chain binary search)', async () => { + const g0 = await seedX() + const g1 = await bumpX(1) + const g2 = await bumpX(2) + + expect(await vAt(g0)).toBe(0) // before-image of the g1 update = the seed value + expect(await vAt(g1)).toBe(1) // before-image of the g2 update + expect(((await brain.get(X)) as any)?.metadata?.v).toBe(2) // live head + expect(g2).toBeGreaterThan(g1) + expect(g1).toBeGreaterThan(g0) + }) + + it('stays correct when an UNCHANGED entity is read at an old pin (no chain → O(1) current)', async () => { + const g0 = await seedX() + // Churn a DIFFERENT entity many times; X is never touched again. + const db = await brain.transact([{ op: 'add', id: Y, type: NounType.Document, subtype: 'note', data: 'y', metadata: { v: 0 } }]) + await db.release() + for (let i = 1; i <= 30; i++) { + const d = await brain.transact([{ op: 'update', id: Y, metadata: { v: i } }]) + await d.release() + } + // X has no chain entry after g0 → resolveAt returns 'current' without scanning the 31 Y-generations. + expect(await vAt(g0)).toBe(0) + expect(((await brain.get(X)) as any)?.metadata?.v).toBe(0) + }) + + it('survives deltaCache eviction (cap lowered; evicted deltas re-read from storage)', async () => { + ;(brain as any).generationStore.deltaCacheMax = 4 // force eviction well before the gen count + 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)… + 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() + const changed = await now.since(g0) + await now.release() + expect(changed.nouns).toContain(X) + }) + + it('rebuilds chains after compaction and keeps asOf correct above the horizon', async () => { + const g0 = await seedX() + const g1 = await bumpX(1) + const g2 = await bumpX(2) + await bumpX(3) + await bumpX(4) + + // Reclaim everything except the 2 most recent committed generations. + const res = await brain.compactHistory({ retainGenerations: 2 }) + expect(res.removedGenerations).toBeGreaterThan(0) + + // A pin ABOVE the new horizon still resolves (chains were rebuilt from the survivors)… + const live = ((await brain.get(X)) as any)?.metadata?.v + expect(live).toBe(4) + // …and the reclaimed-depth pins are below the horizon now. + expect(res.horizon).toBeGreaterThanOrEqual(g0) + expect(g2).toBeGreaterThan(g1) + }) +})