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

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

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

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

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

View file

@ -110,13 +110,29 @@ What you get:
- **`brain.transactionLog({ from?, to?, limit? })`** — the commit log over an - **`brain.transactionLog({ from?, to?, limit? })`** — the commit log over an
**inclusive** generation/`Date` window (contrast `since`'s exclusive lower **inclusive** generation/`Date` window (contrast `since`'s exclusive lower
bound); `limit` applies last. bound); `limit` applies last.
- **`brain.compactHistory({ retainGenerations, retainMs })`** — explicit - **Every write is versioned (Model-B).** History granularity is **per-write**:
retention. Compaction never breaks a pinned read. `diff`/`since` throw EVERY write — `transact()` AND a single-operation `add`/`update`/`remove`/
`GenerationCompactedError` below the horizon; `history` truncates to it. `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 The precise guarantees are documented in:
`transact()` commits; single-operation writes advance the clock but produce no
historical records) are documented in:
- [docs/concepts/consistency-model.md](docs/concepts/consistency-model.md) — - [docs/concepts/consistency-model.md](docs/concepts/consistency-model.md) —
the guarantees, each proven by a dedicated test in 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` | | `brain.stats().indexHealth.hnsw` | `brain.stats().indexHealth.vector` |
| Storage adapter contract `saveHNSWData()` / `getHNSWData()` | `saveVectorIndexData()` / `getVectorIndexData()` | | Storage adapter contract `saveHNSWData()` / `getHNSWData()` | `saveVectorIndexData()` / `getVectorIndexData()` |
| Cache category `'hnsw'` (`CacheProvider` union) | `'vectors'` | | 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): Mechanical renames as one command (run from your repo root, review the diff):

View file

