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

perf(8.0): represent the committed-generation ledger as an interval set

Browse source 
committedGens was a number[] with one element per committed generation. Every
single-op write reserves a distinct generation, so an insert-built 1B corpus
held a ~1B-element array resident regardless of provider — the MVCC sibling of
the storage-cache gate.

Replace it with a sorted disjoint interval set (committedRanges). With no
compaction gaps the whole ledger collapses to a single [start,end] pair, so
resident size is O(number-of-gaps) not O(writes). reservedGens becomes a
generator; point-resolution binary-searches the ranges; compaction trims the
oldest prefix with a defensive prefix-invariant guard. Behaviour is identical
across asOf/diff/since/history/transactionLog — unit 1718 + integration 607 +
the focused MVCC suite green.
This commit is contained in:
David Snelling 2026-06-29 16:05:03 -07:00
parent b6beb7f96a
commit 93f61dbc79
1 changed files with 177 additions and 53 deletions
Download patch file Download diff file Expand all files Collapse all files

230
src/db/generationStore.ts
View file

@ -112,8 +112,24 @@ export class GenerationStore {
/** Compaction horizon — record-sets ≤ this are reclaimed. */
private horizonGen = 0
/** Sorted list of committed generations whose record dirs exist. */
private committedGens: number[] = []
/**
* Committed generations whose record dirs exist, stored as a SORTED, DISJOINT,
* ascending list of INCLUSIVE `[start, end]` intervals (a run-length set).
*
* Committed generations come from the monotonic `counter++`, so they are dense
* contiguous integers EXCEPT where {@link compact} punches a hole (reclaiming
* an oldest contiguous prefix). Storing them as intervals makes the resident
* size O(number-of-compaction-gaps) — typically a SINGLE interval
* `[firstGen, lastGen]` — instead of one array element per generation. A
* billion-insert corpus (every single-op reserves a distinct generation) would
* otherwise hold a ~1B-element array resident for the process lifetime; the
* interval form collapses that to O(1). All access goes through the helpers
* ({@link appendCommittedGen}, {@link committedGensAsc}, {@link committedCount},
* {@link removeCommittedUpTo}, {@link lastCommittedGen},
* {@link largestReservedAtOrBefore}) so the multiset and ordering are identical
* to the old flat array at every point.
*/
private committedRanges: Array<[number, number]> = []
/** Delta cache, keyed by generation (lazily loaded from `tx.json`). `bytes`
* is the serialized record-set size, summed by {@link historyBytes} for the
* `maxBytes` / adaptive retention caps (0 for generations written before
@ -127,7 +143,8 @@ export class GenerationStore {
* 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}
* the pin, instead of linearly scanning the global committed-generation set
* ({@link committedRanges})
* (which is O(database-age) — confirmed by the scalability spike: a read of an
* unchanged entity at an old pin scaled 11.9x for 10x history depth). This
* mirrors cor's `delta_history` chain. Built lazily once (under the mutex),
@ -165,7 +182,7 @@ export class GenerationStore {
* Crucially, these pending generations participate in point-in-time
* resolution EXACTLY like committed ones ({@link resolveAt}, the per-id
* chains, {@link changedBetween}, {@link hasCommittedAfter} all read the
* union via {@link reservedGens}; before-images resolve from this buffer
* union via {@link reservedGensAsc}; before-images resolve from this buffer
* while pending, from disk once flushed). That is what lets the SYNCHRONOUS
* `now()` pin freeze against un-flushed single-ops with no forced flush.
*
@ -252,10 +269,13 @@ export class GenerationStore {
}
let rolledBack = 0
const committedGens: number[] = []
// Coalesce the ascending on-disk committed gens into interval form: each
// contiguous run becomes a single `[start, end]` range (appendCommittedGen
// extends the last range on a +1 step, opens a new one across a gap).
this.committedRanges = []
for (const gen of [...seenGens].sort((a, b) => a - b)) {
if (gen <= this.committed) {
committedGens.push(gen)
this.appendCommittedGen(gen)
} else if (!options?.readOnly) {
// Uncommitted (crashed) generation. A transact generation is rolled
// back by RESTORING its before-images (its before-image + execute are
@ -267,10 +287,9 @@ export class GenerationStore {
rolledBack++
}
// Reader mode: leave orphan dirs alone; resolution ignores them because
// committedGens only includes generations ≤ the manifest watermark.
// committedRanges only includes generations ≤ the manifest watermark.
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()
@ -551,7 +570,7 @@ export class GenerationStore {
// -- 6. Post-commit bookkeeping ---------------------------------------
this.committed = gen
this.committedGens.push(gen)
this.appendCommittedGen(gen)
this.setDelta(gen, {
nouns: new Set(nouns),
verbs: new Set(verbs),
@ -609,7 +628,7 @@ export class GenerationStore {
* manifest fsync** (that synchronous fsync is the 3-5x write regression this
* design avoids).
* - The generation is immediately visible to point-in-time reads (chains +
* {@link reservedGens}), so a synchronous `now()` pin freezes against it
* {@link reservedGensAsc}), so a synchronous `now()` pin freezes against it
* with no forced flush.
* - {@link flushPendingSingleOps} later persists the buffer to disk in one
* fsync (async group-commit). A crash before that flush loses only the
@ -787,7 +806,7 @@ export class GenerationStore {
for (const gen of gens) {
const buf = this.pendingBuffer.get(gen)
if (!buf) continue
this.committedGens.push(gen)
this.appendCommittedGen(gen)
this.setDelta(gen, {
nouns: new Set(buf.nouns.keys()),
verbs: new Set(buf.verbs.keys()),
@ -836,15 +855,121 @@ export class GenerationStore {
}
}
/** @returns Committed ∪ pending generations, ascending. Pending generations
* are always greater than every committed one (reserved after the last
* commit; `transact()`/`compact()` flush the pending tier first), so the
* concatenation is already sorted. This is the union historical reads
* resolve over so un-flushed single-ops are visible to pins/`asOf`. */
private reservedGens(): number[] {
return this.pendingGens.length === 0
? this.committedGens
: [...this.committedGens, ...this.pendingGens]
// ==========================================================================
// Committed-generation interval set (run-length representation)
// ==========================================================================
/**
* @description Record a newly committed generation. `gen` is ALWAYS greater
* than every committed generation (the monotonic `counter++`), so it either
* extends the last interval (the normal contiguous append, `gen ===
* lastEnd + 1`) or opens a fresh single-element interval (the first
* generation, or the first commit after a {@link compact} gap).
* @param gen - The just-committed generation.
*/
private appendCommittedGen(gen: number): void {
const ranges = this.committedRanges
const last = ranges[ranges.length - 1]
if (last !== undefined && gen === last[1] + 1) {
last[1] = gen
} else {
ranges.push([gen, gen])
}
}
/** @returns The newest committed generation, or `undefined` when none exist. */
private lastCommittedGen(): number | undefined {
const ranges = this.committedRanges
return ranges.length ? ranges[ranges.length - 1][1] : undefined
}
/** @returns How many committed generations the interval set represents. */
private committedCount(): number {
let total = 0
for (const [start, end] of this.committedRanges) total += end - start + 1
return total
}
/**
* @description Yield every committed generation ascending — the exact multiset
* (and order) the old flat `committedGens` array held.
*/
private *committedGensAsc(): IterableIterator<number> {
for (const [start, end] of this.committedRanges) {
for (let g = start; g <= end; g++) yield g
}
}
/**
* @description Yield committed ∪ pending generations, ascending. Pending
* generations are always greater than every committed one (reserved after the
* last commit; `transact()`/`compact()` flush the pending tier first), so the
* committed-then-pending concatenation is already sorted — identical to the old
* `[...committedGens, ...pendingGens]`. This is the union historical reads
* resolve over so un-flushed single-ops are visible to pins/`asOf`.
*/
private *reservedGensAsc(): IterableIterator<number> {
yield* this.committedGensAsc()
for (const g of this.pendingGens) yield g
}
/**
* @description Remove every committed generation `<= maxGen`. {@link compact}
* always reclaims an oldest CONTIGUOUS prefix of the committed gens, so this
* walk over the front of {@link committedRanges} is exact: drop whole ranges
* that end at or below `maxGen`, then clip the first surviving range up to
* `maxGen + 1` if `maxGen` falls inside it, and stop.
* @param maxGen - Inclusive upper bound of the reclaimed prefix.
*/
private removeCommittedUpTo(maxGen: number): void {
const ranges = this.committedRanges
let drop = 0 // count of leading ranges fully reclaimed
while (drop < ranges.length) {
const range = ranges[drop]
if (range[1] <= maxGen) {
drop++ // whole range is at or below the cut → reclaim it
} else {
if (range[0] <= maxGen) range[0] = maxGen + 1 // clip the straddling range
break // this range survives; nothing newer is below the cut either
}
}
if (drop > 0) ranges.splice(0, drop)
}
/**
* @description The largest reserved (committed ∪ pending) generation
* `<= gen` — exactly what `largestAtOrBefore([...committedGens, ...pendingGens],
* gen)` returned, in O(log ranges + scanned pending) instead of materializing
* the union. Pending generations are the newest reserved gens (all greater than
* every committed one), so the largest pending `<= gen`, when one exists,
* dominates every committed gen and is the answer; otherwise the answer is the
* largest committed gen `<= gen`, found by binary-searching {@link committedRanges}
* for the rightmost range whose `start <= gen` and clamping `gen` into it.
* @param gen - The upper bound (inclusive).
* @returns The largest reserved generation `<= gen`, or `undefined` when every
* reserved generation is greater than `gen`.
*/
private largestReservedAtOrBefore(gen: number): number | undefined {
// Pending (ascending, all > every committed): the largest one ≤ gen wins.
let bestPending: number | undefined
for (const p of this.pendingGens) {
if (p <= gen) bestPending = p
else break
}
if (bestPending !== undefined) return bestPending
// No pending ≤ gen → the largest committed gen ≤ gen. Binary-search for the
// rightmost range whose start ≤ gen; the answer is min(gen, that range.end).
const ranges = this.committedRanges
let lo = 0
let hi = ranges.length // first range index whose start is > gen
while (lo < hi) {
const mid = (lo + hi) >>> 1
if (ranges[mid][0] <= gen) lo = mid + 1
else hi = mid
}
if (lo === 0) return undefined // every committed range starts above gen
const [, end] = ranges[lo - 1]
return Math.min(gen, end)
}
// ==========================================================================
@ -864,7 +989,7 @@ export class GenerationStore {
const last =
this.pendingGens.length > 0
? this.pendingGens[this.pendingGens.length - 1]
: this.committedGens[this.committedGens.length - 1]
: this.lastCommittedGen()
return last !== undefined && last > gen
}
@ -943,7 +1068,7 @@ export class GenerationStore {
/**
* @description Ensure the per-id history chains ({@link nounChains} /
* {@link verbChains}) reflect every committed generation. Built once, lazily,
* under the commit mutex (so no concurrent commit mutates {@link committedGens}
* under the commit mutex (so no concurrent commit mutates {@link committedRanges}
* mid-build); thereafter maintained incrementally by {@link commit} and
* invalidated by {@link compact}. Concurrent first-callers share one build.
* O(committed generations) the first time; O(1) afterwards.
@ -955,9 +1080,9 @@ export class GenerationStore {
if (this.chainsReady) return
this.nounChains.clear()
this.verbChains.clear()
// reservedGens (committed ∪ pending) is sorted ascending, so chains
// reservedGensAsc (committed ∪ pending) is sorted ascending, so chains
// accrue in ascending order and un-flushed single-ops are indexed too.
for (const g of this.reservedGens()) {
for (const g of this.reservedGensAsc()) {
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)
@ -971,7 +1096,7 @@ export class GenerationStore {
/** 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}). */
* (the eventual build reads `gen` from {@link committedRanges}). */
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)
@ -997,7 +1122,7 @@ export class GenerationStore {
async changedBetween(fromGen: number, toGen: number): Promise<ChangedIds> {
const nouns = new Set<string>()
const verbs = new Set<string>()
for (const gen of this.reservedGens()) {
for (const gen of this.reservedGensAsc()) {
if (gen <= fromGen || gen > toGen) continue
const delta = await this.getDelta(gen)
for (const id of delta.nouns) nouns.add(id)
@ -1028,7 +1153,7 @@ export class GenerationStore {
toGen: number
): Promise<number[]> {
const result: number[] = []
for (const gen of this.reservedGens()) {
for (const gen of this.reservedGensAsc()) {
if (gen <= fromGen || gen > toGen) continue
const delta = await this.getDelta(gen)
const touched = kind === 'noun' ? delta.nouns : delta.verbs
@ -1126,10 +1251,10 @@ export class GenerationStore {
* @param gen - The pinned generation.
*/
async commitTimestampAtOrBefore(gen: number): Promise<number | null> {
// reservedGens (committed ∪ pending) is sorted ascending — binary-search
// the largest reserved generation ≤ gen (O(log n), not an O(database-age)
// backward scan). Pending single-op generations carry timestamps too.
const candidate = largestAtOrBefore(this.reservedGens(), gen)
// The largest reserved (committed ∪ pending) generation ≤ gen, found in
// O(log ranges) over the interval set (not an O(database-age) backward
// scan). Pending single-op generations carry timestamps too.
const candidate = this.largestReservedAtOrBefore(gen)
if (candidate === undefined) return null
const delta = await this.getDelta(candidate)
return delta.timestamp
@ -1205,7 +1330,7 @@ export class GenerationStore {
*/
async historyBytes(): Promise<number> {
let total = 0
for (const gen of this.committedGens) {
for (const gen of this.committedGensAsc()) {
total += (await this.getDelta(gen)).bytes
}
return total
@ -1241,13 +1366,15 @@ export class GenerationStore {
maxGenerations === undefined && maxAge === undefined && maxBytes === undefined
// Running totals the caps are evaluated against; updated as we reclaim.
let remainingCount = this.committedGens.length
let remainingCount = this.committedCount()
let remainingBytes = maxBytes !== undefined ? await this.historyBytes() : 0
// Snapshot the committed gens ascending so the reclaim loop iterates safely
// while removeCommittedUpTo mutates the interval set afterwards.
const removed: number[] = []
for (const gen of [...this.committedGens]) {
for (const gen of [...this.committedGensAsc()]) {
// Pins are always exempt: never reclaim a generation a live pin needs.
if (gen > minPinned) break // committedGens ascending → nothing newer is eligible either
if (gen > minPinned) break // committedGensAsc ascending → nothing newer is eligible either
if (!noCaps) {
const violatesCount = maxGenerations !== undefined && remainingCount > maxGenerations
const violatesBytes = maxBytes !== undefined && remainingBytes > maxBytes
@ -1268,11 +1395,23 @@ export class GenerationStore {
}
if (removed.length > 0) {
const removedSet = new Set(removed)
this.committedGens = this.committedGens.filter((gen) => !removedSet.has(gen))
// The reclaim loop walks committedGensAsc() oldest-first and stops at the
// first survivor, so `removed` is the oldest CONTIGUOUS prefix of the
// committed gens — every committed gen ≤ max(removed) was reclaimed, which
// is exactly what removeCommittedUpTo strips. Guard that prefix invariant.
const highestRemoved = Math.max(...removed)
const countBefore = this.committedCount()
this.removeCommittedUpTo(highestRemoved)
if (this.committedCount() !== countBefore - removed.length) {
throw new Error(
`[GenerationStore] compaction invariant violated: reclaimed ${removed.length} ` +
`generation(s) but committedCount dropped by ${countBefore - this.committedCount()} ` +
`(the reclaimed set is not the oldest contiguous prefix)`
)
}
// 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, highestRemoved)
const manifest: GenerationManifest = {
version: 1,
generation: this.committed,
@ -1479,18 +1618,3 @@ function firstGenerationAfter(chain: number[], gen: number): number | undefined
}
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
}