From 5c3bb2c86479494b0092777b0da335aee6e51036 Mon Sep 17 00:00:00 2001 From: David Snelling Date: Mon, 22 Jun 2026 15:19:58 -0700 Subject: [PATCH] feat(8.0): Model-B per-write generation-stamping + adaptive retention knob MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- RELEASES.md | 31 +- docs/ADR-001-generational-mvcc.md | 34 +- docs/api/README.md | 22 +- docs/concepts/consistency-model.md | 20 +- docs/guides/snapshots-and-time-travel.md | 41 +- src/brainy.ts | 313 +++++++--- src/db/db.ts | 12 +- src/db/generationStore.ts | 549 ++++++++++++++++-- src/db/types.ts | 69 ++- src/types/brainy.types.ts | 62 +- tests/benchmarks/model-b-scalability.spike.ts | 6 +- tests/integration/db-mvcc.test.ts | 94 ++- tests/integration/db-temporal.test.ts | 27 +- tests/unit/db/generation-chain.test.ts | 2 +- tests/unit/db/generationStore.test.ts | 143 ++++- 15 files changed, 1207 insertions(+), 218 deletions(-) diff --git a/RELEASES.md b/RELEASES.md index 9e74a8d9..45b7f7b1 100644 --- a/RELEASES.md +++ b/RELEASES.md @@ -110,13 +110,29 @@ What you get: - **`brain.transactionLog({ from?, to?, limit? })`** — the commit log over an **inclusive** generation/`Date` window (contrast `since`'s exclusive lower bound); `limit` applies last. -- **`brain.compactHistory({ retainGenerations, retainMs })`** — explicit - retention. Compaction never breaks a pinned read. `diff`/`since` throw - `GenerationCompactedError` below the horizon; `history` truncates to it. +- **Every write is versioned (Model-B).** History granularity is **per-write**: + EVERY write — `transact()` AND a single-operation `add`/`update`/`remove`/ + `relate` — is its own immutable generation. A `now()` pin always freezes + against later writes, and `asOf`/`since`/`diff`/`history`/`transactionLog` + reflect single-ops exactly like transacts. `transact()` groups several + operations into ONE atomic generation. Single-op history durability is **async + group-commit** (the live write is acknowledged immediately; its before-image + is batched to disk in one fsync) — a hard crash can lose only the last + un-flushed window's *history*, never live data, and a crash mid-flush is + recovered by drop-without-restore. A freshly-initialized brain is + `generation() === 0` with an empty `transactionLog()` (init-time + infrastructure is the un-versioned baseline); the first user write is gen 1. +- **`new Brainy({ retention })`** — the retention knob governs auto-compaction on + `flush()`/`close()`: **unset → ADAPTIVE** (disk/RAM-pressure byte budget, + zero-config; a coordinator can drive it via `brain.setRetentionBudget(bytes)`) + · **`'all'`** → unbounded · **`{ maxGenerations?, maxAge?, maxBytes? }`** → + explicit caps (reclaim oldest-unpinned while ANY cap is exceeded). Live pins + are ALWAYS exempt. +- **`brain.compactHistory({ maxGenerations?, maxAge?, maxBytes? })`** — manual + reclaim on the same caps. Compaction never breaks a pinned read. `diff`/`since` + throw `GenerationCompactedError` below the horizon; `history` truncates to it. -The precise guarantees (and the honest limits — history granularity is -`transact()` commits; single-operation writes advance the clock but produce no -historical records) are documented in: +The precise guarantees are documented in: - [docs/concepts/consistency-model.md](docs/concepts/consistency-model.md) — the guarantees, each proven by a dedicated test in @@ -228,6 +244,9 @@ Hard renames, no aliases. Every pair below is verified against the 8.0 source. | `brain.stats().indexHealth.hnsw` | `brain.stats().indexHealth.vector` | | Storage adapter contract `saveHNSWData()` / `getHNSWData()` | `saveVectorIndexData()` / `getVectorIndexData()` | | Cache category `'hnsw'` (`CacheProvider` union) | `'vectors'` | +| `config.history = { retainGenerations, retainMs, autoCompact }` | `config.retention` — `'all'` \| `'adaptive'` \| `{ maxGenerations?, maxAge?, maxBytes?, budgetBytes?, autoCompact? }`; **unset → adaptive** (was keep-100/7-days). The fields are now **caps**, not floors. | +| `compactHistory({ retainGenerations, retainMs })` | `compactHistory({ maxGenerations, maxAge, maxBytes })` — `retainGenerations`→`maxGenerations`, `retainMs`→`maxAge` (now upper-bound CAPS), plus new `maxBytes`. | +| `CompactHistoryOptions.retainGenerations` / `.retainMs` | `.maxGenerations` / `.maxAge` (+ `.maxBytes`) | Mechanical renames as one command (run from your repo root, review the diff): diff --git a/docs/ADR-001-generational-mvcc.md b/docs/ADR-001-generational-mvcc.md index d6e32441..b4c4d8dd 100644 --- a/docs/ADR-001-generational-mvcc.md +++ b/docs/ADR-001-generational-mvcc.md @@ -149,12 +149,22 @@ overlay entities carry no embeddings (`with()` never invokes the embedder), so a "full" index query over an overlay would silently exclude the overlay's own entities. Commit with `transact()` to get the full surface. -**History granularity.** Generation *records* are written per `transact()` -batch only. Single-operation writes advance the counter (so watermarks and -CAS stay sound) but do not stage before-images: they remain visible through -earlier pins and are not reported by `db.since()`. Code that needs pinned -isolation across its writes uses `transact()` — that is the documented 8.0 -contract, stated rather than papered over. +**History granularity (Model-B).** EVERY write is its own immutable generation +— `transact()` batches AND single-operation `add`/`update`/`remove`/`relate`. +Single-ops stage a before-image and are reported by `db.since()`/`asOf()`/ +`diff()`/`history()` exactly like transacts; a pin always freezes against later +writes. `transact()` groups several operations into ONE atomic generation. + +Single-op history durability is **async group-commit**: the live write hits +canonical storage immediately (acknowledged), while its before-image is buffered +and persisted to disk in one batched fsync on a size/timer trigger (or forced by +`flush()`/`close()`/`transact()`/`compactHistory()`). The buffer participates in +point-in-time resolution exactly like on-disk generations, so the synchronous +`now()` freezes with no forced flush. A hard crash before the flush loses only +the buffered *history* of the last window — never live data — and a crash +*mid-flush* is recovered by **drop-without-restore** (the partial generation's +before-images are discarded, never replayed, because the live write was already +acknowledged; restoring them would silently revert it). ### Pinning, retention, compaction @@ -162,11 +172,17 @@ contract, stated rather than papered over. `pin(generation)` on every registered `VersionedIndexProvider`, whose explicit pin lifetime overrides any time-based snapshot retention the provider has). -- `compactHistory({ retainGenerations?, retainMs? })` reclaims record-sets. - A record-set `N` is reclaimed only when `N` is at or below **every** live +- The constructor **`retention`** knob governs auto-compaction (on `flush()`/ + `close()`): unset → ADAPTIVE (disk/RAM-pressure byte budget, zero-config; + driven by a coordinator's `budgetBytes` or a local `os.freemem` probe) · + `'all'` → unbounded · `{ maxGenerations?, maxAge?, maxBytes? }` → explicit + CAPS. `compactHistory({ maxGenerations?, maxAge?, maxBytes? })` reclaims + manually on the same caps — the oldest unpinned record-sets are reclaimed + while ANY supplied cap is exceeded. +- A record-set `N` is reclaimed 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, both must allow the reclaim. + pin. Live pins are ALWAYS exempt, in every retention mode. - The manifest records the **horizon** (highest reclaimed generation). Generations below the horizon are unreachable; `asOf()` on them throws `GenerationCompactedError`. The horizon itself stays reachable, resolved diff --git a/docs/api/README.md b/docs/api/README.md index fa35ff29..5af84dec 100644 --- a/docs/api/README.md +++ b/docs/api/README.md @@ -901,16 +901,18 @@ that generation), once per `Db`, freed on `release()`. **Throws:** `GenerationCompactedError` when the generation's records were reclaimed by `compactHistory()`. -**History granularity:** only `transact()` batches produce historical -records; single-operation writes advance the clock but stay visible through -earlier pins. See the [consistency model](../concepts/consistency-model.md). +**History granularity:** every write is its own immutable generation — +`transact()` batches AND single-operation writes — so a pin always freezes and +every write is addressable via `asOf()`. See the +[consistency model](../concepts/consistency-model.md). --- ### `transactionLog(options?)` → `Promise` -Read the reified transaction log — one entry per committed `transact()` -batch, newest first: `{ generation, timestamp, meta? }`. +Read the reified transaction log — one entry per committed generation (every +`transact()` AND single-op write), newest first: `{ generation, timestamp, +meta? }`. Single-op generations carry no `meta` (it is a `transact()`-only field). ```typescript const [latest] = await brain.transactionLog({ limit: 1 }) @@ -921,13 +923,15 @@ latest.meta // { author: 'order-service', requestId: 'req-9f2' } ### `compactHistory(options?)` → `Promise` -Reclaim historical record-sets that no retention rule and no live `Db` pin -protects. Pinned reads stay correct across compaction, always. +Reclaim historical record-sets that no retention cap and no live `Db` pin +protects. Pinned reads stay correct across compaction, always. (Auto-compaction +on `flush()`/`close()` is governed by the constructor `retention` knob — unset → +adaptive, `'all'` → unbounded, `{ … }` → explicit caps.) ```typescript await brain.compactHistory({ - retainGenerations: 100, // keep the 100 most recent commits - retainMs: 7 * 24 * 60 * 60 * 1000 // and everything from the last 7 days + maxGenerations: 100, // keep at most the 100 most recent generations + maxAge: 7 * 24 * 60 * 60 * 1000 // and only those from the last 7 days }) ``` diff --git a/docs/concepts/consistency-model.md b/docs/concepts/consistency-model.md index db8efd91..bad996c5 100644 --- a/docs/concepts/consistency-model.md +++ b/docs/concepts/consistency-model.md @@ -212,10 +212,12 @@ Historical records cost disk space, so retention is explicit: - Every live `Db` holds a refcounted **pin**; a record-set is never reclaimed while any pin could need it — pinned reads stay correct across compaction, always. -- `brain.compactHistory({ retainGenerations?, retainMs? })` reclaims - everything no retention rule and no pin protects, and records the - **horizon** — `asOf()` below it throws `GenerationCompactedError`, - explicitly, never partial data. +- The **`retention`** knob governs auto-compaction (on every `flush()`/ + `close()`): unset → ADAPTIVE (disk/RAM-pressure, zero-config) · `'all'` → + unbounded · `{ maxGenerations?, maxAge?, maxBytes? }` → explicit caps. + `brain.compactHistory({ maxGenerations?, maxAge?, maxBytes? })` reclaims + manually on the same caps, and records the **horizon** — `asOf()` below it + throws `GenerationCompactedError`, explicitly, never partial data. - To keep a state readable forever, `persist()` it first: snapshots are self-contained and unaffected by compaction of the source store. @@ -357,9 +359,13 @@ Stated plainly, so nothing surprises you in production: ([multi-process model](./multi-process.md)). Transactions are atomic within one writer process — there is no distributed or cross-process transaction coordination. -- **History granularity.** Only `transact()` batches produce historical - records; single-operation writes between commits stay visible through - earlier pins (see above). +- **History granularity.** Every write is its own immutable generation — + `transact()` batches AND single-operation `add`/`update`/`remove`/`relate`. + A pin always freezes against later writes, and every write is addressable + via `asOf()`. (`transact()` groups several operations into ONE atomic + generation; durability of single-op history is batched via async + group-commit — a hard crash can lose only the last un-flushed window's + *history*, never live data.) - **Compacted history is gone.** `asOf()` below the compaction horizon fails explicitly; persist what you must keep. - **Counter persistence is coalesced for single-operation writes.** Durable diff --git a/docs/guides/snapshots-and-time-travel.md b/docs/guides/snapshots-and-time-travel.md index 41e2284a..d9f0e264 100644 --- a/docs/guides/snapshots-and-time-travel.md +++ b/docs/guides/snapshots-and-time-travel.md @@ -135,10 +135,12 @@ await after.release() Three things to remember: -- History granularity is `transact()` commits — single-operation writes - advance the clock but do not produce historical records (see the - [consistency model](../concepts/consistency-model.md)). Use `transact()` - for writes you want to travel back through. +- History granularity is per-write: EVERY write — `transact()` AND a + single-operation `add`/`update`/`remove`/`relate` — is its own immutable + generation, so a pin always freezes against later writes and every write is + individually addressable via `asOf()` (see the + [consistency model](../concepts/consistency-model.md)). Use `transact()` when + you want several operations to share ONE atomic generation. - The first index-accelerated query (semantic search, traversal, cursors, aggregation) at a historical generation builds an in-memory index materialization — O(n at that generation), once per `Db`, freed on @@ -336,20 +338,33 @@ For per-entity write coordination (rather than whole-store history), the ## Keeping history bounded -Historical records cost disk space. Reclaim what no live pin protects: +Under Model-B every write is a generation, so history can grow quickly — +Brainy auto-compacts on every `flush()`/`close()` under the **`retention`** +knob (configured on the constructor): ```typescript -await brain.compactHistory({ - retainGenerations: 100, // keep the 100 most recent commits - retainMs: 7 * 24 * 60 * 60 * 1000 // and everything from the last 7 days -}) +// Zero-config: ADAPTIVE — keep as much history as free disk/RAM allows, +// reclaiming oldest-first under pressure. (This is the default.) +new Brainy({ /* retention unset */ }) + +// Unbounded — never reclaim history (opt in explicitly): +new Brainy({ retention: 'all' }) + +// Explicit CAPS — reclaim oldest-unpinned generations while ANY cap is exceeded: +new Brainy({ retention: { maxGenerations: 1000, maxAge: 7 * 86_400_000, maxBytes: 512 * 1024 ** 2 } }) +``` + +Reclaim manually at any time (the same caps): + +```typescript +await brain.compactHistory({ maxGenerations: 100, maxAge: 7 * 24 * 60 * 60 * 1000 }) ``` Compaction never breaks a pinned read — record-sets are reclaimed only when -no live `Db` could need them. Release views you are done with (including the -ones `transact()` returns), and `persist()` any generation you want to keep -beyond the retention window: snapshots are self-contained and unaffected by -compaction. +no live `Db` could need them (live pins are ALWAYS exempt). Release views you +are done with (including the ones `transact()` returns), and `persist()` any +generation you want to keep beyond the retention window: snapshots are +self-contained and unaffected by compaction. ## From branches to values diff --git a/src/brainy.ts b/src/brainy.ts index d368a01a..6342b7e0 100644 --- a/src/brainy.ts +++ b/src/brainy.ts @@ -121,6 +121,7 @@ import { AggregateMaterializer } from './aggregation/materializer.js' import type { AggregateDefinition, AggregateQueryParams, AggregateResult } from './types/brainy.types.js' import { resolveJsHnswConfig, DEFAULT_RECALL } from './utils/recallPreset.js' import * as fs from 'node:fs' +import * as os from 'node:os' import { Db, type DbHost, type HistoricalQueryHandle } from './db/db.js' import { importGraph, @@ -148,7 +149,7 @@ import type { } from './db/types.js' import { stableDeepEqual } from './db/stableEqual.js' import type { VersionedIndexProvider } from './plugin.js' -import type { Operation } from './transaction/types.js' +import type { Operation, TransactionFunction } from './transaction/types.js' /** * Stopwords for semantic highlighting @@ -208,7 +209,7 @@ type ResolvedBrainyConfig = Required< | 'vector' | 'plugins' | 'integrations' - | 'history' + | 'retention' | 'eagerEmbeddings' > > & @@ -219,7 +220,7 @@ type ResolvedBrainyConfig = Required< | 'vector' | 'plugins' | 'integrations' - | 'history' + | 'retention' | 'eagerEmbeddings' > @@ -365,6 +366,23 @@ export class Brainy implements BrainyInterface { private _tripleIntelligence?: TripleIntelligenceSystem private _vfs?: VirtualFileSystem private _vfsInitialized = false // Track VFS init completion separately + /** + * 8.0 MVCC: single-op writes create generations (Model-B) only AFTER init + * completes. Init-time infrastructure writes (the VFS root directory, etc.) + * form the generation-0 baseline of a freshly-materialized brain and are NOT + * versioned — so a brand-new brain reports `generation() === 0` / empty + * `transactionLog()`, and the first USER write is generation 1. Enabled at + * the tail of `init()`; see {@link persistSingleOp}. + */ + private _generationStampingActive = false + /** + * 8.0 MVCC: the coordinator-driven adaptive history byte budget for this + * brain (set via {@link setRetentionBudget}; e.g. cor's `ResourceManager` + * fair-shares one box-wide budget across co-located instances). Overrides the + * local `os.freemem` probe while `retention` is adaptive. `undefined` = use + * the probe. See {@link adaptiveHistoryBudgetBytes}. + */ + private _retentionBudgetBytes?: number private _hub?: IntegrationHub // Integration Hub for external tools private _pendingMigrationRunner?: MigrationRunner // Deferred migration runner for large datasets private _aggregationIndex?: AggregationIndex // Incremental aggregation engine @@ -899,6 +917,13 @@ export class Brainy implements BrainyInterface { await this._vfs.init() this._vfsInitialized = true // Mark VFS as fully initialized + // 8.0 MVCC: infrastructure bootstrap (VFS root, etc.) is now the + // generation-0 baseline. From here, every single-op write is a Model-B + // generation. (Writers only — readers never stamp; a no-op for them.) + if (!this.isReadOnly) { + this._generationStampingActive = true + } + // Eager embedding initialization. // // Adaptive default (8.0): the WASM embedding engine eagerly initializes @@ -1233,6 +1258,43 @@ export class Brainy implements BrainyInterface { return uuidv7() } + /** + * @description Persist one single-operation write (`add`/`update`/`remove`/ + * `relate`/`updateRelation`/`unrelate` and the per-item calls of their `*Many` + * variants) as its OWN immutable generation — the Model-B "every write + * versioned" contract. Wraps the write's existing `executeTransaction` batch in + * {@link GenerationStore.commitSingleOp}, which captures byte-identical + * before-images of `touched`, runs the batch with the generation watermark + * pinned to the reserved generation (so this write's graph-index ops stamp cor + * with that one distinct per-op generation, via the unchanged + * {@link graphWriteGeneration}), and buffers the history for async + * group-commit. A crash before the flush loses only that buffered history, not + * the acknowledged live write (drop-without-restore recovery). + * + * @param touched - The entity/relationship ids the write creates, updates, or + * deletes — the before-image + per-id-chain set. + * @param run - The single-op's existing operation batch builder (the + * `tx => {…}` body previously passed straight to `executeTransaction`). + */ + private async persistSingleOp( + touched: { nouns?: string[]; verbs?: string[] }, + run: TransactionFunction + ): Promise { + if (!this._generationStampingActive) { + // Init-time / infrastructure baseline write (e.g. the VFS root): apply + // WITHOUT creating a generation. Generation 0 is the freshly-materialized + // brain (bootstrap included); the first USER write is generation 1. + await this.generationStore.runWithoutGeneration(() => + this.transactionManager.executeTransaction(run) + ) + return + } + await this.generationStore.commitSingleOp({ + touched, + execute: () => this.transactionManager.executeTransaction(run) + }) + } + async add(params: AddParams): Promise { this.assertWritable('add') await this.ensureInitialized() @@ -1375,9 +1437,10 @@ export class Brainy implements BrainyInterface { } } - // Execute atomically with transaction system + // Execute atomically with transaction system, generation-stamped as one + // immutable Model-B generation (before-image = the absent/create sentinel). // All operations succeed or all rollback - prevents partial failures - await this.transactionManager.executeTransaction(async (tx) => { + await this.persistSingleOp({ nouns: [id] }, async (tx) => { // Operation 1: Save metadata FIRST (TypeAwareStorage caching) // isNew=true: skip pre-read for rollback (entity doesn't exist yet) tx.addOperation( @@ -2248,8 +2311,9 @@ export class Brainy implements BrainyInterface { metadata: newMetadata } - // Execute atomically with transaction system - await this.transactionManager.executeTransaction(async (tx) => { + // Execute atomically with transaction system, generation-stamped as one + // immutable Model-B generation (before-image = the entity's prior state). + await this.persistSingleOp({ nouns: [params.id] }, async (tx) => { // Operation 1: Update metadata FIRST (updates type cache) tx.addOperation( new UpdateNounMetadataOperation(this.storage, params.id, updatedMetadata) @@ -2358,8 +2422,12 @@ export class Brainy implements BrainyInterface { const targetVerbs = await this.storage.getVerbsByTarget(id) const allVerbs = [...verbs, ...targetVerbs] - // Execute atomically with transaction system - await this.transactionManager.executeTransaction(async (tx) => { + // Execute atomically with transaction system, generation-stamped as one + // immutable Model-B generation covering the entity AND its cascade-deleted + // relationships (each id's before-image = its prior state). + await this.persistSingleOp( + { nouns: [id], verbs: allVerbs.map((v) => v.id) }, + async (tx) => { // Operation 1: Remove from vector index if (noun) { tx.addOperation( @@ -2769,8 +2837,13 @@ export class Brainy implements BrainyInterface { // entities were existence-checked above) and mirror them onto the verb. const { sourceInt, targetInt } = this.resolveVerbEndpointInts(verb) - // Execute atomically with transaction system - await this.transactionManager.executeTransaction(async (tx) => { + // Reverse-edge id reserved up front (when bidirectional) so the Model-B + // generation's touched-verb set covers both edges this write creates. + const reverseId = params.bidirectional ? uuidv4() : undefined + + // Execute atomically with transaction system, generation-stamped as one + // immutable Model-B generation (before-image of each new edge = absent). + await this.persistSingleOp({ verbs: reverseId ? [id, reverseId] : [id] }, async (tx) => { // Operation 1: Save verb vector data tx.addOperation( new SaveVerbOperation(this.storage, { @@ -2798,8 +2871,7 @@ export class Brainy implements BrainyInterface { ) // Create bidirectional if requested - if (params.bidirectional) { - const reverseId = uuidv4() + if (params.bidirectional && reverseId) { const reverseVerb: GraphVerb = { ...verb, id: reverseId, @@ -2868,8 +2940,9 @@ export class Brainy implements BrainyInterface { // a rollback can re-add through the BigInt addVerb contract. const endpointInts = verb ? this.resolveVerbEndpointInts(verb) : undefined - // Execute atomically with transaction system - await this.transactionManager.executeTransaction(async (tx) => { + // Execute atomically with transaction system, generation-stamped as one + // immutable Model-B generation (before-image = the relationship's state). + await this.persistSingleOp({ verbs: [id] }, async (tx) => { // Operation 1: Remove from graph index if (verb && endpointInts) { tx.addOperation( @@ -2997,7 +3070,7 @@ export class Brainy implements BrainyInterface { // resolution serves both the remove (rollback re-add) and the re-add. const reindexInts = typeChanged ? this.resolveVerbEndpointInts(verbForIndex) : undefined - await this.transactionManager.executeTransaction(async (tx) => { + await this.persistSingleOp({ verbs: [params.id] }, async (tx) => { tx.addOperation( new UpdateVerbMetadataOperation(this.storage, params.id, updatedMetadata) ) @@ -5104,9 +5177,27 @@ export class Brainy implements BrainyInterface { const chunkQueued: string[] = [] const chunkBuilderFailed: Array<{ item: string; error: string }> = [] + // Model-B pre-pass: collect the chunk's touched-id set (entities + their + // cascade-deleted relationships) so the generation captures before-images + // for all of them. Tolerant by design — the builder below still does the + // authoritative per-id load + error handling, and an over-included id whose + // delete is skipped simply gets a before-image equal to its live state (a + // harmless no-op history entry). + const touchedNouns: string[] = [...chunk] + const touchedVerbs: string[] = [] + for (const id of chunk) { + try { + const srcVerbs = await this.storage.getVerbsBySource(id) + const tgtVerbs = await this.storage.getVerbsByTarget(id) + for (const v of [...srcVerbs, ...tgtVerbs]) touchedVerbs.push(v.id) + } catch { + // Ignore — the builder's per-id try/catch is authoritative. + } + } + try { - // Process chunk in single transaction for atomic deletion - await this.transactionManager.executeTransaction(async (tx) => { + // Process chunk as ONE atomic Model-B generation (entities + cascade verbs). + await this.persistSingleOp({ nouns: touchedNouns, verbs: touchedVerbs }, async (tx) => { for (const id of chunk) { try { // Load entity data @@ -5599,13 +5690,13 @@ export class Brainy implements BrainyInterface { /** * @description Read the reified transaction log — one entry per committed - * `transact()` batch, carrying the committed generation, the commit - * timestamp, and the `meta` the transaction was submitted with. Entries - * are returned newest first. Single-operation writes - * (`add`/`update`/`remove`/`relate`/…) advance the generation counter but - * do not append log entries (the documented 8.0 history granularity), and - * `compactHistory()` never rewrites the log — entries may reference - * generations whose record-sets were already reclaimed. + * generation, carrying the committed generation, the commit timestamp, and + * the `meta` the transaction was submitted with. Entries are returned newest + * first. Under Model-B EVERY write is its own generation, so single-operation + * writes (`add`/`update`/`remove`/`relate`/…) appear here too (without `meta` + * — transaction metadata is a `transact()`-only concept). `compactHistory()` + * never rewrites the log — entries may reference generations whose + * record-sets were already reclaimed. * * @param options - `from`/`to` (generation number or `Date`) bound the window * to commits in `[from, to]` INCLUSIVE on both ends — a log window names @@ -5754,6 +5845,11 @@ export class Brainy implements BrainyInterface { throw new Error('transact(): provide a non-empty array of transaction operations') } + // Commit any buffered single-op generations FIRST so this transaction's + // generation lands strictly after them and the committed-generation + // sequence stays gap-free (single-ops reserve generations eagerly). + await this.generationStore.flushPendingSingleOps() + // Early CAS check — avoids expensive planning (embedding) on a stale // expectation. The generation store re-checks authoritatively under the // commit mutex; this one is purely a fast-fail. @@ -5796,9 +5892,9 @@ export class Brainy implements BrainyInterface { * @description Open an immutable `Db` view of PAST state: * * - **`number`** — a generation: pins it on this store; reads resolve - * through the generational record layer. History granularity is - * `transact()` commits — single-operation writes between commits do not - * produce records and remain visible through earlier pins. + * through the generational record layer. Under Model-B every write is its + * own generation, so single-operation writes are individually addressable + * too (a pin always freezes against later single-op writes). * - **`Date`** — a wall-clock instant: resolved via the transaction log to * the newest generation committed at or before it, then pinned as above. * - **`string`** — a snapshot directory previously produced by @@ -6128,57 +6224,114 @@ export class Brainy implements BrainyInterface { * keep (the GC backstop eventually releases leaked ones, but until then * compaction retains the history they protect). * - * @param options - `retainGenerations` (keep the N most recent committed - * generations) and/or `retainMs` (keep everything committed within the - * window). Both supplied = both must allow reclaim. Neither = reclaim - * everything unpinned. + * @param options - Explicit retention CAPS (`maxGenerations` / `maxAge` / + * `maxBytes`). Reclaim the oldest unpinned generations while ANY supplied + * cap is exceeded. Neither = reclaim everything unpinned. * @returns Count of reclaimed record-sets and the new horizon. * @example - * await brain.compactHistory({ retainGenerations: 100, retainMs: 7 * 86_400_000 }) + * await brain.compactHistory({ maxGenerations: 100, maxAge: 7 * 86_400_000 }) */ async compactHistory(options?: CompactHistoryOptions): Promise { this.assertWritable('compactHistory') await this.ensureInitialized() + // Persist buffered single-op generations first so compaction reasons over a + // complete on-disk history (the pending tier is never itself reclaimable). + await this.generationStore.flushPendingSingleOps() return this.generationStore.compact(options) } /** - * @description Resolve the effective generational-history retention policy - * from `config.history`, filling per-field defaults so a partial object - * inherits the rest. This is the single read site for the policy referenced - * by {@link autoCompactHistory}. - * - * Defaults (zero-config): keep the 100 most recent committed generations - * AND everything committed within the last 7 days; auto-compact on - * `flush()` / `close()`. A generation is reclaimed only when it is older - * than BOTH bounds and no live `Db` pin protects it. - * - * @returns The resolved policy: retention bounds plus the `autoCompact` flag. + * @description Drive the adaptive retention byte budget at runtime — the + * settable input a machine-level coordinator (e.g. cor's `ResourceManager`, + * which fair-shares one box-wide history budget across co-located instances) + * pushes whenever the box's free-resource picture changes. Takes effect at the + * next auto-compaction (flush/close). Only meaningful while `retention` is + * adaptive (unset or `'adaptive'`); ignored under `'all'` or explicit caps. + * @param bytes - The new per-brain history byte budget (non-negative). + * @example brain.setRetentionBudget(2 * 1024 ** 3) // 2 GiB for this brain */ - private resolveHistoryPolicy(): { - retainGenerations: number - retainMs: number + setRetentionBudget(bytes: number): void { + if (!Number.isFinite(bytes) || bytes < 0) { + throw new RangeError(`setRetentionBudget(): bytes must be a non-negative finite number (got ${bytes})`) + } + this._retentionBudgetBytes = bytes + } + + /** + * @description Resolve the effective `retention` policy from `config.retention`. + * The single read site for the policy {@link autoCompactHistory} applies. + * + * - `'all'` → `{ mode: 'all' }` (never reclaim history). + * - unset / `'adaptive'` → `{ mode: 'adaptive' }` (budget-driven; the budget + * is `setRetentionBudget()`/`budgetBytes` when set, else a local probe). + * - `{ … }` → `{ mode: 'explicit', maxGenerations?, maxAge?, maxBytes? }`. + * + * `autoCompact` defaults to `true` in every mode. + * @returns The normalized retention policy. + */ + private resolveRetentionPolicy(): { + mode: 'all' | 'adaptive' | 'explicit' + maxGenerations?: number + maxAge?: number + maxBytes?: number + budgetBytes?: number autoCompact: boolean } { - const history = this.config.history + const retention = this.config.retention + if (retention === 'all') { + return { mode: 'all', autoCompact: true } + } + if (retention === undefined || retention === 'adaptive') { + return { mode: 'adaptive', autoCompact: true } + } return { - retainGenerations: history?.retainGenerations ?? 100, - retainMs: history?.retainMs ?? 604_800_000, // 7 days - autoCompact: history?.autoCompact ?? true + mode: 'explicit', + maxGenerations: retention.maxGenerations, + maxAge: retention.maxAge, + maxBytes: retention.maxBytes, + budgetBytes: retention.budgetBytes, + autoCompact: retention.autoCompact ?? true } } /** - * @description Run history compaction with the resolved retention policy if - * `config.history.autoCompact` is on (the default). Invoked from `flush()` - * and `close()` so generational record-sets cannot accumulate unbounded - * across a long-lived writer's lifetime. + * @description Compute the adaptive history byte budget for this brain: the + * coordinator-driven value when present (`setRetentionBudget()` / + * `config.retention.budgetBytes` — e.g. cor's fair-shared box budget), else a + * conservative LOCAL probe of free resources. Standalone brainy must not be + * cor-dependent, so the fallback uses `os.freemem()` (and is intentionally + * generous — a single instance owning the box can keep a lot of history). + * @returns The byte budget; `Infinity` only if no signal is available. + */ + private adaptiveHistoryBudgetBytes(driven?: number): number { + if (driven !== undefined) return driven + if (this._retentionBudgetBytes !== undefined) return this._retentionBudgetBytes + // Local single-instance probe: allot a quarter of currently-free RAM to + // history. Bounded, zero-config, and never reclaims pinned generations. + try { + const free = os.freemem() + if (Number.isFinite(free) && free > 0) return Math.floor(free / 4) + } catch { + // os unavailable (exotic runtime) — fall through to unbounded. + } + return Infinity + } + + /** + * @description Run history compaction under the resolved `retention` policy + * when `autoCompact` is on (the default). Invoked from `flush()` and + * `close()` so generational record-sets cannot accumulate unbounded across a + * long-lived writer's lifetime. + * + * - `'all'` → returns without reclaiming (index compaction for speed still + * runs elsewhere; history is decoupled and kept). + * - `'adaptive'` → reclaim oldest-unpinned history down to the byte budget. + * - `'explicit'` → apply the supplied `maxGenerations`/`maxAge`/`maxBytes` caps. * * Read-only instances and an explicit `autoCompact: false` skip silently. - * Pinned generations (live `Db` values) are never reclaimed — the safety - * invariant lives in {@link GenerationStore.compact}. Failures are logged - * and swallowed: compaction is housekeeping and must never fail a flush or - * a clean shutdown. + * Pinned generations are never reclaimed ({@link GenerationStore.compact}). + * Failures are logged and swallowed — compaction is housekeeping and must + * never fail a flush or a clean shutdown. */ private async autoCompactHistory(): Promise { // Nothing to compact on a read-only instance or before init wired up the @@ -6186,15 +6339,22 @@ export class Brainy implements BrainyInterface { if (this.isReadOnly || !this.generationStore) { return } - const policy = this.resolveHistoryPolicy() - if (!policy.autoCompact) { + const policy = this.resolveRetentionPolicy() + if (!policy.autoCompact || policy.mode === 'all') { return } try { - await this.generationStore.compact({ - retainGenerations: policy.retainGenerations, - retainMs: policy.retainMs - }) + if (policy.mode === 'adaptive') { + const budget = this.adaptiveHistoryBudgetBytes(policy.budgetBytes) + if (budget === Infinity) return // no pressure signal → keep everything this pass + await this.generationStore.compact({ maxBytes: budget }) + } else { + await this.generationStore.compact({ + maxGenerations: policy.maxGenerations, + maxAge: policy.maxAge, + maxBytes: policy.maxBytes + }) + } } catch (error) { console.warn( `Auto-compaction of generational history failed (non-fatal): ${ @@ -7964,6 +8124,11 @@ export class Brainy implements BrainyInterface { console.log('Flushing Brainy indexes and caches to disk...') const startTime = Date.now() + // 8.0 MVCC: persist the in-memory pending single-op generation history to + // disk first (async group-commit), so a flush makes every single-op write's + // history durable, not just the live data. + await this.generationStore.flushPendingSingleOps() + // Flush all components in parallel for performance await Promise.all([ // 1. Flush storage adapter counts (entity/verb counts by type) @@ -7991,7 +8156,7 @@ export class Brainy implements BrainyInterface { this.generationStore.persistCounterNow() ]) - // 6. Auto-compact generational history per config.history (default on). + // 6. Auto-compact generational history per config.retention (default on). // Runs after the flush so durable state is in place; respects live // Db pins and an explicit autoCompact: false. await this.autoCompactHistory() @@ -11730,9 +11895,10 @@ export class Brainy implements BrainyInterface { // `recall` preset + `persistMode` (folded in from 7.x's hnswPersistMode). // See BRAINY-8.0-RENAME-COORDINATION § A.2 + § G.2. vector: config?.vector ?? undefined, - // Generational-history retention — defaults applied at the read site - // (resolveHistoryPolicy) so a partial object inherits per-field defaults. - history: config?.history ?? undefined, + // Generational-history retention — the `retention` knob; defaults applied + // at the read site (resolveRetentionPolicy) so a partial object inherits + // per-field defaults and `undefined` resolves to ADAPTIVE. + retention: config?.retention ?? undefined, // Embedding initialization — left undefined when omitted so the adaptive // default resolves at the init() read site (eager when the WASM embedder // is the active one on a non-reader writer outside unit tests). Explicit @@ -12270,7 +12436,14 @@ export class Brainy implements BrainyInterface { * This ensures deferred persistence mode data is saved */ async close(): Promise { - // Phase 0: Auto-compact generational history per config.history (default + // Phase 0a: Persist buffered single-op generation history (async + // group-commit) before anything else, so a clean close never drops history + // the caller already observed. No-op when nothing is pending or read-only. + if (this.generationStore && !this.isReadOnly) { + await this.generationStore.flushPendingSingleOps() + } + + // Phase 0b: Auto-compact generational history per config.retention (default // on) BEFORE the generation store closes below. Respects live Db pins and // an explicit autoCompact: false; no-op on read-only instances. await this.autoCompactHistory() diff --git a/src/db/db.ts b/src/db/db.ts index 560c3111..7a4ce858 100644 --- a/src/db/db.ts +++ b/src/db/db.ts @@ -346,9 +346,12 @@ export class Db { const limit = params.limit ?? 10 const offset = params.offset ?? 0 - // Ids whose live-index answer is NOT valid at this generation. + // Ids whose live-index answer is NOT valid at this generation. The upper + // bound is the full reserved watermark generation() — NOT committedGeneration() + // — so un-flushed single-op (Model-B) writes that changed an id after this + // pin are overlaid too (they participate in resolution via the pending tier). const changedNouns = historical - ? (await this.host.store.changedBetween(this.gen, this.host.store.committedGeneration())).nouns + ? (await this.host.store.changedBetween(this.gen, this.host.store.generation())).nouns : [] const overlayNouns = this.overlay?.nouns ?? new Map | null>() const excluded = new Set([...changedNouns, ...overlayNouns.keys()]) @@ -480,8 +483,11 @@ export class Db { const limit = params.limit ?? 100 const offset = params.offset ?? 0 + // Upper bound is the full reserved watermark generation() (see find()): an + // un-flushed single-op write that touched a relationship after this pin is + // overlaid too, via the pending tier. const changedVerbs = historical - ? (await this.host.store.changedBetween(this.gen, this.host.store.committedGeneration())).verbs + ? (await this.host.store.changedBetween(this.gen, this.host.store.generation())).verbs : [] const overlayVerbs = this.overlay?.verbs ?? new Map | null>() const excluded = new Set([...changedVerbs, ...overlayVerbs.keys()]) diff --git a/src/db/generationStore.ts b/src/db/generationStore.ts index f23b6a2c..6f98ddae 100644 --- a/src/db/generationStore.ts +++ b/src/db/generationStore.ts @@ -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; verbs: Set; timestamp: number } + { nouns: Set; verbs: Set; 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; verbs: Map; timestamp: number } + >() + /** Pending timer-coalesced flush handle (cleared on flush/close). */ + private pendingFlushTimer: ReturnType | 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() @@ -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 { + // 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 + }): 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() + 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() + 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): Promise { + 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//` record + * set (`prev/.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 { + 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() // 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//prev/.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 { + 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 { const nouns = new Set() const verbs = new Set() - 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 { 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 { - // 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; verbs: Set; timestamp: number }> { + ): Promise<{ nouns: Set; verbs: Set; 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; verbs: Set; timestamp: number } + entry: { nouns: Set; verbs: Set; 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 { + 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 { 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 { 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 { + 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 { 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//...`. Returns `null` for paths that don't match. diff --git a/src/db/types.ts b/src/db/types.ts index 31420b84..80a1959d 100644 --- a/src/db/types.ts +++ b/src/db/types.ts @@ -138,21 +138,34 @@ export interface TransactReceipt { // ============================================================================ /** - * @description Options for `brain.compactHistory()`. When both options are - * supplied a generation must satisfy *both* retention rules to be reclaimed - * (the stricter retention wins). With no options, every generation not + * @description Options for `brain.compactHistory()` — explicit retention + * **caps** (Model-B `retention` knob). Each supplied field is an upper bound: + * the oldest unpinned generations are reclaimed while **ANY** supplied cap is + * exceeded (predictable ops — the more you constrain, the more is reclaimed). + * Live `Db` pins are ALWAYS exempt. With no options, every generation not * protected by a live pin is eligible. + * + * (8.0 rename: the pre-release `retainGenerations`/`retainMs` *floors* became + * `maxGenerations`/`maxAge` *caps* — the `max*` naming signals the semantics.) */ export interface CompactHistoryOptions { /** - * Keep at least this many of the most recent committed generations - * (`0` = keep none beyond what live pins protect). + * Keep at most this many of the most recent committed generations — reclaim + * the oldest unpinned generations while the count exceeds it (`0` = reclaim + * everything not protected by a live pin). */ - retainGenerations?: number + maxGenerations?: number /** - * Keep every generation committed within the last `retainMs` milliseconds. + * Keep only generations committed within the last `maxAge` milliseconds — + * reclaim unpinned generations older than the window. */ - retainMs?: number + maxAge?: number + /** + * Keep total generational-history bytes at or below this — reclaim the oldest + * unpinned generations while the total exceeds it. Byte accounting is the sum + * of each surviving generation's serialized record set (`GenerationDelta.bytes`). + */ + maxBytes?: number } /** @@ -176,11 +189,10 @@ export interface CompactHistoryResult { /** * @description Result of `db.since(priorDb)` — the ids touched by committed - * transactions in the generation interval `(priorDb.generation, - * db.generation]`. Granularity note: single-operation writes performed - * outside `transact()` bump the generation counter but do not produce - * generation records, so they are not reflected here (documented 8.0 - * history granularity — see `docs/ADR-001-generational-mvcc.md`). + * generations in the interval `(priorDb.generation, db.generation]`. Under + * Model-B EVERY write is its own generation, so single-operation writes + * (`add`/`update`/`remove`/`relate` outside `transact()`) ARE reflected here, + * exactly like `transact()` commits. */ export interface ChangedIds { /** Entity (noun) ids touched in the interval, sorted. */ @@ -219,7 +231,8 @@ export interface DiffResult { * @description One version of a single entity or relationship in its * {@link EntityHistory} — the materialized state at a committed generation that * changed it. `value` is `null` when the id did not exist at that version - * (created-after / removed). Granularity is `transact()` commits. + * (created-after / removed). Granularity is per-write (every `transact()` AND + * single-op write is its own generation — Model-B). */ export interface HistoryVersion { /** The committed generation that produced this version. */ @@ -271,9 +284,11 @@ export interface GenerationRecord { /** * @description The per-generation delta persisted at - * `_generations//tx.json` — written and fsynced *before* any operation of - * the transaction executes, so crash recovery always knows exactly which ids - * to restore from before-images. + * `_generations//tx.json`. For a `transact()` generation it is written and + * fsynced *before* any operation of the transaction executes, so crash + * recovery always knows exactly which ids to restore from before-images. For a + * single-operation (Model-B) generation it is written at group-commit flush + * time with `groupCommit: true` — see that flag. */ export interface GenerationDelta { /** The generation this delta belongs to. */ @@ -286,6 +301,26 @@ export interface GenerationDelta { nouns: string[] /** Relationship ids touched by this generation. */ verbs: string[] + /** + * `true` for a Model-B single-operation generation persisted by the async + * group-commit flush (`GenerationStore.flushPendingSingleOps`). It marks the + * **drop-without-restore** crash-recovery contract: the live write already + * mutated canonical storage and was acknowledged to the caller BEFORE the + * flush, so an uncommitted (crashed-mid-flush) generation with this flag set + * must be DISCARDED — its before-images must NEVER be restored, or recovery + * would silently revert acknowledged user writes. Absent/`false` on a + * `transact()` generation, whose before-images ARE the durable undo log and + * are restored on rollback (the before-image + execute are one atomic unit). + */ + groupCommit?: boolean + /** + * Serialized byte size of this generation's record set (the `prev/.json` + * before-images plus this delta), recorded at write time so `maxBytes` / + * adaptive retention can bound total history without a storage size API. Read + * back lazily by `GenerationStore.historyBytes()`. Absent on generations + * written before byte accounting existed (treated as 0). + */ + bytes?: number } /** diff --git a/src/types/brainy.types.ts b/src/types/brainy.types.ts index d8eaa94a..2a6f58ac 100644 --- a/src/types/brainy.types.ts +++ b/src/types/brainy.types.ts @@ -1538,33 +1538,49 @@ export interface BrainyConfig { } /** - * Generational-history retention policy (8.0 MVCC). + * Generational-history retention policy (8.0 MVCC — the `retention` knob). * - * Every `transact()` produces an immutable generation record-set that - * serves historical reads (`asOf()`, pinned `Db` values). Without - * compaction those record-sets accumulate forever, so Brainy - * **auto-compacts on every `flush()` and `close()`** with this policy. - * A generation is reclaimed only when it is older than BOTH bounds - * (and never while a live `Db` pin protects it): + * Under Model-B EVERY write (`transact()` AND single-op `add`/`update`/ + * `remove`/`relate`) produces an immutable generation record-set serving + * historical reads (`asOf()`, pinned `Db` values). Without compaction those + * accumulate, so Brainy **auto-compacts on every `flush()` and `close()`**. + * Live `Db` pins are ALWAYS exempt from reclamation, in every mode. * - * - `retainGenerations` — keep at least the N most recent committed - * generations (default: `100`). - * - `retainMs` — keep everything committed within the window - * (default: 7 days). + * Modes: + * - **unset → ADAPTIVE (default):** the retention horizon tracks + * disk/RAM pressure. A machine-level byte budget governs how much history + * is kept — driven by `budgetBytes` when a coordinator (e.g. cor's + * `ResourceManager`, fair-shared across co-located instances) sets it, else + * a local `os.freemem`/filesystem-free probe. Oldest-unpinned history is + * reclaimed when the budget is exceeded. Zero-config. + * - **`'all'` → unbounded:** never reclaim history (index compaction for + * query speed still runs — the decouple). The legacy keep-everything + * behavior, now explicit opt-in. + * - **`{ maxGenerations?, maxAge?, maxBytes? }` → explicit CAPS:** reclaim + * the oldest unpinned generations while ANY supplied cap is exceeded + * (predictable ops). `maxAge` in ms; `maxBytes` total history bytes. * - * Escape hatches: raise either bound for longer time-travel windows, - * or set `autoCompact: false` to manage history manually via - * `brain.compactHistory()`. Long-term archives belong in - * `db.persist(path)` snapshots, which compaction never touches. + * `autoCompact: false` disables the automatic flush/close compaction (manage + * manually via `brain.compactHistory()`). `budgetBytes` is the settable + * adaptive byte budget a coordinator drives (also via `brain.setRetentionBudget()`). + * Long-term archives belong in `db.persist(path)` snapshots, which compaction + * never touches. */ - history?: { - /** Keep at least this many recent committed generations (default: 100). */ - retainGenerations?: number - /** Keep generations committed within this window in ms (default: 604800000 = 7 days). */ - retainMs?: number - /** Run compaction automatically on flush()/close() (default: true). */ - autoCompact?: boolean - } + retention?: + | 'all' + | 'adaptive' + | { + /** Keep at most this many recent committed generations (cap). */ + maxGenerations?: number + /** Keep generations committed within this window in ms (cap). */ + maxAge?: number + /** Keep total generational-history bytes at or below this (cap). */ + maxBytes?: number + /** Adaptive byte budget for this brain, driven by a coordinator (e.g. cor). */ + budgetBytes?: number + /** Run compaction automatically on flush()/close() (default: true). */ + autoCompact?: boolean + } // Memory management options maxQueryLimit?: number // Override auto-detected query result limit (max: 100000) diff --git a/tests/benchmarks/model-b-scalability.spike.ts b/tests/benchmarks/model-b-scalability.spike.ts index 93fed800..5433d9f6 100644 --- a/tests/benchmarks/model-b-scalability.spike.ts +++ b/tests/benchmarks/model-b-scalability.spike.ts @@ -58,7 +58,7 @@ async function mkBrain(backend: 'memory' | 'filesystem', dir?: string): Promise< storage, eagerEmbeddings: false, // never load WASM — we pass precomputed vectors // Keep history fully retained for the measurement (no mid-run compaction skewing depth). - history: { autoCompact: false } as any, + retention: { autoCompact: false } as any, index: { m: 16, efConstruction: 200, efSearch: 50 }, cache: { maxSize: 1000, ttl: 3600 } } as BrainyConfig) @@ -176,13 +176,13 @@ async function main() { await b.flush() const beforeBytes = allocBytes(`${dir}/_generations`) const t = process.hrtime.bigint() - const res = await b.compactHistory({ retainGenerations: 100 } as any) + const res = await b.compactHistory({ maxGenerations: 100 } as any) const compMs = Number(process.hrtime.bigint() - t) / 1e6 await b.flush() const afterBytes = allocBytes(`${dir}/_generations`) await b.close() rmSync(dir, { recursive: true, force: true }) - say(`N=${N.toLocaleString()} gens → compactHistory({retainGenerations:100}): ${compMs.toFixed(0)} ms, removed ${(res as any)?.removedGenerations ?? '?'} gens`) + say(`N=${N.toLocaleString()} gens → compactHistory({maxGenerations:100}): ${compMs.toFixed(0)} ms, removed ${(res as any)?.removedGenerations ?? '?'} gens`) say(` _generations footprint: ${(beforeBytes / 1024 / 1024).toFixed(1)} MiB → ${(afterBytes / 1024 / 1024).toFixed(1)} MiB allocated`) results.compaction.push({ N, compactMs: compMs, removed: (res as any)?.removedGenerations, beforeBytes, afterBytes }) } diff --git a/tests/integration/db-mvcc.test.ts b/tests/integration/db-mvcc.test.ts index 696e4537..a5fab68d 100644 --- a/tests/integration/db-mvcc.test.ts +++ b/tests/integration/db-mvcc.test.ts @@ -1118,12 +1118,17 @@ describe('8.0 Db API — generational MVCC', () => { await db.release() }) - it('transactionLog() returns committed entries newest first, with meta and limit', async () => { + it('transactionLog() includes single-op AND transact generations, newest first, with meta and limit', async () => { const brain = await openMemoryBrain() - // Nothing committed yet — the log is empty (single-op writes do not log). + // Model-B: a single-op write is its OWN generation and IS logged (no meta — + // tx metadata is a transact()-only concept). It is generation 1 on a fresh + // brain (init-time infrastructure writes are the un-versioned gen-0 baseline). await brain.add({ id: uid('txlog-solo'), type: NounType.Document, data: 'solo', vector: vec(99), subtype: 'note' }) - expect(await brain.transactionLog()).toEqual([]) + const soloLog = await brain.transactionLog() + expect(soloLog.map((entry) => entry.generation)).toEqual([1]) + expect(soloLog[0].meta).toBeUndefined() + const soloGen = 1 const first = await brain.transact( [{ op: 'add', id: uid('txlog-a'), type: NounType.Document, data: 'a', vector: vec(100), metadata: {} }], @@ -1136,14 +1141,17 @@ describe('8.0 Db API — generational MVCC', () => { const third = await brain.transact([{ op: 'update', id: uid('txlog-a'), metadata: { v: 3 } }]) const entries = await brain.transactionLog() + // Newest first: the three transacts, then the single-op solo write (gen 1). expect(entries.map((entry) => entry.generation)).toEqual([ third.generation, second.generation, - first.generation + first.generation, + soloGen ]) expect(entries[1].meta).toEqual({ author: 'job-2' }) expect(entries[2].meta).toEqual({ author: 'job-1' }) - expect(entries[0].meta).toBeUndefined() + expect(entries[0].meta).toBeUndefined() // third() had no meta + expect(entries[3].meta).toBeUndefined() // the single-op solo write carries no meta for (const entry of entries) { expect(entry.timestamp).toBeGreaterThan(0) } @@ -1156,6 +1164,82 @@ describe('8.0 Db API — generational MVCC', () => { await third.release() }) + // ========================================================================== + // 8b. Model-B single-op generation-stamping (every write versioned) + // ========================================================================== + it('Model-B — a now() pin freezes against single-op add/update/remove (the Model-A hole is closed)', async () => { + const brain = await openMemoryBrain() + const a = uid('mb-a') + const b = uid('mb-b') + // Seed `a` with a single-op add, then pin the current generation. + await brain.add({ id: a, type: NounType.Document, data: 'a', vector: vec(1), metadata: { v: 1 } }) + const pin = brain.now() + + // Single-op mutations AFTER the pin: update `a`, add `b`. + await brain.update({ id: a, metadata: { v: 2 } }) + await brain.add({ id: b, type: NounType.Document, data: 'b', vector: vec(2), metadata: { v: 1 } }) + + // The pin is frozen at its generation — single-op writes do not leak through + // it (pre-Model-B, the pin tracked the live single-op mutation). + expect((await pin.get(a))?.metadata?.v).toBe(1) + expect(await pin.get(b)).toBeNull() + expect((await brain.get(a))?.metadata?.v).toBe(2) + expect((await brain.get(b))?.metadata?.v).toBe(1) + expect(pin.isHistorical()).toBe(true) + + // A single-op REMOVE also leaves the pin untouched. + await brain.remove(a) + expect((await pin.get(a))?.metadata?.v).toBe(1) + expect(await brain.get(a)).toBeNull() + await pin.release() + }) + + it('Model-B — historical find() overlays an un-flushed single-op write (overlay bound is generation, not committed)', async () => { + const brain = await openMemoryBrain() + const a = uid('ov-a') + const b = uid('ov-b') + await ( + await brain.transact([ + { op: 'add', id: a, type: NounType.Document, data: 'a', vector: vec(1), metadata: { v: 1 } }, + { op: 'add', id: b, type: NounType.Document, data: 'b', vector: vec(2), metadata: { v: 1 } } + ]) + ).release() + const at1 = await brain.asOf(1) + + // A single-op REMOVE of `b` lands AFTER the pin and is NOT flushed (pending). + await brain.remove(b) + + const liveIds = (await brain.find({})).map((r) => r.id) + const pastIds = (await at1.find({})).map((r) => r.id) + // Live: `b` is gone. Historical (pinned at gen 1): the un-flushed removal is + // overlaid out, so `b` is still present at its pinned state. + expect(liveIds).toContain(a) + expect(liveIds).not.toContain(b) + expect(pastIds).toContain(a) + expect(pastIds).toContain(b) + await at1.release() + }) + + it('Model-B retention — explicit caps reclaim single-op history; committed history survives reopen', async () => { + const { brain, dir } = await openFsBrain() + const a = uid('ret-a') + await brain.add({ id: a, type: NounType.Document, data: 'a', vector: vec(1), metadata: { v: 1 } }) + for (let v = 2; v <= 6; v++) await brain.update({ id: a, metadata: { v } }) + await brain.flush() // persist the per-write generations to disk + expect(brain.generation()).toBe(6) + + // Cap to the 2 most recent generations — older single-op history is reclaimed. + const res = await brain.compactHistory({ maxGenerations: 2 }) + expect(res.removedGenerations).toBeGreaterThan(0) + expect(res.horizon).toBeGreaterThan(0) + await brain.close() + + // Reopen: the survivors + live value are intact; below-horizon asOf throws. + const { brain: reopened } = await openFsBrain(dir) + expect((await reopened.get(a))?.metadata?.v).toBe(6) + await expect(reopened.asOf(1)).rejects.toBeInstanceOf(GenerationCompactedError) + }) + // ========================================================================== // 9. asOf() / released-Db error paths (Y.15 spot-checks) // ========================================================================== diff --git a/tests/integration/db-temporal.test.ts b/tests/integration/db-temporal.test.ts index 137f24c0..4eee90b9 100644 --- a/tests/integration/db-temporal.test.ts +++ b/tests/integration/db-temporal.test.ts @@ -330,8 +330,8 @@ describe('8.0 Db API — temporal range verbs', () => { await dbAtG2.release() }) - // 7. Granularity ------------------------------------------------------------- - it('granularity: single-operation writes (outside transact) are invisible to the temporal verbs', async () => { + // 7. Granularity (Model-B) --------------------------------------------------- + it('granularity: single-operation writes ARE versioned and visible to the temporal verbs', async () => { const brain = await openMemoryBrain() const a = uid('gran-a') const r1 = await brain.transact([ @@ -340,13 +340,26 @@ describe('8.0 Db API — temporal range verbs', () => { await r1.release() expect((await brain.transactionLog()).length).toBe(1) - // A single-op write bumps the counter but writes no generation record/log entry. + // Model-B: a single-op write is its OWN immutable generation — logged, + // diffable, and time-travelable, exactly like a transact() of one op. await brain.update({ id: a, metadata: { v: 2 } }) - expect((await brain.transactionLog()).length).toBe(1) // unchanged — not logged + // The single-op update appended a generation/log entry. + expect((await brain.transactionLog()).length).toBe(2) + expect(brain.generation()).toBe(r1.generation + 1) + + // diff sees the single-op update as a modification of `a`. const d = await brain.diff(r1.generation, brain.generation()) - expect(d.added.nouns).toEqual([]) // the single-op update is not a recorded change - expect(d.modified.nouns).toEqual([]) + expect(d.added.nouns).toEqual([]) + expect(d.modified.nouns).toEqual([a]) + + // And asOf() resolves both states: v=1 at the transact gen, v=2 at the single-op gen. + const atTransact = await brain.asOf(r1.generation) + const atSingleOp = await brain.asOf(brain.generation()) + expect((await atTransact.get(a))?.metadata?.v).toBe(1) + expect((await atSingleOp.get(a))?.metadata?.v).toBe(2) + await atTransact.release() + await atSingleOp.release() }) // 8. Compaction policy contrast --------------------------------------------- @@ -363,7 +376,7 @@ describe('8.0 Db API — temporal range verbs', () => { ).release() } // gens 1..6, all pins released - await brain.compactHistory({ retainGenerations: 2 }) // raise the horizon + await brain.compactHistory({ maxGenerations: 2 }) // raise the horizon const horizon = generationStoreOf(brain).horizon() expect(horizon).toBeGreaterThan(0) diff --git a/tests/unit/db/generation-chain.test.ts b/tests/unit/db/generation-chain.test.ts index 4effee72..1435e6a9 100644 --- a/tests/unit/db/generation-chain.test.ts +++ b/tests/unit/db/generation-chain.test.ts @@ -99,7 +99,7 @@ describe('generation history chains (Model-B scalability)', () => { await bumpX(4) // Reclaim everything except the 2 most recent committed generations. - const res = await brain.compactHistory({ retainGenerations: 2 }) + const res = await brain.compactHistory({ maxGenerations: 2 }) expect(res.removedGenerations).toBeGreaterThan(0) // A pin ABOVE the new horizon still resolves (chains were rebuilt from the survivors)… diff --git a/tests/unit/db/generationStore.test.ts b/tests/unit/db/generationStore.test.ts index f847b92a..12fce749 100644 --- a/tests/unit/db/generationStore.test.ts +++ b/tests/unit/db/generationStore.test.ts @@ -8,7 +8,7 @@ * resolution, and crash rollback of uncommitted generations. */ -import { describe, it, expect, beforeEach } from 'vitest' +import { describe, it, expect, beforeEach, vi } from 'vitest' import { MemoryStorage } from '../../../src/storage/adapters/memoryStorage.js' import { GenerationStore, @@ -253,12 +253,12 @@ describe('db/GenerationStore', () => { expect(remaining).toEqual([]) }) - it('retainGenerations keeps the most recent record-sets', async () => { + it('maxGenerations cap keeps the most recent record-sets', async () => { await commitWrite(ID_A, 1) await commitWrite(ID_A, 2) await commitWrite(ID_A, 3) - const result = await store.compact({ retainGenerations: 2 }) + const result = await store.compact({ maxGenerations: 2 }) expect(result.removedGenerations).toBe(1) expect(result.horizon).toBe(1) @@ -330,4 +330,141 @@ describe('db/GenerationStore', () => { expect(manifest.generation).toBe(2) }) }) + + // ========================================================================== + // Model-B single-op generation-stamping (per-write history + group-commit) + // ========================================================================== + describe('Model-B single-op generations', () => { + /** Apply one single-op write as its own generation (live write + buffered history). */ + async function singleOpWrite(id: string, version: number): Promise { + const { generation } = await store.commitSingleOp({ + touched: { nouns: [id] }, + execute: async () => { + await storage.saveNounMetadata(id, metadataFixture(version)) + } + }) + return generation + } + + it('each single-op write is its own generation, resolvable BEFORE any flush', async () => { + const g1 = await singleOpWrite(ID_A, 1) + const g2 = await singleOpWrite(ID_A, 2) + expect(g1).toBe(1) + expect(g2).toBe(2) + // Counter advanced per write; nothing persisted to disk yet. + expect(store.generation()).toBe(2) + expect(store.committedGeneration()).toBe(0) + // Reads resolve through the in-memory pending tier with NO forced flush: + // the state as-of g1 is v1 (the before-image buffered by g2). + const atG1 = await store.resolveAt('noun', ID_A, g1) + expect(atG1.source).toBe('record') + expect(((atG1 as { metadata: { version: number } }).metadata).version).toBe(1) + // hasCommittedAfter counts pending → a now()-style pin at g1 is historical. + expect(store.hasCommittedAfter(g1)).toBe(true) + }) + + it('flush persists pending single-op generations (groupCommit-marked) and advances committed', async () => { + await singleOpWrite(ID_A, 1) + await singleOpWrite(ID_A, 2) + await store.flushPendingSingleOps() + expect(store.committedGeneration()).toBe(2) + const delta = await storage.readRawObject(`${GENERATIONS_PREFIX}/2/tx.json`) + expect(delta.groupCommit).toBe(true) + expect(typeof delta.bytes).toBe('number') + // asOf resolution survives the flush (now reading before-images from disk). + const atG1 = await store.resolveAt('noun', ID_A, 1) + expect(((atG1 as { metadata: { version: number } }).metadata).version).toBe(1) + }) + + it('🛑 a crash mid group-commit flush DROPS the generation WITHOUT reverting the live write', async () => { + // gen 1: X = v1, flushed (durable). + await singleOpWrite(ID_A, 1) + await store.flushPendingSingleOps() + expect(store.committedGeneration()).toBe(1) + + // gen 2: X = v2 acknowledged to canonical storage; its history is buffered. + await singleOpWrite(ID_A, 2) + expect((await storage.readNounRaw(ID_A)).metadata).toMatchObject({ version: 2 }) + + // Crash before the manifest rename of the group-commit flush: the gen-2 + // dir (before-image v1) lands on disk; the manifest stays at gen 1. + store.setCommitFaultInjector((phase) => { + if (phase === 'before-manifest-rename') throw new Error('simulated crash mid-flush') + }) + await expect(store.flushPendingSingleOps()).rejects.toThrow('simulated crash mid-flush') + store.setCommitFaultInjector(undefined) + expect(await storage.readRawObject(`${GENERATIONS_PREFIX}/2/tx.json`)).not.toBeNull() + expect((await storage.readRawObject(MANIFEST_PATH)).generation).toBe(1) + + // Recover on a FRESH store (process restart: in-memory pending is gone). + const reopened = new GenerationStore(storage) + const result = await reopened.open() + expect(result.rolledBackGenerations).toBe(1) + expect(reopened.committedGeneration()).toBe(1) + + // THE CORRUPTION TRAP: drop-without-restore — the acknowledged live write + // (v2) is NOT reverted to the before-image (v1). Only the crashed flush + // window's HISTORY is lost; live data stays correct. + expect((await storage.readNounRaw(ID_A)).metadata).toMatchObject({ version: 2 }) + // The orphan dir is gone and the dropped generation number is never reissued. + expect(await storage.readRawObject(`${GENERATIONS_PREFIX}/2/tx.json`)).toBeNull() + expect(reopened.generation()).toBeGreaterThanOrEqual(2) + }) + + it('a clean reopen replays committed single-op history (asOf survives restart)', async () => { + await singleOpWrite(ID_A, 1) + await singleOpWrite(ID_A, 2) + await store.flushPendingSingleOps() + + const reopened = new GenerationStore(storage) + await reopened.open() + expect(reopened.committedGeneration()).toBe(2) + const atG1 = await reopened.resolveAt('noun', ID_A, 1) + expect(((atG1 as { metadata: { version: number } }).metadata).version).toBe(1) + }) + }) + + // ========================================================================== + // Retention CAPS (Model-B `retention` knob) + // ========================================================================== + describe('retention caps', () => { + async function manyGens(n: number): Promise { + for (let v = 1; v <= n; v++) await commitWrite(ID_A, v) + } + + it('maxBytes reclaims the oldest generations until total history fits the budget', async () => { + await manyGens(5) + const total = await store.historyBytes() + expect(total).toBeGreaterThan(0) + const perGen = total / 5 + const budget = Math.floor(perGen * 2.5) + const result = await store.compact({ maxBytes: budget }) + expect(result.removedGenerations).toBeGreaterThan(0) + expect(await store.historyBytes()).toBeLessThanOrEqual(budget) + }) + + it('maxAge keeps generations within the window and reclaims older ones', async () => { + await manyGens(3) + // A wide window keeps everything. + expect((await store.compact({ maxAge: 60_000 })).removedGenerations).toBe(0) + // Advance the clock 10s past the commits, then a 1s window reclaims all + // three (fake timers make the age comparison deterministic). + vi.useFakeTimers() + try { + vi.setSystemTime(Date.now() + 10_000) + expect((await store.compact({ maxAge: 1_000 })).removedGenerations).toBe(3) + } finally { + vi.useRealTimers() + } + }) + + it('no caps reclaims every unpinned generation; a pin protects newer ones', async () => { + await manyGens(3) + store.pin(2) // protects generations > 2 (i.e. gen 3) from reclamation + const result = await store.compact() + // gens 1 and 2 are ≤ the pin → reclaimable; gen 3 is pin-protected. + expect(result.removedGenerations).toBe(2) + store.release(2) + }) + }) })