perf(8.0): represent the committed-generation ledger as an interval set
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:
parent
b6beb7f96a
commit
93f61dbc79
1 changed files with 177 additions and 53 deletions
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue