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

feat(8.0): Model-B per-write generation-stamping + adaptive retention knob

Browse source 
Every write — transact() AND single-op add/update/remove/relate — is now its
own immutable generation (Model-B), so a now() pin always freezes and
asOf/since/diff/history/transactionLog reflect single-ops exactly like
transacts. Closes the Model-A hole where pins did not freeze against single-op
writes.

Generation-stamping:
- GenerationStore.commitSingleOp: a one-operation commitTransaction with
  deferred durability. Wired into add/update/remove/relate/updateRelation/
  unrelate + removeMany (the *Many and VFS paths delegate to these).
- Async group-commit (flushPendingSingleOps): the live write is acknowledged
  immediately; its before-image is buffered in an in-memory pending tier that
  resolveAt/chains/changedBetween/tx-log read like on-disk generations, so the
  synchronous now() freezes with no forced flush. One fsync per window
  (triggers: size / 50ms timer / flush / close / transact / compactHistory).
- Crash recovery is drop-without-restore for group-commit generations (marked
  groupCommit:true): a crash mid-flush discards the partial generation and
  never restores its before-images, which would otherwise revert the
  already-acknowledged live write.
- Init-time infrastructure (the VFS root) is the un-versioned generation-0
  baseline: a fresh brain reports generation()===0 and an empty
  transactionLog(); the first user write is generation 1.
- Historical find()/related() overlay bound is the full reserved watermark
  (generation()), so un-flushed single-op writes are overlaid too.

Retention knob:
- config `history` -> `retention`: 'all' | 'adaptive' |
  { maxGenerations?, maxAge?, maxBytes?, budgetBytes?, autoCompact? }; unset ->
  adaptive (disk/RAM pressure, zero-config). CompactHistoryOptions floors ->
  caps (retainGenerations->maxGenerations, retainMs->maxAge, +maxBytes):
  reclaim oldest-unpinned while ANY cap is exceeded; pins always exempt.
