feat(8.0): full query surface at historical generations via ephemeral index materialization

Historical Db values (now()/asOf() pins that history has moved past) now
serve the COMPLETE query surface - vector/hybrid search, graph traversal,
cursor pagination, and aggregation - by materializing ephemeral in-memory
indexes over the exact at-generation record set. The historical-query
throw is gone; NotYetSupportedAtHistoricalGenerationError is deleted.

Materializer (Brainy.materializeAtGeneration):
- Copies the at-G record set (live bytes for ids untouched since the pin,
  immutable before-images otherwise) into a fresh MemoryStorage; a final
  reconciliation pass under the commit mutex makes the copy exact even
  when transactions commit mid-build.
- Opens a read-only Brainy over the copy: init rebuilds the metadata and
  graph-adjacency indexes from the records; the vector index is built by
  inserting every at-G vector (the at-G HNSW graph never existed on disk,
  so there is nothing to restore). Host embedder and aggregate definitions
  are shared - no second model load, aggregates backfill at-G values.
- Cost is the documented contract: O(n at G) time and memory, ONCE per Db
  (handle cached; freed by release(), with a FinalizationRegistry backstop
  that also closes leaked readers). A native VersionedIndexProvider serves
  the same reads from retained segments with no rebuild.

Db routing (src/db/db.ts): metadata-level find()/related() keep the free
record path; index-only dimensions (query/vector/near/connected/cursor/
aggregate/includeRelations/non-metadata modes) route to the cached
materialization; unsupported where-operators on the record path re-route
there too instead of erroring. Speculative with() overlays keep the one
honest boundary - SpeculativeOverlayError (overlay entities carry no
embeddings, so index reads over them would be silently incomplete);
metadata find()/get()/filter related() work on overlays.

UpdateParams.vector contract now honored: an explicit pre-computed vector
applies directly (with dimension validation) in update() and transact
update ops, re-indexing HNSW - previously it was silently ignored unless
data also changed.

GraphAdjacencyIndex: adjacency now derives from the two verb-id LSM trees
filtered through the live-verb tombstone set (entity->entity edge trees
deleted - they carried no verb ids, so removeVerb could never tombstone
them and traversal served stale neighbors forever). Neighbor reads batch-
load live verbs via the unified cache; addVerb seeds the cache.

Proofs (tests/integration/db-mvcc.test.ts, 24 green): historical vector
search finds old vector placement including since-deleted entities;
historical graph traversal walks the old wiring after a rewire; historical
aggregation computes at-G group values; asOf() pins get the same surface;
the materialization builds once per Db and release() closes the ephemeral
reader (it refuses reads afterwards); overlays throw the documented error.
ADR-001 updated to the no-throws historical model.
This commit is contained in:
David Snelling 2026-06-11 08:12:11 -07:00
parent 8f93add705
commit e5feae4104
11 changed files with 869 additions and 340 deletions

View file

