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

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

Browse source 
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
Download patch file Download diff file Expand all files Collapse all files

209
src/db/generationStore.ts
View file

@ -120,6 +120,35 @@ export class GenerationStore {
{ 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. */
private readonly pins = new Map<number, number>()
@ -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<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,13 +773,12 @@ export class GenerationStore {
* @param gen - The pinned generation.
*/
async commitTimestampAtOrBefore(gen: number): Promise<number | null> {
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<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
// ==========================================================================
@ -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<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
}