fix: O(1) adaptive retention accounting + historyStats fleet audit

Under default adaptive retention, every flush() recomputed total history
bytes by walking EVERY committed generation's delta — O(all generations)
with disk re-reads past the 4096-entry delta-cache bound. On a production
brain with 70,000+ accumulated generations this turned every write into a
full-tail scan (60-100s writes, escalating with history growth), even
though the free-RAM budget never tripped and nothing was ever reclaimed
(SELF-GENERATIONS-GROWTH).

historyBytes() now maintains a running total: seeded by one walk on first
use, then updated incrementally at both commit paths (+bytes) and the
compaction reclaim loop (−bytes), dropped on reopenAfterRestore. The
adaptive retention check on every flush is O(1). Invariant regression-
pinned: running total ≡ fresh walk through transact commits, single-op
group commits, and compaction.

New brain.historyStats() (exported HistoryStats): read-only generation
count / bytes / generation+timestamp range / horizon / retention mode /
effective budget — the one-call per-brain fleet audit for retention
exposure.
This commit is contained in:
David Snelling 2026-07-18 10:51:37 -07:00
parent 4fcef7b8ed
commit 6207e48b51
6 changed files with 228 additions and 5 deletions

View file

@ -257,6 +257,15 @@ export class GenerationStore {
*/
private deltaCacheMax = 4096
/**
* Running total of on-disk history bytes across committed generations
* `null` until {@link historyBytes} pays its one seeding walk. Maintained
* incrementally at commit/reclaim so the adaptive retention check on every
* flush() is O(1), never a tail re-walk. Never updated by cache re-reads
* ({@link setDelta} inserts are cache population, not new history).
*/
private historyBytesTotal: number | null = null
/**
* Model-B per-write group-commit the in-memory PENDING tier.
*
@ -483,6 +492,40 @@ export class GenerationStore {
return this.horizonGen
}
/**
* @description Read-only history footprint for fleet audits: how much
* generational history this store holds on disk. `bytes` pays (and seeds)
* the one-time {@link historyBytes} walk on first call subsequent calls
* are O(1). The oldest/newest timestamps come from those generations'
* deltas (cache-bounded reads).
* @returns Counts, bytes, generation range, and the compaction horizon.
*/
async historyStats(): Promise<{
generations: number
bytes: number
oldestGeneration: number | null
newestGeneration: number | null
oldestTimestamp: number | null
newestTimestamp: number | null
horizon: number
}> {
let oldest: number | null = null
let newest: number | null = null
for (const gen of this.committedGensAsc()) {
if (oldest === null) oldest = gen
newest = gen
}
return {
generations: this.committedCount(),
bytes: await this.historyBytes(),
oldestGeneration: oldest,
newestGeneration: newest,
oldestTimestamp: oldest !== null ? (await this.getDelta(oldest)).timestamp : null,
newestTimestamp: newest !== null ? (await this.getDelta(newest)).timestamp : null,
horizon: this.horizonGen
}
}
/**
* @description Read one generation's persisted before-image records the
* compaction fallback for generations written before deltas carried
@ -849,6 +892,9 @@ export class GenerationStore {
timestamp,
bytes: delta.bytes ?? 0
})
if (this.historyBytesTotal !== null) {
this.historyBytesTotal += delta.bytes ?? 0
}
this.extendChains(gen, nouns, verbs)
const logEntry: TxLogEntry = { generation: gen, timestamp, ...(args.meta && { meta: args.meta }) }
await this.storage.appendTxLogLine(JSON.stringify(logEntry))
@ -1302,6 +1348,9 @@ export class GenerationStore {
timestamp: buf.timestamp,
bytes: genBytes.get(gen) ?? 0
})
if (this.historyBytesTotal !== null) {
this.historyBytesTotal += genBytes.get(gen) ?? 0
}
this.pendingBuffer.delete(gen)
}
this.pendingGens = []
@ -2122,17 +2171,26 @@ export class GenerationStore {
/**
* @description Total serialized bytes of the ON-DISK generational history
* the sum of every committed generation's recorded `bytes`. Backs the
* `maxBytes` and adaptive retention caps. Reads each committed generation's
* delta (cached; a re-read only for cache-evicted ones) O(committed
* generations), bounded by retention itself and invoked only at compaction
* time. Pending (un-flushed) generations are excluded (they are not on disk).
* `maxBytes` and adaptive retention caps. O(1) after the first call: the
* total is computed by ONE walk over committed deltas, then maintained
* incrementally at every commit (+bytes) and reclaim (bytes) and dropped on
* a wholesale state replacement (restore). Without the running total, the
* adaptive auto-compaction on every flush() re-walked the ENTIRE history
* O(committed generations) file reads per flush past the delta-cache bound
* which is how a 70k-generation production brain turned every write into a
* full-tail scan (SELF-GENERATIONS-GROWTH). Pending (un-flushed) generations
* are excluded (they are not on disk).
* @returns The total on-disk history byte count.
*/
async historyBytes(): Promise<number> {
if (this.historyBytesTotal !== null) {
return this.historyBytesTotal
}
let total = 0
for (const gen of this.committedGensAsc()) {
total += (await this.getDelta(gen)).bytes
}
this.historyBytesTotal = total
return total
}
@ -2206,6 +2264,9 @@ export class GenerationStore {
await this.storage.removeRawPrefix(`${GENERATIONS_PREFIX}/${gen}`)
this.deltaCache.delete(gen)
if (this.historyBytesTotal !== null) {
this.historyBytesTotal -= delta.bytes
}
// AFTER the record-set is gone (over-count-only crash ordering):
// release its history references and reclaim any blob left with zero
@ -2266,6 +2327,9 @@ export class GenerationStore {
async reopenAfterRestore(floorGeneration: number): Promise<void> {
await this.withMutex(async () => {
this.deltaCache.clear()
// The running history-byte total describes the REPLACED store — drop it;
// the next historyBytes() re-seeds with one walk over the new state.
this.historyBytesTotal = null
// A wholesale state replacement invalidates any buffered single-op
// history — discard the pending tier (its live writes are gone with the
// replaced store).