feat(update-seam): first-class update operation through the provider seam
Brainy's planner emits a single UpdateInMetadataIndexOperation / UpdateVerbInGraphIndexOperation when the registered provider announces the `update-op` capability (capabilities has 'update-op' AND the method exists — both halves), and the legacy remove+add pair otherwise. A provider whose capability set claims what the instance lacks is refused at registration with a typed ProviderCapabilityMismatchError — never a silent fallback. - MetadataIndexProvider.updateIndex(id, before, after, generation?) and GraphIndexProvider.updateVerb(id, before, after, generation) — optional, rollback symmetric by construction (update(a,b) undone by update(b,a)). - Emission at update(), planTxUpdate() and updateRelation()'s graph leg. - transact() gains op:'updateRelation' (TxUpdateRelationOperation), born batchable; updateRelation()'s record build is shared with the planner. - Both paths live for one overlap release; the pair path retires with the provider-side compensation layer in the following cut. Pinned in tests/integration/update-op-emission.test.ts (7 pins).
This commit is contained in:
parent
7c8c8be30c
commit
0c028dfc81
11 changed files with 1090 additions and 66 deletions
301
src/brainy.ts
301
src/brainy.ts
|
|
@ -103,7 +103,12 @@ import {
|
|||
UpdateNounMetadataOperation,
|
||||
UpdateVerbMetadataOperation,
|
||||
DeleteNounMetadataOperation,
|
||||
DeleteVerbMetadataOperation
|
||||
DeleteVerbMetadataOperation,
|
||||
UpdateInMetadataIndexOperation,
|
||||
UpdateVerbInGraphIndexOperation,
|
||||
metadataUpdateOpProvider,
|
||||
graphUpdateOpProvider,
|
||||
assertUpdateCapabilityCoherent
|
||||
} from './transaction/operations/index.js'
|
||||
import {
|
||||
BaseOperationalMode,
|
||||
|
|
@ -196,7 +201,7 @@ import {
|
|||
} from './events/changeFeed.js'
|
||||
import { isDeterministicEmbedMode } from './embeddings/deterministicEmbedMode.js'
|
||||
import { GenerationConflictError, StoreInconsistentError } from './db/errors.js'
|
||||
import { BrainyError, GraphIndexNotReadyError, MetadataIndexNotReadyError, MigrationInProgressError, VectorIndexNotReadyError } from './errors/brainyError.js'
|
||||
import { BrainyError, GraphIndexNotReadyError, MetadataIndexNotReadyError, MigrationInProgressError, VectorIndexNotReadyError, ProviderCapabilityMismatchError } from './errors/brainyError.js'
|
||||
import { assessIndexReadiness } from './utils/indexReadiness.js'
|
||||
import { reconstructNounWrapper } from './db/factLog.js'
|
||||
import { asBrainyFieldRefusal } from './db/fieldAddressing.js'
|
||||
|
|
@ -1276,6 +1281,11 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
entityIdMapper: entityIdMapperFactory ? entityIdMapperFactory(this.storage) : undefined,
|
||||
})
|
||||
}
|
||||
// Registration-time refusal (before any write can run): a provider
|
||||
// whose `capabilities` set claims 'update-op' while its instance lacks
|
||||
// `updateIndex` is a typed, loud refusal — never a silent fallback
|
||||
// discovered only at the first write.
|
||||
assertUpdateCapabilityCoherent(this.metadataIndex, 'metadata')
|
||||
|
||||
// Provider: graph index factory
|
||||
const graphFactory = this.pluginRegistry.getProvider<(storage: StorageAdapter) => any>('graphIndex')
|
||||
|
|
@ -1290,6 +1300,9 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
])
|
||||
this.graphIndex = graphIndex
|
||||
}
|
||||
// Same registration-time refusal, graph half (see the metadata-index
|
||||
// check above).
|
||||
assertUpdateCapabilityCoherent(this.graphIndex, 'graph')
|
||||
|
||||
// Fact-log v2 mint seam: after-image records carry minted dense ints,
|
||||
// and the ONE authority for those assignments is the metadata index's
|
||||
|
|
@ -1682,6 +1695,14 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
if (error instanceof Error && (error as Error & { code?: string }).code === 'BRAINY_WRITER_LOCKED') {
|
||||
throw error
|
||||
}
|
||||
// Same rationale, provider-capability half: a registration-time refusal
|
||||
// (a provider's `capabilities` set lies about implementing 'update-op')
|
||||
// carries a machine-readable `.type`/`.family`/`.missingMethod` — the
|
||||
// whole point of the typed-error family — so it must not be flattened
|
||||
// into a message-only generic Error either.
|
||||
if (error instanceof ProviderCapabilityMismatchError) {
|
||||
throw error
|
||||
}
|
||||
throw new Error(`Failed to initialize Brainy: ${error}`)
|
||||
}
|
||||
}
|
||||
|
|
@ -3619,12 +3640,23 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
createdBy: existing.createdBy,
|
||||
metadata: existing.metadata // CRITICAL: keep as nested 'metadata' property!
|
||||
}
|
||||
tx.addOperation(
|
||||
new RemoveFromMetadataIndexOperation(this.metadataIndex, params.id, removalMetadata, this.indexWriteGeneration)
|
||||
)
|
||||
tx.addOperation(
|
||||
new AddToMetadataIndexOperation(this.metadataIndex, params.id, entityForIndexing, this.indexWriteGeneration)
|
||||
)
|
||||
// ONE atomic metadata-index leg when the provider announces the
|
||||
// update-op capability (both halves of the check pass), else the
|
||||
// legacy remove-old/add-new pair — the one-train overlap this
|
||||
// release (see MetadataIndexProvider.updateIndex in src/plugin.ts).
|
||||
const metadataUpdateProvider = metadataUpdateOpProvider(this.metadataIndex)
|
||||
if (metadataUpdateProvider) {
|
||||
tx.addOperation(
|
||||
new UpdateInMetadataIndexOperation(metadataUpdateProvider, params.id, removalMetadata, entityForIndexing, this.indexWriteGeneration)
|
||||
)
|
||||
} else {
|
||||
tx.addOperation(
|
||||
new RemoveFromMetadataIndexOperation(this.metadataIndex, params.id, removalMetadata, this.indexWriteGeneration)
|
||||
)
|
||||
tx.addOperation(
|
||||
new AddToMetadataIndexOperation(this.metadataIndex, params.id, entityForIndexing, this.indexWriteGeneration)
|
||||
)
|
||||
}
|
||||
}, casPrecommit, this._changeFeed.hasListeners
|
||||
? [
|
||||
{
|
||||
|
|
@ -4837,16 +4869,103 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
|
||||
validateUpdateRelationParams(params)
|
||||
|
||||
|
||||
const existing = await this.storage.getVerb(params.id)
|
||||
if (!existing) {
|
||||
throw new RelationNotFoundError(params.id)
|
||||
}
|
||||
|
||||
// Legacy stored shapes carried the verb type under `type` instead of the
|
||||
// canonical `verb` field — read both, canonical first.
|
||||
const existingRec: HNSWVerbWithMetadata & { type?: VerbType } = existing
|
||||
const newVerbType = params.type ?? existingRec.verb ?? existingRec.type
|
||||
const { typeChanged, verbForIndex, updatedMetadata } = this.buildUpdateRelationRecord(params, existing)
|
||||
|
||||
// 8.0 BigInt boundary: endpoints are unchanged across a type swap, so one
|
||||
// resolution serves both the remove (rollback re-add) and the re-add —
|
||||
// only needed on the legacy pair path below; the update-op path never
|
||||
// touches endpoint ints (the provider already holds the mapping).
|
||||
const reindexInts = typeChanged ? this.resolveVerbEndpointInts(verbForIndex) : undefined
|
||||
|
||||
await this.persistSingleOp({ verbs: [params.id] }, async (tx) => {
|
||||
tx.addOperation(
|
||||
new UpdateVerbMetadataOperation(this.storage, params.id, updatedMetadata)
|
||||
)
|
||||
|
||||
// If the verb type changed, re-index in graph adjacency so traversal-by-type
|
||||
// stays consistent. The id is preserved across the swap. ONE atomic
|
||||
// update-op leg when the graph provider announces the capability, else
|
||||
// the legacy remove-old/add-new pair — the one-train overlap this
|
||||
// release.
|
||||
if (typeChanged && reindexInts) {
|
||||
const graphUpdateProvider = graphUpdateOpProvider(this.graphIndex)
|
||||
if (graphUpdateProvider) {
|
||||
tx.addOperation(
|
||||
new UpdateVerbInGraphIndexOperation(graphUpdateProvider, existing, verbForIndex, this.graphWriteGeneration)
|
||||
)
|
||||
} else {
|
||||
tx.addOperation(
|
||||
new RemoveFromGraphIndexOperation(
|
||||
this.graphIndex, existing, reindexInts, this.graphWriteGeneration
|
||||
)
|
||||
)
|
||||
tx.addOperation(
|
||||
new AddToGraphIndexOperation(
|
||||
this.graphIndex, verbForIndex, reindexInts,
|
||||
this.graphWriteGeneration,
|
||||
(verbInt) => this.cacheVerbInt(verbInt, params.id)
|
||||
)
|
||||
)
|
||||
}
|
||||
}
|
||||
},
|
||||
undefined,
|
||||
this._changeFeed.hasListeners
|
||||
? [
|
||||
{
|
||||
kind: 'relation',
|
||||
op: 'updateRelation',
|
||||
id: params.id,
|
||||
relation: {
|
||||
id: params.id,
|
||||
from: verbForIndex.sourceId,
|
||||
to: verbForIndex.targetId,
|
||||
type: String(verbForIndex.verb ?? verbForIndex.type),
|
||||
...(verbForIndex.metadata && {
|
||||
metadata: verbForIndex.metadata as Record<string, unknown>
|
||||
})
|
||||
}
|
||||
}
|
||||
]
|
||||
: undefined)
|
||||
}
|
||||
|
||||
/**
|
||||
* Pure record-building core shared by `updateRelation()` and
|
||||
* `planTxUpdateRelation()`: given the caller-resolved before-image verb,
|
||||
* runs subtype enforcement and computes the merged v2 metadata record and
|
||||
* the graph-index view of the after-image — WITHOUT touching storage. The
|
||||
* actual read of the before-image (`storage.getVerb` for the single-op
|
||||
* path; batch-state-aware resolution for the transact planner, so a verb
|
||||
* updated or created earlier in the SAME batch is visible) stays with the
|
||||
* caller — the two differ, mirroring why `update()`/`planTxUpdate` each
|
||||
* own their own entity read rather than sharing one.
|
||||
*
|
||||
* @param params - The update params.
|
||||
* @param existingRec - The verb's REAL before-image (legacy stored shapes
|
||||
* carried the verb type under `type` instead of the canonical `verb`
|
||||
* field — this reads both, canonical first).
|
||||
* @returns `typeChanged` (whether the effective verb type differs from the
|
||||
* before-image), the merged `newMetadata` user bag, the full
|
||||
* `updatedMetadata` v2 record to persist, and `verbForIndex` — the
|
||||
* graph-index view of the after-image (its `id` matches
|
||||
* `existingRec.id` — updates never change a verb's id).
|
||||
*/
|
||||
private buildUpdateRelationRecord(
|
||||
params: UpdateRelationParams<T>,
|
||||
existingRec: GraphVerb
|
||||
): {
|
||||
typeChanged: boolean
|
||||
newMetadata: Record<string, unknown>
|
||||
updatedMetadata: Record<string, unknown>
|
||||
verbForIndex: GraphVerb
|
||||
} {
|
||||
const newVerbType = (params.type ?? existingRec.verb ?? existingRec.type) as VerbType
|
||||
|
||||
// Subtype pairing enforcement on update (7.30.0). The effective verb type after
|
||||
// the update may have changed; we check against the new type and the resulting
|
||||
|
|
@ -4898,7 +5017,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
|
||||
// Build the verb view used by the graph index — top-level fields mirror relate()'s.
|
||||
const verbForIndex: GraphVerb = {
|
||||
id: params.id,
|
||||
id: existingRec.id,
|
||||
vector: existingRec.vector,
|
||||
sourceId: existingRec.sourceId,
|
||||
targetId: existingRec.targetId,
|
||||
|
|
@ -4916,51 +5035,12 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
createdAt: existingRec.createdAt
|
||||
}
|
||||
|
||||
// 8.0 BigInt boundary: endpoints are unchanged across a type swap, so one
|
||||
// resolution serves both the remove (rollback re-add) and the re-add.
|
||||
const reindexInts = typeChanged ? this.resolveVerbEndpointInts(verbForIndex) : undefined
|
||||
|
||||
await this.persistSingleOp({ verbs: [params.id] }, async (tx) => {
|
||||
tx.addOperation(
|
||||
new UpdateVerbMetadataOperation(this.storage, params.id, updatedMetadata)
|
||||
)
|
||||
|
||||
// If the verb type changed, re-index in graph adjacency so traversal-by-type
|
||||
// stays consistent. The id is preserved across the swap.
|
||||
if (typeChanged && reindexInts) {
|
||||
tx.addOperation(
|
||||
new RemoveFromGraphIndexOperation(
|
||||
this.graphIndex, existing, reindexInts, this.graphWriteGeneration
|
||||
)
|
||||
)
|
||||
tx.addOperation(
|
||||
new AddToGraphIndexOperation(
|
||||
this.graphIndex, verbForIndex, reindexInts,
|
||||
this.graphWriteGeneration,
|
||||
(verbInt) => this.cacheVerbInt(verbInt, params.id)
|
||||
)
|
||||
)
|
||||
}
|
||||
},
|
||||
undefined,
|
||||
this._changeFeed.hasListeners
|
||||
? [
|
||||
{
|
||||
kind: 'relation',
|
||||
op: 'updateRelation',
|
||||
id: params.id,
|
||||
relation: {
|
||||
id: params.id,
|
||||
from: verbForIndex.sourceId,
|
||||
to: verbForIndex.targetId,
|
||||
type: String(verbForIndex.verb ?? verbForIndex.type),
|
||||
...(verbForIndex.metadata && {
|
||||
metadata: verbForIndex.metadata as Record<string, unknown>
|
||||
})
|
||||
}
|
||||
}
|
||||
]
|
||||
: undefined)
|
||||
return {
|
||||
typeChanged,
|
||||
newMetadata: newMetadata as Record<string, unknown>,
|
||||
updatedMetadata,
|
||||
verbForIndex
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -7958,6 +8038,9 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
entityIdMapper: entityIdMapperFactory ? entityIdMapperFactory(this.storage) : undefined,
|
||||
})
|
||||
}
|
||||
// Registration-time refusal, re-run on every re-adoption (see init()'s
|
||||
// matching check) — before any write can run against the recreated index.
|
||||
assertUpdateCapabilityCoherent(this.metadataIndex, 'metadata')
|
||||
await this.metadataIndex.init()
|
||||
|
||||
// Re-resolve the graph index the same way init() does (provider factory,
|
||||
|
|
@ -7972,6 +8055,8 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
} else {
|
||||
this.graphIndex = await this.storage.getGraphIndex()
|
||||
}
|
||||
// Same registration-time refusal, graph half.
|
||||
assertUpdateCapabilityCoherent(this.graphIndex, 'graph')
|
||||
this.wireGraphIdResolver()
|
||||
|
||||
// Reset dimensions
|
||||
|
|
@ -10159,6 +10244,9 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
case 'relate':
|
||||
plan.ids.push(await this.planTxRelate(op, state, plan))
|
||||
break
|
||||
case 'updateRelation':
|
||||
plan.ids.push(await this.planTxUpdateRelation(op, state, plan))
|
||||
break
|
||||
case 'unrelate':
|
||||
plan.ids.push(await this.planTxUnrelate(op, state, plan))
|
||||
break
|
||||
|
|
@ -10530,10 +10618,21 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
new ReplaceInVectorIndexOperation(this.index, params.id, existing.vector, vector, this.indexWriteGeneration)
|
||||
)
|
||||
}
|
||||
plan.operations.push(
|
||||
new RemoveFromMetadataIndexOperation(this.metadataIndex, params.id, removalMetadata, this.indexWriteGeneration),
|
||||
new AddToMetadataIndexOperation(this.metadataIndex, params.id, entityForIndexing, this.indexWriteGeneration)
|
||||
)
|
||||
// ONE atomic metadata-index leg when the provider announces the
|
||||
// update-op capability, else the legacy remove-old/add-new pair — same
|
||||
// branch as update() (see the class note above planTxUpdate's Object 5-6
|
||||
// sibling in update()).
|
||||
const metadataUpdateProviderTx = metadataUpdateOpProvider(this.metadataIndex)
|
||||
if (metadataUpdateProviderTx) {
|
||||
plan.operations.push(
|
||||
new UpdateInMetadataIndexOperation(metadataUpdateProviderTx, params.id, removalMetadata, entityForIndexing, this.indexWriteGeneration)
|
||||
)
|
||||
} else {
|
||||
plan.operations.push(
|
||||
new RemoveFromMetadataIndexOperation(this.metadataIndex, params.id, removalMetadata, this.indexWriteGeneration),
|
||||
new AddToMetadataIndexOperation(this.metadataIndex, params.id, entityForIndexing, this.indexWriteGeneration)
|
||||
)
|
||||
}
|
||||
plan.touchedNouns.push(params.id)
|
||||
|
||||
// The full planGetEntity view, passed whole — a subset view makes the
|
||||
|
|
@ -10861,6 +10960,82 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
return id
|
||||
}
|
||||
|
||||
/**
|
||||
* Plan one `{ op: 'updateRelation' }` — mirror of `updateRelation()`,
|
||||
* batch-aware: resolves the before-image against "current state + batch
|
||||
* so far" (a `relate` or an earlier `updateRelation` on the SAME id
|
||||
* earlier in this batch is visible), so the op is genuinely batchable —
|
||||
* the graph counterpart of `planTxUpdate`. Returns the relationship id.
|
||||
*/
|
||||
private async planTxUpdateRelation(
|
||||
op: Extract<TxOperation<T>, { op: 'updateRelation' }>,
|
||||
state: TxPlanState,
|
||||
plan: PlannedTransact
|
||||
): Promise<string> {
|
||||
const { op: _discriminator, ...rawParams } = op
|
||||
validateUpdateRelationParams(rawParams as UpdateRelationParams<T>)
|
||||
const params = rawParams as UpdateRelationParams<T>
|
||||
|
||||
const existing = state.removedVerbs.has(params.id)
|
||||
? null
|
||||
: (state.verbs.get(params.id) ?? (await this.storage.getVerb(params.id)))
|
||||
if (!existing) {
|
||||
throw new RelationNotFoundError(params.id)
|
||||
}
|
||||
|
||||
const { typeChanged, verbForIndex, updatedMetadata } = this.buildUpdateRelationRecord(params, existing)
|
||||
|
||||
plan.operations.push(
|
||||
new UpdateVerbMetadataOperation(this.storage, params.id, updatedMetadata)
|
||||
)
|
||||
|
||||
// If the verb type changed, re-index in graph adjacency — same branch as
|
||||
// updateRelation(): ONE atomic update-op leg when the graph provider
|
||||
// announces the capability, else the legacy remove-old/add-new pair.
|
||||
if (typeChanged) {
|
||||
const graphUpdateProvider = graphUpdateOpProvider(this.graphIndex)
|
||||
if (graphUpdateProvider) {
|
||||
plan.operations.push(
|
||||
new UpdateVerbInGraphIndexOperation(graphUpdateProvider, existing, verbForIndex, this.graphWriteGeneration)
|
||||
)
|
||||
} else {
|
||||
plan.operations.push(
|
||||
// Endpoint ints resolve at EXECUTE time — mirror of planTxRelate/
|
||||
// planTxUnrelate: the verb (or its endpoints) may have been
|
||||
// created earlier in this same batch.
|
||||
new RemoveFromGraphIndexOperation(
|
||||
this.graphIndex, existing, () => this.resolveVerbEndpointInts(verbForIndex), this.graphWriteGeneration
|
||||
),
|
||||
new AddToGraphIndexOperation(
|
||||
this.graphIndex, verbForIndex, () => this.resolveVerbEndpointInts(verbForIndex),
|
||||
this.graphWriteGeneration,
|
||||
(verbInt) => this.cacheVerbInt(verbInt, params.id)
|
||||
)
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
plan.touchedVerbs.push(params.id)
|
||||
state.verbs.set(params.id, verbForIndex)
|
||||
|
||||
if (this._changeFeed.hasListeners) {
|
||||
plan.changeEvents.push({
|
||||
kind: 'relation',
|
||||
op: 'updateRelation',
|
||||
id: params.id,
|
||||
relation: {
|
||||
id: params.id,
|
||||
from: verbForIndex.sourceId,
|
||||
to: verbForIndex.targetId,
|
||||
type: String(verbForIndex.verb ?? verbForIndex.type),
|
||||
...(verbForIndex.metadata && { metadata: verbForIndex.metadata as Record<string, unknown> })
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
return params.id
|
||||
}
|
||||
|
||||
/** Plan one `{ op: 'unrelate' }` — mirror of `unrelate()`. Returns the relationship id. */
|
||||
private async planTxUnrelate(
|
||||
op: Extract<TxOperation<T>, { op: 'unrelate' }>,
|
||||
|
|
|
|||
Reference in a new issue