- brain.setRetentionBudget(bytes) drives the adaptive byte budget at runtime
  (a coordinator's fair-share input). Per-generation bytes recorded in each
  delta enable historyBytes() introspection without a storage size API.

Tests: per-write generation resolution, pin freeze vs add/update/remove,
drop-without-restore corruption-trap (fault injector), clean-reopen replay,
maxBytes/maxAge/no-cap reclamation, retention-then-reopen. 107 db/generation/
temporal tests green, tsc clean. Docs (ADR-001, consistency-model, snapshots
guide, api reference, RELEASES) updated to per-write granularity + retention.
This commit is contained in:
David Snelling 2026-06-22 15:19:58 -07:00
parent afac7f9662
commit 5c3bb2c864
15 changed files with 1207 additions and 218 deletions
Download patch file Download diff file Expand all files Collapse all files

549
src/db/generationStore.ts
View file

@ -114,10 +114,13 @@ export class GenerationStore {
/** Sorted list of committed generations whose record dirs exist. */
private committedGens: number[] = []
/** Delta cache, keyed by generation (lazily loaded from `tx.json`). */
/** 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
* byte accounting, or for un-flushed pending generations). */
private readonly deltaCache = new Map<
number,
{ nouns: Set<string>; verbs: Set<string>; timestamp: number }
{ nouns: Set<string>; verbs: Set<string>; timestamp: number; bytes: number }
>()
/**
@ -149,6 +152,44 @@ export class GenerationStore {
*/
private deltaCacheMax = 4096
/**
* Model-B per-write group-commit — the in-memory PENDING tier.
*
* Each single-operation write reserves its OWN generation (the locked
* cross-project contract: "every write versioned; a distinct generation per
* single-op"), captures byte-identical before-images, and buffers them here
* instead of paying a synchronous per-write manifest fsync (the 3-5x write
* regression). {@link flushPendingSingleOps} persists the whole buffer to
* disk in ONE fsync (durability batching — NOT a generation collapse).
*
* 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
* while pending, from disk once flushed). That is what lets the SYNCHRONOUS
* `now()` pin freeze against un-flushed single-ops with no forced flush.
*
* Durability boundary: a hard crash before a flush loses only the buffered
* *history* of un-flushed writes (live data is already in canonical storage);
* a crash mid-flush is recovered by drop-without-restore (see
* {@link GenerationDelta.groupCommit}).
*/
private pendingGens: number[] = []
private readonly pendingBuffer = new Map<
number,
{ nouns: Map<string, GenerationRecord>; verbs: Map<string, GenerationRecord>; timestamp: number }
>()
/** Pending timer-coalesced flush handle (cleared on flush/close). */
private pendingFlushTimer: ReturnType<typeof setTimeout> | null = null
/**
* Flush the pending tier once it holds this many generations (size trigger),
* complementing the {@link PENDING_FLUSH_DELAY_MS} timer trigger. Not
* `readonly` so tests can drive the boundary deterministically.
*/
private pendingFlushThreshold = 256
/** Coalescing window (ms) before an idle pending tier is flushed. */
private static readonly PENDING_FLUSH_DELAY_MS = 50
/** Live pin refcounts, keyed by pinned generation. */
private readonly pins = new Map<number, number>()
@ -216,8 +257,13 @@ export class GenerationStore {
if (gen <= this.committed) {
committedGens.push(gen)
} else if (!options?.readOnly) {
// Uncommitted (crashed) transaction: restore before-images, drop the dir.
await this.rollBackUncommittedGeneration(gen)
// Uncommitted (crashed) generation. A transact generation is rolled
// back by RESTORING its before-images (its before-image + execute are
// one atomic unit). A single-op group-commit generation is DROPPED
// WITHOUT RESTORE — its live write already landed and was acknowledged
// before the flush, so restoring would silently revert it. The branch
// is decided by the persisted delta's `groupCommit` flag.
await this.recoverUncommittedGeneration(gen)
rolledBack++
}
// Reader mode: leave orphan dirs alone; resolution ignores them because
@ -252,6 +298,10 @@ export class GenerationStore {
* from `brain.close()`.
*/
async close(): Promise<void> {
// Persist any buffered single-op history before detaching, so a clean close
// never loses generations the caller already observed.
this.clearPendingFlushTimer()
await this.flushPendingSingleOps()
this.storage.setGenerationBumpHook(undefined)
await this.persistCounterNow()
}
@ -446,11 +496,13 @@ export class GenerationStore {
try {
// -- 3. Before-images + delta (the durable undo log) ------------------
const stagedPaths: string[] = []
let recordBytes = 0 // serialized record-set size, for retention accounting
for (const id of nouns) {
const prev = await this.storage.readNounRaw(id)
const recordPath = `${dir}/prev/${id}.json`
const record: GenerationRecord = { kind: 'noun', metadata: prev.metadata, vector: prev.vector }
await this.storage.writeRawObject(recordPath, record)
recordBytes += serializedBytes(record)
stagedPaths.push(recordPath)
}
for (const id of verbs) {
@ -458,6 +510,7 @@ export class GenerationStore {
const recordPath = `${dir}/prev/${id}.json`
const record: GenerationRecord = { kind: 'verb', metadata: prev.metadata, vector: prev.vector }
await this.storage.writeRawObject(recordPath, record)
recordBytes += serializedBytes(record)
stagedPaths.push(recordPath)
}
const delta: GenerationDelta = {
@ -467,6 +520,7 @@ export class GenerationStore {
nouns,
verbs
}
delta.bytes = recordBytes + serializedBytes(delta)
const deltaPath = `${dir}/tx.json`
await this.storage.writeRawObject(deltaPath, delta)
stagedPaths.push(deltaPath)
@ -498,7 +552,12 @@ export class GenerationStore {
// -- 6. Post-commit bookkeeping ---------------------------------------
this.committed = gen
this.committedGens.push(gen)
this.setDelta(gen, { nouns: new Set(nouns), verbs: new Set(verbs), timestamp })
this.setDelta(gen, {
nouns: new Set(nouns),
verbs: new Set(verbs),
timestamp,
bytes: 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))
@ -530,6 +589,263 @@ export class GenerationStore {
})
}
// ==========================================================================
// Single-operation commit (Model-B per-write group-commit)
// ==========================================================================
/**
* @description Commit ONE single-operation write (`add`/`update`/`remove`/
* `relate`/`updateRelation`/`unrelate` and their `*Many` per-item calls) as
* its own immutable generation — the Model-B "every write versioned"
* contract. Structurally a one-operation {@link commitTransaction} with
* **deferred durability**:
*
* - Under the commit mutex it reserves generation `N`, captures byte-identical
* before-images of every touched id (so the pin/`asOf` read source is
* exact), runs `execute()` (the existing single-op `executeTransaction`
* batch) with the storage bump hook suppressed, then BUFFERS the
* before-images in the in-memory pending tier and returns — **no per-write
* 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
* 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
* buffered *history*; live data is already in canonical storage. A crash
* mid-flush is recovered by drop-without-restore
* (see {@link GenerationDelta.groupCommit}).
*
* The per-write generation is read by graph-index ops through the same
* `generation()` watermark `transact()` uses — because this method, like
* {@link commitTransaction}, holds the mutex across the counter bump AND
* `execute()`, so the watermark equals `N` for the duration of the write (a
* distinct generation threaded to cor per single-op, the locked contract).
*
* @param args.touched - The ids this write creates/updates/deletes, by kind.
* @param args.execute - Runs the single-op's existing operation batch.
* @returns The reserved generation and its timestamp.
*/
async commitSingleOp(args: {
touched: { nouns?: string[]; verbs?: string[] }
execute: () => Promise<void>
}): Promise<{ generation: number; timestamp: number }> {
return this.withMutex(async () => {
const nouns = args.touched.nouns ? [...new Set(args.touched.nouns)] : []
const verbs = args.touched.verbs ? [...new Set(args.touched.verbs)] : []
const gen = ++this.counter
const timestamp = Date.now()
// Before-images BEFORE execute overwrites live storage (mirrors
// commitTransaction's staging, but buffered in memory — durability is
// deferred to the flush). readNounRaw/readVerbRaw on an absent id return
// {metadata:null, vector:null} = the create sentinel.
const nounBefore = new Map<string, GenerationRecord>()
for (const id of nouns) {
const prev = await this.storage.readNounRaw(id)
nounBefore.set(id, { kind: 'noun', metadata: prev.metadata, vector: prev.vector })
}
const verbBefore = new Map<string, GenerationRecord>()
for (const id of verbs) {
const prev = await this.storage.readVerbRaw(id)
verbBefore.set(id, { kind: 'verb', metadata: prev.metadata, vector: prev.vector })
}
// Execute the live write. inTransact suppresses the storage bump hook so
// the counter is not double-advanced (this generation already reserved
// it) — the same suppression transact() uses.
this.inTransact = true
try {
await args.execute()
} catch (err) {
this.inTransact = false
// Live write failed: nothing buffered, nothing on disk. Return the
// reservation (when no concurrent bump consumed a later number) so
// generation() is unchanged.
if (this.counter === gen) this.counter = gen - 1
throw err
}
this.inTransact = false
// Buffer the pending generation + make it instantly visible to reads.
this.pendingBuffer.set(gen, { nouns: nounBefore, verbs: verbBefore, timestamp })
this.pendingGens.push(gen)
this.extendChains(gen, nouns, verbs)
this.schedulePendingFlush()
return { generation: gen, timestamp }
})
}
/**
* @description Run a write WITHOUT creating a generation — for init-time /
* infrastructure writes (e.g. the VFS root directory) that constitute the
* generation-0 baseline of a freshly-materialized brain rather than a user
* mutation. The storage bump hook is suppressed (`inTransact`) so the
* generation counter does not advance, so a brand-new brain reports
* `generation() === 0` and an empty `transactionLog()`, and the first USER
* write is generation 1. Mirrors Datomic semantics where the empty database
* value is the starting point and bootstrap is not a transaction.
* @param execute - The infrastructure write to apply to the baseline.
*/
async runWithoutGeneration(execute: () => Promise<void>): Promise<void> {
this.inTransact = true
try {
await execute()
} finally {
this.inTransact = false
}
}
/**
* @description Persist the in-memory pending single-op generations to disk in
* ONE fsync (async group-commit) and advance the committed watermark. Called
* on the size/timer triggers and forced by `flush()`/`close()`/`transact()`/
* `compactHistory()` (so generations stay strictly ordered and durable before
* any of those). A no-op when the pending tier is empty.
*
* Each pending generation is written as a normal `_generations/<N>/` record
* set (`prev/<id>.json` before-images + `tx.json` delta) but with the delta's
* `groupCommit: true` flag set — the drop-without-restore recovery marker.
* The single manifest rename to the highest generation commits the whole
* batch atomically (committed iff `gen ≤ manifest.generation`).
*/
async flushPendingSingleOps(): Promise<void> {
if (this.pendingGens.length === 0) return
return this.withMutex(async () => {
if (this.pendingGens.length === 0) return
this.clearPendingFlushTimer()
const gens = [...this.pendingGens].sort((a, b) => a - b)
const stagedPaths: string[] = []
const logEntries: TxLogEntry[] = []
const genBytes = new Map<number, number>() // for the post-commit setDelta
for (const gen of gens) {
const buf = this.pendingBuffer.get(gen)
if (!buf) continue
const dir = `${GENERATIONS_PREFIX}/${gen}`
const nounIds: string[] = []
const verbIds: string[] = []
let recordBytes = 0
for (const [id, record] of buf.nouns) {
const path = `${dir}/prev/${id}.json`
await this.storage.writeRawObject(path, record)
recordBytes += serializedBytes(record)
stagedPaths.push(path)
nounIds.push(id)
}
for (const [id, record] of buf.verbs) {
const path = `${dir}/prev/${id}.json`
await this.storage.writeRawObject(path, record)
recordBytes += serializedBytes(record)
stagedPaths.push(path)
verbIds.push(id)
}
const delta: GenerationDelta = {
generation: gen,
timestamp: buf.timestamp,
groupCommit: true,
nouns: nounIds,
verbs: verbIds
}
delta.bytes = recordBytes + serializedBytes(delta)
genBytes.set(gen, delta.bytes)
const deltaPath = `${dir}/tx.json`
await this.storage.writeRawObject(deltaPath, delta)
stagedPaths.push(deltaPath)
logEntries.push({ generation: gen, timestamp: buf.timestamp })
}
// ONE fsync for the whole window — the durability-batching win.
await this.storage.syncRawObjects(stagedPaths)
// Test-only crash simulation: a throwing injector here leaves the staged
// group-commit generation dirs on disk with NO manifest advance — the
// exact "crashed mid-flush" state recovery must DROP-WITHOUT-RESTORE
// (the pending in-memory state would be lost in a real crash; we abandon
// this store and recover from disk on the next open()).
if (this.commitFaultInjector) this.commitFaultInjector('before-manifest-rename')
// Commit point: persist the counter, then atomically rename the manifest
// to the highest pending generation — that commits ALL of them at once.
const highest = gens[gens.length - 1]
const highestTs = this.pendingBuffer.get(highest)?.timestamp ?? Date.now()
await this.persistCounterUnlocked()
const manifest: GenerationManifest = {
version: 1,
generation: highest,
committedAt: new Date(highestTs).toISOString(),
horizon: this.horizonGen
}
await this.storage.writeRawObject(MANIFEST_PATH, manifest)
await this.storage.syncRawObjects([MANIFEST_PATH])
// Move pending → committed. Chains already carry these gens; seed the
// bounded delta cache from the buffer so the next read avoids a re-read,
// then drop the in-memory before-images.
this.committed = highest
for (const gen of gens) {
const buf = this.pendingBuffer.get(gen)
if (!buf) continue
this.committedGens.push(gen)
this.setDelta(gen, {
nouns: new Set(buf.nouns.keys()),
verbs: new Set(buf.verbs.keys()),
timestamp: buf.timestamp,
bytes: genBytes.get(gen) ?? 0
})
this.pendingBuffer.delete(gen)
}
this.pendingGens = []
for (const entry of logEntries) {
await this.storage.appendTxLogLine(JSON.stringify(entry))
}
})
}
/** Schedule a coalesced pending-tier flush (size trigger fires immediately on
* the next microtask; otherwise a {@link PENDING_FLUSH_DELAY_MS} timer). Both
* defer outside the current mutex section so the flush can re-acquire it. */
private schedulePendingFlush(): void {
if (this.pendingGens.length >= this.pendingFlushThreshold) {
void Promise.resolve()
.then(() => this.flushPendingSingleOps())
.catch((err) =>
prodLog.warn(`[GenerationStore] pending single-op flush failed: ${(err as Error).message}`)
)
return
}
if (this.pendingFlushTimer !== null) return
this.pendingFlushTimer = setTimeout(() => {
this.pendingFlushTimer = null
void this.flushPendingSingleOps().catch((err) =>
prodLog.warn(`[GenerationStore] pending single-op flush failed: ${(err as Error).message}`)
)
}, GenerationStore.PENDING_FLUSH_DELAY_MS)
// A background flush must not keep the process alive (Node only).
const t = this.pendingFlushTimer as unknown as { unref?: () => void }
if (t && typeof t.unref === 'function') t.unref()
}
/** Cancel any scheduled pending-tier flush timer. */
private clearPendingFlushTimer(): void {
if (this.pendingFlushTimer !== null) {
clearTimeout(this.pendingFlushTimer)
this.pendingFlushTimer = null
}
}
/** @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]
}
// ==========================================================================
// Point-in-time resolution
// ==========================================================================
@ -541,7 +857,13 @@ export class GenerationStore {
* @param gen - The pinned generation.
*/
hasCommittedAfter(gen: number): boolean {
const last = this.committedGens[this.committedGens.length - 1]
// Pending single-op generations count: a now() pin must report itself
// historical the moment any later write (committed OR un-flushed) lands,
// else the Model-A "pin doesn't freeze against single-ops" hole reopens.
const last =
this.pendingGens.length > 0
? this.pendingGens[this.pendingGens.length - 1]
: this.committedGens[this.committedGens.length - 1]
return last !== undefined && last > gen
}
@ -582,9 +904,7 @@ export class GenerationStore {
// 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
const record = await this.readBeforeImage(kind, candidate, id)
if (record === null) {
// The record-set exists (candidate is committed) but the id's
// before-image is missing — only possible through external tampering.
@ -599,6 +919,26 @@ export class GenerationStore {
return { source: 'record', metadata: record.metadata, vector: record.vector }
}
/**
* @description Read the before-image of `id` stored by generation `gen` from
* the right tier: the in-memory pending buffer while the generation is
* un-flushed, or `_generations/<gen>/prev/<id>.json` on disk once committed.
* Returns `null` when the record-set has no before-image for `id`.
*/
private async readBeforeImage(
kind: 'noun' | 'verb',
gen: number,
id: string
): Promise<GenerationRecord | null> {
const pending = this.pendingBuffer.get(gen)
if (pending) {
return (kind === 'noun' ? pending.nouns : pending.verbs).get(id) ?? null
}
return (await this.storage.readRawObject(
`${GENERATIONS_PREFIX}/${gen}/prev/${id}.json`
)) as GenerationRecord | null
}
/**
* @description Ensure the per-id history chains ({@link nounChains} /
* {@link verbChains}) reflect every committed generation. Built once, lazily,
@ -614,8 +954,9 @@ export class GenerationStore {
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) {
// reservedGens (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()) {
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)
@ -655,7 +996,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.committedGens) {
for (const gen of this.reservedGens()) {
if (gen <= fromGen || gen > toGen) continue
const delta = await this.getDelta(gen)
for (const id of delta.nouns) nouns.add(id)
@ -686,7 +1027,7 @@ export class GenerationStore {
toGen: number
): Promise<number[]> {
const result: number[] = []
for (const gen of this.committedGens) {
for (const gen of this.reservedGens()) {
if (gen <= fromGen || gen > toGen) continue
const delta = await this.getDelta(gen)
const touched = kind === 'noun' ? delta.nouns : delta.verbs
@ -723,9 +1064,10 @@ export class GenerationStore {
/**
* @description Resolve a wall-clock timestamp to the generation that was
* committed at-or-before it, via the tx-log. Resolution granularity is
* transact commits — single-operation writes between commits are not
* individually addressable (documented 8.0 history granularity).
* committed at-or-before it, via the tx-log. Under Model-B every write —
* `transact()` AND single-op — has its own tx-log entry, so resolution is
* per-write (the tx-log includes un-flushed single-op generations too, so
* `asOf(Date)` is consistent with the rest of the temporal surface).
* @param timestampMs - Milliseconds since epoch.
* @returns The resolved generation (`0` when `timestampMs` predates the
* first commit) and the matched tx-log entry when one exists.
@ -763,6 +1105,16 @@ export class GenerationStore {
}
if (entry.generation <= this.committed) entries.push(entry)
}
// Include un-flushed single-op generations so the tx-log is consistent with
// the rest of the temporal surface (`since`/`diff`/`asOf`/`changedBetween`
// already resolve over the pending tier). Reads are thus flush-agnostic —
// whether the async group-commit has run yet never changes results, only
// crash durability. Pending single-ops are always newer than every
// committed generation, so they extend the ascending list.
for (const gen of this.pendingGens) {
const buf = this.pendingBuffer.get(gen)
if (buf) entries.push({ generation: gen, timestamp: buf.timestamp })
}
return entries
}
@ -773,9 +1125,10 @@ export class GenerationStore {
* @param gen - The pinned generation.
*/
async commitTimestampAtOrBefore(gen: number): Promise<number | 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)
// 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)
if (candidate === undefined) return null
const delta = await this.getDelta(candidate)
return delta.timestamp
@ -783,9 +1136,23 @@ export class GenerationStore {
private async getDelta(
gen: number
): Promise<{ nouns: Set<string>; verbs: Set<string>; timestamp: number }> {
): Promise<{ nouns: Set<string>; verbs: Set<string>; timestamp: number; bytes: number }> {
const cached = this.deltaCache.get(gen)
if (cached) return cached
// A pending (un-flushed) generation has no tx.json on disk yet — derive its
// delta from the in-memory before-image buffer. Not cached: the bounded
// cache is for the disk path; the buffer is the source of truth here.
// bytes = 0: pending generations are not on disk, so they do not count
// toward the on-disk history byte budget ({@link historyBytes}).
const pending = this.pendingBuffer.get(gen)
if (pending) {
return {
nouns: new Set(pending.nouns.keys()),
verbs: new Set(pending.verbs.keys()),
timestamp: pending.timestamp,
bytes: 0
}
}
const delta = (await this.storage.readRawObject(
`${GENERATIONS_PREFIX}/${gen}/tx.json`
)) as GenerationDelta | null
@ -798,7 +1165,8 @@ export class GenerationStore {
const entry = {
nouns: new Set(delta.nouns),
verbs: new Set(delta.verbs),
timestamp: delta.timestamp
timestamp: delta.timestamp,
bytes: delta.bytes ?? 0
}
this.setDelta(gen, entry)
return entry
@ -811,7 +1179,7 @@ export class GenerationStore {
*/
private setDelta(
gen: number,
entry: { nouns: Set<string>; verbs: Set<string>; timestamp: number }
entry: { nouns: Set<string>; verbs: Set<string>; timestamp: number; bytes: number }
): void {
this.deltaCache.set(gen, entry)
while (this.deltaCache.size > this.deltaCacheMax) {
@ -826,34 +1194,75 @@ export class GenerationStore {
// ==========================================================================
/**
* @description Reclaim generation record-sets past the retention policy.
* A generation directory `N` may be removed only when `N` is at or below
* **every** live pin (deleting `N` can only break readers pinned *below*
* `N`, because resolution reads before-images from generations strictly
* greater than the pin). With both retention options supplied, a
* generation must satisfy both to be eligible.
* @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).
* @returns The total on-disk history byte count.
*/
async historyBytes(): Promise<number> {
let total = 0
for (const gen of this.committedGens) {
total += (await this.getDelta(gen)).bytes
}
return total
}
/**
* @description Reclaim generation record-sets per the retention CAPS. A
* generation is removed only when it is at or below **every** live pin
* (deleting `N` can only break readers pinned *below* `N`, because resolution
* reads before-images from generations strictly greater than the pin) — pins
* are ALWAYS exempt. Among the unpinned generations, the OLDEST are reclaimed
* while **ANY** supplied cap is exceeded:
*
* @param options - Retention policy (see {@link CompactHistoryOptions}).
* - `maxGenerations` — reclaim while the committed-generation count exceeds it.
* - `maxAge` — reclaim generations older than `now − maxAge`.
* - `maxBytes` — reclaim while total on-disk history bytes exceed it.
*
* Because reclamation is oldest-first and every cap is monotone in age, once
* the oldest unpinned generation violates no cap, no newer one does either, so
* the scan stops. With no options, every unpinned generation is reclaimed.
*
* @param options - Retention caps (see {@link CompactHistoryOptions}).
* @returns Count of removed record-sets and the new horizon.
*/
async compact(options?: CompactHistoryOptions): Promise<CompactHistoryResult> {
return this.withMutex(async () => {
const minPinned = this.minPinnedGeneration()
const retainGenerations = options?.retainGenerations ?? 0
const countWatermark = this.committed - retainGenerations
const timeCutoff =
options?.retainMs !== undefined ? Date.now() - options.retainMs : Infinity
const maxGenerations = options?.maxGenerations
const maxAge = options?.maxAge
const maxBytes = options?.maxBytes
const ageCutoff = maxAge !== undefined ? Date.now() - maxAge : undefined
const noCaps =
maxGenerations === undefined && maxAge === undefined && maxBytes === undefined
// Running totals the caps are evaluated against; updated as we reclaim.
let remainingCount = this.committedGens.length
let remainingBytes = maxBytes !== undefined ? await this.historyBytes() : 0
const removed: number[] = []
for (const gen of [...this.committedGens]) {
if (gen > countWatermark) continue
if (gen > minPinned) continue
if (timeCutoff !== Infinity) {
const delta = await this.getDelta(gen)
if (delta.timestamp >= timeCutoff) continue
// Pins are always exempt: never reclaim a generation a live pin needs.
if (gen > minPinned) break // committedGens ascending → nothing newer is eligible either
if (!noCaps) {
const violatesCount = maxGenerations !== undefined && remainingCount > maxGenerations
const violatesBytes = maxBytes !== undefined && remainingBytes > maxBytes
let violatesAge = false
if (ageCutoff !== undefined) {
const delta = await this.getDelta(gen)
violatesAge = delta.timestamp < ageCutoff
}
// Oldest-first: once the oldest unpinned gen trips no cap, none newer do.
if (!violatesCount && !violatesBytes && !violatesAge) break
}
const genBytes = maxBytes !== undefined ? (await this.getDelta(gen)).bytes : 0
await this.storage.removeRawPrefix(`${GENERATIONS_PREFIX}/${gen}`)
this.deltaCache.delete(gen)
remainingCount--
remainingBytes -= genBytes
removed.push(gen)
}
@ -892,6 +1301,12 @@ export class GenerationStore {
async reopenAfterRestore(floorGeneration: number): Promise<void> {
await this.withMutex(async () => {
this.deltaCache.clear()
// A wholesale state replacement invalidates any buffered single-op
// history — discard the pending tier (its live writes are gone with the
// replaced store).
this.clearPendingFlushTimer()
this.pendingGens = []
this.pendingBuffer.clear()
this.opened = false
// open() re-reads counter/manifest and re-registers the bump hook.
await this.open()
@ -907,10 +1322,44 @@ export class GenerationStore {
// ==========================================================================
/**
* Restore the before-images of an uncommitted generation to the canonical
* entity paths, then remove its record directory. Restores are idempotent
* (writing identical bytes / deleting already-absent files), so recovery
* is safe at every crash point of the commit protocol.
* Recover one uncommitted (crashed) generation, branching on whether it is a
* single-op group-commit generation (drop-without-restore) or a `transact()`
* generation (restore before-images). The decision reads the generation's
* `tx.json` `groupCommit` flag:
*
* - **`groupCommit: true` → DROP** the directory, never restore. The single-op
* write already mutated canonical storage and was acknowledged to the caller
* before the flush; restoring its before-images would silently revert an
* acknowledged user write (the data-corruption trap). Only the *history* of
* the crashed flush window is lost — live data is correct as-is.
* - **absent/`false` (a transact generation) → RESTORE** then drop (the
* existing atomic-rollback path).
* - **`tx.json` unreadable/missing → DROP** (safe for both): a transact
* fsyncs `tx.json` BEFORE it executes, so an unreadable delta means the
* execute never ran → live storage is unchanged → dropping is a no-op on
* live state, while a partial group-commit must be dropped anyway.
*/
private async recoverUncommittedGeneration(gen: number): Promise<void> {
const dir = `${GENERATIONS_PREFIX}/${gen}`
const delta = (await this.storage.readRawObject(`${dir}/tx.json`)) as GenerationDelta | null
if (delta === null || delta.groupCommit === true) {
// Drop-without-restore (group-commit, or an indeterminate partial dir).
await this.storage.removeRawPrefix(dir)
prodLog.warn(
`[GenerationStore] dropped uncommitted ${
delta === null ? 'indeterminate' : 'group-commit'
} generation ${gen} (live data left intact; history of that flush window discarded)`
)
return
}
await this.rollBackUncommittedGeneration(gen)
}
/**
* Restore the before-images of an uncommitted `transact()` generation to the
* canonical entity paths, then remove its record directory. Restores are
* idempotent (writing identical bytes / deleting already-absent files), so
* recovery is safe at every crash point of the transact commit protocol.
*/
private async rollBackUncommittedGeneration(gen: number): Promise<void> {
const dir = `${GENERATIONS_PREFIX}/${gen}`
@ -962,6 +1411,22 @@ export class GenerationStore {
}
}
/**
* @description Approximate serialized byte size of a record or delta, for the
* Model-B retention byte caps (`maxBytes` / adaptive). The JSON string length
* (chars ≈ bytes) is a deliberately cheap soft estimate — retention is a budget,
* not exact accounting — and avoids a storage size API. Returns 0 on any
* serialization failure (e.g. a cyclic structure, which records never are).
* @param obj - The record or delta about to be persisted.
*/
function serializedBytes(obj: unknown): number {
try {
return JSON.stringify(obj)?.length ?? 0
} catch {
return 0
}
}
/**
* @description Extract the generation number from a record path of the form
* `_generations/<N>/...`. Returns `null` for paths that don't match.