@ -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 so a "full" index query over an overlay would silently exclude the
overlay's own entities. Commit with `transact()` to get the full surface. overlay's own entities. Commit with `transact()` to get the full surface.
**History granularity.** Generation *records* are written per `transact()` **History granularity (Model-B).** EVERY write is its own immutable generation
batch only. Single-operation writes advance the counter (so watermarks and `transact()` batches AND single-operation `add`/`update`/`remove`/`relate`.
CAS stay sound) but do not stage before-images: they remain visible through Single-ops stage a before-image and are reported by `db.since()`/`asOf()`/
earlier pins and are not reported by `db.since()`. Code that needs pinned `diff()`/`history()` exactly like transacts; a pin always freezes against later
isolation across its writes uses `transact()` — that is the documented 8.0 writes. `transact()` groups several operations into ONE atomic generation.
contract, stated rather than papered over.
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 ### Pinning, retention, compaction
@ -162,11 +172,17 @@ contract, stated rather than papered over.
`pin(generation)` on every registered `VersionedIndexProvider`, whose `pin(generation)` on every registered `VersionedIndexProvider`, whose
explicit pin lifetime overrides any time-based snapshot retention the explicit pin lifetime overrides any time-based snapshot retention the
provider has). provider has).
- `compactHistory({ retainGenerations?, retainMs? })` reclaims record-sets. - The constructor **`retention`** knob governs auto-compaction (on `flush()`/
A record-set `N` is reclaimed only when `N` is at or below **every** live `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 pin — deleting `N` can only break readers pinned *below* `N`, because
resolution reads before-images from generations strictly greater than the 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). - The manifest records the **horizon** (highest reclaimed generation).
Generations below the horizon are unreachable; `asOf()` on them throws Generations below the horizon are unreachable; `asOf()` on them throws
`GenerationCompactedError`. The horizon itself stays reachable, resolved `GenerationCompactedError`. The horizon itself stays reachable, resolved

View file

@ -901,16 +901,18 @@ that generation), once per `Db`, freed on `release()`.
**Throws:** `GenerationCompactedError` when the generation's records were reclaimed by `compactHistory()`. **Throws:** `GenerationCompactedError` when the generation's records were reclaimed by `compactHistory()`.
**History granularity:** only `transact()` batches produce historical **History granularity:** every write is its own immutable generation —
records; single-operation writes advance the clock but stay visible through `transact()` batches AND single-operation writes — so a pin always freezes and
earlier pins. See the [consistency model](../concepts/consistency-model.md). every write is addressable via `asOf()`. See the
[consistency model](../concepts/consistency-model.md).
--- ---
### `transactionLog(options?)``Promise<TxLogEntry[]>` ### `transactionLog(options?)``Promise<TxLogEntry[]>`
Read the reified transaction log — one entry per committed `transact()` Read the reified transaction log — one entry per committed generation (every
batch, newest first: `{ generation, timestamp, meta? }`. `transact()` AND single-op write), newest first: `{ generation, timestamp,
meta? }`. Single-op generations carry no `meta` (it is a `transact()`-only field).
```typescript ```typescript
const [latest] = await brain.transactionLog({ limit: 1 }) const [latest] = await brain.transactionLog({ limit: 1 })
@ -921,13 +923,15 @@ latest.meta // { author: 'order-service', requestId: 'req-9f2' }
### `compactHistory(options?)``Promise<CompactHistoryResult>` ### `compactHistory(options?)``Promise<CompactHistoryResult>`
Reclaim historical record-sets that no retention rule and no live `Db` pin Reclaim historical record-sets that no retention cap and no live `Db` pin
protects. Pinned reads stay correct across compaction, always. 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 ```typescript
await brain.compactHistory({ await brain.compactHistory({
retainGenerations: 100, // keep the 100 most recent commits maxGenerations: 100, // keep at most the 100 most recent generations
retainMs: 7 * 24 * 60 * 60 * 1000 // and everything from the last 7 days maxAge: 7 * 24 * 60 * 60 * 1000 // and only those from the last 7 days
}) })
``` ```

View file

@ -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 - Every live `Db` holds a refcounted **pin**; a record-set is never
reclaimed while any pin could need it — pinned reads stay correct across reclaimed while any pin could need it — pinned reads stay correct across
compaction, always. compaction, always.
- `brain.compactHistory({ retainGenerations?, retainMs? })` reclaims - The **`retention`** knob governs auto-compaction (on every `flush()`/
everything no retention rule and no pin protects, and records the `close()`): unset → ADAPTIVE (disk/RAM-pressure, zero-config) · `'all'`
**horizon**`asOf()` below it throws `GenerationCompactedError`, unbounded · `{ maxGenerations?, maxAge?, maxBytes? }` → explicit caps.
explicitly, never partial data. `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 - To keep a state readable forever, `persist()` it first: snapshots are
self-contained and unaffected by compaction of the source store. 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 ([multi-process model](./multi-process.md)). Transactions are atomic
within one writer process — there is no distributed or cross-process within one writer process — there is no distributed or cross-process
transaction coordination. transaction coordination.
- **History granularity.** Only `transact()` batches produce historical - **History granularity.** Every write is its own immutable generation —
records; single-operation writes between commits stay visible through `transact()` batches AND single-operation `add`/`update`/`remove`/`relate`.
earlier pins (see above). 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 - **Compacted history is gone.** `asOf()` below the compaction horizon
fails explicitly; persist what you must keep. fails explicitly; persist what you must keep.
- **Counter persistence is coalesced for single-operation writes.** Durable - **Counter persistence is coalesced for single-operation writes.** Durable

View file

@ -135,10 +135,12 @@ await after.release()
Three things to remember: Three things to remember:
- History granularity is `transact()` commits — single-operation writes - History granularity is per-write: EVERY write — `transact()` AND a
advance the clock but do not produce historical records (see the single-operation `add`/`update`/`remove`/`relate` — is its own immutable
[consistency model](../concepts/consistency-model.md)). Use `transact()` generation, so a pin always freezes against later writes and every write is
for writes you want to travel back through. 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, - The first index-accelerated query (semantic search, traversal, cursors,
aggregation) at a historical generation builds an in-memory index aggregation) at a historical generation builds an in-memory index
materialization — O(n at that generation), once per `Db`, freed on 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 ## 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 ```typescript
await brain.compactHistory({ // Zero-config: ADAPTIVE — keep as much history as free disk/RAM allows,
retainGenerations: 100, // keep the 100 most recent commits // reclaiming oldest-first under pressure. (This is the default.)
retainMs: 7 * 24 * 60 * 60 * 1000 // and everything from the last 7 days 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 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 no live `Db` could need them (live pins are ALWAYS exempt). Release views you
ones `transact()` returns), and `persist()` any generation you want to keep are done with (including the ones `transact()` returns), and `persist()` any
beyond the retention window: snapshots are self-contained and unaffected by generation you want to keep beyond the retention window: snapshots are
compaction. self-contained and unaffected by compaction.
## From branches to values ## From branches to values

View file

@ -121,6 +121,7 @@ import { AggregateMaterializer } from './aggregation/materializer.js'
import type { AggregateDefinition, AggregateQueryParams, AggregateResult } from './types/brainy.types.js' import type { AggregateDefinition, AggregateQueryParams, AggregateResult } from './types/brainy.types.js'
import { resolveJsHnswConfig, DEFAULT_RECALL } from './utils/recallPreset.js' import { resolveJsHnswConfig, DEFAULT_RECALL } from './utils/recallPreset.js'
import * as fs from 'node:fs' import * as fs from 'node:fs'
import * as os from 'node:os'
import { Db, type DbHost, type HistoricalQueryHandle } from './db/db.js' import { Db, type DbHost, type HistoricalQueryHandle } from './db/db.js'
import { import {
importGraph, importGraph,
@ -148,7 +149,7 @@ import type {
} from './db/types.js' } from './db/types.js'
import { stableDeepEqual } from './db/stableEqual.js' import { stableDeepEqual } from './db/stableEqual.js'
import type { VersionedIndexProvider } from './plugin.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 * Stopwords for semantic highlighting
@ -208,7 +209,7 @@ type ResolvedBrainyConfig = Required<
| 'vector' | 'vector'
| 'plugins' | 'plugins'
| 'integrations' | 'integrations'
| 'history' | 'retention'
| 'eagerEmbeddings' | 'eagerEmbeddings'
> >
> & > &
@ -219,7 +220,7 @@ type ResolvedBrainyConfig = Required<
| 'vector' | 'vector'
| 'plugins' | 'plugins'
| 'integrations' | 'integrations'
| 'history' | 'retention'
| 'eagerEmbeddings' | 'eagerEmbeddings'
> >
@ -365,6 +366,23 @@ export class Brainy<T = any> implements BrainyInterface<T> {
private _tripleIntelligence?: TripleIntelligenceSystem private _tripleIntelligence?: TripleIntelligenceSystem
private _vfs?: VirtualFileSystem private _vfs?: VirtualFileSystem
private _vfsInitialized = false // Track VFS init completion separately 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 _hub?: IntegrationHub // Integration Hub for external tools
private _pendingMigrationRunner?: MigrationRunner // Deferred migration runner for large datasets private _pendingMigrationRunner?: MigrationRunner // Deferred migration runner for large datasets
private _aggregationIndex?: AggregationIndex // Incremental aggregation engine private _aggregationIndex?: AggregationIndex // Incremental aggregation engine
@ -899,6 +917,13 @@ export class Brainy<T = any> implements BrainyInterface<T> {
await this._vfs.init() await this._vfs.init()
this._vfsInitialized = true // Mark VFS as fully initialized 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. // Eager embedding initialization.
// //
// Adaptive default (8.0): the WASM embedding engine eagerly initializes // Adaptive default (8.0): the WASM embedding engine eagerly initializes
@ -1233,6 +1258,43 @@ export class Brainy<T = any> implements BrainyInterface<T> {
return uuidv7() 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<void>
): Promise<void> {
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<T>): Promise<string> { async add(params: AddParams<T>): Promise<string> {
this.assertWritable('add') this.assertWritable('add')
await this.ensureInitialized() await this.ensureInitialized()
@ -1375,9 +1437,10 @@ export class Brainy<T = any> implements BrainyInterface<T> {
} }
} }
// 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 // 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) // Operation 1: Save metadata FIRST (TypeAwareStorage caching)
// isNew=true: skip pre-read for rollback (entity doesn't exist yet) // isNew=true: skip pre-read for rollback (entity doesn't exist yet)
tx.addOperation( tx.addOperation(
@ -2248,8 +2311,9 @@ export class Brainy<T = any> implements BrainyInterface<T> {
metadata: newMetadata metadata: newMetadata
} }
// Execute atomically with transaction system // Execute atomically with transaction system, generation-stamped as one
await this.transactionManager.executeTransaction(async (tx) => { // 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) // Operation 1: Update metadata FIRST (updates type cache)
tx.addOperation( tx.addOperation(
new UpdateNounMetadataOperation(this.storage, params.id, updatedMetadata) new UpdateNounMetadataOperation(this.storage, params.id, updatedMetadata)
@ -2358,8 +2422,12 @@ export class Brainy<T = any> implements BrainyInterface<T> {
const targetVerbs = await this.storage.getVerbsByTarget(id) const targetVerbs = await this.storage.getVerbsByTarget(id)
const allVerbs = [...verbs, ...targetVerbs] const allVerbs = [...verbs, ...targetVerbs]
// Execute atomically with transaction system // Execute atomically with transaction system, generation-stamped as one
await this.transactionManager.executeTransaction(async (tx) => { // 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 // Operation 1: Remove from vector index
if (noun) { if (noun) {
tx.addOperation( tx.addOperation(
@ -2769,8 +2837,13 @@ export class Brainy<T = any> implements BrainyInterface<T> {
// entities were existence-checked above) and mirror them onto the verb. // entities were existence-checked above) and mirror them onto the verb.
const { sourceInt, targetInt } = this.resolveVerbEndpointInts(verb) const { sourceInt, targetInt } = this.resolveVerbEndpointInts(verb)
// Execute atomically with transaction system // Reverse-edge id reserved up front (when bidirectional) so the Model-B
await this.transactionManager.executeTransaction(async (tx) => { // 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 // Operation 1: Save verb vector data
tx.addOperation( tx.addOperation(
new SaveVerbOperation(this.storage, { new SaveVerbOperation(this.storage, {
@ -2798,8 +2871,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
) )
// Create bidirectional if requested // Create bidirectional if requested
if (params.bidirectional) { if (params.bidirectional && reverseId) {
const reverseId = uuidv4()
const reverseVerb: GraphVerb = { const reverseVerb: GraphVerb = {
...verb, ...verb,
id: reverseId, id: reverseId,
@ -2868,8 +2940,9 @@ export class Brainy<T = any> implements BrainyInterface<T> {
// a rollback can re-add through the BigInt addVerb contract. // a rollback can re-add through the BigInt addVerb contract.
const endpointInts = verb ? this.resolveVerbEndpointInts(verb) : undefined const endpointInts = verb ? this.resolveVerbEndpointInts(verb) : undefined
// Execute atomically with transaction system // Execute atomically with transaction system, generation-stamped as one
await this.transactionManager.executeTransaction(async (tx) => { // immutable Model-B generation (before-image = the relationship's state).
await this.persistSingleOp({ verbs: [id] }, async (tx) => {
// Operation 1: Remove from graph index // Operation 1: Remove from graph index
if (verb && endpointInts) { if (verb && endpointInts) {
tx.addOperation( tx.addOperation(
@ -2997,7 +3070,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
// resolution serves both the remove (rollback re-add) and the re-add. // resolution serves both the remove (rollback re-add) and the re-add.
const reindexInts = typeChanged ? this.resolveVerbEndpointInts(verbForIndex) : undefined const reindexInts = typeChanged ? this.resolveVerbEndpointInts(verbForIndex) : undefined
await this.transactionManager.executeTransaction(async (tx) => { await this.persistSingleOp({ verbs: [params.id] }, async (tx) => {
tx.addOperation( tx.addOperation(
new UpdateVerbMetadataOperation(this.storage, params.id, updatedMetadata) new UpdateVerbMetadataOperation(this.storage, params.id, updatedMetadata)
) )
@ -5104,9 +5177,27 @@ export class Brainy<T = any> implements BrainyInterface<T> {
const chunkQueued: string[] = [] const chunkQueued: string[] = []
const chunkBuilderFailed: Array<{ item: string; error: 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 { try {
// Process chunk in single transaction for atomic deletion // Process chunk as ONE atomic Model-B generation (entities + cascade verbs).
await this.transactionManager.executeTransaction(async (tx) => { await this.persistSingleOp({ nouns: touchedNouns, verbs: touchedVerbs }, async (tx) => {
for (const id of chunk) { for (const id of chunk) {
try { try {
// Load entity data // Load entity data
@ -5599,13 +5690,13 @@ export class Brainy<T = any> implements BrainyInterface<T> {
/** /**
* @description Read the reified transaction log one entry per committed * @description Read the reified transaction log one entry per committed
* `transact()` batch, carrying the committed generation, the commit * generation, carrying the committed generation, the commit timestamp, and
* timestamp, and the `meta` the transaction was submitted with. Entries * the `meta` the transaction was submitted with. Entries are returned newest
* are returned newest first. Single-operation writes * first. Under Model-B EVERY write is its own generation, so single-operation
* (`add`/`update`/`remove`/`relate`/) advance the generation counter but * writes (`add`/`update`/`remove`/`relate`/) appear here too (without `meta`
* do not append log entries (the documented 8.0 history granularity), and * transaction metadata is a `transact()`-only concept). `compactHistory()`
* `compactHistory()` never rewrites the log entries may reference * never rewrites the log entries may reference generations whose
* generations whose record-sets were already reclaimed. * record-sets were already reclaimed.
* *
* @param options - `from`/`to` (generation number or `Date`) bound the window * @param options - `from`/`to` (generation number or `Date`) bound the window
* to commits in `[from, to]` INCLUSIVE on both ends a log window names * to commits in `[from, to]` INCLUSIVE on both ends a log window names
@ -5754,6 +5845,11 @@ export class Brainy<T = any> implements BrainyInterface<T> {
throw new Error('transact(): provide a non-empty array of transaction operations') 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 // Early CAS check — avoids expensive planning (embedding) on a stale
// expectation. The generation store re-checks authoritatively under the // expectation. The generation store re-checks authoritatively under the
// commit mutex; this one is purely a fast-fail. // commit mutex; this one is purely a fast-fail.
@ -5796,9 +5892,9 @@ export class Brainy<T = any> implements BrainyInterface<T> {
* @description Open an immutable `Db` view of PAST state: * @description Open an immutable `Db` view of PAST state:
* *
* - **`number`** a generation: pins it on this store; reads resolve * - **`number`** a generation: pins it on this store; reads resolve
* through the generational record layer. History granularity is * through the generational record layer. Under Model-B every write is its
* `transact()` commits single-operation writes between commits do not * own generation, so single-operation writes are individually addressable
* produce records and remain visible through earlier pins. * too (a pin always freezes against later single-op writes).
* - **`Date`** a wall-clock instant: resolved via the transaction log to * - **`Date`** a wall-clock instant: resolved via the transaction log to
* the newest generation committed at or before it, then pinned as above. * the newest generation committed at or before it, then pinned as above.
* - **`string`** a snapshot directory previously produced by * - **`string`** a snapshot directory previously produced by
@ -6128,57 +6224,114 @@ export class Brainy<T = any> implements BrainyInterface<T> {
* keep (the GC backstop eventually releases leaked ones, but until then * keep (the GC backstop eventually releases leaked ones, but until then
* compaction retains the history they protect). * compaction retains the history they protect).
* *
* @param options - `retainGenerations` (keep the N most recent committed * @param options - Explicit retention CAPS (`maxGenerations` / `maxAge` /
* generations) and/or `retainMs` (keep everything committed within the * `maxBytes`). Reclaim the oldest unpinned generations while ANY supplied
* window). Both supplied = both must allow reclaim. Neither = reclaim * cap is exceeded. Neither = reclaim everything unpinned.
* everything unpinned.
* @returns Count of reclaimed record-sets and the new horizon. * @returns Count of reclaimed record-sets and the new horizon.
* @example * @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<CompactHistoryResult> { async compactHistory(options?: CompactHistoryOptions): Promise<CompactHistoryResult> {
this.assertWritable('compactHistory') this.assertWritable('compactHistory')
await this.ensureInitialized() 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) return this.generationStore.compact(options)
} }
/** /**
* @description Resolve the effective generational-history retention policy * @description Drive the adaptive retention byte budget at runtime the
* from `config.history`, filling per-field defaults so a partial object * settable input a machine-level coordinator (e.g. cor's `ResourceManager`,
* inherits the rest. This is the single read site for the policy referenced * which fair-shares one box-wide history budget across co-located instances)
* by {@link autoCompactHistory}. * pushes whenever the box's free-resource picture changes. Takes effect at the
* * next auto-compaction (flush/close). Only meaningful while `retention` is
* Defaults (zero-config): keep the 100 most recent committed generations * adaptive (unset or `'adaptive'`); ignored under `'all'` or explicit caps.
* AND everything committed within the last 7 days; auto-compact on * @param bytes - The new per-brain history byte budget (non-negative).
* `flush()` / `close()`. A generation is reclaimed only when it is older * @example brain.setRetentionBudget(2 * 1024 ** 3) // 2 GiB for this brain
* than BOTH bounds and no live `Db` pin protects it.
*
* @returns The resolved policy: retention bounds plus the `autoCompact` flag.
*/ */
private resolveHistoryPolicy(): { setRetentionBudget(bytes: number): void {
retainGenerations: number if (!Number.isFinite(bytes) || bytes < 0) {
retainMs: number 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 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 { return {
retainGenerations: history?.retainGenerations ?? 100, mode: 'explicit',
retainMs: history?.retainMs ?? 604_800_000, // 7 days maxGenerations: retention.maxGenerations,
autoCompact: history?.autoCompact ?? true maxAge: retention.maxAge,
maxBytes: retention.maxBytes,
budgetBytes: retention.budgetBytes,
autoCompact: retention.autoCompact ?? true
} }
} }
/** /**
* @description Run history compaction with the resolved retention policy if * @description Compute the adaptive history byte budget for this brain: the
* `config.history.autoCompact` is on (the default). Invoked from `flush()` * coordinator-driven value when present (`setRetentionBudget()` /
* and `close()` so generational record-sets cannot accumulate unbounded * `config.retention.budgetBytes` e.g. cor's fair-shared box budget), else a
* across a long-lived writer's lifetime. * 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. * Read-only instances and an explicit `autoCompact: false` skip silently.
* Pinned generations (live `Db` values) are never reclaimed the safety * Pinned generations are never reclaimed ({@link GenerationStore.compact}).
* invariant lives in {@link GenerationStore.compact}. Failures are logged * Failures are logged and swallowed compaction is housekeeping and must
* and swallowed: compaction is housekeeping and must never fail a flush or * never fail a flush or a clean shutdown.
* a clean shutdown.
*/ */
private async autoCompactHistory(): Promise<void> { private async autoCompactHistory(): Promise<void> {
// Nothing to compact on a read-only instance or before init wired up the // Nothing to compact on a read-only instance or before init wired up the
@ -6186,15 +6339,22 @@ export class Brainy<T = any> implements BrainyInterface<T> {
if (this.isReadOnly || !this.generationStore) { if (this.isReadOnly || !this.generationStore) {
return return
} }
const policy = this.resolveHistoryPolicy() const policy = this.resolveRetentionPolicy()
if (!policy.autoCompact) { if (!policy.autoCompact || policy.mode === 'all') {
return return
} }
try { try {
await this.generationStore.compact({ if (policy.mode === 'adaptive') {
retainGenerations: policy.retainGenerations, const budget = this.adaptiveHistoryBudgetBytes(policy.budgetBytes)
retainMs: policy.retainMs 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) { } catch (error) {
console.warn( console.warn(
`Auto-compaction of generational history failed (non-fatal): ${ `Auto-compaction of generational history failed (non-fatal): ${
@ -7964,6 +8124,11 @@ export class Brainy<T = any> implements BrainyInterface<T> {
console.log('Flushing Brainy indexes and caches to disk...') console.log('Flushing Brainy indexes and caches to disk...')
const startTime = Date.now() 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 // Flush all components in parallel for performance
await Promise.all([ await Promise.all([
// 1. Flush storage adapter counts (entity/verb counts by type) // 1. Flush storage adapter counts (entity/verb counts by type)
@ -7991,7 +8156,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
this.generationStore.persistCounterNow() 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 // Runs after the flush so durable state is in place; respects live
// Db pins and an explicit autoCompact: false. // Db pins and an explicit autoCompact: false.
await this.autoCompactHistory() await this.autoCompactHistory()
@ -11730,9 +11895,10 @@ export class Brainy<T = any> implements BrainyInterface<T> {
// `recall` preset + `persistMode` (folded in from 7.x's hnswPersistMode). // `recall` preset + `persistMode` (folded in from 7.x's hnswPersistMode).
// See BRAINY-8.0-RENAME-COORDINATION § A.2 + § G.2. // See BRAINY-8.0-RENAME-COORDINATION § A.2 + § G.2.
vector: config?.vector ?? undefined, vector: config?.vector ?? undefined,
// Generational-history retention — defaults applied at the read site // Generational-history retention — the `retention` knob; defaults applied
// (resolveHistoryPolicy) so a partial object inherits per-field defaults. // at the read site (resolveRetentionPolicy) so a partial object inherits
history: config?.history ?? undefined, // per-field defaults and `undefined` resolves to ADAPTIVE.
retention: config?.retention ?? undefined,
// Embedding initialization — left undefined when omitted so the adaptive // Embedding initialization — left undefined when omitted so the adaptive
// default resolves at the init() read site (eager when the WASM embedder // 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 // is the active one on a non-reader writer outside unit tests). Explicit
@ -12270,7 +12436,14 @@ export class Brainy<T = any> implements BrainyInterface<T> {
* This ensures deferred persistence mode data is saved * This ensures deferred persistence mode data is saved
*/ */
async close(): Promise<void> { async close(): Promise<void> {
// 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 // on) BEFORE the generation store closes below. Respects live Db pins and
// an explicit autoCompact: false; no-op on read-only instances. // an explicit autoCompact: false; no-op on read-only instances.
await this.autoCompactHistory() await this.autoCompactHistory()

View file

@ -346,9 +346,12 @@ export class Db<T = any> {
const limit = params.limit ?? 10 const limit = params.limit ?? 10
const offset = params.offset ?? 0 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 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<string, Entity<T> | null>() const overlayNouns = this.overlay?.nouns ?? new Map<string, Entity<T> | null>()
const excluded = new Set<string>([...changedNouns, ...overlayNouns.keys()]) const excluded = new Set<string>([...changedNouns, ...overlayNouns.keys()])
@ -480,8 +483,11 @@ export class Db<T = any> {
const limit = params.limit ?? 100 const limit = params.limit ?? 100
const offset = params.offset ?? 0 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 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<string, Relation<T> | null>() const overlayVerbs = this.overlay?.verbs ?? new Map<string, Relation<T> | null>()
const excluded = new Set<string>([...changedVerbs, ...overlayVerbs.keys()]) const excluded = new Set<string>([...changedVerbs, ...overlayVerbs.keys()])

View file

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

View file

@ -138,21 +138,34 @@ export interface TransactReceipt {
// ============================================================================ // ============================================================================
/** /**
* @description Options for `brain.compactHistory()`. When both options are * @description Options for `brain.compactHistory()` explicit retention
* supplied a generation must satisfy *both* retention rules to be reclaimed * **caps** (Model-B `retention` knob). Each supplied field is an upper bound:
* (the stricter retention wins). With no options, every generation not * 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. * 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 { export interface CompactHistoryOptions {
/** /**
* Keep at least this many of the most recent committed generations * Keep at most this many of the most recent committed generations reclaim
* (`0` = keep none beyond what live pins protect). * 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 * @description Result of `db.since(priorDb)` the ids touched by committed
* transactions in the generation interval `(priorDb.generation, * generations in the interval `(priorDb.generation, db.generation]`. Under
* db.generation]`. Granularity note: single-operation writes performed * Model-B EVERY write is its own generation, so single-operation writes
* outside `transact()` bump the generation counter but do not produce * (`add`/`update`/`remove`/`relate` outside `transact()`) ARE reflected here,
* generation records, so they are not reflected here (documented 8.0 * exactly like `transact()` commits.
* history granularity see `docs/ADR-001-generational-mvcc.md`).
*/ */
export interface ChangedIds { export interface ChangedIds {
/** Entity (noun) ids touched in the interval, sorted. */ /** 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 * @description One version of a single entity or relationship in its
* {@link EntityHistory} the materialized state at a committed generation that * {@link EntityHistory} the materialized state at a committed generation that
* changed it. `value` is `null` when the id did not exist at that version * 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<T = any> { export interface HistoryVersion<T = any> {
/** The committed generation that produced this version. */ /** The committed generation that produced this version. */
@ -271,9 +284,11 @@ export interface GenerationRecord {
/** /**
* @description The per-generation delta persisted at * @description The per-generation delta persisted at
* `_generations/<N>/tx.json` written and fsynced *before* any operation of * `_generations/<N>/tx.json`. For a `transact()` generation it is written and
* the transaction executes, so crash recovery always knows exactly which ids * fsynced *before* any operation of the transaction executes, so crash
* to restore from before-images. * 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 { export interface GenerationDelta {
/** The generation this delta belongs to. */ /** The generation this delta belongs to. */
@ -286,6 +301,26 @@ export interface GenerationDelta {
nouns: string[] nouns: string[]
/** Relationship ids touched by this generation. */ /** Relationship ids touched by this generation. */
verbs: string[] 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/<id>.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
} }
/** /**

View file

@ -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 * Under Model-B EVERY write (`transact()` AND single-op `add`/`update`/
* serves historical reads (`asOf()`, pinned `Db` values). Without * `remove`/`relate`) produces an immutable generation record-set serving
* compaction those record-sets accumulate forever, so Brainy * historical reads (`asOf()`, pinned `Db` values). Without compaction those
* **auto-compacts on every `flush()` and `close()`** with this policy. * accumulate, so Brainy **auto-compacts on every `flush()` and `close()`**.
* A generation is reclaimed only when it is older than BOTH bounds * Live `Db` pins are ALWAYS exempt from reclamation, in every mode.
* (and never while a live `Db` pin protects it):
* *
* - `retainGenerations` keep at least the N most recent committed * Modes:
* generations (default: `100`). * - **unset ADAPTIVE (default):** the retention horizon tracks
* - `retainMs` keep everything committed within the window * disk/RAM pressure. A machine-level byte budget governs how much history
* (default: 7 days). * 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, * `autoCompact: false` disables the automatic flush/close compaction (manage
* or set `autoCompact: false` to manage history manually via * manually via `brain.compactHistory()`). `budgetBytes` is the settable
* `brain.compactHistory()`. Long-term archives belong in * adaptive byte budget a coordinator drives (also via `brain.setRetentionBudget()`).
* `db.persist(path)` snapshots, which compaction never touches. * Long-term archives belong in `db.persist(path)` snapshots, which compaction
* never touches.
*/ */
history?: { retention?:
/** Keep at least this many recent committed generations (default: 100). */ | 'all'
retainGenerations?: number | 'adaptive'
/** Keep generations committed within this window in ms (default: 604800000 = 7 days). */ | {
retainMs?: number /** Keep at most this many recent committed generations (cap). */
/** Run compaction automatically on flush()/close() (default: true). */ maxGenerations?: number
autoCompact?: boolean /** 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 // Memory management options
maxQueryLimit?: number // Override auto-detected query result limit (max: 100000) maxQueryLimit?: number // Override auto-detected query result limit (max: 100000)

View file

@ -58,7 +58,7 @@ async function mkBrain(backend: 'memory' | 'filesystem', dir?: string): Promise<
storage, storage,
eagerEmbeddings: false, // never load WASM — we pass precomputed vectors eagerEmbeddings: false, // never load WASM — we pass precomputed vectors
// Keep history fully retained for the measurement (no mid-run compaction skewing depth). // 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 }, index: { m: 16, efConstruction: 200, efSearch: 50 },
cache: { maxSize: 1000, ttl: 3600 } cache: { maxSize: 1000, ttl: 3600 }
} as BrainyConfig) } as BrainyConfig)
@ -176,13 +176,13 @@ async function main() {
await b.flush() await b.flush()
const beforeBytes = allocBytes(`${dir}/_generations`) const beforeBytes = allocBytes(`${dir}/_generations`)
const t = process.hrtime.bigint() 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 const compMs = Number(process.hrtime.bigint() - t) / 1e6
await b.flush() await b.flush()
const afterBytes = allocBytes(`${dir}/_generations`) const afterBytes = allocBytes(`${dir}/_generations`)
await b.close() await b.close()
rmSync(dir, { recursive: true, force: true }) 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`) 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 }) results.compaction.push({ N, compactMs: compMs, removed: (res as any)?.removedGenerations, beforeBytes, afterBytes })
} }

View file

@ -1118,12 +1118,17 @@ describe('8.0 Db API — generational MVCC', () => {
await db.release() 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() 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' }) 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( const first = await brain.transact(
[{ op: 'add', id: uid('txlog-a'), type: NounType.Document, data: 'a', vector: vec(100), metadata: {} }], [{ 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 third = await brain.transact([{ op: 'update', id: uid('txlog-a'), metadata: { v: 3 } }])
const entries = await brain.transactionLog() 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([ expect(entries.map((entry) => entry.generation)).toEqual([
third.generation, third.generation,
second.generation, second.generation,
first.generation first.generation,
soloGen
]) ])
expect(entries[1].meta).toEqual({ author: 'job-2' }) expect(entries[1].meta).toEqual({ author: 'job-2' })
expect(entries[2].meta).toEqual({ author: 'job-1' }) 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) { for (const entry of entries) {
expect(entry.timestamp).toBeGreaterThan(0) expect(entry.timestamp).toBeGreaterThan(0)
} }
@ -1156,6 +1164,82 @@ describe('8.0 Db API — generational MVCC', () => {
await third.release() 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) // 9. asOf() / released-Db error paths (Y.15 spot-checks)
// ========================================================================== // ==========================================================================

View file

@ -330,8 +330,8 @@ describe('8.0 Db API — temporal range verbs', () => {
await dbAtG2.release() await dbAtG2.release()
}) })
// 7. Granularity ------------------------------------------------------------- // 7. Granularity (Model-B) ---------------------------------------------------
it('granularity: single-operation writes (outside transact) are invisible to the temporal verbs', async () => { it('granularity: single-operation writes ARE versioned and visible to the temporal verbs', async () => {
const brain = await openMemoryBrain() const brain = await openMemoryBrain()
const a = uid('gran-a') const a = uid('gran-a')
const r1 = await brain.transact([ const r1 = await brain.transact([
@ -340,13 +340,26 @@ describe('8.0 Db API — temporal range verbs', () => {
await r1.release() await r1.release()
expect((await brain.transactionLog()).length).toBe(1) 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 } }) 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()) 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.added.nouns).toEqual([])
expect(d.modified.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 --------------------------------------------- // 8. Compaction policy contrast ---------------------------------------------
@ -363,7 +376,7 @@ describe('8.0 Db API — temporal range verbs', () => {
).release() ).release()
} // gens 1..6, all pins released } // 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() const horizon = generationStoreOf(brain).horizon()
expect(horizon).toBeGreaterThan(0) expect(horizon).toBeGreaterThan(0)

View file

@ -99,7 +99,7 @@ describe('generation history chains (Model-B scalability)', () => {
await bumpX(4) await bumpX(4)
// Reclaim everything except the 2 most recent committed generations. // 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) expect(res.removedGenerations).toBeGreaterThan(0)
// A pin ABOVE the new horizon still resolves (chains were rebuilt from the survivors)… // A pin ABOVE the new horizon still resolves (chains were rebuilt from the survivors)…

View file

@ -8,7 +8,7 @@
* resolution, and crash rollback of uncommitted generations. * 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 { MemoryStorage } from '../../../src/storage/adapters/memoryStorage.js'
import { import {
GenerationStore, GenerationStore,
@ -253,12 +253,12 @@ describe('db/GenerationStore', () => {
expect(remaining).toEqual([]) 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, 1)
await commitWrite(ID_A, 2) await commitWrite(ID_A, 2)
await commitWrite(ID_A, 3) 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.removedGenerations).toBe(1)
expect(result.horizon).toBe(1) expect(result.horizon).toBe(1)
@ -330,4 +330,141 @@ describe('db/GenerationStore', () => {
expect(manifest.generation).toBe(2) 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<number> {
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<void> {
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)
})
})
}) })