diff --git a/src/db/generationStore.ts b/src/db/generationStore.ts index 95493117..a94a333e 100644 --- a/src/db/generationStore.ts +++ b/src/db/generationStore.ts @@ -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 { + 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 { + 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, verbs: Iterable): 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 { const nouns = new Set() const verbs = new Set() - 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 { 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 { - // 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 { 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 -}