feat(plugin): every provider write surface carries the real committed generation
The provider contract (metadata addToIndex/removeFromIndex, vector addItem/removeItem, id-mapper getOrAssign/remove) gains an optional trailing generation — evaluated lazily at operation execute time (the graph surface's thunk pattern, generalized), threaded from all 17 construction sites: undefined during generation-0 bootstrap, the real committed generation everywhere else. Optional = additive: no existing provider or caller breaks; native delta logs that stamped literal zero start hearing truth. JS twins accept the parameter with parity notes. Pins: provider doubles capture and assert nonzero monotonic generations across add/update/remove on both surfaces.
This commit is contained in:
parent
3484107462
commit
2d532684b4
7 changed files with 583 additions and 63 deletions
|
|
@ -531,9 +531,24 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
* store has assigned the batch generation by then; for single-op writes it
|
||||
* reads the post-write watermark. The arrow body reads `generationStore`
|
||||
* lazily, so it is safe to define before `init()` assigns the store.
|
||||
* Metadata/vector index writes use the bootstrap-honest twin
|
||||
* {@link indexWriteGeneration} below.
|
||||
*/
|
||||
private readonly graphWriteGeneration = (): bigint =>
|
||||
BigInt(this.generationStore.generation())
|
||||
/**
|
||||
* The metadata/vector twin of {@link graphWriteGeneration}, honest about
|
||||
* bootstrap: while generation stamping is inactive (init-time
|
||||
* infrastructure writes, e.g. the VFS root, applied via
|
||||
* `runWithoutGeneration`) there IS no commit generation — this resolves to
|
||||
* `undefined` so a provider records "unstamped", never a fabricated 0.
|
||||
* The graph thunk keeps its non-optional `bigint` contract (no graph
|
||||
* writes occur during bootstrap).
|
||||
*/
|
||||
private readonly indexWriteGeneration = (): bigint | undefined =>
|
||||
this._generationStampingActive
|
||||
? BigInt(this.generationStore.generation())
|
||||
: undefined
|
||||
/** Lazily built host surface shared by every `Db` value of this brain. */
|
||||
private _dbHost?: DbHost<T>
|
||||
/**
|
||||
|
|
@ -1995,7 +2010,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
})
|
||||
)
|
||||
tx.addOperation(
|
||||
new ReplaceInVectorIndexOperation(this.index, id, oldVector, newVector)
|
||||
new ReplaceInVectorIndexOperation(this.index, id, oldVector, newVector, this.indexWriteGeneration)
|
||||
)
|
||||
})
|
||||
await this.clearPendingEmbed(id)
|
||||
|
|
@ -2479,13 +2494,13 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
// inserts the real vector.
|
||||
if (!deferringEmbed) {
|
||||
tx.addOperation(
|
||||
new AddToVectorIndexOperation(this.index, id, vector)
|
||||
new AddToVectorIndexOperation(this.index, id, vector, this.indexWriteGeneration)
|
||||
)
|
||||
}
|
||||
|
||||
// Operation 4: Add to metadata index
|
||||
tx.addOperation(
|
||||
new AddToMetadataIndexOperation(this.metadataIndex, id, entityForIndexing)
|
||||
new AddToMetadataIndexOperation(this.metadataIndex, id, entityForIndexing, this.indexWriteGeneration)
|
||||
)
|
||||
}
|
||||
|
||||
|
|
@ -3180,7 +3195,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
// flickered in production — is a pure no-op), else remove+add
|
||||
// adjacent within the single op.
|
||||
tx.addOperation(
|
||||
new ReplaceInVectorIndexOperation(this.index, params.id, existing.vector, vector)
|
||||
new ReplaceInVectorIndexOperation(this.index, params.id, existing.vector, vector, this.indexWriteGeneration)
|
||||
)
|
||||
}
|
||||
|
||||
|
|
@ -3210,10 +3225,10 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
metadata: existing.metadata // CRITICAL: keep as nested 'metadata' property!
|
||||
}
|
||||
tx.addOperation(
|
||||
new RemoveFromMetadataIndexOperation(this.metadataIndex, params.id, removalMetadata)
|
||||
new RemoveFromMetadataIndexOperation(this.metadataIndex, params.id, removalMetadata, this.indexWriteGeneration)
|
||||
)
|
||||
tx.addOperation(
|
||||
new AddToMetadataIndexOperation(this.metadataIndex, params.id, entityForIndexing)
|
||||
new AddToMetadataIndexOperation(this.metadataIndex, params.id, entityForIndexing, this.indexWriteGeneration)
|
||||
)
|
||||
}, casPrecommit, this._changeFeed.hasListeners
|
||||
? [
|
||||
|
|
@ -3298,14 +3313,14 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
// Operation 1: Remove from vector index
|
||||
if (noun) {
|
||||
tx.addOperation(
|
||||
new RemoveFromVectorIndexOperation(this.index, id, noun.vector)
|
||||
new RemoveFromVectorIndexOperation(this.index, id, noun.vector, this.indexWriteGeneration)
|
||||
)
|
||||
}
|
||||
|
||||
// Operation 2: Remove from metadata index
|
||||
if (metadata) {
|
||||
tx.addOperation(
|
||||
new RemoveFromMetadataIndexOperation(this.metadataIndex, id, metadata)
|
||||
new RemoveFromMetadataIndexOperation(this.metadataIndex, id, metadata, this.indexWriteGeneration)
|
||||
)
|
||||
}
|
||||
|
||||
|
|
@ -3409,8 +3424,14 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
verb: Pick<GraphVerb, 'sourceId' | 'targetId'> & { sourceInt?: bigint; targetInt?: bigint }
|
||||
): { sourceInt: bigint; targetInt: bigint } {
|
||||
const idMapper = this.metadataIndex.getIdMapper()
|
||||
const sourceInt = BigInt(idMapper.getOrAssign(verb.sourceId))
|
||||
const targetInt = BigInt(idMapper.getOrAssign(verb.targetId))
|
||||
// Thread the write generation into any mint: a native mapper stamps the
|
||||
// assignment record with the real watermark instead of a literal 0.
|
||||
// Evaluated HERE (mint time) — at execute time inside a batch this is the
|
||||
// in-flight commit generation; at plan time it is the pre-batch watermark
|
||||
// (truthful: the mint happened before the batch committed).
|
||||
const generation = this.indexWriteGeneration()
|
||||
const sourceInt = BigInt(idMapper.getOrAssign(verb.sourceId, generation))
|
||||
const targetInt = BigInt(idMapper.getOrAssign(verb.targetId, generation))
|
||||
verb.sourceInt = sourceInt
|
||||
verb.targetInt = targetInt
|
||||
return { sourceInt, targetInt }
|
||||
|
|
@ -7122,13 +7143,13 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
// Add delete operations to transaction
|
||||
if (noun) {
|
||||
tx.addOperation(
|
||||
new RemoveFromVectorIndexOperation(this.index, id, noun.vector)
|
||||
new RemoveFromVectorIndexOperation(this.index, id, noun.vector, this.indexWriteGeneration)
|
||||
)
|
||||
}
|
||||
|
||||
if (metadata) {
|
||||
tx.addOperation(
|
||||
new RemoveFromMetadataIndexOperation(this.metadataIndex, id, metadata)
|
||||
new RemoveFromMetadataIndexOperation(this.metadataIndex, id, metadata, this.indexWriteGeneration)
|
||||
)
|
||||
}
|
||||
|
||||
|
|
@ -9248,7 +9269,9 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
}
|
||||
// 'absent' / vectorless / wrong-dim → skip (not vector-rankable at this gen).
|
||||
if (Array.isArray(vec) && vec.length === dim) {
|
||||
ints.push(BigInt(idMapper.getInt(id) ?? idMapper.getOrAssign(id)))
|
||||
// Mint-now fallback stamps the CURRENT committed watermark (the mint
|
||||
// happens now, regardless of the historical G being materialized).
|
||||
ints.push(BigInt(idMapper.getInt(id) ?? idMapper.getOrAssign(id, this.indexWriteGeneration())))
|
||||
rows.push(vec)
|
||||
}
|
||||
}
|
||||
|
|
@ -9492,8 +9515,8 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
new SaveNounOperation(this.storage, { id, vector, connections: new Map(), level: 0 }, isNew),
|
||||
...(deferringEmbed
|
||||
? []
|
||||
: [new AddToVectorIndexOperation(this.index, id, vector)]),
|
||||
new AddToMetadataIndexOperation(this.metadataIndex, id, entityForIndexing)
|
||||
: [new AddToVectorIndexOperation(this.index, id, vector, this.indexWriteGeneration)]),
|
||||
new AddToMetadataIndexOperation(this.metadataIndex, id, entityForIndexing, this.indexWriteGeneration)
|
||||
)
|
||||
plan.touchedNouns.push(id)
|
||||
plan.postCommit.push(() => {
|
||||
|
|
@ -9670,12 +9693,12 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
// ONE atomic vector-index leg — same law as update(): the row must
|
||||
// never be absent from vector search during an update (see
|
||||
// ReplaceInVectorIndexOperation).
|
||||
new ReplaceInVectorIndexOperation(this.index, params.id, existing.vector, vector)
|
||||
new ReplaceInVectorIndexOperation(this.index, params.id, existing.vector, vector, this.indexWriteGeneration)
|
||||
)
|
||||
}
|
||||
plan.operations.push(
|
||||
new RemoveFromMetadataIndexOperation(this.metadataIndex, params.id, removalMetadata),
|
||||
new AddToMetadataIndexOperation(this.metadataIndex, params.id, entityForIndexing)
|
||||
new RemoveFromMetadataIndexOperation(this.metadataIndex, params.id, removalMetadata, this.indexWriteGeneration),
|
||||
new AddToMetadataIndexOperation(this.metadataIndex, params.id, entityForIndexing, this.indexWriteGeneration)
|
||||
)
|
||||
plan.touchedNouns.push(params.id)
|
||||
|
||||
|
|
@ -9755,10 +9778,10 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
}
|
||||
|
||||
if (noun) {
|
||||
plan.operations.push(new RemoveFromVectorIndexOperation(this.index, id, noun.vector))
|
||||
plan.operations.push(new RemoveFromVectorIndexOperation(this.index, id, noun.vector, this.indexWriteGeneration))
|
||||
}
|
||||
if (metadata) {
|
||||
plan.operations.push(new RemoveFromMetadataIndexOperation(this.metadataIndex, id, metadata))
|
||||
plan.operations.push(new RemoveFromMetadataIndexOperation(this.metadataIndex, id, metadata, this.indexWriteGeneration))
|
||||
}
|
||||
// Pre-read metadata rides along: the count decrement must not depend on
|
||||
// re-reading the record being removed (see remove()).
|
||||
|
|
|
|||
|
|
@ -405,8 +405,15 @@ export class JsHnswVectorIndex implements VectorIndexProvider {
|
|||
|
||||
/**
|
||||
* Add a vector to the index
|
||||
*
|
||||
* @param generation - Brainy's commit generation for this write (contract
|
||||
* parity with `VectorIndexProvider.addItem`). This JS index serves "now"
|
||||
* only — no per-record delta log, no natural slot — so the value is
|
||||
* accepted and ignored; a native provider stamps its durable records
|
||||
* with it. The JS twin adopts stamping with the watermark train.
|
||||
*/
|
||||
public async addItem(item: VectorDocument): Promise<string> {
|
||||
public async addItem(item: VectorDocument, generation?: bigint): Promise<string> {
|
||||
void generation // Contract parity — the JS index keeps no per-write log.
|
||||
// Check if item is defined
|
||||
if (!item) {
|
||||
throw new Error('Item is undefined or null')
|
||||
|
|
@ -771,8 +778,13 @@ export class JsHnswVectorIndex implements VectorIndexProvider {
|
|||
* `'immediate'` persists their connections now; `'deferred'` marks them
|
||||
* dirty for the next flush. The system record (entry point + maxLevel) is
|
||||
* NOT rewritten — an in-place update changes neither.
|
||||
*
|
||||
* @param generation - Brainy's commit generation for this write (contract
|
||||
* parity with the feature-detected `updateItem` provider capability).
|
||||
* Accepted and ignored — the JS index keeps no per-write log.
|
||||
*/
|
||||
public async updateItem(item: VectorDocument): Promise<void> {
|
||||
public async updateItem(item: VectorDocument, generation?: bigint): Promise<void> {
|
||||
void generation // Contract parity — the JS index keeps no per-write log.
|
||||
if (!item) {
|
||||
throw new Error('Item is undefined or null')
|
||||
}
|
||||
|
|
@ -1212,8 +1224,14 @@ export class JsHnswVectorIndex implements VectorIndexProvider {
|
|||
|
||||
/**
|
||||
* Remove an item from the index
|
||||
*
|
||||
* @param generation - Brainy's commit generation for this removal (contract
|
||||
* parity with `VectorIndexProvider.removeItem`). Accepted and ignored —
|
||||
* this JS index removes immediately; a native provider records the
|
||||
* tombstone at this generation.
|
||||
*/
|
||||
public async removeItem(id: string): Promise<boolean> {
|
||||
public async removeItem(id: string, generation?: bigint): Promise<boolean> {
|
||||
void generation // Contract parity — the JS index keeps no per-write log.
|
||||
if (!this.nouns.has(id)) {
|
||||
return false
|
||||
}
|
||||
|
|
|
|||
|
|
@ -277,8 +277,35 @@ export interface MetadataIndexProvider {
|
|||
*/
|
||||
isMigrating?(): boolean
|
||||
|
||||
addToIndex(id: string, entityOrMetadata: any, skipFlush?: boolean, deferWrites?: boolean): Promise<void>
|
||||
removeFromIndex(id: string, metadata?: any): Promise<void>
|
||||
/**
|
||||
* @description Index one entity's metadata.
|
||||
* @param id - The entity's UUID.
|
||||
* @param entityOrMetadata - Entity structure or plain metadata bag.
|
||||
* @param skipFlush - Transactional atomicity: defer the flush to the commit seam.
|
||||
* @param deferWrites - Batch mode: buffer postings for a later flush.
|
||||
* @param generation - OPTIONAL (additive) — Brainy's commit generation for
|
||||
* this write: the SAME u64 counter {@link GraphIndexProvider.addVerb}
|
||||
* carries, resolved at operation-execute time. A provider with per-record
|
||||
* delta logs stamps it onto the durable record so its watermark
|
||||
* ("this projection reflects generation N") is derivable from real data —
|
||||
* never a literal 0. `undefined` means the caller genuinely has no commit
|
||||
* generation for this write (rebuild-from-canonical scans, bootstrap
|
||||
* writes before generation stamping activates); a provider must treat
|
||||
* that as "unstamped", not as generation 0. The built-in JS manager
|
||||
* accepts and ignores it (single live view, no per-record log).
|
||||
*/
|
||||
addToIndex(id: string, entityOrMetadata: any, skipFlush?: boolean, deferWrites?: boolean, generation?: bigint): Promise<void>
|
||||
/**
|
||||
* @description Remove one entity from the index.
|
||||
* @param id - The entity's UUID.
|
||||
* @param metadata - The entity's metadata (targets exact postings; absent → full scan).
|
||||
* @param generation - OPTIONAL (additive) — Brainy's commit generation for
|
||||
* this removal, same contract as {@link MetadataIndexProvider.addToIndex}:
|
||||
* a provider with per-record delta logs records the tombstone at this
|
||||
* generation (so as-of reads before it still see the entity); the JS
|
||||
* manager removes immediately and ignores it.
|
||||
*/
|
||||
removeFromIndex(id: string, metadata?: any, generation?: bigint): Promise<void>
|
||||
|
||||
getIds(field: string, value: any): Promise<string[]>
|
||||
/**
|
||||
|
|
@ -368,7 +395,14 @@ export interface MetadataIndexProvider {
|
|||
* the ceiling on the JS path), so `Number(bigint)` narrowing is lossless.
|
||||
*/
|
||||
getIdMapper(): {
|
||||
getOrAssign(uuid: string): number
|
||||
/**
|
||||
* Resolve-or-mint the entity's int. `generation` is OPTIONAL (additive):
|
||||
* Brainy's commit generation current at mint time, so a mapper with
|
||||
* per-record delta logs stamps the assignment record with a real
|
||||
* watermark instead of a literal 0. Ignored when the uuid is already
|
||||
* assigned (assignments are append-only) and by the JS mapper.
|
||||
*/
|
||||
getOrAssign(uuid: string, generation?: bigint): number
|
||||
getInt(uuid: string): number | undefined
|
||||
getUuid(intId: number): string | undefined
|
||||
}
|
||||
|
|
@ -1052,8 +1086,33 @@ export interface VectorIndexProvider {
|
|||
*/
|
||||
readonly name: string
|
||||
|
||||
addItem(item: VectorDocument): Promise<string>
|
||||
removeItem(id: string): Promise<boolean>
|
||||
/**
|
||||
* @description Insert one vector.
|
||||
* @param item - The vector document (`id` + `vector`).
|
||||
* @param generation - OPTIONAL (additive) — Brainy's commit generation for
|
||||
* this write: the SAME u64 counter the graph provider's
|
||||
* `addVerb(..., generation)` carries (and that `search`'s as-of
|
||||
* `options.generation` reads back), resolved at operation-execute time.
|
||||
* A provider with per-record delta logs / segment stamps records it so
|
||||
* its watermark reflects real data — never a literal 0. `undefined` =
|
||||
* the caller has no commit generation (rebuild-from-canonical, the
|
||||
* at-generation materializer's ephemeral reader); treat as "unstamped",
|
||||
* not generation 0. The built-in JS index accepts and ignores it (it
|
||||
* serves "now" only). The feature-detected `updateItem` capability (see
|
||||
* `src/transaction/operations/IndexOperations.ts`) carries the same
|
||||
* optional trailing generation.
|
||||
*/
|
||||
addItem(item: VectorDocument, generation?: bigint): Promise<string>
|
||||
/**
|
||||
* @description Remove one vector by id.
|
||||
* @param id - The entity's UUID.
|
||||
* @param generation - OPTIONAL (additive) — Brainy's commit generation for
|
||||
* this removal, same contract as {@link VectorIndexProvider.addItem}: a
|
||||
* provider with durable delete records stamps the tombstone at this
|
||||
* generation (as-of reads before it still see the vector); the JS index
|
||||
* removes immediately and ignores it.
|
||||
*/
|
||||
removeItem(id: string, generation?: bigint): Promise<boolean>
|
||||
search(
|
||||
queryVector: Vector,
|
||||
k?: number,
|
||||
|
|
@ -1199,10 +1258,29 @@ export interface EntityIdMapperProvider {
|
|||
* stays compatible — `restore()` falls back to `init()` when this is absent.
|
||||
*/
|
||||
rebuild?(): Promise<void>
|
||||
getOrAssign(uuid: string): number
|
||||
/**
|
||||
* @description Resolve-or-mint the entity's interned int (append-only:
|
||||
* once assigned, a uuid's int never changes and is never recycled).
|
||||
* @param uuid - The entity's UUID.
|
||||
* @param generation - OPTIONAL (additive) — Brainy's commit generation
|
||||
* current at mint time (the same u64 counter the graph/metadata write
|
||||
* surfaces carry). A mapper with per-record delta logs stamps the
|
||||
* assignment record with this real watermark instead of a literal 0.
|
||||
* Ignored when the uuid is already assigned, and by the JS mapper
|
||||
* (which keeps no per-record log).
|
||||
*/
|
||||
getOrAssign(uuid: string, generation?: bigint): number
|
||||
getUuid(intId: number): string | undefined
|
||||
getInt(uuid: string): number | undefined
|
||||
remove(uuid: string): boolean
|
||||
/**
|
||||
* @description Remove the uuid's mapping (the int stays reserved).
|
||||
* @param uuid - The entity's UUID.
|
||||
* @param generation - OPTIONAL (additive) — Brainy's commit generation for
|
||||
* this removal: a mapper with a per-key version chain tombstones the
|
||||
* mapping at this generation (as-of reads before it still resolve);
|
||||
* the JS mapper removes immediately and ignores it.
|
||||
*/
|
||||
remove(uuid: string, generation?: bigint): boolean
|
||||
flush(): Promise<void>
|
||||
clear(): Promise<void>
|
||||
getAllIntIds(): number[]
|
||||
|
|
|
|||
|
|
@ -56,16 +56,33 @@ function resolveVectorProviderId(index: VectorIndexProvider): string {
|
|||
* or timing trace see which engine actually ran, never a fossil name from
|
||||
* whichever engine happened to be active when this op class was written.
|
||||
*
|
||||
* Generation: `generationFn` is resolved at execute time (not construction) so
|
||||
* the write is stamped at the transaction's in-flight commit generation —
|
||||
* which the generation store only assigns once the batch begins executing.
|
||||
* The same generation is reused for the rollback removal, so an add and its
|
||||
* undo reference one watermark in a provider's per-record delta log (the
|
||||
* exact pattern the graph operations established).
|
||||
*
|
||||
* Rollback strategy:
|
||||
* - Remove item from index
|
||||
*/
|
||||
export class AddToVectorIndexOperation implements Operation {
|
||||
readonly name: string
|
||||
|
||||
/**
|
||||
* @param index - The vector-index provider (JS HNSW or native).
|
||||
* @param id - The entity's UUID.
|
||||
* @param vector - The vector to index.
|
||||
* @param generationFn - OPTIONAL: resolves the commit generation to stamp
|
||||
* this write at, evaluated when the operation executes (see class note).
|
||||
* Absent -> the provider receives no generation (undefined), never a
|
||||
* fabricated 0.
|
||||
*/
|
||||
constructor(
|
||||
private readonly index: VectorIndexProvider,
|
||||
private readonly id: string,
|
||||
private readonly vector: number[]
|
||||
private readonly vector: number[],
|
||||
private readonly generationFn?: () => bigint | undefined
|
||||
) {
|
||||
this.name = `AddToVectorIndex(${resolveVectorProviderId(index)})`
|
||||
}
|
||||
|
|
@ -74,14 +91,18 @@ export class AddToVectorIndexOperation implements Operation {
|
|||
// Check if item already exists (for rollback decision)
|
||||
const existed = await this.itemExists(this.id)
|
||||
|
||||
// Stamp this write at the in-flight commit generation; reuse it for the
|
||||
// rollback so add + undo reference the same watermark.
|
||||
const generation = this.generationFn?.()
|
||||
|
||||
// Add to index
|
||||
await this.index.addItem({ id: this.id, vector: this.vector })
|
||||
await this.index.addItem({ id: this.id, vector: this.vector }, generation)
|
||||
|
||||
// Return rollback action
|
||||
return async () => {
|
||||
if (!existed) {
|
||||
// Remove newly added item
|
||||
await this.index.removeItem(this.id)
|
||||
await this.index.removeItem(this.id, generation)
|
||||
}
|
||||
// If item existed before, we don't rollback (update is OK)
|
||||
// This prevents index corruption from removing pre-existing items
|
||||
|
|
@ -131,22 +152,34 @@ export class AddToVectorIndexOperation implements Operation {
|
|||
export class RemoveFromVectorIndexOperation implements Operation {
|
||||
readonly name: string
|
||||
|
||||
/**
|
||||
* @param index - The vector-index provider (JS HNSW or native).
|
||||
* @param id - The entity's UUID.
|
||||
* @param vector - The removed vector (required for rollback re-add).
|
||||
* @param generationFn - Resolves the commit generation for this removal,
|
||||
* evaluated when the operation executes; reused for the rollback re-add
|
||||
* so the round trip references one watermark.
|
||||
*/
|
||||
constructor(
|
||||
private readonly index: VectorIndexProvider,
|
||||
private readonly id: string,
|
||||
private readonly vector: number[] // Required for rollback
|
||||
private readonly vector: number[], // Required for rollback
|
||||
private readonly generationFn?: () => bigint | undefined
|
||||
) {
|
||||
this.name = `RemoveFromVectorIndex(${resolveVectorProviderId(index)})`
|
||||
}
|
||||
|
||||
async execute(): Promise<RollbackAction> {
|
||||
// Resolve the removal generation once; reuse it for the rollback re-add.
|
||||
const generation = this.generationFn?.()
|
||||
|
||||
// Remove from index
|
||||
await this.index.removeItem(this.id)
|
||||
await this.index.removeItem(this.id, generation)
|
||||
|
||||
// Return rollback action
|
||||
return async () => {
|
||||
// Re-add item with original vector
|
||||
await this.index.addItem({ id: this.id, vector: this.vector })
|
||||
await this.index.addItem({ id: this.id, vector: this.vector }, generation)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -198,11 +231,22 @@ export class RemoveFromVectorIndexOperation implements Operation {
|
|||
export class ReplaceInVectorIndexOperation implements Operation {
|
||||
readonly name: string
|
||||
|
||||
/**
|
||||
* @param index - The vector-index provider (JS HNSW or native).
|
||||
* @param id - The entity's UUID.
|
||||
* @param oldVector - The pre-update vector (required for rollback).
|
||||
* @param newVector - The replacement vector.
|
||||
* @param generationFn - Resolves the commit generation to stamp this write
|
||||
* at, evaluated when the operation executes and reused across both
|
||||
* execute branches AND the rollback — one watermark for the whole
|
||||
* replace round trip.
|
||||
*/
|
||||
constructor(
|
||||
private readonly index: VectorIndexProvider,
|
||||
private readonly id: string,
|
||||
private readonly oldVector: number[], // Required for rollback
|
||||
private readonly newVector: number[]
|
||||
private readonly newVector: number[],
|
||||
private readonly generationFn?: () => bigint | undefined
|
||||
) {
|
||||
this.name = `ReplaceInVectorIndex(${resolveVectorProviderId(index)})`
|
||||
}
|
||||
|
|
@ -210,32 +254,36 @@ export class ReplaceInVectorIndexOperation implements Operation {
|
|||
async execute(): Promise<RollbackAction> {
|
||||
// Feature-detect the in-place capability — optional on the provider
|
||||
// contract, like `getItem`/`setPersistMode` (Brainy's JS HNSW index
|
||||
// ships it; a native provider may not have yet).
|
||||
// ships it; a native provider may not have yet). The capability carries
|
||||
// the same optional trailing generation as the required write surface.
|
||||
const index = this.index as VectorIndexProvider & {
|
||||
updateItem?: (item: { id: string; vector: number[] }) => Promise<void>
|
||||
updateItem?: (item: { id: string; vector: number[] }, generation?: bigint) => Promise<void>
|
||||
}
|
||||
|
||||
// One commit generation for the whole replace (both branches + rollback).
|
||||
const generation = this.generationFn?.()
|
||||
|
||||
if (typeof index.updateItem === 'function') {
|
||||
// Atomic path: one in-place call, the row never leaves the index.
|
||||
await index.updateItem({ id: this.id, vector: this.newVector })
|
||||
await index.updateItem({ id: this.id, vector: this.newVector }, generation)
|
||||
|
||||
return async () => {
|
||||
// Restore the declared before-state in place (see class JSDoc for
|
||||
// the item-did-not-exist posture).
|
||||
await index.updateItem!({ id: this.id, vector: this.oldVector })
|
||||
await index.updateItem!({ id: this.id, vector: this.oldVector }, generation)
|
||||
}
|
||||
}
|
||||
|
||||
// Fallback seam: remove+add ADJACENT within this single op — no other
|
||||
// transaction operation can interleave between them (see class JSDoc).
|
||||
await this.index.removeItem(this.id)
|
||||
await this.index.addItem({ id: this.id, vector: this.newVector })
|
||||
await this.index.removeItem(this.id, generation)
|
||||
await this.index.addItem({ id: this.id, vector: this.newVector }, generation)
|
||||
|
||||
return async () => {
|
||||
// updateItem-style restore via the same adjacent pair, back to the
|
||||
// declared before-state.
|
||||
await this.index.removeItem(this.id)
|
||||
await this.index.addItem({ id: this.id, vector: this.oldVector })
|
||||
await this.index.removeItem(this.id, generation)
|
||||
await this.index.addItem({ id: this.id, vector: this.oldVector }, generation)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -243,26 +291,43 @@ export class ReplaceInVectorIndexOperation implements Operation {
|
|||
/**
|
||||
* Add to metadata index with rollback support
|
||||
*
|
||||
* Generation: `generationFn` is resolved at execute time (not construction) —
|
||||
* see {@link AddToVectorIndexOperation}'s class note; the same generation is
|
||||
* reused for the rollback removal so add + undo reference one watermark in a
|
||||
* provider's per-record delta log.
|
||||
*
|
||||
* Rollback strategy:
|
||||
* - Remove item from index
|
||||
*/
|
||||
export class AddToMetadataIndexOperation implements Operation {
|
||||
readonly name = 'AddToMetadataIndex'
|
||||
|
||||
/**
|
||||
* @param index - The metadata-index manager (JS baseline or a registered provider).
|
||||
* @param id - The entity's UUID.
|
||||
* @param entity - Entity or metadata structure to index.
|
||||
* @param generationFn - Resolves the commit generation to stamp this write
|
||||
* at, evaluated when the operation executes.
|
||||
*/
|
||||
constructor(
|
||||
private readonly index: MetadataIndexManager,
|
||||
private readonly id: string,
|
||||
private readonly entity: any // Entity or metadata structure
|
||||
private readonly entity: any, // Entity or metadata structure
|
||||
private readonly generationFn?: () => bigint | undefined
|
||||
) {}
|
||||
|
||||
async execute(): Promise<RollbackAction> {
|
||||
// Stamp this write at the in-flight commit generation; reuse it for the
|
||||
// rollback so add + undo reference the same watermark.
|
||||
const generation = this.generationFn?.()
|
||||
|
||||
// Add to metadata index (skipFlush=true for transaction atomicity)
|
||||
await this.index.addToIndex(this.id, this.entity, true)
|
||||
await this.index.addToIndex(this.id, this.entity, true, false, generation)
|
||||
|
||||
// Return rollback action
|
||||
return async () => {
|
||||
// Remove from metadata index
|
||||
await this.index.removeFromIndex(this.id, this.entity)
|
||||
await this.index.removeFromIndex(this.id, this.entity, generation)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -270,26 +335,41 @@ export class AddToMetadataIndexOperation implements Operation {
|
|||
/**
|
||||
* Remove from metadata index with rollback support
|
||||
*
|
||||
* Generation: resolved at execute time and reused for the rollback re-add —
|
||||
* one watermark for the removal round trip (see
|
||||
* {@link AddToMetadataIndexOperation}).
|
||||
*
|
||||
* Rollback strategy:
|
||||
* - Re-add item to index with original metadata
|
||||
*/
|
||||
export class RemoveFromMetadataIndexOperation implements Operation {
|
||||
readonly name = 'RemoveFromMetadataIndex'
|
||||
|
||||
/**
|
||||
* @param index - The metadata-index manager (JS baseline or a registered provider).
|
||||
* @param id - The entity's UUID.
|
||||
* @param entity - The entity/metadata being removed (required for rollback).
|
||||
* @param generationFn - Resolves the commit generation for this removal,
|
||||
* evaluated when the operation executes.
|
||||
*/
|
||||
constructor(
|
||||
private readonly index: MetadataIndexManager,
|
||||
private readonly id: string,
|
||||
private readonly entity: any // Required for rollback
|
||||
private readonly entity: any, // Required for rollback
|
||||
private readonly generationFn?: () => bigint | undefined
|
||||
) {}
|
||||
|
||||
async execute(): Promise<RollbackAction> {
|
||||
// Resolve the removal generation once; reuse it for the rollback re-add.
|
||||
const generation = this.generationFn?.()
|
||||
|
||||
// Remove from metadata index
|
||||
await this.index.removeFromIndex(this.id, this.entity)
|
||||
await this.index.removeFromIndex(this.id, this.entity, generation)
|
||||
|
||||
// Return rollback action
|
||||
return async () => {
|
||||
// Re-add with original metadata (skipFlush=true)
|
||||
await this.index.addToIndex(this.id, this.entity, true)
|
||||
await this.index.addToIndex(this.id, this.entity, true, false, generation)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -358,7 +438,7 @@ export class AddToGraphIndexOperation implements Operation {
|
|||
// Stamp this edge at the in-flight commit generation; reuse it for the
|
||||
// rollback so add + undo reference the same watermark. Endpoint ints
|
||||
// resolve HERE — after any same-batch adds have applied.
|
||||
const generation = this.generationFn()
|
||||
const generation = this.generationFn?.()
|
||||
const { sourceInt, targetInt } = resolveEndpointInts(this.endpointInts)
|
||||
const verbInt = await this.index.addVerb(this.verb, sourceInt, targetInt, generation)
|
||||
this.onVerbInt?.(verbInt)
|
||||
|
|
@ -407,7 +487,7 @@ export class RemoveFromGraphIndexOperation implements Operation {
|
|||
// Resolve the removal generation once; reuse it for the rollback re-add.
|
||||
// Endpoint ints resolve HERE (after any same-batch adds applied) and are
|
||||
// captured for the rollback, whose re-add must use the same mappings.
|
||||
const generation = this.generationFn()
|
||||
const generation = this.generationFn?.()
|
||||
const { sourceInt, targetInt } = resolveEndpointInts(this.endpointInts)
|
||||
await this.index.removeVerb(this.verb.id, generation)
|
||||
|
||||
|
|
@ -431,13 +511,20 @@ export class BatchAddToVectorIndexOperation implements Operation {
|
|||
|
||||
private operations: AddToVectorIndexOperation[]
|
||||
|
||||
/**
|
||||
* @param index - The vector-index provider (JS HNSW or native).
|
||||
* @param items - The vectors to index.
|
||||
* @param generationFn - Resolves the commit generation shared by every item
|
||||
* in the batch, evaluated when the operations execute.
|
||||
*/
|
||||
constructor(
|
||||
index: VectorIndexProvider,
|
||||
items: Array<{ id: string; vector: number[] }>
|
||||
items: Array<{ id: string; vector: number[] }>,
|
||||
generationFn?: () => bigint | undefined
|
||||
) {
|
||||
this.name = `BatchAddToVectorIndex(${resolveVectorProviderId(index)})`
|
||||
this.operations = items.map(
|
||||
item => new AddToVectorIndexOperation(index, item.id, item.vector)
|
||||
item => new AddToVectorIndexOperation(index, item.id, item.vector, generationFn)
|
||||
)
|
||||
}
|
||||
|
||||
|
|
@ -472,12 +559,19 @@ export class BatchAddToMetadataIndexOperation implements Operation {
|
|||
|
||||
private operations: AddToMetadataIndexOperation[]
|
||||
|
||||
/**
|
||||
* @param index - The metadata-index manager (JS baseline or a registered provider).
|
||||
* @param items - The entities to index.
|
||||
* @param generationFn - Resolves the commit generation shared by every item
|
||||
* in the batch, evaluated when the operations execute.
|
||||
*/
|
||||
constructor(
|
||||
index: MetadataIndexManager,
|
||||
items: Array<{ id: string; entity: any }>
|
||||
items: Array<{ id: string; entity: any }>,
|
||||
generationFn?: () => bigint | undefined
|
||||
) {
|
||||
this.operations = items.map(
|
||||
item => new AddToMetadataIndexOperation(index, item.id, item.entity)
|
||||
item => new AddToMetadataIndexOperation(index, item.id, item.entity, generationFn)
|
||||
)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -164,8 +164,15 @@ export class EntityIdMapper implements EntityIdMapperProvider {
|
|||
* would exceed that, throws {@link EntityIdSpaceExceeded} so the caller
|
||||
* loudly migrates to cor's binary mapper with `idSpace: 'u64'`
|
||||
* rather than silently truncating entity ids.
|
||||
*
|
||||
* @param generation - Brainy's commit generation current at mint time
|
||||
* (contract parity with the `EntityIdMapperProvider` surface). This JS
|
||||
* mapper keeps a snapshot file, not a per-record delta log, so there is
|
||||
* no natural slot to store it — accepted and ignored; a native mapper
|
||||
* stamps its assignment records with it.
|
||||
*/
|
||||
getOrAssign(uuid: string): number {
|
||||
getOrAssign(uuid: string, generation?: bigint): number {
|
||||
void generation // Contract parity — no per-record log in the JS mapper.
|
||||
const existing = this.uuidToInt.get(uuid)
|
||||
if (existing !== undefined) {
|
||||
return existing
|
||||
|
|
@ -226,8 +233,14 @@ export class EntityIdMapper implements EntityIdMapperProvider {
|
|||
|
||||
/**
|
||||
* Remove mapping for UUID
|
||||
*
|
||||
* @param generation - Brainy's commit generation for this removal (contract
|
||||
* parity with the `EntityIdMapperProvider` surface). Accepted and ignored —
|
||||
* this JS mapper removes immediately; a native mapper tombstones the
|
||||
* mapping at this generation in its version chain.
|
||||
*/
|
||||
remove(uuid: string): boolean {
|
||||
remove(uuid: string, generation?: bigint): boolean {
|
||||
void generation // Contract parity — no per-key version chain in the JS mapper.
|
||||
const intId = this.uuidToInt.get(uuid)
|
||||
if (intId === undefined) {
|
||||
return false
|
||||
|
|
|
|||
|
|
@ -1459,8 +1459,16 @@ export class MetadataIndexManager implements MetadataIndexProvider {
|
|||
* @param id - Entity ID
|
||||
* @param entityOrMetadata - Either full entity structure or plain metadata (backward compat)
|
||||
* @param skipFlush - Skip automatic flush (used during batch operations)
|
||||
* @param deferWrites - Batch mode: buffer postings for a later flush
|
||||
* @param generation - Brainy's commit generation for this write (see the
|
||||
* {@link import('../plugin.js').MetadataIndexProvider} contract). This JS
|
||||
* manager keeps a single live view with no per-record delta log, so it
|
||||
* has no slot to store it — the value is accepted for contract parity
|
||||
* and forwarded to the shared id mapper (an injected native mapper
|
||||
* stamps its assignment records with it; the JS mapper ignores it).
|
||||
* The JS twin adopts full per-write stamping with the watermark train.
|
||||
*/
|
||||
async addToIndex(id: string, entityOrMetadata: any, skipFlush: boolean = false, deferWrites: boolean = false): Promise<void> {
|
||||
async addToIndex(id: string, entityOrMetadata: any, skipFlush: boolean = false, deferWrites: boolean = false, generation?: bigint): Promise<void> {
|
||||
const fields = this.extractIndexableFields(entityOrMetadata)
|
||||
|
||||
// Sanity check for excessive indexed fields (indicates possible data issue)
|
||||
|
|
@ -1508,7 +1516,10 @@ export class MetadataIndexManager implements MetadataIndexProvider {
|
|||
// element, so a scalar overwrite (last-value-wins) would index only the final
|
||||
// element and `contains` would miss the rest.
|
||||
if (this.columnStore) {
|
||||
const entityIntId = this.idMapper.getOrAssign(id)
|
||||
// Thread the commit generation into the mint: an injected native mapper
|
||||
// stamps the assignment record's delta log with the real watermark
|
||||
// instead of a literal 0 (the JS mapper accepts and ignores it).
|
||||
const entityIntId = this.idMapper.getOrAssign(id, generation)
|
||||
const fieldsMap: Record<string, unknown> = {}
|
||||
for (const { field, value } of fields) {
|
||||
if (field === '__words__') {
|
||||
|
|
@ -1600,8 +1611,13 @@ export class MetadataIndexManager implements MetadataIndexProvider {
|
|||
*
|
||||
* @param id - Entity ID to remove
|
||||
* @param metadata - Optional entity or metadata structure (if not provided, requires scanning all fields - slow!)
|
||||
* @param generation - Brainy's commit generation for this removal (see the
|
||||
* {@link import('../plugin.js').MetadataIndexProvider} contract). Accepted
|
||||
* for contract parity — this JS manager removes immediately (no tombstone
|
||||
* chain) and forwards it to the shared id mapper's `remove`, where an
|
||||
* injected native mapper tombstones the mapping at this generation.
|
||||
*/
|
||||
async removeFromIndex(id: string, metadata?: any): Promise<void> {
|
||||
async removeFromIndex(id: string, metadata?: any, generation?: bigint): Promise<void> {
|
||||
if (metadata) {
|
||||
const fields = this.extractIndexableFields(metadata)
|
||||
|
||||
|
|
@ -1625,7 +1641,9 @@ export class MetadataIndexManager implements MetadataIndexProvider {
|
|||
// Clean up ID mapper — must happen AFTER column store removal since it uses
|
||||
// idMapper.getInt(id). Prevents deleted IDs from persisting in the mapper
|
||||
// universe, which would cause ne/exists:false queries to return deleted entities.
|
||||
this.idMapper.remove(id)
|
||||
// The generation rides along so a native mapper tombstones the mapping at
|
||||
// the real commit watermark (the JS mapper ignores it).
|
||||
this.idMapper.remove(id, generation)
|
||||
await this.idMapper.flush()
|
||||
}
|
||||
|
||||
|
|
|
|||
276
tests/unit/plugin/provider-generation.test.ts
Normal file
276
tests/unit/plugin/provider-generation.test.ts
Normal file
|
|
@ -0,0 +1,276 @@
|
|||
/**
|
||||
* Generation threading to the metadata-index and vector-index provider write
|
||||
* surfaces — the counterpart of the graph pins in
|
||||
* tests/unit/transaction/graphIndexOperations-generation.test.ts.
|
||||
*
|
||||
* The provider contract gained an optional trailing `generation?: bigint` on
|
||||
* `MetadataIndexProvider.addToIndex`/`removeFromIndex`,
|
||||
* `VectorIndexProvider.addItem`/`removeItem` (+ the feature-detected
|
||||
* `updateItem`), and the id-mapper's `getOrAssign`/`remove`. A native provider
|
||||
* with per-record delta logs stamps its durable records with it — so the value
|
||||
* arriving MUST be the real commit generation (nonzero, monotonic), never a
|
||||
* fabricated 0 and never absent on the coordinator's write paths.
|
||||
*
|
||||
* Two layers of pins:
|
||||
* 1. End-to-end: provider doubles registered via the plugin system capture
|
||||
* the generation argument during brain.add()/update()/remove() and it
|
||||
* must equal the committed watermark (`brain.now().generation`).
|
||||
* 2. Operation layer: execute-time (not construction-time) resolution, and
|
||||
* one shared generation across an op's forward + rollback halves.
|
||||
*/
|
||||
import { describe, it, expect, afterEach } from 'vitest'
|
||||
import { Brainy, NounType } from '../../../src/index.js'
|
||||
import { MetadataIndexManager } from '../../../src/utils/metadataIndex.js'
|
||||
import {
|
||||
AddToVectorIndexOperation,
|
||||
RemoveFromVectorIndexOperation,
|
||||
ReplaceInVectorIndexOperation,
|
||||
AddToMetadataIndexOperation,
|
||||
RemoveFromMetadataIndexOperation
|
||||
} from '../../../src/transaction/operations/IndexOperations.js'
|
||||
import type { VectorIndexProvider } from '../../../src/plugin.js'
|
||||
|
||||
const V = () => Array.from({ length: 384 }, () => Math.random())
|
||||
|
||||
type Captured = { method: string; id: string; generation: bigint | undefined }
|
||||
|
||||
const brains: Brainy[] = []
|
||||
afterEach(async () => {
|
||||
for (const b of brains.splice(0)) await b.close().catch(() => {})
|
||||
})
|
||||
|
||||
/** Metadata manager subclass that records the generation of every write. */
|
||||
function makeCapturingMetadataFactory(calls: Captured[]) {
|
||||
return (storage: any) => {
|
||||
class CapturingManager extends MetadataIndexManager {
|
||||
async addToIndex(id: string, entityOrMetadata: any, skipFlush = false, deferWrites = false, generation?: bigint): Promise<void> {
|
||||
calls.push({ method: 'addToIndex', id, generation })
|
||||
return super.addToIndex(id, entityOrMetadata, skipFlush, deferWrites, generation)
|
||||
}
|
||||
async removeFromIndex(id: string, metadata?: any, generation?: bigint): Promise<void> {
|
||||
calls.push({ method: 'removeFromIndex', id, generation })
|
||||
return super.removeFromIndex(id, metadata, generation)
|
||||
}
|
||||
}
|
||||
return new CapturingManager(storage)
|
||||
}
|
||||
}
|
||||
|
||||
/** Minimal vector-index double capturing the generation of every write. */
|
||||
function makeCapturingVectorFactory(calls: Captured[]) {
|
||||
return () => {
|
||||
const items = new Map<string, number[]>()
|
||||
const double: VectorIndexProvider & { updateItem(item: { id: string; vector: number[] }, generation?: bigint): Promise<void> } = {
|
||||
name: 'capture-double',
|
||||
async addItem(item, generation) {
|
||||
calls.push({ method: 'addItem', id: item.id, generation })
|
||||
items.set(item.id, item.vector as number[])
|
||||
return item.id
|
||||
},
|
||||
async removeItem(id, generation) {
|
||||
calls.push({ method: 'removeItem', id, generation })
|
||||
return items.delete(id)
|
||||
},
|
||||
async updateItem(item, generation) {
|
||||
calls.push({ method: 'updateItem', id: item.id, generation })
|
||||
items.set(item.id, item.vector)
|
||||
},
|
||||
async search() { return [] },
|
||||
size: () => items.size,
|
||||
clear: () => { items.clear() },
|
||||
async rebuild() {},
|
||||
async flush() { return 0 },
|
||||
getPersistMode: () => 'deferred' as const
|
||||
}
|
||||
return double
|
||||
}
|
||||
}
|
||||
|
||||
async function makeBrain(plugin: any): Promise<Brainy> {
|
||||
const brain = new Brainy({
|
||||
storage: { type: 'memory' },
|
||||
requireSubtype: false,
|
||||
silent: true,
|
||||
plugins: []
|
||||
})
|
||||
brain.use(plugin)
|
||||
await brain.init()
|
||||
brains.push(brain)
|
||||
return brain
|
||||
}
|
||||
|
||||
describe('Metadata-index provider — real commit generation on every write (end-to-end)', () => {
|
||||
it('add()/update()/remove() pass the nonzero, monotonic commit generation to addToIndex/removeFromIndex', async () => {
|
||||
const calls: Captured[] = []
|
||||
const brain = await makeBrain({
|
||||
name: 'capture-metadata',
|
||||
activate: async (ctx: any) => {
|
||||
ctx.registerProvider('metadataIndex', makeCapturingMetadataFactory(calls))
|
||||
return true
|
||||
}
|
||||
})
|
||||
|
||||
const id = await brain.add({ data: 'one', type: NounType.Concept, metadata: { k: 'a' }, vector: V() })
|
||||
const addCall = calls.find((c) => c.method === 'addToIndex' && c.id === id)
|
||||
expect(addCall).toBeDefined()
|
||||
expect(typeof addCall!.generation).toBe('bigint')
|
||||
expect(addCall!.generation!).toBeGreaterThan(0n)
|
||||
// Committed watermark after a single-op write IS this write's generation.
|
||||
expect(addCall!.generation!).toBe(BigInt(brain.now().generation))
|
||||
|
||||
calls.length = 0
|
||||
await brain.update({ id, metadata: { k: 'b' } })
|
||||
const updRemove = calls.find((c) => c.method === 'removeFromIndex' && c.id === id)
|
||||
const updAdd = calls.find((c) => c.method === 'addToIndex' && c.id === id)
|
||||
expect(updRemove?.generation).toBeDefined()
|
||||
expect(updAdd?.generation).toBeDefined()
|
||||
// One commit → the remove-old + add-new legs share one watermark.
|
||||
expect(updAdd!.generation!).toBe(updRemove!.generation!)
|
||||
expect(updAdd!.generation!).toBe(BigInt(brain.now().generation))
|
||||
const updateGen = updAdd!.generation!
|
||||
expect(updateGen).toBeGreaterThan(0n)
|
||||
|
||||
calls.length = 0
|
||||
await brain.remove(id)
|
||||
const rmCall = calls.find((c) => c.method === 'removeFromIndex' && c.id === id)
|
||||
expect(rmCall?.generation).toBeDefined()
|
||||
expect(rmCall!.generation!).toBeGreaterThan(updateGen) // monotonic
|
||||
expect(rmCall!.generation!).toBe(BigInt(brain.now().generation))
|
||||
})
|
||||
|
||||
it('transact() adds stamp the batch receipt generation', async () => {
|
||||
const calls: Captured[] = []
|
||||
const brain = await makeBrain({
|
||||
name: 'capture-metadata-tx',
|
||||
activate: async (ctx: any) => {
|
||||
ctx.registerProvider('metadataIndex', makeCapturingMetadataFactory(calls))
|
||||
return true
|
||||
}
|
||||
})
|
||||
|
||||
// Bootstrap honesty: init-time infrastructure writes (the VFS root) are
|
||||
// applied WITHOUT a generation — the provider must receive undefined,
|
||||
// never a fabricated 0.
|
||||
for (const c of calls) expect(c.generation).toBeUndefined()
|
||||
calls.length = 0
|
||||
|
||||
const db = await brain.transact([
|
||||
{ op: 'add', data: 'tx-one', type: NounType.Concept, vector: V() },
|
||||
{ op: 'add', data: 'tx-two', type: NounType.Concept, vector: V() }
|
||||
] as any)
|
||||
|
||||
const receiptGen = BigInt(db.receipt!.generation)
|
||||
const addGens = calls.filter((c) => c.method === 'addToIndex').map((c) => c.generation)
|
||||
expect(addGens.length).toBeGreaterThanOrEqual(2)
|
||||
for (const g of addGens) expect(g).toBe(receiptGen)
|
||||
})
|
||||
})
|
||||
|
||||
describe('Vector-index provider — real commit generation on every write (end-to-end)', () => {
|
||||
it('add()/update()/remove() pass the nonzero commit generation to addItem/updateItem/removeItem', async () => {
|
||||
const calls: Captured[] = []
|
||||
const brain = await makeBrain({
|
||||
name: 'capture-vector',
|
||||
activate: async (ctx: any) => {
|
||||
ctx.registerProvider('vector', makeCapturingVectorFactory(calls))
|
||||
return true
|
||||
}
|
||||
})
|
||||
|
||||
const id = await brain.add({ data: 'vec', type: NounType.Concept, vector: V() })
|
||||
const addCall = calls.find((c) => c.method === 'addItem' && c.id === id)
|
||||
expect(addCall).toBeDefined()
|
||||
expect(typeof addCall!.generation).toBe('bigint')
|
||||
expect(addCall!.generation!).toBeGreaterThan(0n)
|
||||
expect(addCall!.generation!).toBe(BigInt(brain.now().generation))
|
||||
|
||||
calls.length = 0
|
||||
await brain.update({ id, vector: V() })
|
||||
const updCall = calls.find((c) => c.method === 'updateItem' && c.id === id)
|
||||
expect(updCall?.generation).toBeDefined()
|
||||
expect(updCall!.generation!).toBeGreaterThan(addCall!.generation!) // monotonic
|
||||
expect(updCall!.generation!).toBe(BigInt(brain.now().generation))
|
||||
|
||||
calls.length = 0
|
||||
await brain.remove(id)
|
||||
const rmCall = calls.find((c) => c.method === 'removeItem' && c.id === id)
|
||||
expect(rmCall?.generation).toBeDefined()
|
||||
expect(rmCall!.generation!).toBeGreaterThan(updCall!.generation!)
|
||||
expect(rmCall!.generation!).toBe(BigInt(brain.now().generation))
|
||||
})
|
||||
})
|
||||
|
||||
describe('Index operations — generation threading (operation layer)', () => {
|
||||
function makeVectorSpy() {
|
||||
const calls: Array<{ method: string; generation: bigint | undefined }> = []
|
||||
const index = {
|
||||
name: 'spy',
|
||||
async addItem(_item: any, generation?: bigint) { calls.push({ method: 'addItem', generation }); return 'x' },
|
||||
async removeItem(_id: string, generation?: bigint) { calls.push({ method: 'removeItem', generation }); return true },
|
||||
async updateItem(_item: any, generation?: bigint) { calls.push({ method: 'updateItem', generation }) }
|
||||
} as unknown as VectorIndexProvider
|
||||
return { index, calls }
|
||||
}
|
||||
|
||||
it('vector add/remove/replace resolve the thunk at EXECUTE time and reuse one generation for rollback', async () => {
|
||||
const { index, calls } = makeVectorSpy()
|
||||
let current = 1n
|
||||
const op = new AddToVectorIndexOperation(index, 'id-1', [1, 2], () => current)
|
||||
current = 42n // assigned after construction, read at execute
|
||||
const rollback = await op.execute()
|
||||
expect(calls[0]).toEqual({ method: 'addItem', generation: 42n })
|
||||
current = 77n // rollback must NOT re-read — one watermark per round trip
|
||||
await rollback()
|
||||
expect(calls[1]).toEqual({ method: 'removeItem', generation: 42n })
|
||||
|
||||
calls.length = 0
|
||||
const rm = new RemoveFromVectorIndexOperation(index, 'id-1', [1, 2], () => 7n)
|
||||
const rb2 = await rm.execute()
|
||||
await rb2()
|
||||
expect(calls).toEqual([
|
||||
{ method: 'removeItem', generation: 7n },
|
||||
{ method: 'addItem', generation: 7n }
|
||||
])
|
||||
|
||||
calls.length = 0
|
||||
const rep = new ReplaceInVectorIndexOperation(index, 'id-1', [1, 2], [3, 4], () => 9n)
|
||||
const rb3 = await rep.execute()
|
||||
await rb3()
|
||||
expect(calls).toEqual([
|
||||
{ method: 'updateItem', generation: 9n },
|
||||
{ method: 'updateItem', generation: 9n }
|
||||
])
|
||||
})
|
||||
|
||||
it('metadata add/remove pass the resolved generation through both halves', async () => {
|
||||
const calls: Array<{ method: string; generation: bigint | undefined }> = []
|
||||
const manager = {
|
||||
async addToIndex(_id: string, _e: any, _s?: boolean, _d?: boolean, generation?: bigint) {
|
||||
calls.push({ method: 'addToIndex', generation })
|
||||
},
|
||||
async removeFromIndex(_id: string, _m?: any, generation?: bigint) {
|
||||
calls.push({ method: 'removeFromIndex', generation })
|
||||
}
|
||||
} as unknown as MetadataIndexManager
|
||||
|
||||
const add = new AddToMetadataIndexOperation(manager, 'id-1', { type: 'x' }, () => 11n)
|
||||
const rb = await add.execute()
|
||||
await rb()
|
||||
const rm = new RemoveFromMetadataIndexOperation(manager, 'id-1', { type: 'x' }, () => 12n)
|
||||
const rb2 = await rm.execute()
|
||||
await rb2()
|
||||
expect(calls).toEqual([
|
||||
{ method: 'addToIndex', generation: 11n },
|
||||
{ method: 'removeFromIndex', generation: 11n },
|
||||
{ method: 'removeFromIndex', generation: 12n },
|
||||
{ method: 'addToIndex', generation: 12n }
|
||||
])
|
||||
})
|
||||
|
||||
it('omitted thunk (legacy caller) → provider receives undefined, never a fabricated 0', async () => {
|
||||
const { index, calls } = makeVectorSpy()
|
||||
const op = new AddToVectorIndexOperation(index, 'id-1', [1, 2])
|
||||
await op.execute()
|
||||
expect(calls[0]).toEqual({ method: 'addItem', generation: undefined })
|
||||
})
|
||||
})
|
||||
Loading…
Add table
Add a link
Reference in a new issue