David Snelling
73500e7d10
transact()'s delete legs (direct unrelate and the noun-remove cascade) hand the SAME verb object to the graph-retraction op and the metadata-retraction op. The metadata leg sanitized at PLAN time, when the verb was still clean, so the wrap returned the same reference — then the graph op's execute-time endpoint resolution (deliberately deferred for same-batch forward refs) mirrored BigInt sourceInt/targetInt onto the shared object, and the metadata op crossed the seam with them. A strict provider rightly refuses that crossing, so every transact-wrapped edge delete aborted; direct unrelate() resolves ints at build time, before its sanitize, which is why no existing gate saw it. The JSON-safe view now lives in a shared leaf (utils/jsonSafeIndexMetadata) and is applied INSIDE AddToMetadataIndexOperation and RemoveFromMetadataIndexOperation at execute and rollback time — the one place no plan-vs-execute ordering can bypass. Pins: the fleet repro, the cascade shape, a mixed batch, and unit pins that mutate the entity after construction against a strict seam (5 red before, 5 green after).
683 lines
28 KiB
TypeScript
683 lines
28 KiB
TypeScript
/**
|
|
* Index Operations with Rollback Support
|
|
*
|
|
* Provides transactional operations for all indexes:
|
|
* - VectorIndexProvider (the JS HNSW fallback, or a native acceleration provider)
|
|
* - MetadataIndexManager (roaring bitmap filtering)
|
|
* - GraphAdjacencyIndex (LSM-tree graph storage)
|
|
*
|
|
* Each operation can be executed and rolled back atomically.
|
|
*/
|
|
|
|
import type { VectorIndexProvider, GraphIndexProvider } from '../../plugin.js'
|
|
import type { MetadataIndexManager } from '../../utils/metadataIndex.js'
|
|
import type { GraphVerb } from '../../coreTypes.js'
|
|
import type { Operation, RollbackAction } from '../types.js'
|
|
import { isZeroNormVector } from '../../utils/distance.js'
|
|
import { jsonSafeIndexMetadata } from '../../utils/jsonSafeIndexMetadata.js'
|
|
import { prodLog } from '../../utils/logger.js'
|
|
|
|
/**
|
|
* Backend identity stamped into an operation's emitted `name` string (e.g.
|
|
* `AddToVectorIndex(hnsw-js)`), resolved from the provider's own
|
|
* {@link VectorIndexProvider.name} self-report.
|
|
*
|
|
* These operation classes are backend-neutral (the vector index they wrap may
|
|
* be Brainy's own JS HNSW fallback OR a native acceleration provider), but
|
|
* their names surface directly in consumer-visible transaction journals and
|
|
* timings. `name` is REQUIRED at the TypeScript level — but a native provider
|
|
* instance compiled against the previous (pre-required) contract can still
|
|
* reach this function at runtime without it. That legacy case is tolerated,
|
|
* never crashed on and never silently mislabeled: it resolves to
|
|
* `'unknown-provider'` and emits ONE loud warning naming the missing contract
|
|
* field, so the fix (implement `name`) is discoverable rather than a silent
|
|
* fossil label in every journal line thereafter.
|
|
*/
|
|
const warnedMissingName = new WeakSet<VectorIndexProvider>()
|
|
|
|
function resolveVectorProviderId(index: VectorIndexProvider): string {
|
|
const name = (index as { name?: unknown }).name
|
|
if (typeof name === 'string') return name
|
|
if (!warnedMissingName.has(index)) {
|
|
warnedMissingName.add(index)
|
|
console.warn(
|
|
'[vector-index] provider is missing the required `name` field (VectorIndexProvider.name, ' +
|
|
'required since 8.10.0) — stamping "unknown-provider" in transaction op names until the ' +
|
|
'provider declares its own identity.'
|
|
)
|
|
}
|
|
return 'unknown-provider'
|
|
}
|
|
|
|
/**
|
|
* Add to the vector index with rollback support.
|
|
*
|
|
* Backend-neutral: `index` is whatever the `'vector'` provider factory
|
|
* returns — Brainy's own JS HNSW fallback, or a native acceleration provider
|
|
* (e.g. DiskANN). The emitted `name` stamps the active backend (see
|
|
* {@link resolveVectorProviderId}) so operators reading a transaction journal
|
|
* 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 generationFn?: () => bigint | undefined
|
|
) {
|
|
this.name = `AddToVectorIndex(${resolveVectorProviderId(index)})`
|
|
}
|
|
|
|
async execute(): Promise<RollbackAction> {
|
|
// THE ZERO-NORM LAW (the live provider-write seam's belt): a zero-norm
|
|
// vector is not a vector — it never crosses an engine boundary. This
|
|
// engine's own cosine distance treats an all-zero vector safely (a
|
|
// zero-norm operand always scores MAXIMUM distance, see
|
|
// {@link isZeroNormVector}'s JSDoc), but a downstream engine serving
|
|
// squared-euclidean distance cannot tell it apart from a legitimate
|
|
// origin point — a false attractor that silently darkens real results.
|
|
// The canonical write already landed (SaveNoun/SaveNounMetadata
|
|
// operations are staged ahead of this one in every caller) — only the
|
|
// INDEX INSERT is refused here, loudly, never a throw. A length-0
|
|
// vector is the unrelated "unvectored" shape and is skipped silently
|
|
// (the same contract callers already rely on for deferred embeds).
|
|
if (this.vector.length === 0) {
|
|
return async () => {}
|
|
}
|
|
if (isZeroNormVector(this.vector)) {
|
|
prodLog.warn(
|
|
`[vector-index] refusing to index a zero-norm vector for entity ${this.id} — ` +
|
|
`a zero-norm vector is not a vector and never crosses an engine boundary ` +
|
|
`(the canonical write is unaffected; only the vector-index insert is skipped)`
|
|
)
|
|
return async () => {}
|
|
}
|
|
|
|
// 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 }, generation)
|
|
|
|
// Return rollback action
|
|
return async () => {
|
|
if (!existed) {
|
|
// Remove newly added item
|
|
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
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Check if item exists in index.
|
|
*
|
|
* `getItem` is an optional, feature-detected provider capability — see the
|
|
* VectorIndexProvider docs; it is intentionally absent from the required
|
|
* contract (Brainy's JS HNSW index omits it). When the capability is
|
|
* missing the answer must be `false`, not `true`: treating unknowable
|
|
* pre-existence as "existed" made every rollback skip removeItem, leaving
|
|
* phantom entries in the index after a failed transaction. The safe default
|
|
* is to remove what this operation added — update flows pair this op with a
|
|
* RemoveFromVectorIndexOperation whose own rollback restores the prior
|
|
* vector, so reverse-order rollback reconstructs the original state either
|
|
* way.
|
|
*/
|
|
private async itemExists(id: string): Promise<boolean> {
|
|
const index = this.index as VectorIndexProvider & {
|
|
getItem?: (id: string) => Promise<unknown>
|
|
}
|
|
if (typeof index.getItem !== 'function') return false
|
|
try {
|
|
const item = await index.getItem(id)
|
|
return item !== undefined && item !== null
|
|
} catch {
|
|
return false
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Remove from the vector index with rollback support.
|
|
*
|
|
* Backend-neutral: see {@link AddToVectorIndexOperation} — `index` may be the
|
|
* JS HNSW fallback or a native acceleration provider; the emitted `name`
|
|
* stamps the active backend.
|
|
*
|
|
* Rollback strategy:
|
|
* - Re-add item to index with original vector
|
|
*
|
|
* Note: Requires storing the vector for rollback
|
|
*/
|
|
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 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, generation)
|
|
|
|
// Return rollback action
|
|
return async () => {
|
|
// Re-add item with original vector
|
|
await this.index.addItem({ id: this.id, vector: this.vector }, generation)
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Replace an item's vector in the vector index as ONE atomic transaction leg —
|
|
* the row is never absent from vector search during an update.
|
|
*
|
|
* Backend-neutral: see {@link AddToVectorIndexOperation} — `index` may be the
|
|
* JS HNSW fallback or a native acceleration provider; the emitted `name`
|
|
* stamps the active backend.
|
|
*
|
|
* Why this op exists: update flows historically staged a
|
|
* {@link RemoveFromVectorIndexOperation} followed by an
|
|
* {@link AddToVectorIndexOperation} as two separately-awaited operations.
|
|
* Between them the row was in NEITHER index — dark to semantic recall while
|
|
* perfectly visible to metadata reads (a transient-invisibility window that
|
|
* stretched to seconds in a production deployment). The structural cure is a
|
|
* single leg that never removes without simultaneously re-inserting.
|
|
*
|
|
* Execution strategy (feature-detected, in preference order):
|
|
* 1. Provider exposes `updateItem` → ONE in-place call. The provider swaps
|
|
* the vector without the row ever leaving its index, and an element-wise
|
|
* UNCHANGED vector (the type-only-update production shape) is a pure
|
|
* no-op on its side.
|
|
* 2. Provider without `updateItem` (a native provider that has not shipped
|
|
* it yet) → `removeItem` + `addItem` executed ADJACENT within this single
|
|
* op. Still strictly better than the historical pair: no other transaction
|
|
* operation can interleave between the two calls. This is a temporary
|
|
* seam — the native side of the pair is expected to ship its own
|
|
* `updateItem` so path 1 applies everywhere; when it does, this fallback
|
|
* becomes dead code that costs nothing.
|
|
*
|
|
* Rollback strategy (mirrors the execute branch that ran):
|
|
* - `updateItem` path → `updateItem` back to `oldVector`.
|
|
* - Fallback path → `removeItem` + `addItem` back to `oldVector`.
|
|
*
|
|
* Rollback semantics when the item did not exist at execute time: this op's
|
|
* contract is that the caller read the entity and its CURRENT vector
|
|
* (`oldVector`) before staging — update flows only stage it for existing
|
|
* rows. If the item was somehow absent, execute() inserts it (`updateItem`
|
|
* delegates to add; the fallback's remove is a no-op before its add), and
|
|
* rollback restores `oldVector` rather than removing — the same posture as
|
|
* {@link RemoveFromVectorIndexOperation}'s unconditional re-add: by
|
|
* constructing the op with `oldVector` the caller DECLARED the before-state,
|
|
* and rollback reconstructs that declared state instead of silently deciding
|
|
* the row should vanish.
|
|
*/
|
|
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 generationFn?: () => bigint | undefined
|
|
) {
|
|
this.name = `ReplaceInVectorIndex(${resolveVectorProviderId(index)})`
|
|
}
|
|
|
|
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). 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[] }, generation?: bigint) => Promise<void>
|
|
}
|
|
|
|
// One commit generation for the whole replace (both branches + rollback).
|
|
const generation = this.generationFn?.()
|
|
|
|
// THE ZERO-NORM LAW (see AddToVectorIndexOperation's matching JSDoc): a
|
|
// real all-zero replacement vector must never land in the index — refuse
|
|
// loudly, canonical write unaffected. The row must not be left stale
|
|
// either: if it was genuinely indexed under `oldVector`, remove it
|
|
// rather than pretend the old vector still describes the row. A
|
|
// length-0 `newVector` (the unrelated "unvectored" shape) is handled the
|
|
// same way, silently — no caller today reaches this with an empty
|
|
// replacement (update() rejects a dimension-mismatched empty vector),
|
|
// but the seam stays consistent in case one ever legitimately does.
|
|
if (isZeroNormVector(this.newVector) || this.newVector.length === 0) {
|
|
const wasIndexed = this.oldVector.length > 0 && !isZeroNormVector(this.oldVector)
|
|
if (isZeroNormVector(this.newVector)) {
|
|
prodLog.warn(
|
|
`[vector-index] refusing to replace with a zero-norm vector for entity ${this.id} — ` +
|
|
`a zero-norm vector is not a vector and never crosses an engine boundary ` +
|
|
`(the canonical write is unaffected; the row is removed from the vector index instead)`
|
|
)
|
|
}
|
|
if (wasIndexed) {
|
|
await this.index.removeItem(this.id, generation)
|
|
}
|
|
return async () => {
|
|
// Restore the declared before-state.
|
|
if (wasIndexed) {
|
|
await this.index.addItem({ id: this.id, vector: this.oldVector }, generation)
|
|
}
|
|
}
|
|
}
|
|
|
|
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 }, generation)
|
|
|
|
return async () => {
|
|
// Restore the declared before-state in place (see class JSDoc for
|
|
// the item-did-not-exist posture). A length-0 oldVector means the row
|
|
// was never actually indexed before this op ran (a length-0 vector is
|
|
// never a legal index member — see EmptyVectorIndexError) — there is
|
|
// no in-place "restore to empty" for the provider to perform, so
|
|
// rollback removes the row instead, leaving the same "not indexed"
|
|
// state the row was in before execute().
|
|
if (this.oldVector.length > 0) {
|
|
await index.updateItem!({ id: this.id, vector: this.oldVector }, generation)
|
|
} else {
|
|
await this.index.removeItem(this.id, 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, 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. Same length-0 carve-out as the updateItem
|
|
// path above: an empty oldVector was never a legal index member, so
|
|
// rollback just leaves the row removed rather than attempting an
|
|
// illegal empty re-add.
|
|
await this.index.removeItem(this.id, generation)
|
|
if (this.oldVector.length > 0) {
|
|
await this.index.addItem({ id: this.id, vector: this.oldVector }, generation)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 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 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?.()
|
|
|
|
// The JSON-safe view is taken HERE, per crossing, never at construction:
|
|
// the entity reference this op holds can be mutated between plan and
|
|
// execute (a graph op's execute-time endpoint-int resolution mirrors
|
|
// BigInts onto a shared verb object) — see jsonSafeIndexMetadata's
|
|
// module doc.
|
|
await this.index.addToIndex(
|
|
this.id, jsonSafeIndexMetadata(this.entity), true, false, generation
|
|
)
|
|
|
|
// Return rollback action
|
|
return async () => {
|
|
// Remove from metadata index
|
|
await this.index.removeFromIndex(
|
|
this.id, jsonSafeIndexMetadata(this.entity), generation
|
|
)
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 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 generationFn?: () => bigint | undefined
|
|
) {}
|
|
|
|
async execute(): Promise<RollbackAction> {
|
|
// Resolve the removal generation once; reuse it for the rollback re-add.
|
|
const generation = this.generationFn?.()
|
|
|
|
// Sanitized per crossing, never at construction — transact()'s delete
|
|
// legs hand this op the SAME verb object the graph-retraction op's
|
|
// execute-time endpoint resolution mutates (BigInt sourceInt/targetInt),
|
|
// so a plan-time view aliases the pollution. See jsonSafeIndexMetadata's
|
|
// module doc.
|
|
await this.index.removeFromIndex(
|
|
this.id, jsonSafeIndexMetadata(this.entity), generation
|
|
)
|
|
|
|
// Return rollback action
|
|
return async () => {
|
|
// Re-add with original metadata (skipFlush=true)
|
|
await this.index.addToIndex(
|
|
this.id, jsonSafeIndexMetadata(this.entity), true, false, generation
|
|
)
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Add verb to graph index with rollback support
|
|
*
|
|
* Rollback strategy:
|
|
* - Remove verb from graph index
|
|
*
|
|
* 8.0 u64 contract: the coordinator resolves both endpoint ints via the
|
|
* shared idMapper (`getOrAssign`) and passes them alongside the verb; the
|
|
* provider returns the interned verb int, which is surfaced through the
|
|
* optional `onVerbInt` callback so the coordinator can feed its warm cache.
|
|
*
|
|
* Generation: `generationFn` is resolved at execute time (not construction) so
|
|
* the edge 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-generation edge chain.
|
|
*/
|
|
/**
|
|
* A verb's interned endpoint ints — eager (already resolved), or a thunk
|
|
* evaluated when the operation EXECUTES. The lazy form exists for
|
|
* `transact()` forward references: a relate whose endpoint is added in the
|
|
* SAME batch cannot resolve ints at plan time (the entity does not exist yet
|
|
* — a strict native id mapper rightly refuses to assign, and even a permissive
|
|
* one would leak the assignment if the batch is rejected at precommit).
|
|
* Deferring to execute time resolves after the batch's add operations have
|
|
* applied, mirroring how `generationFn` is already evaluated lazily.
|
|
*/
|
|
export type VerbEndpointInts =
|
|
| { sourceInt: bigint; targetInt: bigint }
|
|
| (() => { sourceInt: bigint; targetInt: bigint })
|
|
|
|
/** Resolve a {@link VerbEndpointInts} at execution time. */
|
|
function resolveEndpointInts(
|
|
endpoints: VerbEndpointInts
|
|
): { sourceInt: bigint; targetInt: bigint } {
|
|
return typeof endpoints === 'function' ? endpoints() : endpoints
|
|
}
|
|
|
|
export class AddToGraphIndexOperation implements Operation {
|
|
readonly name = 'AddToGraphIndex'
|
|
|
|
/**
|
|
* @param index - The graph-index provider (JS baseline or native).
|
|
* @param verb - The verb to index (`sourceInt`/`targetInt` mirrored on it).
|
|
* @param endpointInts - The endpoints' interned ints — eager, or a thunk
|
|
* evaluated at execute time (REQUIRED for transact forward references;
|
|
* see {@link VerbEndpointInts}).
|
|
* @param generationFn - Resolves the commit generation to stamp this edge at,
|
|
* evaluated when the operation executes (see class note).
|
|
* @param onVerbInt - Optional hook invoked with the interned verb int
|
|
* returned by the provider (feeds the coordinator's verb-int warm cache).
|
|
*/
|
|
constructor(
|
|
private readonly index: GraphIndexProvider,
|
|
private readonly verb: GraphVerb,
|
|
private readonly endpointInts: VerbEndpointInts,
|
|
private readonly generationFn: () => bigint,
|
|
private readonly onVerbInt?: (verbInt: bigint) => void
|
|
) {}
|
|
|
|
async execute(): Promise<RollbackAction> {
|
|
// 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 { sourceInt, targetInt } = resolveEndpointInts(this.endpointInts)
|
|
const verbInt = await this.index.addVerb(this.verb, sourceInt, targetInt, generation)
|
|
this.onVerbInt?.(verbInt)
|
|
|
|
// Return rollback action
|
|
return async () => {
|
|
// Remove verb from graph index
|
|
await this.index.removeVerb(this.verb.id, generation)
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Remove verb from graph index with rollback support
|
|
*
|
|
* Rollback strategy:
|
|
* - Re-add verb to graph index
|
|
*
|
|
* 8.0 u64 contract: rollback re-adds through `addVerb(verb, sourceInt,
|
|
* targetInt, generation)`, so the coordinator resolves the endpoint ints up
|
|
* front (while the entity → int mappings are guaranteed to still exist). The
|
|
* removal generation is resolved at execute time and reused for the rollback
|
|
* re-add, so the round trip references one watermark.
|
|
*/
|
|
export class RemoveFromGraphIndexOperation implements Operation {
|
|
readonly name = 'RemoveFromGraphIndex'
|
|
|
|
/**
|
|
* @param index - The graph-index provider (JS baseline or native).
|
|
* @param verb - The verb being removed (required for rollback re-add).
|
|
* @param endpointInts - The endpoints' interned ints for the rollback
|
|
* re-add — eager, or a thunk evaluated at execute time (required when the
|
|
* verb or its endpoints were created in the SAME transact batch; see
|
|
* {@link VerbEndpointInts}).
|
|
* @param generationFn - Resolves the commit generation for this removal,
|
|
* evaluated when the operation executes.
|
|
*/
|
|
constructor(
|
|
private readonly index: GraphIndexProvider,
|
|
private readonly verb: GraphVerb, // Required for rollback
|
|
private readonly endpointInts: VerbEndpointInts,
|
|
private readonly generationFn: () => bigint
|
|
) {}
|
|
|
|
async execute(): Promise<RollbackAction> {
|
|
// 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 { sourceInt, targetInt } = resolveEndpointInts(this.endpointInts)
|
|
await this.index.removeVerb(this.verb.id, generation)
|
|
|
|
// Return rollback action
|
|
return async () => {
|
|
// Re-add verb with original data
|
|
await this.index.addVerb(this.verb, sourceInt, targetInt, generation)
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Batch operation: Add multiple items to the vector index (backend-neutral —
|
|
* see {@link AddToVectorIndexOperation}).
|
|
*
|
|
* Useful for bulk imports with transaction support.
|
|
* Rolls back all items if any fail.
|
|
*/
|
|
export class BatchAddToVectorIndexOperation implements Operation {
|
|
readonly name: string
|
|
|
|
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[] }>,
|
|
generationFn?: () => bigint | undefined
|
|
) {
|
|
this.name = `BatchAddToVectorIndex(${resolveVectorProviderId(index)})`
|
|
this.operations = items.map(
|
|
item => new AddToVectorIndexOperation(index, item.id, item.vector, generationFn)
|
|
)
|
|
}
|
|
|
|
async execute(): Promise<RollbackAction> {
|
|
const rollbackActions: RollbackAction[] = []
|
|
|
|
// Execute all operations
|
|
for (const op of this.operations) {
|
|
const rollback = await op.execute()
|
|
if (rollback) {
|
|
rollbackActions.push(rollback)
|
|
}
|
|
}
|
|
|
|
// Return combined rollback action
|
|
return async () => {
|
|
// Execute all rollbacks in reverse order
|
|
for (let i = rollbackActions.length - 1; i >= 0; i--) {
|
|
await rollbackActions[i]()
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Batch operation: Add multiple entities to metadata index
|
|
*
|
|
* Useful for bulk imports with transaction support.
|
|
*/
|
|
export class BatchAddToMetadataIndexOperation implements Operation {
|
|
readonly name = 'BatchAddToMetadataIndex'
|
|
|
|
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 }>,
|
|
generationFn?: () => bigint | undefined
|
|
) {
|
|
this.operations = items.map(
|
|
item => new AddToMetadataIndexOperation(index, item.id, item.entity, generationFn)
|
|
)
|
|
}
|
|
|
|
async execute(): Promise<RollbackAction> {
|
|
const rollbackActions: RollbackAction[] = []
|
|
|
|
// Execute all operations
|
|
for (const op of this.operations) {
|
|
const rollback = await op.execute()
|
|
if (rollback) {
|
|
rollbackActions.push(rollback)
|
|
}
|
|
}
|
|
|
|
// Return combined rollback action
|
|
return async () => {
|
|
// Execute all rollbacks in reverse order
|
|
for (let i = rollbackActions.length - 1; i >= 0; i--) {
|
|
await rollbackActions[i]()
|
|
}
|
|
}
|
|
}
|
|
}
|