perf(8.0): per-id history chains for O(log) historical reads + bounded delta cache

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.
This commit is contained in:
David Snelling 2026-06-22 13:47:33 -07:00
parent f3e69110f0
commit ceed70d7be
2 changed files with 291 additions and 30 deletions

View file

@ -120,6 +120,35 @@ export class GenerationStore {
{ nouns: Set<string>; verbs: Set<string>; timestamp: number } { nouns: Set<string>; verbs: Set<string>; 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<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
/**
* 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. */ /** Live pin refcounts, keyed by pinned generation. */
private readonly pins = new Map<number, number>() private readonly pins = new Map<number, number>()
@ -196,6 +225,9 @@ export class GenerationStore {
this.counter = Math.max(this.counter, gen) this.counter = Math.max(this.counter, gen)
} }
this.committedGens = committedGens 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) { if (rolledBack > 0) {
prodLog.warn( prodLog.warn(
@ -466,7 +498,8 @@ export class GenerationStore {
// -- 6. Post-commit bookkeeping --------------------------------------- // -- 6. Post-commit bookkeeping ---------------------------------------
this.committed = gen this.committed = gen
this.committedGens.push(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 }) } const logEntry: TxLogEntry = { generation: gen, timestamp, ...(args.meta && { meta: args.meta }) }
await this.storage.appendTxLogLine(JSON.stringify(logEntry)) await this.storage.appendTxLogLine(JSON.stringify(logEntry))
@ -535,11 +568,20 @@ export class GenerationStore {
| { source: 'absent' } | { source: 'absent' }
| { source: 'record'; metadata: any; vector: any | null } | { source: 'record'; metadata: any; vector: any | null }
> { > {
for (const candidate of this.committedGens) { await this.ensureChains()
if (candidate <= gen) continue const chain = (kind === 'noun' ? this.nounChains : this.verbChains).get(id)
const delta = await this.getDelta(candidate) if (chain === undefined) {
const touched = kind === 'noun' ? delta.nouns : delta.verbs // The id was never touched by any committed generation → the live
if (!touched.has(id)) continue // storage state IS its state at `gen`.
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( const record = (await this.storage.readRawObject(
`${GENERATIONS_PREFIX}/${candidate}/prev/${id}.json` `${GENERATIONS_PREFIX}/${candidate}/prev/${id}.json`
)) as GenerationRecord | null )) as GenerationRecord | null
@ -556,7 +598,51 @@ export class GenerationStore {
} }
return { source: 'record', metadata: record.metadata, vector: record.vector } return { source: 'record', metadata: record.metadata, vector: record.vector }
} }
return { source: 'current' }
/**
* @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<void> {
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<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)
}
/** 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,14 +773,13 @@ export class GenerationStore {
* @param gen - The pinned generation. * @param gen - The pinned generation.
*/ */
async commitTimestampAtOrBefore(gen: number): Promise<number | null> { async commitTimestampAtOrBefore(gen: number): Promise<number | null> {
for (let i = this.committedGens.length - 1; i >= 0; i--) { // committedGens is sorted ascending — binary-search the largest committed
const candidate = this.committedGens[i] // generation ≤ gen (O(log n), not an O(database-age) backward scan).
if (candidate > gen) continue const candidate = largestAtOrBefore(this.committedGens, gen)
if (candidate === undefined) return null
const delta = await this.getDelta(candidate) const delta = await this.getDelta(candidate)
return delta.timestamp return delta.timestamp
} }
return null
}
private async getDelta( private async getDelta(
gen: number gen: number
@ -715,10 +800,27 @@ export class GenerationStore {
verbs: new Set(delta.verbs), verbs: new Set(delta.verbs),
timestamp: delta.timestamp timestamp: delta.timestamp
} }
this.deltaCache.set(gen, entry) this.setDelta(gen, entry)
return 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<string>; verbs: Set<string>; 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 // Compaction
// ========================================================================== // ==========================================================================
@ -758,6 +860,8 @@ export class GenerationStore {
if (removed.length > 0) { if (removed.length > 0) {
const removedSet = new Set(removed) const removedSet = new Set(removed)
this.committedGens = this.committedGens.filter((gen) => !removedSet.has(gen)) 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) this.horizonGen = Math.max(this.horizonGen, ...removed)
const manifest: GenerationManifest = { const manifest: GenerationManifest = {
version: 1, version: 1,
@ -879,3 +983,48 @@ function recordIdFromPath(path: string): string | null {
const match = /[/\\]prev[/\\]([^/\\]+)\.json$/.exec(path) const match = /[/\\]prev[/\\]([^/\\]+)\.json$/.exec(path)
return match ? match[1] : null 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<string, number[]>, 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
}

View file

@ -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<number> {
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<number> {
const db = await brain.transact([{ op: 'update', id: X, metadata: { v } }])
await db.release()
return brain.generation()
}
const vAt = async (gen: number): Promise<number | undefined> => {
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)
})
})