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

@ -123,17 +123,31 @@ While nothing has committed past G, *every* read on the `Db` delegates to
the live fast paths untouched — `now()` adds no read overhead until history the live fast paths untouched — `now()` adds no read overhead until history
actually moves. actually moves.
**Honesty boundary.** `get()`, metadata-level `find()`, and `related()` are **Two read paths, one result set.** `get()`, metadata-level `find()`, and
fully correct at any reachable pinned generation. Index-accelerated filter-based `related()` resolve directly through the record layer at any
dimensions (semantic/vector search, graph traversal, cursors, aggregation) reachable pinned generation — no extra cost beyond scanning the deltas of
are served by live indexes that represent only the current generation, so later commits. Index-accelerated dimensions (semantic/vector search, graph
at a historical generation they throw traversal, cursors, aggregation) are served by **at-generation index
`NotYetSupportedAtHistoricalGenerationError` — never silently-wrong materialization**: the first such query on a historical `Db` copies the
results. The escape hatch is `persist()` + `Brainy.load()`, which rebuilds exact at-G record set (live bytes for ids untouched since the pin,
indexes from the snapshot and supports the full query surface. Rebuilding before-images for the rest; a final reconciliation pass runs under the
indexes from a pinned record set in-place (the standard LSM answer) is the commit mutex so transactions racing the copy cannot skew it) into an
documented follow-up; versioned native providers already expose ephemeral in-memory store and opens a read-only engine over it — the same
`isGenerationVisible()` for provider-accelerated historical reads. vector/metadata/graph index classes the live brain uses, sharing the host's
embedder and aggregate definitions. The handle is cached on the `Db` and
freed by `release()`.
**Cost, stated plainly:** materialization is O(n at G) time and memory,
once per `Db`. That is the open-core price of historical index queries. A
native `VersionedIndexProvider` (`isGenerationVisible()` + pins over
retained LSM segments) serves the same reads with no rebuild at all — the
materializer is the correctness baseline, the provider is the accelerator.
**The one remaining boundary.** Speculative `with()` overlays throw
`SpeculativeOverlayError` for index-accelerated queries and `persist()`:
overlay entities carry no embeddings (`with()` never invokes the embedder),
so a "full" index query over an overlay would silently exclude the
overlay's own entities. Commit with `transact()` to get the full surface.
**History granularity.** Generation *records* are written per `transact()` **History granularity.** Generation *records* are written per `transact()`
batch only. Single-operation writes advance the counter (so watermarks and batch only. Single-operation writes advance the counter (so watermarks and
@ -230,7 +244,8 @@ only counter values that nothing durable ever referenced.
| Crash after manifest rename, before tx-log append | Transaction kept (committed); tx-log misses one advisory line; `asOf(Date)` resolution for that commit falls back to neighboring entries. | | Crash after manifest rename, before tx-log append | Transaction kept (committed); tx-log misses one advisory line; `asOf(Date)` resolution for that commit falls back to neighboring entries. |
| Batch fails mid-execution (no crash) | TransactionManager operation rollback + staging-directory removal + reservation return; generation unchanged. | | Batch fails mid-execution (no crash) | TransactionManager operation rollback + staging-directory removal + reservation return; generation unchanged. |
| `asOf()` below the compaction horizon | `GenerationCompactedError` — explicit, never partial data. | | `asOf()` below the compaction horizon | `GenerationCompactedError` — explicit, never partial data. |
| Index-accelerated query at a historical pin | `NotYetSupportedAtHistoricalGenerationError` — explicit, never silently-wrong results. | | Index-accelerated query on a `with()` overlay | `SpeculativeOverlayError` — explicit, never silently-incomplete results (overlay entities carry no embeddings). |
| `persist()` of a view history has moved past | `GenerationConflictError` — a snapshot captures current bytes; persist before further writes. |
| Torn trailing tx-log line (crashed append) | Tolerated; unparseable lines are skipped by readers. | | Torn trailing tx-log line (crashed append) | Tolerated; unparseable lines are skipped by readers. |
## Lineage ## Lineage
@ -257,6 +272,8 @@ boring where it counts:
- Every commit pays O(ids touched) extra writes (before-images + delta + - Every commit pays O(ids touched) extra writes (before-images + delta +
manifest). Single-operation writes pay only an in-memory counter bump manifest). Single-operation writes pay only an in-memory counter bump
with coalesced persistence. with coalesced persistence.
- Historical index-accelerated queries are an explicit error until the - The full query surface works at every reachable pinned generation.
asOf-rebuild ships; correctness-critical historical reads (`get`, Record-path reads (`get`, metadata `find`, filter `related`) are
metadata `find`, `related`) work today and forever at any reachable pin. effectively free; index-accelerated historical queries pay a one-time
O(n at G) materialization per `Db` on the open-core path (freed on
`release()`), and run rebuild-free on a native `VersionedIndexProvider`.

View file

@ -12,7 +12,7 @@ Brainy has **3 main indexes** at the top level, each with multiple sub-indexes m
|-------|---------|----------------|------------|---------------|------------------| |-------|---------|----------------|------------|---------------|------------------|
| **TypeAwareVectorIndex** | Type-aware vector similarity search | 42 type-specific hierarchical graphs | O(log n) search | `src/hnsw/typeAwareHNSWIndex.ts` | ✅ Line 403 | | **TypeAwareVectorIndex** | Type-aware vector similarity search | 42 type-specific hierarchical graphs | O(log n) search | `src/hnsw/typeAwareHNSWIndex.ts` | ✅ Line 403 |
| **MetadataIndexManager** | Fast metadata filtering | Chunked sparse indices with bloom filters + zone maps + roaring bitmaps | O(1) exact, O(log n) ranges | `src/utils/metadataIndex.ts` | ✅ Line 2318 | | **MetadataIndexManager** | Fast metadata filtering | Chunked sparse indices with bloom filters + zone maps + roaring bitmaps | O(1) exact, O(log n) ranges | `src/utils/metadataIndex.ts` | ✅ Line 2318 |
| **GraphAdjacencyIndex** | Relationship traversal | 4 LSM-trees + bidirectional adjacency maps | O(1) per hop | `src/graph/graphAdjacencyIndex.ts` | ✅ Line 389 | | **GraphAdjacencyIndex** | Relationship traversal | 2 verb-id LSM-trees + tombstone-filtered adjacency derivation | O(degree) per hop | `src/graph/graphAdjacencyIndex.ts` | ✅ Line 389 |
### Sub-Indexes (Level 2) ### Sub-Indexes (Level 2)
@ -847,7 +847,7 @@ Where:
|-------|-------------------|-------| |-------|-------------------|-------|
| **MetadataIndexManager** | ~100 bytes | Depends on field count and cardinality (RoaringBitmap32 compression) | | **MetadataIndexManager** | ~100 bytes | Depends on field count and cardinality (RoaringBitmap32 compression) |
| **TypeAwareVectorIndex** | ~1.5 KB | Vector (384 dims × 4 bytes) + graph connections across 42 type-specific indexes | | **TypeAwareVectorIndex** | ~1.5 KB | Vector (384 dims × 4 bytes) + graph connections across 42 type-specific indexes |
| **GraphAdjacencyIndex** | ~50 bytes per relationship | Bidirectional references + metadata in 4 LSM-trees | | **GraphAdjacencyIndex** | ~50 bytes per relationship | Bidirectional verb-id references in 2 LSM-trees |
**Total overhead**: ~1.6 KB per entity + ~50 bytes per relationship **Total overhead**: ~1.6 KB per entity + ~50 bytes per relationship
@ -932,7 +932,7 @@ All have rebuild() methods and are covered by lazy loading:
Automatically managed by parent rebuild(): Automatically managed by parent rebuild():
- **42 type-specific vector indexes** (one per NounType) - **42 type-specific vector indexes** (one per NounType)
- **6 metadata components** (ChunkManager, EntityIdMapper, FieldTypeInference, Field Sparse Indexes, Sorted Indexes) - **6 metadata components** (ChunkManager, EntityIdMapper, FieldTypeInference, Field Sparse Indexes, Sorted Indexes)
- **4 LSM-trees** (lsmTreeSource, lsmTreeTarget, lsmTreeVerbsBySource, lsmTreeVerbsByTarget) - **2 LSM-trees** (lsmTreeVerbsBySource, lsmTreeVerbsByTarget — the verb set is the single adjacency source of truth; neighbor reads derive from live verbs so removals are honored)
- **In-memory graph structures** (sourceIndex, targetIndex, verbIndex) - **In-memory graph structures** (sourceIndex, targetIndex, verbIndex)
### Lazy Loading ### Lazy Loading

View file

@ -106,12 +106,10 @@ 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 } from './utils/recallPreset.js' import { resolveJsHnswConfig } from './utils/recallPreset.js'
import * as fs from 'node:fs' 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 { GenerationStore } from './db/generationStore.js'
import { import { GenerationConflictError } from './db/errors.js'
GenerationConflictError, import { MemoryStorage } from './storage/adapters/memoryStorage.js'
NotYetSupportedAtHistoricalGenerationError
} from './db/errors.js'
import type { import type {
CompactHistoryOptions, CompactHistoryOptions,
CompactHistoryResult, CompactHistoryResult,
@ -478,10 +476,19 @@ export class Brainy<T = any> implements BrainyInterface<T> {
// Apply any init-time configuration overrides // Apply any init-time configuration overrides
if (overrides) { if (overrides) {
const { dimensions, ...configOverrides } = 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 = {
...this.config, ...this.config,
...configOverrides, ...configOverrides,
storage: { ...this.config.storage, ...configOverrides.storage }, storage: mergedStorage,
index: { ...this.config.index, ...configOverrides.index }, index: { ...this.config.index, ...configOverrides.index },
verbose: configOverrides.verbose ?? this.config.verbose, verbose: configOverrides.verbose ?? this.config.verbose,
silent: configOverrides.silent ?? this.config.silent 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) 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 let vector = existing.vector
const newType = params.type || existing.type if (params.vector) {
const needsReindexing = params.data || params.type if (this.dimensions && params.vector.length !== this.dimensions) {
if (params.data) { throw new Error(
vector = params.vector || (await this.embed(params.data)) `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 // Always update the noun with new metadata
const newMetadata = params.merge !== false 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` * @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 * view of it O(1), no I/O. The returned view keeps serving the FULL
* state no matter what commits afterwards: `get()` and metadata-level * query surface at exactly this state no matter what commits afterwards:
* `find()`/`related()` stay correct at the pinned generation forever * `get()` and metadata-level `find()`/`related()` resolve from immutable
* (resolved from immutable generation records), while index-accelerated * generation records at no extra cost, while index-accelerated queries
* queries throw the documented historical-query error once history moves * (semantic/vector search, graph traversal, cursors, aggregation) are
* past the pin. * 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 * Call `db.release()` when done pins gate `compactHistory()`. A
* `FinalizationRegistry` backstop releases leaked pins at GC time, but * `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), relationFromRecord: (id, record) => this.relationFromGenerationRecord(id, record),
persistPinned: (targetPath, generation) => persistPinned: (targetPath, generation) =>
this.persistPinnedGeneration(targetPath, generation), this.persistPinnedGeneration(targetPath, generation),
materializeAt: (generation) => this.materializeAtGeneration(generation),
pinGeneration: (generation) => this.pinGeneration(generation), pinGeneration: (generation) => this.pinGeneration(generation),
releaseGeneration: (generation) => this.releaseGeneration(generation), releaseGeneration: (generation) => this.releaseGeneration(generation),
registerDbForFinalization: (db, generation, closeOnRelease) => { 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 * compaction) plus a `pin()` on every versioned index provider the
* explicit pin lifetime overrides any time-based snapshot retention the * explicit pin lifetime overrides any time-based snapshot retention the
* provider has. Provider visibility (`isGenerationVisible`) is evaluated * provider has. Provider visibility (`isGenerationVisible`) is evaluated
* at pin time per the locked read-routing rule; provider-accelerated * at pin time per the locked read-routing rule. Without a versioned
* at-generation reads engage with the asOf-rebuild follow-up until then * provider, historical reads resolve from canonical generation records
* historical reads always resolve from canonical generation records * and the at-generation index materialization (strictly correct,
* (strictly correct, provider-independent). * provider-independent); a provider that retains the pinned generation
* serves the same reads from its segments without rebuild.
*/ */
private pinGeneration(generation: number): void { private pinGeneration(generation: number): void {
this.generationStore.pin(generation) 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 * compaction, and no counter write can interleave with the hard-link
* walk, and the counter is durably persisted into the snapshot. * walk, and the counter is durably persisted into the snapshot.
* *
* @throws NotYetSupportedAtHistoricalGenerationError when `generation` is * @throws GenerationConflictError when `generation` is no longer the
* no longer the store's latest a snapshot captures current bytes, so * store's latest a snapshot captures current bytes, so persist the
* persist the view before further writes (pin with `brain.now()`, * view before further writes (pin with `brain.now()`, persist, then
* persist, then mutate). * mutate).
*/ */
private async persistPinnedGeneration(targetPath: string, generation: number): Promise<void> { private async persistPinnedGeneration(targetPath: string, generation: number): Promise<void> {
await this.ensureInitialized() await this.ensureInitialized()
await this.flush() await this.flush()
await this.generationStore.snapshotWith(async () => { await this.generationStore.snapshotWith(async () => {
if (generation !== this.generationStore.generation()) { if (generation !== this.generationStore.generation()) {
throw new NotYetSupportedAtHistoricalGenerationError( throw new GenerationConflictError(generation, this.generationStore.generation())
'persist of a historical generation',
generation,
this.generationStore.generation()
)
} }
await this.storage.snapshotToDirectory(targetPath) 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 ------------------------------------------------------ // --- Transact planner ------------------------------------------------------
/** /**
@ -5236,11 +5405,21 @@ export class Brainy<T = any> implements BrainyInterface<T> {
throw new RevisionConflictError(params.id, params.ifRev, currentRev) 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 let vector = existing.vector
const needsReindexing = params.data || params.type if (params.vector) {
if (params.data) { if (this.dimensions && params.vector.length !== this.dimensions) {
vector = params.vector || (await this.embed(params.data)) 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 = const newMetadata =
params.merge !== false params.merge !== false
@ -9361,18 +9540,15 @@ export class Brainy<T = any> implements BrainyInterface<T> {
* Setup storage * Setup storage
*/ */
private async setupStorage(): Promise<BaseStorage> { private async setupStorage(): Promise<BaseStorage> {
const rawStorage = this.config.storage as unknown
// If the caller passed a pre-constructed storage adapter (e.g. // 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 // through the factory. Otherwise the factory's `type === 'auto'` branch
// would silently create a FileSystemStorage at `./brainy-data`, ignoring // would silently create a FileSystemStorage at `./brainy-data`, ignoring
// the instance and surprising anyone who wrote `new MemoryStorage()` // the instance and surprising anyone who wrote `new MemoryStorage()`
// expecting it to be honoured. // expecting it to be honoured.
if (rawStorage && typeof rawStorage === 'object' && 'init' in (rawStorage as any) && if (isStorageAdapterInstance(this.config.storage)) {
typeof (rawStorage as any).init === 'function' && return this.config.storage as BaseStorage
typeof (rawStorage as any).saveNoun === 'function') {
return rawStorage as BaseStorage
} }
const storageConfig = (this.config.storage || {}) as Record<string, unknown> 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 — // Validate storage configuration. Brainy 8.0 ships two adapters only —
// FileSystemStorage and MemoryStorage — per BR-BRAINY-80-STORAGE-SIMPLIFY. // FileSystemStorage and MemoryStorage — per BR-BRAINY-80-STORAGE-SIMPLIFY.
// Cloud backup remains supported via operator tooling (db.persist() + // Cloud backup remains supported via operator tooling (db.persist() +
// gsutil / aws s3 cp / rclone / azcopy). // gsutil / aws s3 cp / rclone / azcopy). Pre-constructed adapter
if (config?.storage?.type && !['auto', 'memory', 'filesystem'].includes(config.storage.type)) { // 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( 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; ` + `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.` `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.CLUSTER_SIZE ||
process.env.KUBERNETES_SERVICE_HOST // Running in K8s 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 not explicitly configured but environment suggests distributed
if (!config && (envEnabled || storageImpliesDistributed)) { if (!config && envEnabled) {
return { return {
enabled: true, enabled: true,
nodeId: process.env.HOSTNAME || process.env.NODE_ID || `node-${Date.now()}`, 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 // Re-export types for convenience
export * from './types/brainy.types.js' export * from './types/brainy.types.js'
export { NounType, VerbType } from './types/graphTypes.js' export { NounType, VerbType } from './types/graphTypes.js'

View file

@ -12,13 +12,27 @@
* **Read semantics.** While no transaction has committed past the pinned * **Read semantics.** While no transaction has committed past the pinned
* generation, every read delegates straight to the live brain the pinned * generation, every read delegates straight to the live brain the pinned
* view IS the current state, and the existing fast paths serve it untouched. * view IS the current state, and the existing fast paths serve it untouched.
* Once later transactions commit, the `Db` resolves reads through the * Once later transactions commit, the `Db` serves the FULL query surface at
* generational record layer: `get()` and metadata-level `find()`/`related()` * the pinned generation through two complementary paths:
* remain fully correct at the pinned generation (changed ids are resolved *
* from immutable before-images; unchanged ids still ride the live fast * - `get()`, metadata-level `find()`, and filter-based `related()` resolve
* path), while index-accelerated queries (semantic/vector search) throw the * through the generational record layer (changed ids from immutable
* documented {@link NotYetSupportedAtHistoricalGenerationError} an honest * before-images; unchanged ids still ride the live fast path) no
* boundary, never silently-wrong results. * materialization cost.
* - Index-accelerated queries (semantic/vector search, graph traversal,
* cursors, aggregation) are served by **at-generation index
* materialization**: on first use, the host rebuilds ephemeral in-memory
* indexes (the same vector/metadata/graph index classes the live brain
* uses) over the exact at-generation record set, caches them on this `Db`,
* and frees them on {@link Db.release}. This costs O(n at G) time and
* memory ONCE per `Db` the open-core price of historical index queries;
* a native {@link ../plugin.js VersionedIndexProvider} serves the same
* reads from its retained segments without any rebuild.
*
* Speculative `with()` overlays keep one honest boundary: overlay entities
* carry no embeddings, so index-accelerated queries on overlays throw
* {@link SpeculativeOverlayError} instead of returning silently-incomplete
* results.
* *
* **History granularity.** Generation records are written per `transact()` * **History granularity.** Generation records are written per `transact()`
* batch. Single-operation writes (`add`/`update`/`relate`/ outside * batch. Single-operation writes (`add`/`update`/`relate`/ outside
@ -43,11 +57,27 @@ import type {
Result Result
} from '../types/brainy.types.js' } from '../types/brainy.types.js'
import { v4 as uuidv4 } from '../universal/uuid.js' import { v4 as uuidv4 } from '../universal/uuid.js'
import { NotYetSupportedAtHistoricalGenerationError } from './errors.js' import { SpeculativeOverlayError } from './errors.js'
import type { GenerationStore } from './generationStore.js' import type { GenerationStore } from './generationStore.js'
import type { ChangedIds, TransactReceipt, TxOperation } from './types.js' import type { ChangedIds, TransactReceipt, TxOperation } from './types.js'
import { entityMatchesFind, resolveEntityField, UnsupportedWhereOperatorError } from './whereMatcher.js' import { entityMatchesFind, resolveEntityField, UnsupportedWhereOperatorError } from './whereMatcher.js'
/**
* @description The query surface a historical materialization exposes back
* to its `Db`: the full live read pipeline of an ephemeral reader brain
* whose storage holds the exact at-generation record set. Produced by
* {@link DbHost.materializeAt}; closed (indexes freed) via `close()` when
* the owning `Db` is released.
*/
export interface HistoricalQueryHandle<T = any> {
/** Full `find()` surface (vector/semantic/graph/cursor/aggregate) at the pinned generation. */
find(query: string | FindParams<T>): Promise<Result<T>[]>
/** Full `getRelations()` surface (including cursor pagination) at the pinned generation. */
getRelations(paramsOrId?: string | GetRelationsParams): Promise<Relation<T>[]>
/** Free the materialized indexes (closes the ephemeral reader). Idempotent. */
close(): Promise<void>
}
/** /**
* @description The speculative overlay carried by a `db.with()` value: * @description The speculative overlay carried by a `db.with()` value:
* per-id replacement entities/relations (`null` = tombstone). Reads on the * per-id replacement entities/relations (`null` = tombstone). Reads on the
@ -86,6 +116,13 @@ export interface DbHost<T = any> {
relationFromRecord(id: string, record: { metadata: any; vector: any | null }): Relation<T> | null relationFromRecord(id: string, record: { metadata: any; vector: any | null }): Relation<T> | null
/** Snapshot the store at `generation` into `targetPath` (throws if the pin is stale). */ /** Snapshot the store at `generation` into `targetPath` (throws if the pin is stale). */
persistPinned(targetPath: string, generation: number): Promise<void> persistPinned(targetPath: string, generation: number): Promise<void>
/**
* Build the at-`generation` index materialization: an ephemeral in-memory
* reader over the exact record set at that generation, serving the full
* query surface. O(n at G) called at most once per `Db` (cached by the
* caller) and freed via the returned handle's `close()`.
*/
materializeAt(generation: number): Promise<HistoricalQueryHandle<T>>
/** Add one refcounted pin (store + versioned providers) on `generation`. */ /** Add one refcounted pin (store + versioned providers) on `generation`. */
pinGeneration(generation: number): void pinGeneration(generation: number): void
/** Release one refcounted pin (store + versioned providers) on `generation`. */ /** Release one refcounted pin (store + versioned providers) on `generation`. */
@ -134,6 +171,11 @@ export class Db<T = any> {
private readonly overlay?: SpeculativeOverlay<T> private readonly overlay?: SpeculativeOverlay<T>
private readonly closeOnRelease?: () => Promise<void> private readonly closeOnRelease?: () => Promise<void>
private isReleased = false private isReleased = false
/**
* Cached at-generation index materialization (built lazily by the first
* index-accelerated query at a historical pin; freed on release()).
*/
private materializedHandle?: Promise<HistoricalQueryHandle<T>>
/** /**
* Per-operation receipt (committed generation, timestamp, resolved ids in * Per-operation receipt (committed generation, timestamp, resolved ids in
@ -220,25 +262,31 @@ export class Db<T = any> {
} }
/** /**
* @description Metadata-level `find()` **as of this view's generation**. * @description The full `find()` surface **as of this view's generation**.
* *
* At the current generation (nothing committed past the pin, no overlay) * Routing:
* this delegates to `brain.find()` untouched the full query surface, * - **Current generation, no overlay** delegates to `brain.find()`
* including semantic search. At a historical generation, or on a * untouched (the existing fast paths).
* speculative view, only the metadata dimensions are supported (`type`, * - **Historical generation** metadata dimensions (`type`, `subtype`,
* `subtype`, `where`, `service`, `excludeVFS`, `orderBy`/`order`, * `where`, `service`, `excludeVFS`, `orderBy`/`order`,
* `limit`/`offset`): results are computed by combining the live index for * `limit`/`offset`) are computed from the generational record layer
* unchanged ids with per-entity evaluation of the same filters against * directly (no materialization cost): the live index serves unchanged
* generation records / overlay entries set-correct at the pinned * ids, and per-entity evaluation of the same filters covers
* generation. Index-only dimensions (`query`, `vector`, `near`, * record-resolved ids set-correct at the pinned generation.
* `connected`, `cursor`, `aggregate`, ) throw * Index-accelerated dimensions (`query`, `vector`, `near`, `connected`,
* {@link NotYetSupportedAtHistoricalGenerationError}. * `cursor`, `aggregate`, `includeRelations`, non-metadata search modes)
* are served by the at-generation index materialization built lazily
* on first use (O(n at G) time and memory, once per `Db`; see the module
* doc), then cached until {@link Db.release}.
* - **Speculative overlay** metadata dimensions only; index-accelerated
* dimensions throw {@link SpeculativeOverlayError} (overlay entities
* carry no embeddings see the error's documentation).
* *
* Result ordering is deterministic when `orderBy` is supplied; without it, * Result ordering is deterministic when `orderBy` is supplied; without it,
* overlay/record-resolved matches follow the live-index matches. * record-path results list live-index matches before record-resolved
* matches, while materialized results follow the live engine's ordering.
* *
* @param query - A `FindParams` object, or a string (semantic search * @param query - A `FindParams` object, or a string (semantic search).
* current-generation views only).
* @returns Matching results as of this generation (score `1.0` for * @returns Matching results as of this generation (score `1.0` for
* record-resolved metadata matches, mirroring live metadata-only finds). * record-resolved metadata matches, mirroring live metadata-only finds).
*/ */
@ -252,7 +300,12 @@ export class Db<T = any> {
return this.host.find(query) return this.host.find(query)
} }
this.assertMetadataOnlyFind(params) if (this.overlay) {
this.assertOverlayCompatibleFind(params)
} else if (findRequiresIndexes(params)) {
const materialized = await this.materialize()
return materialized.find(params)
}
const limit = params.limit ?? 10 const limit = params.limit ?? 10
const offset = params.offset ?? 0 const offset = params.offset ?? 0
@ -307,41 +360,48 @@ export class Db<T = any> {
return merged.slice(offset, offset + limit) return merged.slice(offset, offset + limit)
} catch (err) { } catch (err) {
if (err instanceof UnsupportedWhereOperatorError) { if (err instanceof UnsupportedWhereOperatorError) {
throw new NotYetSupportedAtHistoricalGenerationError( // The record-path's per-entity matcher doesn't implement this
// where-operator. On an overlay that is a hard boundary; at a
// historical generation the materialization serves it via the full
// live query engine.
if (this.overlay) {
throw new SpeculativeOverlayError(
`metadata find with where-operator '${err.operator}'`, `metadata find with where-operator '${err.operator}'`,
this.gen, this.gen
this.host.store.generation()
) )
} }
const materialized = await this.materialize()
return materialized.find(params)
}
throw err throw err
} }
} }
/** /**
* @description Semantic (vector) search **at the current generation only**. * @description Semantic (vector) search **as of this view's generation**
* Convenience for `find({ query, ...options })`. Vector indexes serve the * convenience for `find({ query, ...options })`. At the current generation
* live state; at a historical generation, or on a speculative overlay * the live vector index serves it directly; at a historical generation the
* (whose entities carry no embeddings), this throws the documented * at-generation index materialization serves it (built lazily on first
* {@link NotYetSupportedAtHistoricalGenerationError} persist the * use O(n at G) once per `Db`, freed on release; see the module doc).
* generation you need and `Brainy.load()` it for the full query surface. * Speculative overlays throw {@link SpeculativeOverlayError}: overlay
* entities carry no embeddings, so the result would silently exclude them.
* *
* @param query - Natural-language search query. * @param query - Natural-language search query.
* @param options - Additional `FindParams` (limit, filters, ). * @param options - Additional `FindParams` (limit, filters, ).
* @returns Semantic search results (current-generation views only). * @returns Semantic search results as of this generation.
* @throws NotYetSupportedAtHistoricalGenerationError at historical * @throws SpeculativeOverlayError on speculative `with()` overlays.
* generations and on speculative overlays.
*/ */
async search(query: string, options?: Omit<FindParams<T>, 'query'>): Promise<Result<T>[]> { async search(query: string, options?: Omit<FindParams<T>, 'query'>): Promise<Result<T>[]> {
this.assertUsable('search') this.assertUsable('search')
if (!this.isHistorical() && !this.overlay) { if (this.overlay) {
throw new SpeculativeOverlayError('vector search', this.gen)
}
if (!this.isHistorical()) {
return this.host.find({ ...(options ?? {}), query }) return this.host.find({ ...(options ?? {}), query })
} }
throw new NotYetSupportedAtHistoricalGenerationError( const materialized = await this.materialize()
this.overlay ? 'vector search (speculative overlay)' : 'vector search', return materialized.find({ ...(options ?? {}), query })
this.gen,
this.host.store.generation()
)
} }
/** /**
@ -351,8 +411,9 @@ export class Db<T = any> {
* resolved from generation records; overlay relations (speculative * resolved from generation records; overlay relations (speculative
* `relate`/`unrelate`) and cascade tombstones (relations touching a * `relate`/`unrelate`) and cascade tombstones (relations touching a
* speculatively deleted entity) are applied on top. Cursor pagination is * speculatively deleted entity) are applied on top. Cursor pagination is
* index-only and throws the documented historical error on * index-only: at a historical generation it is served by the
* non-current/speculative views. * at-generation index materialization (O(n at G) once per `Db`); on a
* speculative overlay it throws {@link SpeculativeOverlayError}.
* *
* @param paramsOrId - An entity id (shorthand for `{ from: id }`) or a * @param paramsOrId - An entity id (shorthand for `{ from: id }`) or a
* `GetRelationsParams` filter object. * `GetRelationsParams` filter object.
@ -370,11 +431,11 @@ export class Db<T = any> {
} }
if (params.cursor !== undefined) { if (params.cursor !== undefined) {
throw new NotYetSupportedAtHistoricalGenerationError( if (this.overlay) {
'cursor pagination on related()', throw new SpeculativeOverlayError('cursor pagination on related()', this.gen)
this.gen, }
this.host.store.generation() const materialized = await this.materialize()
) return materialized.getRelations(params)
} }
const limit = params.limit ?? 100 const limit = params.limit ?? 100
@ -449,8 +510,8 @@ export class Db<T = any> {
* (`with()` never invokes the embedder). * (`with()` never invokes the embedder).
* - `remove` cascades on the read side: relations touching the removed * - `remove` cascades on the read side: relations touching the removed
* entity disappear from `related()` without being enumerated. * entity disappear from `related()` without being enumerated.
* - `relate` deduplicates against edges visible in this view when verb * - `relate` deduplicates against the edges visible in this view (overlay
* history allows the check, mirroring `brain.relate()`. * first, then the record layer), mirroring `brain.relate()`.
* - Brain-level vocabulary/subtype enforcement does not run overlays * - Brain-level vocabulary/subtype enforcement does not run overlays
* never persist, so there is nothing to protect. * never persist, so there is nothing to protect.
* *
@ -541,8 +602,8 @@ export class Db<T = any> {
if (!from) throw new Error(`with(): source entity ${op.from} not found`) if (!from) throw new Error(`with(): source entity ${op.from} not found`)
if (!to) throw new Error(`with(): target entity ${op.to} not found`) if (!to) throw new Error(`with(): target entity ${op.to} not found`)
// Dedupe against the view (overlay first, then committed edges // Dedupe against the view (overlay first, then the edges visible
// when verb history allows the read) — mirror of relate(). // at this generation via the record layer) — mirror of relate().
let duplicate: Relation<T> | undefined let duplicate: Relation<T> | undefined
for (const relation of overlay.verbs.values()) { for (const relation of overlay.verbs.values()) {
if (relation && relation.from === op.from && relation.to === op.to && relation.type === op.type) { if (relation && relation.from === op.from && relation.to === op.to && relation.type === op.type) {
@ -551,14 +612,8 @@ export class Db<T = any> {
} }
} }
if (!duplicate) { if (!duplicate) {
try {
const existing = await this.related({ from: op.from, type: op.type }) const existing = await this.related({ from: op.from, type: op.type })
duplicate = existing.find((relation) => relation.to === op.to) duplicate = existing.find((relation) => relation.to === op.to)
} catch (err) {
if (!(err instanceof NotYetSupportedAtHistoricalGenerationError)) throw err
// Verb history can't serve the check at this generation —
// proceed with overlay-only dedupe (documented).
}
} }
if (duplicate) break if (duplicate) break
@ -653,22 +708,20 @@ export class Db<T = any> {
* `persist()` requires this view to still be the store's **latest** * `persist()` requires this view to still be the store's **latest**
* generation (snapshotting captures current bytes): pin with `brain.now()` * generation (snapshotting captures current bytes): pin with `brain.now()`
* and persist before further writes. A view that history has moved past * and persist before further writes. A view that history has moved past
* throws {@link NotYetSupportedAtHistoricalGenerationError}; speculative * throws {@link GenerationConflictError}; speculative overlays throw
* overlays cannot be persisted (commit them with `brain.transact()` first). * {@link SpeculativeOverlayError} (commit them with `brain.transact()`
* first).
* *
* @param path - Absolute directory for the snapshot (created; must be * @param path - Absolute directory for the snapshot (created; must be
* empty or absent). * empty or absent).
* @throws NotYetSupportedAtHistoricalGenerationError when this view is no * @throws GenerationConflictError when this view is no longer the latest
* longer the latest generation, or is speculative. * generation.
* @throws SpeculativeOverlayError when this view is a speculative overlay.
*/ */
async persist(path: string): Promise<void> { async persist(path: string): Promise<void> {
this.assertUsable('persist') this.assertUsable('persist')
if (this.overlay) { if (this.overlay) {
throw new NotYetSupportedAtHistoricalGenerationError( throw new SpeculativeOverlayError('persist', this.gen)
'persist of a speculative overlay',
this.gen,
this.host.store.generation()
)
} }
await this.host.persistPinned(path, this.gen) await this.host.persistPinned(path, this.gen)
} }
@ -679,17 +732,24 @@ export class Db<T = any> {
/** /**
* @description Release this view's pin (store + versioned index * @description Release this view's pin (store + versioned index
* providers). Idempotent. After release, every read throws. Views from * providers) and free its at-generation index materialization, if one was
* built. Idempotent. After release, every read throws. Views from
* `Brainy.load()` / `asOf(path)` also close their underlying read-only * `Brainy.load()` / `asOf(path)` also close their underlying read-only
* brain here. A `FinalizationRegistry` backstop releases leaked pins when * brain here. A `FinalizationRegistry` backstop releases leaked pins (and
* a `Db` is garbage-collected, but explicit release is what makes * materializations) when a `Db` is garbage-collected, but explicit
* `compactHistory()` deterministic prefer it. * release is what makes `compactHistory()` deterministic prefer it.
*/ */
async release(): Promise<void> { async release(): Promise<void> {
if (this.isReleased) return if (this.isReleased) return
this.isReleased = true this.isReleased = true
this.host.unregisterDbFromFinalization(this) this.host.unregisterDbFromFinalization(this)
this.host.releaseGeneration(this.gen) this.host.releaseGeneration(this.gen)
if (this.materializedHandle) {
// A failed materialization already surfaced to the query that
// triggered it; release() must still succeed.
const handle = await this.materializedHandle.catch(() => null)
if (handle) await handle.close()
}
if (this.closeOnRelease) { if (this.closeOnRelease) {
await this.closeOnRelease() await this.closeOnRelease()
} }
@ -704,6 +764,32 @@ export class Db<T = any> {
return this.host.store.hasCommittedAfter(this.gen) return this.host.store.hasCommittedAfter(this.gen)
} }
/**
* Build (once) and cache the at-generation index materialization. The
* first index-accelerated query at a historical pin pays the O(n at G)
* build; every later one reuses the cached handle until release(). The
* GC-backstop registration is refreshed so a leaked `Db` also closes its
* materialized reader (whose flush timers would otherwise outlive it).
*/
private materialize(): Promise<HistoricalQueryHandle<T>> {
if (!this.materializedHandle) {
this.materializedHandle = this.host.materializeAt(this.gen).then((handle) => {
this.host.unregisterDbFromFinalization(this)
this.host.registerDbForFinalization(this, this.gen, async () => {
await handle.close()
if (this.closeOnRelease) await this.closeOnRelease()
})
return handle
})
// A failed build must not poison the cache — the next query retries.
this.materializedHandle = this.materializedHandle.catch((err) => {
this.materializedHandle = undefined
throw err
})
}
return this.materializedHandle
}
/** Throw on use-after-release. */ /** Throw on use-after-release. */
private assertUsable(method: string): void { private assertUsable(method: string): void {
if (this.isReleased) { if (this.isReleased) {
@ -715,11 +801,26 @@ export class Db<T = any> {
} }
/** /**
* Reject find() dimensions that require index machinery the record layer * Reject find() dimensions that require index machinery a speculative
* cannot answer at a historical generation / for a speculative overlay. * overlay cannot answer (overlay entities carry no embeddings and are in
* no index see {@link SpeculativeOverlayError}).
*/ */
private assertMetadataOnlyFind(params: FindParams<T>): void { private assertOverlayCompatibleFind(params: FindParams<T>): void {
const unsupported: Array<[string, boolean]> = [ for (const [capability, present] of indexOnlyFindDimensions(params)) {
if (present) {
throw new SpeculativeOverlayError(capability, this.gen)
}
}
}
}
/**
* @description The find() dimensions that only index machinery can answer
* (everything beyond metadata filters + ordering + offset/limit windows),
* as `[capability, present]` pairs for the given params.
*/
function indexOnlyFindDimensions(params: FindParams): Array<[string, boolean]> {
return [
['semantic query', params.query !== undefined], ['semantic query', params.query !== undefined],
['vector search', params.vector !== undefined], ['vector search', params.vector !== undefined],
['proximity search (near)', params.near !== undefined], ['proximity search (near)', params.near !== undefined],
@ -733,16 +834,16 @@ export class Db<T = any> {
(params.searchMode !== undefined && params.searchMode !== 'auto') (params.searchMode !== undefined && params.searchMode !== 'auto')
] ]
] ]
for (const [capability, present] of unsupported) { }
if (present) {
throw new NotYetSupportedAtHistoricalGenerationError( /**
capability, * @description Whether the find() params include any index-only dimension
this.gen, * the routing predicate that decides between the record path (cheap,
this.host.store.generation() * metadata-only) and the at-generation index materialization at historical
) * generations.
} */
} function findRequiresIndexes(params: FindParams): boolean {
} return indexOnlyFindDimensions(params).some(([, present]) => present)
} }
/** /**

View file

@ -5,18 +5,22 @@
* Three failure modes have dedicated classes so callers can branch on them * Three failure modes have dedicated classes so callers can branch on them
* with `instanceof` instead of string matching: * with `instanceof` instead of string matching:
* *
* - {@link GenerationConflictError} a `transact()` compare-and-swap * - {@link GenerationConflictError} the store's current generation differs
* (`ifAtGeneration`) observed a different current generation than the * from what the caller's operation requires: a `transact()`
* caller expected. The standard retry pattern is: re-read via * compare-and-swap (`ifAtGeneration`) observed a different generation than
* `brain.now()`, re-derive the transaction, and re-submit. * expected, or `db.persist()` was called on a view that history has moved
* - {@link NotYetSupportedAtHistoricalGenerationError} an index-accelerated * past (snapshots capture current bytes, so persisting requires the view
* query (vector / hybrid / graph-traversal search) was issued against a * to still be the latest generation). The standard retry pattern is:
* `Db` pinned at a generation the live indexes no longer represent. * re-read via `brain.now()`, re-derive the operation, and re-submit.
* `get()` and metadata-filter `find()` remain fully supported at pinned * - {@link SpeculativeOverlayError} an index-accelerated query (vector /
* generations; index-accelerated queries at historical generations require * hybrid / graph-traversal search, cursors, aggregation) or `persist()`
* an index rebuild from the pinned record set, which ships as a follow-up * was issued against a speculative `db.with()` overlay. Overlays are pure
* (`asOf` rebuild). The error message names the unsupported capability and * in-memory values whose entities carry no embeddings (`with()` never
* points at the supported alternatives. * invokes the embedder), so a "full" index query over one would silently
* miss the overlay's own entities an honest error beats silently-wrong
* results. Commit the operations with `brain.transact()` to get the full
* query surface. Historical (non-overlay) views do NOT throw this: they
* serve the full query surface via at-generation index materialization.
* - {@link GenerationCompactedError} `asOf()` asked for a generation whose * - {@link GenerationCompactedError} `asOf()` asked for a generation whose
* immutable records were reclaimed by `compactHistory()`. * immutable records were reclaimed by `compactHistory()`.
* *
@ -24,14 +28,19 @@
*/ */
/** /**
* @description Thrown by `brain.transact(ops, { ifAtGeneration })` when the * @description Thrown when an operation requires the store to be at a
* store's current generation does not equal the caller-supplied expectation. * specific generation and it is not:
* This is the whole-store compare-and-swap counterpart to the per-entity
* `ifRev` / `RevisionConflictError` pair: it guarantees that *nothing* was
* committed between the caller's read (`brain.now()`) and this transaction.
* *
* The transaction is rejected before any record is staged the store is * - `brain.transact(ops, { ifAtGeneration })` the whole-store
* compare-and-swap counterpart to the per-entity `ifRev` /
* `RevisionConflictError` pair: it guarantees that *nothing* was committed
* between the caller's read (`brain.now()`) and this transaction. The
* transaction is rejected before any record is staged the store is
* untouched and the generation counter is unchanged. * untouched and the generation counter is unchanged.
* - `db.persist(path)` snapshots capture current bytes, so persisting
* requires the view to still be the store's **latest** generation. A view
* that history has moved past throws this error: pin with `brain.now()`
* and persist before further writes.
* *
* @example * @example
* const db = brain.now() * const db = brain.now()
@ -45,20 +54,21 @@
* } * }
*/ */
export class GenerationConflictError extends Error { export class GenerationConflictError extends Error {
/** The generation the caller expected the store to be at. */ /** The generation the operation required the store to be at. */
public readonly expected: number public readonly expected: number
/** The generation the store was actually at when the transaction arrived. */ /** The generation the store was actually at. */
public readonly actual: number public readonly actual: number
/** /**
* @param expected - The generation supplied via `ifAtGeneration`. * @param expected - The required generation (`ifAtGeneration`, or the
* @param actual - The store's current generation at submission time. * pinned generation of the view being persisted).
* @param actual - The store's current generation.
*/ */
constructor(expected: number, actual: number) { constructor(expected: number, actual: number) {
super( super(
`Generation conflict: transact() was submitted with ifAtGeneration: ${expected}, ` + `Generation conflict: the operation requires the store at generation ${expected}, ` +
`but the store is at generation ${actual}. Another write committed in between. ` + `but it is at generation ${actual}. Another write committed in between. ` +
`Re-read with brain.now(), rebuild the transaction, and retry.` `Re-read with brain.now(), rebuild the operation, and retry.`
) )
this.name = 'GenerationConflictError' this.name = 'GenerationConflictError'
this.expected = expected this.expected = expected
@ -67,45 +77,50 @@ export class GenerationConflictError extends Error {
} }
/** /**
* @description Thrown when an operation on a pinned `Db` requires * @description Thrown when an operation on a speculative `db.with()` overlay
* index-accelerated machinery (vector search, hybrid search, graph * requires index machinery (vector / hybrid / graph-traversal search, cursor
* traversal, cursor pagination, aggregation) that the live indexes cannot * pagination, aggregation) or durability (`persist()`).
* answer for a *historical* generation. Storage-record reads (`get()`) and
* metadata-filter `find()` are always supported at pinned generations.
* *
* Rebuilding indexes from a pinned record set (the standard LSM answer for * Why this single boundary exists: overlays are pure in-memory values
* historical index queries) is the documented `asOf`-rebuild follow-up; this * `with()` never touches disk, the generation counter, or the embedder, so
* error is the honest boundary until it ships. * overlay entities carry **no embeddings**. A "full" vector or hybrid query
* over an overlay would silently exclude the overlay's own entities, and a
* persisted overlay would be a store whose entities cannot be semantically
* searched. An honest error beats silently-wrong results. Commit the
* operations with `brain.transact()` to get the full query surface, or query
* the overlay's base view (historical views serve the complete surface via
* at-generation index materialization).
*
* `get()`, metadata-filter `find()`, and filter-based `related()` remain
* fully supported on overlays.
* *
* @example * @example
* const db = brain.now() * const spec = await brain.now().with([{ op: 'add', type, data: 'draft' }])
* await brain.transact([{ op: 'update', id, metadata: { v: 2 } }]) * await spec.find({ where: { draft: true } }) // ✅ metadata find — supported
* await db.get(id) // ✅ pinned read — supported * await spec.search('semantic query') // ❌ throws this error
* await db.find({ where: { v: 1 } }) // ✅ metadata find — supported * await brain.transact([{ op: 'add', type, data: 'draft' }]) // → full surface
* await db.search('semantic query') // ❌ throws this error
*/ */
export class NotYetSupportedAtHistoricalGenerationError extends Error { export class SpeculativeOverlayError extends Error {
/** The pinned generation the operation was attempted against. */ /** The capability that is not supported on speculative overlays. */
public readonly generation: number
/** The capability that is not supported at historical generations. */
public readonly capability: string public readonly capability: string
/** The overlay's pinned base generation. */
public readonly generation: number
/** /**
* @param capability - Human-readable name of the unsupported capability * @param capability - Human-readable name of the unsupported capability
* (e.g. `'vector search'`, `'graph traversal (connected)'`). * (e.g. `'vector search'`, `'persist'`).
* @param generation - The Db's pinned generation. * @param generation - The overlay's pinned base generation.
* @param currentGeneration - The store's current generation, for context.
*/ */
constructor(capability: string, generation: number, currentGeneration: number) { constructor(capability: string, generation: number) {
super( super(
`${capability} is not yet supported at a historical generation ` + `${capability} is not supported on a speculative with() overlay ` +
`(Db pinned at generation ${generation}; store is at ${currentGeneration}). ` + `(pinned at base generation ${generation}). Overlays are pure in-memory ` +
`Supported at pinned generations: get() and metadata-filter find(). ` + `values whose entities carry no embeddings, so index-accelerated reads ` +
`For index-accelerated queries over historical state, persist() the generation ` + `over them would be silently incomplete. Supported on overlays: get(), ` +
`you need and open it with Brainy.load(path) — the loaded snapshot rebuilds ` + `metadata-filter find(), and filter-based related(). Commit the operations ` +
`its indexes and supports the full query surface.` `with brain.transact() for the full query surface, or query the base view.`
) )
this.name = 'NotYetSupportedAtHistoricalGenerationError' this.name = 'SpeculativeOverlayError'
this.capability = capability this.capability = capability
this.generation = generation this.generation = generation
} }

View file

@ -23,9 +23,11 @@ import type { Entity, FindParams } from '../types/brainy.types.js'
/** /**
* @description Thrown when a `where` clause uses an operator this in-memory * @description Thrown when a `where` clause uses an operator this in-memory
* evaluator does not implement. Callers (historical `find()` on a `Db`) * evaluator does not implement. Callers (historical/speculative `find()` on
* convert this into the documented historical-query error rather than * a `Db`) never let it surface as silently-wrong results: at a historical
* returning silently-wrong results. * generation the query is rerouted through the at-generation index
* materialization (which evaluates the full live operator surface); on a
* speculative overlay it becomes a `SpeculativeOverlayError`.
*/ */
export class UnsupportedWhereOperatorError extends Error { export class UnsupportedWhereOperatorError extends Error {
/** The unrecognized operator name. */ /** The unrecognized operator name. */
@ -253,8 +255,9 @@ export function whereMatches(entity: Entity, where: Record<string, unknown>): bo
* entity. Used by historical and speculative `find()` to decide whether a * entity. Used by historical and speculative `find()` to decide whether a
* changed/overlaid entity belongs in the result set. The caller guarantees * changed/overlaid entity belongs in the result set. The caller guarantees
* the query carries no index-only dimensions (semantic `query`/`vector`, * the query carries no index-only dimensions (semantic `query`/`vector`,
* `connected` traversal, ) those are rejected with the documented * `connected` traversal, ) those are routed to the at-generation index
* historical-query error before evaluation starts. * materialization (historical) or rejected with `SpeculativeOverlayError`
* (overlays) before evaluation starts.
* *
* @param entity - The resolved entity. * @param entity - The resolved entity.
* @param params - The metadata-level find parameters. * @param params - The metadata-level find parameters.

View file

@ -2,8 +2,13 @@
* @module graph/graphAdjacencyIndex * @module graph/graphAdjacencyIndex
* @description GraphAdjacencyIndex billion-scale graph traversal engine. * @description GraphAdjacencyIndex billion-scale graph traversal engine.
* *
* LSM-tree storage reduces memory from 500GB to 1.3GB for 1 billion * Adjacency lives in two verb-id LSM trees (sourceId verbIds,
* relationships while maintaining sub-5ms neighbor lookups. * targetId verbIds) filtered through an in-memory live-verb tombstone set,
* with full verb objects loaded on demand through the unified cache. The
* verb set is the single source of truth: neighbor reads derive from live
* verbs, so removals are visible to every read path immediately. ID-only
* in-memory tracking keeps the resident footprint to the verb-id set (~8
* bytes per relationship) regardless of relationship payload size.
* *
* **8.0 u64 boundary:** this class implements the BigInt * **8.0 u64 boundary:** this class implements the BigInt
* {@link GraphIndexProvider} contract entity ints in, entity/verb ints out. * {@link GraphIndexProvider} contract entity ints in, entity/verb ints out.
@ -58,16 +63,17 @@ export interface GraphIndexStats {
/** /**
* GraphAdjacencyIndex - Billion-scale adjacency list with LSM-tree storage * GraphAdjacencyIndex - Billion-scale adjacency list with LSM-tree storage
* *
* Core innovation: LSM-tree for disk-based storage with bloom filter optimization * Disk-resident verb-id adjacency (LSM trees with bloom filter optimization)
* Memory efficient: 385x less memory (1.3GB vs 500GB for 1B relationships) * plus an in-memory live-verb tombstone set; neighbor reads derive from live
* Performance: Sub-5ms neighbor lookups with bloom filter optimization * verbs via cache-assisted batch loads O(node degree) per lookup, correct
* under verb removal.
*/ */
export class GraphAdjacencyIndex implements GraphIndexProvider { export class GraphAdjacencyIndex implements GraphIndexProvider {
// LSM-tree storage for outgoing and incoming edges // LSM-tree storage for verb ID lookups — the single adjacency source of
private lsmTreeSource: LSMTree // sourceId -> targetIds (outgoing edges) // truth. Neighbor reads derive from these (live-verb filtered via
private lsmTreeTarget: LSMTree // targetId -> sourceIds (incoming edges) // verbIdSet) so removeVerb tombstones are honored by EVERY read path;
// a separate entity→entity edge tree cannot be tombstone-filtered (it
// LSM-tree storage for verb ID lookups (billion-scale optimization) // carries no verb ids) and previously served stale neighbors forever.
private lsmTreeVerbsBySource: LSMTree // sourceId -> verbIds private lsmTreeVerbsBySource: LSMTree // sourceId -> verbIds
private lsmTreeVerbsByTarget: LSMTree // targetId -> verbIds private lsmTreeVerbsByTarget: LSMTree // targetId -> verbIds
@ -126,19 +132,6 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
flushInterval: config.flushInterval ?? 30000 flushInterval: config.flushInterval ?? 30000
} }
// Create LSM-trees for source and target indexes
this.lsmTreeSource = new LSMTree(storage, {
memTableThreshold: 100000,
storagePrefix: 'graph-lsm-source',
enableCompaction: true
})
this.lsmTreeTarget = new LSMTree(storage, {
memTableThreshold: 100000,
storagePrefix: 'graph-lsm-target',
enableCompaction: true
})
// Create LSM-trees for verb ID lookups (billion-scale optimization) // Create LSM-trees for verb ID lookups (billion-scale optimization)
this.lsmTreeVerbsBySource = new LSMTree(storage, { this.lsmTreeVerbsBySource = new LSMTree(storage, {
memTableThreshold: 100000, memTableThreshold: 100000,
@ -155,7 +148,7 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
// Use SAME UnifiedCache as MetadataIndexManager for coordinated memory management // Use SAME UnifiedCache as MetadataIndexManager for coordinated memory management
this.unifiedCache = getGlobalCache() this.unifiedCache = getGlobalCache()
prodLog.info('GraphAdjacencyIndex initialized with LSM-tree storage (4 LSM-trees total)') prodLog.info('GraphAdjacencyIndex initialized with LSM-tree storage (2 LSM-trees total)')
} }
/** /**
@ -210,8 +203,6 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
return return
} }
await this.lsmTreeSource.init()
await this.lsmTreeTarget.init()
await this.lsmTreeVerbsBySource.init() await this.lsmTreeVerbsBySource.init()
await this.lsmTreeVerbsByTarget.init() await this.lsmTreeVerbsByTarget.init()
@ -279,11 +270,13 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
} }
/** /**
* @description Core API neighbor lookup with LSM-tree storage (BigInt * @description Core API neighbor lookup (BigInt boundary). Neighbors
* boundary). O(log n) with bloom filter optimization (90% of queries skip * derive from the node's live verbs (tombstone-filtered verb-id LSM trees
* disk I/O); pagination support for high-degree nodes. The entity int is * + unified-cache batch loads), so `removeVerb()` is honored immediately;
* resolved to a UUID via the shared mapper (unknown int empty result) and * cost is O(node degree) with cache-assisted verb resolution. Pagination
* neighbor UUIDs convert back via `BigInt(getOrAssign(uuid))`. * support for high-degree nodes. The entity int is resolved to a UUID via
* the shared mapper (unknown int empty result) and neighbor UUIDs
* convert back via `BigInt(getOrAssign(uuid))`.
* *
* @param id - Entity int to get neighbors for (from the shared idMapper). * @param id - Entity int to get neighbors for (from the shared idMapper).
* @param options - Optional direction ('both' default) + limit/offset. * @param options - Optional direction ('both' default) + limit/offset.
@ -322,6 +315,13 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
* String-keyed neighbor lookup the internal implementation behind * String-keyed neighbor lookup the internal implementation behind
* {@link getNeighbors}. Kept UUID-based because the LSM trees are keyed by * {@link getNeighbors}. Kept UUID-based because the LSM trees are keyed by
* UUID; only the public contract speaks BigInt. * UUID; only the public contract speaks BigInt.
*
* Neighbors are derived from the node's **live** verbs (the verb-id LSM
* trees filtered through the `verbIdSet` tombstones, then batch-loaded via
* the unified cache) rather than from a separate entityentity edge tree.
* That keeps `removeVerb()` visible to traversal: an entity-level edge
* entry carries no verb id, so it could never be tombstone-filtered, and
* a removed relationship would keep its endpoints "connected" forever.
*/ */
private async getNeighborUuids( private async getNeighborUuids(
id: string, id: string,
@ -335,18 +335,15 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
const direction = options?.direction || 'both' const direction = options?.direction || 'both'
const neighbors = new Set<string>() const neighbors = new Set<string>()
// Query LSM-trees with bloom filter optimization
if (direction !== 'in') { if (direction !== 'in') {
const outgoing = await this.lsmTreeSource.get(id) for (const verb of (await this.liveVerbsForNode(this.lsmTreeVerbsBySource, id)).values()) {
if (outgoing) { neighbors.add(verb.targetId)
outgoing.forEach(neighborId => neighbors.add(neighborId))
} }
} }
if (direction !== 'out') { if (direction !== 'out') {
const incoming = await this.lsmTreeTarget.get(id) for (const verb of (await this.liveVerbsForNode(this.lsmTreeVerbsByTarget, id)).values()) {
if (incoming) { neighbors.add(verb.sourceId)
incoming.forEach(neighborId => neighbors.add(neighborId))
} }
} }
@ -370,6 +367,23 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
return result return result
} }
/**
* @description The live (non-tombstoned) verb objects adjacent to `nodeId`
* in one verb-id LSM tree: read the node's verb-id list, drop ids deleted
* by `removeVerb()` (the `verbIdSet` tombstone filter same rule as
* {@link verbIdsToPaginatedInts}), and batch-load the survivors through the
* unified cache. Shared by {@link getNeighborUuids} for both directions.
* @param tree - `lsmTreeVerbsBySource` (out-edges) or `lsmTreeVerbsByTarget` (in-edges).
* @param nodeId - The node's UUID (LSM trees are UUID-keyed).
* @returns The node's live verbs in that direction, keyed by verb id.
*/
private async liveVerbsForNode(tree: LSMTree, nodeId: string): Promise<Map<string, GraphVerb>> {
const verbIds = (await tree.get(nodeId)) || []
const liveIds = [...new Set(verbIds)].filter(verbId => this.verbIdSet.has(verbId))
if (liveIds.length === 0) return new Map()
return this.getVerbsBatchCached(liveIds)
}
/** /**
* @description Verb ints for all edges originating at `sourceInt` (BigInt * @description Verb ints for all edges originating at `sourceInt` (BigInt
* boundary). O(log n) LSM-tree lookup with bloom filter optimization; * boundary). O(log n) LSM-tree lookup with bloom filter optimization;
@ -569,8 +583,8 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
* Get total relationship count - O(1) operation * Get total relationship count - O(1) operation
*/ */
size(): number { size(): number {
// Use LSM-tree size for accurate count // Use LSM-tree size for accurate count (one entry per indexed verb)
return this.lsmTreeSource.size() return this.lsmTreeVerbsBySource.size()
} }
/** /**
@ -604,13 +618,9 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
uniqueTargetNodes: number uniqueTargetNodes: number
totalNodes: number totalNodes: number
} { } {
const totalRelationships = this.lsmTreeSource.size() const totalRelationships = this.lsmTreeVerbsBySource.size()
const relationshipsByType = Object.fromEntries(this.relationshipCountsByType) const relationshipsByType = Object.fromEntries(this.relationshipCountsByType)
// Get stats from LSM-trees
const sourceStats = this.lsmTreeSource.getStats()
const targetStats = this.lsmTreeTarget.getStats()
// Note: Exact unique node counts would require full LSM-tree scan // Note: Exact unique node counts would require full LSM-tree scan
// Using verbIdSet (ID-only tracking) for memory efficiency // Using verbIdSet (ID-only tracking) for memory efficiency
const uniqueSourceNodes = this.verbIdSet.size const uniqueSourceNodes = this.verbIdSet.size
@ -654,11 +664,14 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
this.verbIdSet.add(verb.id) this.verbIdSet.add(verb.id)
const verbInt = this.internVerbId(verb.id) const verbInt = this.internVerbId(verb.id)
// Add to LSM-trees (outgoing and incoming edges) // Seed the unified cache with the authoritative verb object: neighbor
await this.lsmTreeSource.add(verb.sourceId, verb.targetId) // reads ({@link liveVerbsForNode}) resolve verbs through the cache with a
await this.lsmTreeTarget.add(verb.targetId, verb.sourceId) // storage fallback, so an indexed verb is immediately traversable — and
// freshly written verbs are the likeliest next reads.
this.unifiedCache.set(`graph:verb:${verb.id}`, verb, 'other', 128, 50)
// Add to verbId tracking LSM-trees (billion-scale optimization for getVerbsBySource/Target) // Add to the verb-id adjacency LSM-trees (the single adjacency source of
// truth — neighbor and verb-id reads both derive from these).
await this.lsmTreeVerbsBySource.add(verb.sourceId, verb.id) await this.lsmTreeVerbsBySource.add(verb.sourceId, verb.id)
await this.lsmTreeVerbsByTarget.add(verb.targetId, verb.id) await this.lsmTreeVerbsByTarget.add(verb.targetId, verb.id)
@ -682,9 +695,13 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
/** /**
* @description Remove a relationship from the index by its id string. * @description Remove a relationship from the index by its id string.
* LSM-tree edges persist (tombstone deletion via verbIdSet filtering); the * Deletion is tombstone-based: the verb id leaves `verbIdSet`, and every
* verb's interned int is intentionally retained so previously returned verb * read path (verb-id reads via {@link verbIdsToPaginatedInts}, neighbor
* ints stay resolvable via {@link verbIntsToIds}. * reads via {@link liveVerbsForNode}) filters the append-only LSM trees
* through that set so the verb disappears from traversal immediately
* while the trees stay immutable. The verb's interned int is intentionally
* retained so previously returned verb ints stay resolvable via
* {@link verbIntsToIds}.
* @param verbId - The verb's UUID string. * @param verbId - The verb's UUID string.
* @returns Resolves once the verb no longer appears in reads. * @returns Resolves once the verb no longer appears in reads.
*/ */
@ -709,10 +726,6 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
this.relationshipCountsByType.delete(verbType) this.relationshipCountsByType.delete(verbType)
} }
// Note: LSM-tree edges persist
// Full tombstone deletion can be implemented via compaction
// For now, removed verbs won't appear in queries (verbIndex check)
const elapsed = performance.now() - startTime const elapsed = performance.now() - startTime
// Performance assertion // Performance assertion
@ -794,7 +807,7 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
prodLog.info(`GraphAdjacencyIndex: Rebuild complete in ${rebuildTime}ms`) prodLog.info(`GraphAdjacencyIndex: Rebuild complete in ${rebuildTime}ms`)
prodLog.info(` - Total relationships: ${totalVerbs}`) prodLog.info(` - Total relationships: ${totalVerbs}`)
prodLog.info(` - Memory usage: ${(memoryUsage / 1024 / 1024).toFixed(1)}MB`) prodLog.info(` - Memory usage: ${(memoryUsage / 1024 / 1024).toFixed(1)}MB`)
prodLog.info(` - LSM-tree stats:`, this.lsmTreeSource.getStats()) prodLog.info(` - LSM-tree stats:`, this.lsmTreeVerbsBySource.getStats())
} finally { } finally {
this.isRebuilding = false this.isRebuilding = false
@ -808,8 +821,8 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
let bytes = 0 let bytes = 0
// LSM-tree memory (MemTable + bloom filters + zone maps) // LSM-tree memory (MemTable + bloom filters + zone maps)
const sourceStats = this.lsmTreeSource.getStats() const sourceStats = this.lsmTreeVerbsBySource.getStats()
const targetStats = this.lsmTreeTarget.getStats() const targetStats = this.lsmTreeVerbsByTarget.getStats()
bytes += sourceStats.memTableMemory bytes += sourceStats.memTableMemory
bytes += targetStats.memTableMemory bytes += targetStats.memTableMemory
@ -829,8 +842,8 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
* Get comprehensive statistics * Get comprehensive statistics
*/ */
getStats(): GraphIndexStats { getStats(): GraphIndexStats {
const sourceStats = this.lsmTreeSource.getStats() const sourceStats = this.lsmTreeVerbsBySource.getStats()
const targetStats = this.lsmTreeTarget.getStats() const targetStats = this.lsmTreeVerbsByTarget.getStats()
return { return {
totalRelationships: this.size(), totalRelationships: this.size(),
@ -862,14 +875,8 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
const startTime = Date.now() const startTime = Date.now()
// Flush all 4 LSM-trees in parallel (MemTables → SSTables on disk) // Flush both LSM-trees in parallel (MemTables → SSTables on disk)
await Promise.all([ await Promise.all([
this.lsmTreeSource.flush().then(() => {
prodLog.debug(`GraphAdjacencyIndex: Flushed source tree`)
}),
this.lsmTreeTarget.flush().then(() => {
prodLog.debug(`GraphAdjacencyIndex: Flushed target tree`)
}),
this.lsmTreeVerbsBySource.flush().then(() => { this.lsmTreeVerbsBySource.flush().then(() => {
prodLog.debug(`GraphAdjacencyIndex: Flushed verbs-by-source tree`) prodLog.debug(`GraphAdjacencyIndex: Flushed verbs-by-source tree`)
}), }),
@ -892,11 +899,9 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
this.flushTimer = undefined this.flushTimer = undefined
} }
// Close all 4 LSM-trees (will flush MemTables to SSTables) // Close both LSM-trees (will flush MemTables to SSTables)
if (this.initialized) { if (this.initialized) {
await Promise.all([ await Promise.all([
this.lsmTreeSource.close(),
this.lsmTreeTarget.close(),
this.lsmTreeVerbsBySource.close(), this.lsmTreeVerbsBySource.close(),
this.lsmTreeVerbsByTarget.close(), this.lsmTreeVerbsByTarget.close(),
]) ])
@ -915,8 +920,8 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
return ( return (
!this.isRebuilding && !this.isRebuilding &&
this.lsmTreeSource.isHealthy() && this.lsmTreeVerbsBySource.isHealthy() &&
this.lsmTreeTarget.isHealthy() this.lsmTreeVerbsByTarget.isHealthy()
) )
} }
} }

View file

@ -134,7 +134,7 @@ export { RevisionConflictError } from './transaction/RevisionConflictError.js'
export { Db } from './db/db.js' export { Db } from './db/db.js'
export { export {
GenerationConflictError, GenerationConflictError,
NotYetSupportedAtHistoricalGenerationError, SpeculativeOverlayError,
GenerationCompactedError GenerationCompactedError
} from './db/errors.js' } from './db/errors.js'
export type { export type {

View file

@ -4,7 +4,7 @@
* Beautiful, consistent, type-safe interfaces for the future of neural databases * Beautiful, consistent, type-safe interfaces for the future of neural databases
*/ */
import { Vector } from '../coreTypes.js' import { StorageAdapter, Vector } from '../coreTypes.js'
import { NounType, VerbType } from './graphTypes.js' import { NounType, VerbType } from './graphTypes.js'
// ============= Core Types ============= // ============= Core Types =============
@ -1092,14 +1092,21 @@ export interface BrainyStats {
* Brainy configuration * Brainy configuration
*/ */
export interface BrainyConfig { export interface BrainyConfig {
// Storage configuration /**
storage?: { * Storage configuration: either a config object resolved through the
type: 'auto' | 'memory' | 'filesystem' | 's3' | 'r2' | 'opfs' | 'gcs' * storage factory, or a pre-constructed adapter instance (used directly
* e.g. `storage: new MemoryStorage()`; Brainy's historical-query
* materializer hands a pre-populated instance through this path).
*/
storage?:
| {
type: 'auto' | 'memory' | 'filesystem'
/** Root directory for filesystem storage. Passed through to storage factories /** Root directory for filesystem storage. Passed through to storage factories
* including plugin-provided factories (e.g. Cortex mmap). */ * including plugin-provided factories (e.g. native mmap providers). */
rootDirectory?: string rootDirectory?: string
options?: any options?: any
} }
| StorageAdapter
// Index configuration // Index configuration
index?: { index?: {

View file

@ -26,10 +26,16 @@
* 9. Versioned provider wiring a provider implementing the 4-method * 9. Versioned provider wiring a provider implementing the 4-method
* `VersionedIndexProvider` contract receives balanced pin/release (and a * `VersionedIndexProvider` contract receives balanced pin/release (and a
* visibility consult per pin) as Db values are created and released. * visibility consult per pin) as Db values are created and released.
* 10. Historical full query surface vector search, graph traversal, and
* aggregation at a past pinned generation return at-generation results
* via ephemeral index materialization (for `now()` and `asOf()` pins
* alike, built once per Db and cached); `release()` frees the
* materialization by closing the ephemeral reader; speculative overlays
* keep the documented `SpeculativeOverlayError` boundary.
* *
* All entities carry explicit vectors so the suite never invokes the * All entities carry explicit vectors so the suite never invokes the
* embedding engine (semantic search is exercised only for its documented * embedding engine (vector queries use find({ vector }); semantic string
* historical-generation error). * search on overlays is exercised only for its documented error).
*/ */
import { describe, it, expect, afterEach, vi } from 'vitest' import { describe, it, expect, afterEach, vi } from 'vitest'
@ -37,11 +43,11 @@ import * as fs from 'node:fs'
import * as os from 'node:os' import * as os from 'node:os'
import * as path from 'node:path' import * as path from 'node:path'
import { Brainy } from '../../src/brainy.js' import { Brainy } from '../../src/brainy.js'
import { Db } from '../../src/db/db.js' import { Db, type HistoricalQueryHandle } from '../../src/db/db.js'
import { import {
GenerationCompactedError, GenerationCompactedError,
GenerationConflictError, GenerationConflictError,
NotYetSupportedAtHistoricalGenerationError SpeculativeOverlayError
} from '../../src/db/errors.js' } from '../../src/db/errors.js'
import type { GenerationStore } from '../../src/db/generationStore.js' import type { GenerationStore } from '../../src/db/generationStore.js'
import { GraphAdjacencyIndex } from '../../src/graph/graphAdjacencyIndex.js' import { GraphAdjacencyIndex } from '../../src/graph/graphAdjacencyIndex.js'
@ -472,9 +478,7 @@ describe('8.0 Db API — generational MVCC', () => {
]) ])
const snapDir = path.join(makeTempDir(), 'stale-snapshot') const snapDir = path.join(makeTempDir(), 'stale-snapshot')
await expect(db.persist(snapDir)).rejects.toThrow( await expect(db.persist(snapDir)).rejects.toThrow(GenerationConflictError)
NotYetSupportedAtHistoricalGenerationError
)
await db.release() await db.release()
}) })
@ -870,32 +874,193 @@ describe('8.0 Db API — generational MVCC', () => {
await after.release() await after.release()
}) })
it('index-accelerated queries at a historical generation throw the documented error — never silently-wrong results', async () => { it('historical vector search is served at the pinned generation via index materialization', async () => {
const brain = await openMemoryBrain() const brain = await openMemoryBrain()
// Two one-hot vectors: orthogonal under cosine, so nearest-neighbor
// results are fully deterministic.
const axisA = Array.from({ length: 384 }, (_, i) => (i === 0 ? 1 : 0))
const axisB = Array.from({ length: 384 }, (_, i) => (i === 1 ? 1 : 0))
const axisC = Array.from({ length: 384 }, (_, i) => (i === 2 ? 1 : 0))
await brain.transact([ await brain.transact([
{ op: 'add', id: uid('hist-e'), type: NounType.Document, data: 'x', vector: vec(95), metadata: { v: 1 } } { op: 'add', id: uid('mat-a'), type: NounType.Document, data: 'a', vector: axisA, metadata: { who: 'a' } },
{ op: 'add', id: uid('mat-b'), type: NounType.Document, data: 'b', vector: axisB, metadata: { who: 'b' } },
{ op: 'add', id: uid('mat-del'), type: NounType.Document, data: 'del', vector: axisC, metadata: { who: 'del' } }
]) ])
const db = brain.now() const db = brain.now()
await brain.transact([{ op: 'update', id: uid('hist-e'), metadata: { v: 2 } }])
// Storage-record reads remain fully supported… // Mutate distinctively after the pin: swap a/b's vectors, delete 'del'.
expect(((await db.get(uid('hist-e')))?.metadata as { v: number }).v).toBe(1) await brain.transact([
expect((await db.find({ type: NounType.Document, where: { v: 1 } })).length).toBe(1) { op: 'update', id: uid('mat-a'), vector: axisB },
{ op: 'update', id: uid('mat-b'), vector: axisA },
{ op: 'remove', id: uid('mat-del') }
])
// …index-accelerated dimensions throw the documented, named error. // Live index reflects the swap…
await expect(db.search('anything')).rejects.toThrow( const liveNearA = await brain.find({ vector: axisA, limit: 1 })
NotYetSupportedAtHistoricalGenerationError expect(liveNearA[0].id).toBe(uid('mat-b'))
)
await expect(db.find({ vector: vec(95) })).rejects.toThrow(
NotYetSupportedAtHistoricalGenerationError
)
await expect(db.find({ connected: { from: uid('hist-e') } })).rejects.toThrow(
NotYetSupportedAtHistoricalGenerationError
)
// Released views refuse all reads. // …the pinned view finds the OLD vector placement, including the
// since-deleted entity, via the at-generation materialization.
const pinnedNearA = await db.find({ vector: axisA, limit: 1 })
expect(pinnedNearA[0].id).toBe(uid('mat-a'))
const pinnedNearC = await db.find({ vector: axisC, limit: 1 })
expect(pinnedNearC[0].id).toBe(uid('mat-del'))
// Record-path reads on the same Db stay consistent with the materialization.
expect(((await db.get(uid('mat-del')))?.metadata as { who: string }).who).toBe('del')
// Released views refuse all reads (and free the materialization).
await db.release()
await expect(db.find({ vector: axisA, limit: 1 })).rejects.toThrow('released Db')
})
it('historical graph traversal is served at the pinned generation via index materialization', async () => {
const brain = await openMemoryBrain()
await brain.transact([
{ op: 'add', id: uid('trav-a'), type: NounType.Person, data: 'a', vector: vec(110), metadata: {} },
{ op: 'add', id: uid('trav-b'), type: NounType.Document, data: 'b', vector: vec(111), metadata: {} },
{ op: 'add', id: uid('trav-c'), type: NounType.Document, data: 'c', vector: vec(112), metadata: {} },
{ op: 'relate', from: uid('trav-a'), to: uid('trav-b'), type: VerbType.References }
])
const db = brain.now()
// Rewire the graph after the pin: a→b becomes a→c.
const oldEdge = (await brain.getRelations({ from: uid('trav-a') }))[0]
await brain.transact([
{ op: 'unrelate', id: oldEdge.id },
{ op: 'relate', from: uid('trav-a'), to: uid('trav-c'), type: VerbType.References }
])
// Live traversal sees the new wiring…
const liveConnected = await brain.find({ connected: { from: uid('trav-a') }, limit: 10 })
expect(liveConnected.map((row) => row.id)).toContain(uid('trav-c'))
expect(liveConnected.map((row) => row.id)).not.toContain(uid('trav-b'))
// …the pinned view traverses the OLD graph.
const pinnedConnected = await db.find({ connected: { from: uid('trav-a') }, limit: 10 })
expect(pinnedConnected.map((row) => row.id)).toContain(uid('trav-b'))
expect(pinnedConnected.map((row) => row.id)).not.toContain(uid('trav-c'))
// The record-path related() agrees with the materialized traversal.
const pinnedEdges = await db.related({ from: uid('trav-a') })
expect(pinnedEdges.length).toBe(1)
expect(pinnedEdges[0].to).toBe(uid('trav-b'))
await db.release()
})
it('historical aggregation computes at-generation values via index materialization', async () => {
const brain = await openMemoryBrain()
brain.defineAggregate({
name: 'order_totals',
source: { type: NounType.Document },
groupBy: ['category'],
metrics: {
total: { op: 'sum', field: 'amount' },
orders: { op: 'count' }
}
})
await brain.transact([
{ op: 'add', id: uid('agg-1'), type: NounType.Document, data: 'o1', vector: vec(120), metadata: { category: 'invoice', amount: 10 } },
{ op: 'add', id: uid('agg-2'), type: NounType.Document, data: 'o2', vector: vec(121), metadata: { category: 'invoice', amount: 20 } }
])
const db = brain.now()
// Mutate after the pin: add an order and inflate an existing one.
await brain.transact([
{ op: 'add', id: uid('agg-3'), type: NounType.Document, data: 'o3', vector: vec(122), metadata: { category: 'invoice', amount: 70 } },
{ op: 'update', id: uid('agg-1'), metadata: { amount: 100 } }
])
// Live aggregate reflects the mutations…
const liveRows = await brain.find({ aggregate: 'order_totals' })
const liveInvoice = liveRows.find((row) => (row.metadata as { category: string }).category === 'invoice')
expect((liveInvoice?.metadata as { total: number }).total).toBe(190)
expect((liveInvoice?.metadata as { orders: number }).orders).toBe(3)
// …the pinned view computes the aggregate over the at-G record set.
const pinnedRows = await db.find({ aggregate: 'order_totals' })
const pinnedInvoice = pinnedRows.find((row) => (row.metadata as { category: string }).category === 'invoice')
expect((pinnedInvoice?.metadata as { total: number }).total).toBe(30)
expect((pinnedInvoice?.metadata as { orders: number }).orders).toBe(2)
await db.release()
})
it('asOf() pins serve the materialized surface; the build is cached per Db; release() closes the ephemeral reader', async () => {
const brain = await openMemoryBrain()
// Two one-hot vectors — orthogonal, fully deterministic neighbors.
const axisA = Array.from({ length: 384 }, (_, i) => (i === 0 ? 1 : 0))
const axisB = Array.from({ length: 384 }, (_, i) => (i === 1 ? 1 : 0))
const tx = await brain.transact([
{ op: 'add', id: uid('asof-mat-a'), type: NounType.Document, data: 'a', vector: axisA, metadata: {} },
{ op: 'add', id: uid('asof-mat-b'), type: NounType.Document, data: 'b', vector: axisB, metadata: {} }
])
const pinnedGeneration = tx.generation
await tx.release()
// Swap the vectors after the pin point, then open the past via asOf().
await brain.transact([
{ op: 'update', id: uid('asof-mat-a'), vector: axisB },
{ op: 'update', id: uid('asof-mat-b'), vector: axisA }
])
const db = await brain.asOf(pinnedGeneration)
// The asOf() view serves index-accelerated queries at its pinned
// generation — same materialized surface as a now() pin.
expect((await brain.find({ vector: axisA, limit: 1 }))[0].id).toBe(uid('asof-mat-b'))
expect((await db.find({ vector: axisA, limit: 1 }))[0].id).toBe(uid('asof-mat-a'))
// The O(n at G) build runs ONCE per Db: the second index query reuses
// the cached handle (typed access to the private cache, mirroring
// generationStoreOf above).
const internals = db as unknown as { materializedHandle?: Promise<HistoricalQueryHandle> }
expect(internals.materializedHandle).toBeDefined()
const handle = await internals.materializedHandle!
expect((await db.find({ vector: axisB, limit: 1 }))[0].id).toBe(uid('asof-mat-b'))
expect(await internals.materializedHandle).toBe(handle)
// release() frees the materialization by CLOSING the ephemeral reader —
// the reader itself refuses reads afterwards, proving its indexes and
// timers were torn down (not merely hidden behind the released-Db guard).
await db.release()
await db.release() // idempotent
await expect(db.find({ vector: axisA, limit: 1 })).rejects.toThrow('released Db')
await expect(handle.find({ vector: axisA, limit: 1 })).rejects.toThrow('not initialized')
})
it('speculative overlays keep the one honest boundary: index queries and persist throw SpeculativeOverlayError', async () => {
const brain = await openMemoryBrain()
await brain.transact([
{ op: 'add', id: uid('spec-e'), type: NounType.Document, data: 'x', vector: vec(95), metadata: { v: 1 } }
])
const db = brain.now()
const spec = await db.with([{ op: 'update', id: uid('spec-e'), metadata: { v: 2 } }])
// Metadata reads on the overlay are fully supported…
expect(((await spec.get(uid('spec-e')))?.metadata as { v: number }).v).toBe(2)
expect((await spec.find({ type: NounType.Document, where: { v: 2 } })).length).toBe(1)
// …index-accelerated dimensions and persist throw the named error.
await expect(spec.search('anything')).rejects.toThrow(SpeculativeOverlayError)
await expect(spec.find({ vector: vec(95) })).rejects.toThrow(SpeculativeOverlayError)
await expect(spec.find({ connected: { from: uid('spec-e') } })).rejects.toThrow(
SpeculativeOverlayError
)
await expect(spec.persist(path.join(makeTempDir(), 'overlay-snap'))).rejects.toThrow(
SpeculativeOverlayError
)
// The non-speculative base view is unaffected by the overlay boundary.
expect(((await db.get(uid('spec-e')))?.metadata as { v: number }).v).toBe(1)
await spec.release()
await db.release() await db.release()
await expect(db.get(uid('hist-e'))).rejects.toThrow('released Db')
}) })
it('restore() requires confirmation and replaces state from a snapshot (generation floor preserved)', async () => { it('restore() requires confirmation and replaces state from a snapshot (generation floor preserved)', async () => {

View file

@ -253,31 +253,44 @@ describe('Brainy.update()', () => {
})).rejects.toThrow('invalid NounType') })).rejects.toThrow('invalid NounType')
}) })
it('should not update vector directly via update method', async () => { it('applies an explicit pre-computed vector (UpdateParams.vector contract)', async () => {
// Arrange // Arrange
const id = await brain.add(createAddParams({ const id = await brain.add(createAddParams({
data: 'Test', data: 'Test',
type: 'thing' type: 'thing'
})) }))
// v5.11.1: Need includeVectors to get actual vector
const original = await brain.get(id, { includeVectors: true }) const original = await brain.get(id, { includeVectors: true })
const originalVector = original!.vector const originalVector = original!.vector
// Create a properly dimensioned but different vector // Create a properly dimensioned but different vector
const differentVector = originalVector.map(v => v * 2) const differentVector = originalVector.map(v => v * 2)
// Act - Update with vector param (not supported) // Act — update with an explicit vector and no new data. The
// UpdateParams contract ("New pre-computed vector") applies it
// directly, with no re-embedding (mirrored by transact update ops).
await brain.update({ await brain.update({
id, id,
vector: differentVector vector: differentVector
}) })
// Assert - Vector should not change (update ignores vector param) // Assert — the stored vector is the supplied one.
// v5.11.1: Need includeVectors to check vector
const updated = await brain.get(id, { includeVectors: true }) const updated = await brain.get(id, { includeVectors: true })
expect(updated).not.toBeNull() expect(updated).not.toBeNull()
expect(updated!.vector).toEqual(originalVector) expect(updated!.vector).toEqual(differentVector)
})
it('rejects an explicit vector with mismatched dimensions', async () => {
const id = await brain.add(createAddParams({
data: 'Test',
type: 'thing'
}))
// Param validation rejects wrong dimensionality before the update runs
// (update() also re-checks against the store's actual dimensionality).
await expect(
brain.update({ id, vector: [0.1, 0.2, 0.3] })
).rejects.toThrow(/dimensions?/)
}) })
it('should reject empty update parameters', async () => { it('should reject empty update parameters', async () => {