@ -106,12 +106,10 @@ import { AggregateMaterializer } from './aggregation/materializer.js'
import type { AggregateDefinition, AggregateQueryParams, AggregateResult } from './types/brainy.types.js'
import { resolveJsHnswConfig } from './utils/recallPreset.js'
import * as fs from 'node:fs'
import { Db, type DbHost } from './db/db.js'
import { Db, type DbHost, type HistoricalQueryHandle } from './db/db.js'
import { GenerationStore } from './db/generationStore.js'
import {
GenerationConflictError,
NotYetSupportedAtHistoricalGenerationError
} from './db/errors.js'
import { GenerationConflictError } from './db/errors.js'
import { MemoryStorage } from './storage/adapters/memoryStorage.js'
import type {
CompactHistoryOptions,
CompactHistoryResult,
@ -478,10 +476,19 @@ export class Brainy<T = any> implements BrainyInterface<T> {
// Apply any init-time configuration overrides
if (overrides) {
const { dimensions, ...configOverrides } = overrides
// Storage configs shallow-merge; a pre-constructed adapter instance
// (on either side) is taken whole — spreading an instance would strip
// its prototype methods.
const baseStorage = this.config.storage
const overrideStorage = configOverrides.storage
const mergedStorage =
isStorageAdapterInstance(baseStorage) || isStorageAdapterInstance(overrideStorage)
? (overrideStorage ?? baseStorage)
: { ...baseStorage, ...overrideStorage }
this.config = {
...this.config,
...configOverrides,
storage: { ...this.config.storage, ...configOverrides.storage },
storage: mergedStorage,
index: { ...this.config.index, ...configOverrides.index },
verbose: configOverrides.verbose ?? this.config.verbose,
silent: configOverrides.silent ?? this.config.silent
@ -1688,13 +1695,22 @@ export class Brainy<T = any> implements BrainyInterface<T> {
throw new RevisionConflictError(params.id, params.ifRev, currentRev)
}
// Update vector if data changed
// Resolve the updated vector: an explicit `vector` always wins (the
// UpdateParams contract — a new pre-computed vector, with or without
// new `data`); otherwise new `data` re-embeds; otherwise the existing
// vector is kept. Any vector change re-indexes HNSW below.
let vector = existing.vector
const newType = params.type || existing.type
const needsReindexing = params.data || params.type
if (params.data) {
vector = params.vector || (await this.embed(params.data))
if (params.vector) {
if (this.dimensions && params.vector.length !== this.dimensions) {
throw new Error(
`Vector dimension mismatch: expected ${this.dimensions}, got ${params.vector.length}`
)
}
vector = params.vector
} else if (params.data) {
vector = await this.embed(params.data)
}
const needsReindexing = Boolean(params.data || params.type || params.vector)
// Always update the noun with new metadata
const newMetadata = params.merge !== false
@ -4508,12 +4524,15 @@ export class Brainy<T = any> implements BrainyInterface<T> {
/**
* @description Pin the CURRENT generation and return an immutable `Db`
* view of it O(1), no I/O. The returned view keeps reading exactly this
* state no matter what commits afterwards: `get()` and metadata-level
* `find()`/`related()` stay correct at the pinned generation forever
* (resolved from immutable generation records), while index-accelerated
* queries throw the documented historical-query error once history moves
* past the pin.
* view of it O(1), no I/O. The returned view keeps serving the FULL
* query surface at exactly this state no matter what commits afterwards:
* `get()` and metadata-level `find()`/`related()` resolve from immutable
* generation records at no extra cost, while index-accelerated queries
* (semantic/vector search, graph traversal, cursors, aggregation) are
* served by an at-generation index materialization built lazily on first
* use O(n at G) time and memory once per `Db`, freed on `release()`
* (a native `VersionedIndexProvider` serves the same reads from retained
* segments without rebuild).
*
* Call `db.release()` when done pins gate `compactHistory()`. A
* `FinalizationRegistry` backstop releases leaked pins at GC time, but
@ -4857,6 +4876,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
relationFromRecord: (id, record) => this.relationFromGenerationRecord(id, record),
persistPinned: (targetPath, generation) =>
this.persistPinnedGeneration(targetPath, generation),
materializeAt: (generation) => this.materializeAtGeneration(generation),
pinGeneration: (generation) => this.pinGeneration(generation),
releaseGeneration: (generation) => this.releaseGeneration(generation),
registerDbForFinalization: (db, generation, closeOnRelease) => {
@ -4904,10 +4924,11 @@ export class Brainy<T = any> implements BrainyInterface<T> {
* compaction) plus a `pin()` on every versioned index provider the
* explicit pin lifetime overrides any time-based snapshot retention the
* provider has. Provider visibility (`isGenerationVisible`) is evaluated
* at pin time per the locked read-routing rule; provider-accelerated
* at-generation reads engage with the asOf-rebuild follow-up until then
* historical reads always resolve from canonical generation records
* (strictly correct, provider-independent).
* at pin time per the locked read-routing rule. Without a versioned
* provider, historical reads resolve from canonical generation records
* and the at-generation index materialization (strictly correct,
* provider-independent); a provider that retains the pinned generation
* serves the same reads from its segments without rebuild.
*/
private pinGeneration(generation: number): void {
this.generationStore.pin(generation)
@ -5015,26 +5036,174 @@ export class Brainy<T = any> implements BrainyInterface<T> {
* compaction, and no counter write can interleave with the hard-link
* walk, and the counter is durably persisted into the snapshot.
*
* @throws NotYetSupportedAtHistoricalGenerationError when `generation` is
* no longer the store's latest a snapshot captures current bytes, so
* persist the view before further writes (pin with `brain.now()`,
* persist, then mutate).
* @throws GenerationConflictError when `generation` is no longer the
* store's latest a snapshot captures current bytes, so persist the
* view before further writes (pin with `brain.now()`, persist, then
* mutate).
*/
private async persistPinnedGeneration(targetPath: string, generation: number): Promise<void> {
await this.ensureInitialized()
await this.flush()
await this.generationStore.snapshotWith(async () => {
if (generation !== this.generationStore.generation()) {
throw new NotYetSupportedAtHistoricalGenerationError(
'persist of a historical generation',
generation,
this.generationStore.generation()
)
throw new GenerationConflictError(generation, this.generationStore.generation())
}
await this.storage.snapshotToDirectory(targetPath)
})
}
/**
* Build the at-`generation` index materialization the brain-side half of
* the historical full-query surface (`Db.find`/`Db.search`/`Db.related` at
* past pinned generations; see `src/db/db.ts` and
* `docs/ADR-001-generational-mvcc.md`):
*
* 1. Flush, then enumerate every live entity/relationship id plus every id
* touched by transactions committed after `generation`.
* 2. Resolve each id AT `generation` (live bytes when untouched since the
* pin; immutable before-images otherwise) and copy the raw stored
* objects into a fresh in-memory storage adapter.
* 3. A final reconciliation pass under the store's commit mutex
* re-resolves ids touched by transactions that committed DURING the
* copy, so the materialized set is exactly the at-`generation` record
* set. (Single-operation writes racing the copy follow the documented
* history granularity they remain visible through earlier pins.)
* 4. Open a read-only `Brainy` over that storage: its init reconciliation
* rebuilds the metadata and graph-adjacency indexes by scanning the
* copied records, and the materializer then builds the
* `JsHnswVectorIndex` by inserting every copied at-`generation` vector
* (the persisted-HNSW restore path has nothing to restore the
* at-generation graph never existed on disk). The host's embedder is
* shared so semantic queries embed through the already-loaded model,
* and the host's aggregate definitions are re-registered so
* `find({ aggregate })` computes at-generation values via
* backfill-on-define.
*
* COST document-grade contract: O(n at G) time and memory, ONCE per
* `Db` (the `Db` caches the handle; `db.release()` frees it). This is the
* open-core price of historical index queries. A native
* `VersionedIndexProvider` serves the same reads from its retained
* segments without any rebuild when one is registered, it accelerates
* these queries instead.
*/
private async materializeAtGeneration(generation: number): Promise<HistoricalQueryHandle<T>> {
await this.ensureInitialized()
await this.flush()
const snapshotStorage = new MemoryStorage()
await snapshotStorage.init()
/** Copy one id's at-`generation` state into the snapshot (or ensure absence). */
const copyAt = async (kind: 'noun' | 'verb', id: string): Promise<void> => {
const resolved = await this.generationStore.resolveAt(kind, id, generation)
let record: { metadata: any; vector: any | null } | null = null
if (resolved.source === 'record') {
if (resolved.metadata !== null) {
record = { metadata: resolved.metadata, vector: resolved.vector }
}
} else if (resolved.source === 'current') {
const raw =
kind === 'noun' ? await this.storage.readNounRaw(id) : await this.storage.readVerbRaw(id)
if (raw.metadata !== null) {
record = raw
}
}
// `record === null` — absent at this generation (or a metadata-less
// stored state, which is not a live entity): writing nulls ensures the
// id is absent in the snapshot, which also un-copies ids created by
// transactions that commit mid-build.
if (kind === 'noun') {
await snapshotStorage.writeNounRaw(id, record ?? { metadata: null, vector: null })
} else {
await snapshotStorage.writeVerbRaw(id, record ?? { metadata: null, vector: null })
}
}
// Bulk copy: live ids ids touched after the pinned generation (the
// union covers entities deleted after the pin, which are no longer
// listed live but resolve from before-images).
let watermark = this.generationStore.committedGeneration()
const changed = await this.generationStore.changedBetween(generation, watermark)
const nounIds = new Set<string>(changed.nouns)
const verbIds = new Set<string>(changed.verbs)
for (const path of await this.storage.listRawObjects('entities/nouns')) {
const id = entityIdFromCanonicalPath(path)
if (id) nounIds.add(id)
}
for (const path of await this.storage.listRawObjects('entities/verbs')) {
const id = entityIdFromCanonicalPath(path)
if (id) verbIds.add(id)
}
for (const id of nounIds) await copyAt('noun', id)
for (const id of verbIds) await copyAt('verb', id)
// Reconciliation pass under the commit mutex: nothing can commit during
// this section, so afterwards the snapshot is exactly the at-`generation`
// record set even when transactions committed while the bulk copy ran.
// Reconciled ids join the enumeration sets so the vector-index build
// below sees them too.
await this.generationStore.snapshotWith(async () => {
const committedNow = this.generationStore.committedGeneration()
if (committedNow === watermark) return
const delta = await this.generationStore.changedBetween(watermark, committedNow)
for (const id of delta.nouns) {
nounIds.add(id)
await copyAt('noun', id)
}
for (const id of delta.verbs) {
verbIds.add(id)
await copyAt('verb', id)
}
watermark = committedNow
})
// Open the ephemeral reader: init's index reconciliation rebuilds the
// metadata and graph indexes by scanning the copied records.
const reader = new Brainy<T>({
storage: snapshotStorage,
mode: 'reader',
requireSubtype: false,
silent: true
})
await reader.init()
// Share the host's embedder and dimensionality — semantic queries on the
// materialization must never load a second embedding model.
reader.embedder = this.embedder
reader.dimensions = this.dimensions
// Build the reader's vector index by INSERTING the copied at-generation
// vectors. Unlike the metadata/graph indexes (which rebuild by scanning
// records), `JsHnswVectorIndex.rebuild()` only restores a previously
// persisted graph structure — and the at-generation HNSW graph never
// existed on disk, so the materializer constructs it the honest way:
// one insert per vector (the O(n log n at G) component of the documented
// materialization cost).
for (const id of nounIds) {
const noun = await snapshotStorage.getNoun(id)
if (noun && Array.isArray(noun.vector) && noun.vector.length > 0) {
await reader.index.addItem({ id: noun.id, vector: noun.vector })
}
}
// Re-register the host's aggregate definitions; backfill-on-define
// computes their at-generation values from the materialized record set
// on first query.
if (this._aggregationIndex) {
for (const def of this._aggregationIndex.getDefinitions()) {
reader.defineAggregate(def)
}
}
let closed = false
return {
find: (query) => reader.find(query),
getRelations: (paramsOrId) => reader.getRelations(paramsOrId),
close: async () => {
if (closed) return
closed = true
await reader.close()
}
}
}
// --- Transact planner ------------------------------------------------------
/**
@ -5236,11 +5405,21 @@ export class Brainy<T = any> implements BrainyInterface<T> {
throw new RevisionConflictError(params.id, params.ifRev, currentRev)
}
// Resolve the updated vector — mirror of update(): an explicit `vector`
// always wins, new `data` re-embeds, otherwise the existing vector is
// kept. Any vector change re-indexes HNSW below.
let vector = existing.vector
const needsReindexing = params.data || params.type
if (params.data) {
vector = params.vector || (await this.embed(params.data))
if (params.vector) {
if (this.dimensions && params.vector.length !== this.dimensions) {
throw new Error(
`Vector dimension mismatch: expected ${this.dimensions}, got ${params.vector.length}`
)
}
vector = params.vector
} else if (params.data) {
vector = await this.embed(params.data)
}
const needsReindexing = Boolean(params.data || params.type || params.vector)
const newMetadata =
params.merge !== false
@ -9361,18 +9540,15 @@ export class Brainy<T = any> implements BrainyInterface<T> {
* Setup storage
*/
private async setupStorage(): Promise<BaseStorage> {
const rawStorage = this.config.storage as unknown
// If the caller passed a pre-constructed storage adapter (e.g.
// `storage: new MemoryStorage()`), use it directly instead of routing
// `storage: new MemoryStorage()`, or the historical materializer's
// pre-populated snapshot storage), use it directly instead of routing
// through the factory. Otherwise the factory's `type === 'auto'` branch
// would silently create a FileSystemStorage at `./brainy-data`, ignoring
// the instance and surprising anyone who wrote `new MemoryStorage()`
// expecting it to be honoured.
if (rawStorage && typeof rawStorage === 'object' && 'init' in (rawStorage as any) &&
typeof (rawStorage as any).init === 'function' &&
typeof (rawStorage as any).saveNoun === 'function') {
return rawStorage as BaseStorage
if (isStorageAdapterInstance(this.config.storage)) {
return this.config.storage as BaseStorage
}
const storageConfig = (this.config.storage || {}) as Record<string, unknown>
@ -9492,10 +9668,17 @@ export class Brainy<T = any> implements BrainyInterface<T> {
// Validate storage configuration. Brainy 8.0 ships two adapters only —
// FileSystemStorage and MemoryStorage — per BR-BRAINY-80-STORAGE-SIMPLIFY.
// Cloud backup remains supported via operator tooling (db.persist() +
// gsutil / aws s3 cp / rclone / azcopy).
if (config?.storage?.type && !['auto', 'memory', 'filesystem'].includes(config.storage.type)) {
// gsutil / aws s3 cp / rclone / azcopy). Pre-constructed adapter
// instances bypass the type check (they ARE the storage).
const storageConfig = config?.storage
if (
storageConfig &&
!isStorageAdapterInstance(storageConfig) &&
storageConfig.type &&
!['auto', 'memory', 'filesystem'].includes(storageConfig.type)
) {
throw new Error(
`Invalid storage type: ${config.storage.type}. Brainy 8.0 ships 'auto', 'memory', and 'filesystem' only. ` +
`Invalid storage type: ${storageConfig.type}. Brainy 8.0 ships 'auto', 'memory', and 'filesystem' only. ` +
`Cloud storage adapters (GCS / S3 / R2 / Azure) and OPFS were removed in 8.0; ` +
`back up locally with db.persist() and sync the on-disk artefact with your tool of choice.`
)
@ -10154,14 +10337,8 @@ export class Brainy<T = any> implements BrainyInterface<T> {
process.env.CLUSTER_SIZE ||
process.env.KUBERNETES_SERVICE_HOST // Running in K8s
// Auto-detect based on storage type (S3/R2/GCS implies distributed)
const storageImpliesDistributed =
this.config?.storage?.type === 's3' ||
this.config?.storage?.type === 'r2' ||
this.config?.storage?.type === 'gcs'
// If not explicitly configured but environment suggests distributed
if (!config && (envEnabled || storageImpliesDistributed)) {
if (!config && envEnabled) {
return {
enabled: true,
nodeId: process.env.HOSTNAME || process.env.NODE_ID || `node-${Date.now()}`,
@ -10265,6 +10442,32 @@ export class Brainy<T = any> implements BrainyInterface<T> {
}
}
/**
* @description Extract the entity/relationship id from a canonical storage
* path of the form `entities/(nouns|verbs)/<shard>/<id>/metadata.json`.
* Vector-file and non-canonical paths return `null` the historical
* materializer enumerates each id exactly once, from its metadata file.
* @param path - A storage-root-relative object path.
* @returns The id, or `null` when the path is not a metadata file.
*/
function entityIdFromCanonicalPath(path: string): string | null {
const match = /^entities[/\\](?:nouns|verbs)[/\\][^/\\]+[/\\]([^/\\]+)[/\\]metadata\.json$/.exec(path)
return match ? match[1] : null
}
/**
* @description Narrow `BrainyConfig['storage']` to a pre-constructed storage
* adapter instance (vs. a factory config object). Instances are detected by
* their `init` method config objects are plain data and never carry one.
* @param value - The configured storage value.
* @returns Whether `value` is an adapter instance to use directly.
*/
function isStorageAdapterInstance(
value: BrainyConfig['storage']
): value is StorageAdapter {
return !!value && typeof (value as StorageAdapter).init === 'function'
}
// Re-export types for convenience
export * from './types/brainy.types.js'
export { NounType, VerbType } from './types/graphTypes.js'