feat(recovery): the catchup verdict is consumed; verb rows go live; the metadata rebuild goes online
Three cures on the JS metadata index, one seam:
- THE CATCHUP WIRING. The index computed its three-way watermark verdict at
open and nothing consumed it — after a crash + adopt reopen, find() served
the pre-crash index while canonical reads and counts recovered (caught by
the lifecycle lane's first run). The open path now consumes the verdict:
'adopt' is a no-op, 'catchup' folds the fact window (stamped, committed]
through the index legs — nouns and verbs, remove-then-add, one mechanism
for add and update — and 'rescan' runs the explicit rebuild, each narrated.
The lane's Ch4–6 release-blocking marker comes off: the contract holds.
Bonus root-cause: close() never stamped the projection watermarks (only
flush() did), so any close without a prior flush verdicted a needless
'rescan' on reopen — both doors now stamp.
- THE LIVE VERB PATH. Verb rows entered the metadata index only via rebuild
walks, so every rebuilt store minted phantom/stale verb postings from its
first live relate(). relate()/unrelate()/updateRelation() and remove()'s
cascade now post/retract the verb's row in the same commit as the graph
leg — transact() planners mirror identically — using the exact record
shape the rebuild walk uses, so live and rebuilt populations agree.
- THE ONLINE REBUILD. rebuild() was clear-then-walk — every metadata read
empty for the duration. rebuildMetadataIndexOnline builds a fresh manager
beside the serving one (shared identity, in-memory build, dual-write via
a shadow seam with zero call-site changes), atomically swaps the
reference, and persists exactly once post-swap. A find() polled ~200x
during a 2k-noun rebuild never dropped below its baseline.
repairIndex({ rebuild: ['metadata'] }) uses it automatically.
This commit is contained in:
parent
f8f64780b1
commit
18f172e098
6 changed files with 1275 additions and 117 deletions
472
src/brainy.ts
472
src/brainy.ts
|
|
@ -1385,6 +1385,20 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
])
|
])
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// METADATA WATERMARK CATCHUP: the JS metadata index computed its
|
||||||
|
// three-way watermark verdict inside metadataIndex.init() above,
|
||||||
|
// against the generation store's now-FINAL committed generation (the
|
||||||
|
// crash-recovery fold above — the durable-at-ack replay of acked
|
||||||
|
// writes whose canonical bytes hadn't reached disk — has already run,
|
||||||
|
// and any rolled-back-transaction rebuild just above already brought
|
||||||
|
// every index current, so the verdict is consumed here whether or not
|
||||||
|
// that rebuild ran). Consumed BEFORE the rebuild gate below and BEFORE
|
||||||
|
// this open serves any read — the cure for the class of bug where
|
||||||
|
// canonical get()/counts recover a crash-window write but find()
|
||||||
|
// keeps serving the metadata index's pre-crash state (the index
|
||||||
|
// flushes only periodically, not per-commit).
|
||||||
|
await this.consumeMetadataWatermarkVerdict(generationOpenResult.rolledBackGenerations > 0)
|
||||||
|
|
||||||
// 8.0 versioned-provider replay-gap check: a provider whose persisted
|
// 8.0 versioned-provider replay-gap check: a provider whose persisted
|
||||||
// index generation is behind the storage layer's committed generation
|
// index generation is behind the storage layer's committed generation
|
||||||
// replays the gap itself (post-commit applier contract) — surface the
|
// replays the gap itself (post-commit applier contract) — surface the
|
||||||
|
|
@ -3672,6 +3686,85 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
if (deferringEmbed) this.kickEmbedWorker()
|
if (deferringEmbed) this.kickEmbedWorker()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @description Build the metadata-index retraction operation for one id
|
||||||
|
* (noun or verb) — the null-metadata-safe closure shared by every removal
|
||||||
|
* leg that reaches the metadata index with a possibly-missed pre-read:
|
||||||
|
* `remove()`'s own noun leg, its verb-cascade retractions, `unrelate()`,
|
||||||
|
* and their `transact()`/`planTx*` mirrors (both callers add the returned
|
||||||
|
* operation to their own batch — `tx.addOperation()` for a single-op
|
||||||
|
* transaction, `plan.operations.push()` for a planned `transact()` batch).
|
||||||
|
* THE NULL-METADATA SKIP IS CLOSED (a posting-leak class):
|
||||||
|
* - metadata present → the ordinary, provider-agnostic
|
||||||
|
* `RemoveFromMetadataIndexOperation` (exact per-field retraction).
|
||||||
|
* - metadata absent (a torn pre-read, or the row was already gone) →
|
||||||
|
* a provider exposing `removeEntityById` (the id-keyed contract) gets
|
||||||
|
* exact per-entity retraction via its reverse record; the JS index
|
||||||
|
* gets `removeFromIndex(id)` — safe id-keyed cleanup (deleted bitmap +
|
||||||
|
* id mapper; field statistics reconcile at the next rebuild/repairIndex),
|
||||||
|
* narrated; a native provider WITHOUT the contract is never called
|
||||||
|
* metadata-omitted (that path walks its value space) — the skip is
|
||||||
|
* tracked in the degraded set instead, narrated, so `repairIndex()`
|
||||||
|
* reconciles it (and this method returns `null` — no operation to add).
|
||||||
|
* Silence is the only thing outlawed.
|
||||||
|
* @param id - The noun/verb id being retracted.
|
||||||
|
* @param metadata - The pre-read metadata/entity structure, or falsy when
|
||||||
|
* the read missed.
|
||||||
|
* @param context - Narration prefix identifying the caller/id, e.g.
|
||||||
|
* `remove(${id})` or `remove(${entityId}) cascade unrelate ${verbId}`.
|
||||||
|
* @returns The operation to add to the caller's batch, or `null` when
|
||||||
|
* nothing could be done (already narrated + tracked as degraded).
|
||||||
|
*/
|
||||||
|
private metadataIndexRetractionOp(
|
||||||
|
id: string,
|
||||||
|
metadata: unknown,
|
||||||
|
context: string
|
||||||
|
): Operation | null {
|
||||||
|
if (metadata) {
|
||||||
|
return new RemoveFromMetadataIndexOperation(this.metadataIndex, id, metadata, this.indexWriteGeneration)
|
||||||
|
}
|
||||||
|
const prov = this.metadataIndex as unknown as {
|
||||||
|
removeEntityById?: (id: string) => Promise<void>
|
||||||
|
removeFromIndex?: (id: string, metadata?: unknown, generation?: bigint) => Promise<void>
|
||||||
|
}
|
||||||
|
if (typeof prov.removeEntityById === 'function') {
|
||||||
|
const g = this.indexWriteGeneration
|
||||||
|
return {
|
||||||
|
name: 'RemoveEntityByIdTombstone',
|
||||||
|
execute: async () => {
|
||||||
|
await prov.removeEntityById!(id)
|
||||||
|
return async () => {
|
||||||
|
// Undo of an id-keyed tombstone on an absent row: nothing to
|
||||||
|
// restore (the row had no readable metadata to re-post).
|
||||||
|
void g
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} else if (this.metadataIndex instanceof MetadataIndexManager) {
|
||||||
|
const gv = this.indexWriteGeneration
|
||||||
|
prodLog.warn(
|
||||||
|
`[Brainy] ${context}: no metadata at delete — id-keyed index cleanup ran ` +
|
||||||
|
`(deleted bitmap + id mapper); field statistics reconcile at the next rebuild/repairIndex.`
|
||||||
|
)
|
||||||
|
return {
|
||||||
|
name: 'IdKeyedIndexCleanup',
|
||||||
|
execute: async () => {
|
||||||
|
await prov.removeFromIndex!(id, undefined, typeof gv === 'function' ? gv() : gv)
|
||||||
|
return async () => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
this._indexDegradedIds.add(id)
|
||||||
|
prodLog.warn(
|
||||||
|
`[Brainy] ${context}: no metadata at delete and this provider has no id-keyed ` +
|
||||||
|
`removal — its postings for this id are NOT tombstoned yet (tracked as degraded; ` +
|
||||||
|
`repairIndex() reconciles). Never calling a metadata-omitted native removal: that ` +
|
||||||
|
`path walks the store's value space.`
|
||||||
|
)
|
||||||
|
return null
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Remove an entity and all its relationships
|
* Remove an entity and all its relationships
|
||||||
*
|
*
|
||||||
|
|
@ -3736,61 +3829,11 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Operation 2: Remove from metadata index. THE NULL-METADATA SKIP IS
|
// Operation 2: Remove from metadata index (null-metadata-safe — see
|
||||||
// CLOSED (a posting-leak class, confirmed at this site): when the
|
// metadataIndexRetractionOp's JSDoc for the full closure).
|
||||||
// pre-read missed, the leg no longer silently skips —
|
{
|
||||||
// - a provider exposing removeEntityById (the id-keyed contract)
|
const retractionOp = this.metadataIndexRetractionOp(id, metadata, `remove(${id})`)
|
||||||
// gets it: exact per-entity retraction via its reverse record;
|
if (retractionOp) tx.addOperation(retractionOp)
|
||||||
// - the JS index gets removeFromIndex(id) — safe id-keyed cleanup
|
|
||||||
// (deleted bitmap + id mapper; field stats reconcile at rebuild);
|
|
||||||
// - a NATIVE provider WITHOUT the contract is never called
|
|
||||||
// metadata-omitted (that path walks its value space) — the skip
|
|
||||||
// happens, but NARRATED and tracked in the degraded set so
|
|
||||||
// repairIndex reconciles it. Silence is the only thing outlawed.
|
|
||||||
if (metadata) {
|
|
||||||
tx.addOperation(
|
|
||||||
new RemoveFromMetadataIndexOperation(this.metadataIndex, id, metadata, this.indexWriteGeneration)
|
|
||||||
)
|
|
||||||
} else {
|
|
||||||
const prov = this.metadataIndex as unknown as {
|
|
||||||
removeEntityById?: (id: string) => Promise<void>
|
|
||||||
removeFromIndex?: (id: string, metadata?: unknown, generation?: bigint) => Promise<void>
|
|
||||||
}
|
|
||||||
if (typeof prov.removeEntityById === 'function') {
|
|
||||||
const g = this.indexWriteGeneration
|
|
||||||
tx.addOperation({
|
|
||||||
name: 'RemoveEntityByIdTombstone',
|
|
||||||
execute: async () => {
|
|
||||||
await prov.removeEntityById!(id)
|
|
||||||
return async () => {
|
|
||||||
// Undo of an id-keyed tombstone on an absent row: nothing
|
|
||||||
// to restore (the row had no readable metadata to re-post).
|
|
||||||
void g
|
|
||||||
}
|
|
||||||
}
|
|
||||||
})
|
|
||||||
} else if (this.metadataIndex instanceof MetadataIndexManager) {
|
|
||||||
const gv = this.indexWriteGeneration
|
|
||||||
tx.addOperation({
|
|
||||||
name: 'IdKeyedIndexCleanup',
|
|
||||||
execute: async () => {
|
|
||||||
await prov.removeFromIndex!(id, undefined, typeof gv === 'function' ? gv() : gv)
|
|
||||||
return async () => {}
|
|
||||||
}
|
|
||||||
})
|
|
||||||
prodLog.warn(
|
|
||||||
`[Brainy] remove(${id}): no metadata at delete — id-keyed index cleanup ran ` +
|
|
||||||
`(deleted bitmap + id mapper); field statistics reconcile at the next rebuild/repairIndex.`
|
|
||||||
)
|
|
||||||
} else {
|
|
||||||
this._indexDegradedIds.add(id)
|
|
||||||
prodLog.warn(
|
|
||||||
`[Brainy] remove(${id}): no metadata at delete and this provider has no id-keyed ` +
|
|
||||||
`removal — its postings for this id are NOT tombstoned yet (tracked as degraded; ` +
|
|
||||||
`repairIndex() reconciles). Never calling a metadata-omitted native removal: that ` +
|
|
||||||
`path walks the store's value space.`
|
|
||||||
)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Operation 3: Delete noun (full removal). The pre-read metadata rides
|
// Operation 3: Delete noun (full removal). The pre-read metadata rides
|
||||||
|
|
@ -3808,6 +3851,21 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
tx.addOperation(
|
tx.addOperation(
|
||||||
new RemoveFromGraphIndexOperation(this.graphIndex, verb, { sourceInt, targetInt }, this.graphWriteGeneration)
|
new RemoveFromGraphIndexOperation(this.graphIndex, verb, { sourceInt, targetInt }, this.graphWriteGeneration)
|
||||||
)
|
)
|
||||||
|
// Retract the cascaded relation's metadata-index row too — the
|
||||||
|
// live mirror of what a rebuild would derive for this (now-gone)
|
||||||
|
// edge (mirrors the noun leg above). The whole hydrated verb
|
||||||
|
// (system fields top-level + the custom bag under `metadata`,
|
||||||
|
// same shape `extractIndexableFields` reads for any entity-record
|
||||||
|
// frame) is the before-image — every entry in `allVerbs` was
|
||||||
|
// already successfully hydrated by the reads above, so this is
|
||||||
|
// never metadata-omitted in practice, but the closure stays
|
||||||
|
// defensive rather than assuming.
|
||||||
|
{
|
||||||
|
const cascadeRetractionOp = this.metadataIndexRetractionOp(
|
||||||
|
verb.id, verb, `remove(${id}) cascade unrelate ${verb.id}`
|
||||||
|
)
|
||||||
|
if (cascadeRetractionOp) tx.addOperation(cascadeRetractionOp)
|
||||||
|
}
|
||||||
// Delete verb metadata
|
// Delete verb metadata
|
||||||
tx.addOperation(
|
tx.addOperation(
|
||||||
new DeleteVerbMetadataOperation(this.storage, verb.id)
|
new DeleteVerbMetadataOperation(this.storage, verb.id)
|
||||||
|
|
@ -4641,6 +4699,16 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// Operation 3b: Add the verb's metadata-index row, in the SAME
|
||||||
|
// commit as the graph leg — the live mirror of what rebuild()'s
|
||||||
|
// verb walk already derives (ADR-007 A4: one mechanism, never a
|
||||||
|
// second hand-rolled shape). `verbMetadata` is the exact raw stored
|
||||||
|
// record `SaveVerbMetadataOperation` above just persisted — the same
|
||||||
|
// shape `storage.getVerbMetadata()`/rebuild() read back.
|
||||||
|
tx.addOperation(
|
||||||
|
new AddToMetadataIndexOperation(this.metadataIndex, id, verbMetadata, this.indexWriteGeneration)
|
||||||
|
)
|
||||||
|
|
||||||
// Create bidirectional if requested
|
// Create bidirectional if requested
|
||||||
if (params.bidirectional && reverseId) {
|
if (params.bidirectional && reverseId) {
|
||||||
const reverseVerb: GraphVerb = {
|
const reverseVerb: GraphVerb = {
|
||||||
|
|
@ -4678,6 +4746,13 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
(verbInt) => this.cacheVerbInt(verbInt, reverseId)
|
(verbInt) => this.cacheVerbInt(verbInt, reverseId)
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// Operation 6b: Add the reverse edge's metadata-index row (same
|
||||||
|
// stored shape as the primary edge — SaveVerbMetadataOperation
|
||||||
|
// above persists the same `verbMetadata` object for both).
|
||||||
|
tx.addOperation(
|
||||||
|
new AddToMetadataIndexOperation(this.metadataIndex, reverseId, verbMetadata, this.indexWriteGeneration)
|
||||||
|
)
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
undefined,
|
undefined,
|
||||||
|
|
@ -4760,6 +4835,15 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Operation 1b: Retract the verb's metadata-index row — the live
|
||||||
|
// mirror of remove()'s cascade leg (null-metadata-safe; see
|
||||||
|
// metadataIndexRetractionOp's JSDoc). Nothing to retract when the
|
||||||
|
// pre-read found no verb (already gone / never existed).
|
||||||
|
if (verb) {
|
||||||
|
const retractionOp = this.metadataIndexRetractionOp(id, verb, `unrelate(${id})`)
|
||||||
|
if (retractionOp) tx.addOperation(retractionOp)
|
||||||
|
}
|
||||||
|
|
||||||
// Operation 2: Delete verb metadata (which also deletes vector)
|
// Operation 2: Delete verb metadata (which also deletes vector)
|
||||||
tx.addOperation(
|
tx.addOperation(
|
||||||
new DeleteVerbMetadataOperation(this.storage, id)
|
new DeleteVerbMetadataOperation(this.storage, id)
|
||||||
|
|
@ -4903,6 +4987,23 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
new UpdateVerbMetadataOperation(this.storage, params.id, updatedMetadata)
|
new UpdateVerbMetadataOperation(this.storage, params.id, updatedMetadata)
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// Re-post the verb's metadata-index row — remove the old shape, add
|
||||||
|
// the new one, same commit (the plain pair; there is no update-op
|
||||||
|
// capability for the metadata leg yet — see the GRAPH leg's
|
||||||
|
// typeChanged branch just below for the capability this ISN'T:
|
||||||
|
// that's the graph adjacency's own remove+add, keyed on the verb
|
||||||
|
// TYPE changing; the metadata row updates on EVERY updateRelation()
|
||||||
|
// call, since metadata/subtype/weight/etc. can all change without a
|
||||||
|
// type change). `existing` is the pre-update hydrated verb (already
|
||||||
|
// read above); `updatedMetadata` is the raw stored record just
|
||||||
|
// persisted — the same shape relate()/rebuild() use to add.
|
||||||
|
tx.addOperation(
|
||||||
|
new RemoveFromMetadataIndexOperation(this.metadataIndex, params.id, existing, this.indexWriteGeneration)
|
||||||
|
)
|
||||||
|
tx.addOperation(
|
||||||
|
new AddToMetadataIndexOperation(this.metadataIndex, params.id, updatedMetadata, this.indexWriteGeneration)
|
||||||
|
)
|
||||||
|
|
||||||
// If the verb type changed, re-index in graph adjacency so traversal-by-type
|
// If the verb type changed, re-index in graph adjacency so traversal-by-type
|
||||||
// stays consistent. The id is preserved across the swap.
|
// stays consistent. The id is preserved across the swap.
|
||||||
if (typeChanged && reindexInts) {
|
if (typeChanged && reindexInts) {
|
||||||
|
|
@ -10603,6 +10704,15 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
new RemoveFromGraphIndexOperation(this.graphIndex, verb, () => this.resolveVerbEndpointInts(verb), this.graphWriteGeneration),
|
new RemoveFromGraphIndexOperation(this.graphIndex, verb, () => this.resolveVerbEndpointInts(verb), this.graphWriteGeneration),
|
||||||
new DeleteVerbMetadataOperation(this.storage, verb.id)
|
new DeleteVerbMetadataOperation(this.storage, verb.id)
|
||||||
)
|
)
|
||||||
|
// Retract the cascaded relation's metadata-index row too — the
|
||||||
|
// transact() mirror of remove()'s single-op cascade leg
|
||||||
|
// (null-metadata-safe; see metadataIndexRetractionOp's JSDoc).
|
||||||
|
{
|
||||||
|
const cascadeRetractionOp = this.metadataIndexRetractionOp(
|
||||||
|
verb.id, verb, `transact remove(${id}) cascade unrelate ${verb.id}`
|
||||||
|
)
|
||||||
|
if (cascadeRetractionOp) plan.operations.push(cascadeRetractionOp)
|
||||||
|
}
|
||||||
plan.touchedVerbs.push(verb.id)
|
plan.touchedVerbs.push(verb.id)
|
||||||
state.verbs.delete(verb.id)
|
state.verbs.delete(verb.id)
|
||||||
state.removedVerbs.add(verb.id)
|
state.removedVerbs.add(verb.id)
|
||||||
|
|
@ -10769,7 +10879,10 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
// id mapper to assign an int for an entity that did not exist yet.
|
// id mapper to assign an int for an entity that did not exist yet.
|
||||||
new AddToGraphIndexOperation(this.graphIndex, verb, () => this.resolveVerbEndpointInts(verb), this.graphWriteGeneration, (verbInt) =>
|
new AddToGraphIndexOperation(this.graphIndex, verb, () => this.resolveVerbEndpointInts(verb), this.graphWriteGeneration, (verbInt) =>
|
||||||
this.cacheVerbInt(verbInt, id)
|
this.cacheVerbInt(verbInt, id)
|
||||||
)
|
),
|
||||||
|
// The transact() mirror of relate()'s metadata-index leg — same
|
||||||
|
// commit as the graph leg, same raw stored shape.
|
||||||
|
new AddToMetadataIndexOperation(this.metadataIndex, id, verbMetadata, this.indexWriteGeneration)
|
||||||
)
|
)
|
||||||
plan.touchedVerbs.push(id)
|
plan.touchedVerbs.push(id)
|
||||||
state.verbs.set(id, verb)
|
state.verbs.set(id, verb)
|
||||||
|
|
@ -10811,7 +10924,8 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
new SaveVerbMetadataOperation(this.storage, reverseId, verbMetadata),
|
new SaveVerbMetadataOperation(this.storage, reverseId, verbMetadata),
|
||||||
new AddToGraphIndexOperation(this.graphIndex, reverseVerb, () => this.resolveVerbEndpointInts(reverseVerb), this.graphWriteGeneration, (verbInt) =>
|
new AddToGraphIndexOperation(this.graphIndex, reverseVerb, () => this.resolveVerbEndpointInts(reverseVerb), this.graphWriteGeneration, (verbInt) =>
|
||||||
this.cacheVerbInt(verbInt, reverseId)
|
this.cacheVerbInt(verbInt, reverseId)
|
||||||
)
|
),
|
||||||
|
new AddToMetadataIndexOperation(this.metadataIndex, reverseId, verbMetadata, this.indexWriteGeneration)
|
||||||
)
|
)
|
||||||
plan.touchedVerbs.push(reverseId)
|
plan.touchedVerbs.push(reverseId)
|
||||||
state.verbs.set(reverseId, reverseVerb)
|
state.verbs.set(reverseId, reverseVerb)
|
||||||
|
|
@ -10856,6 +10970,12 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
// may have been created earlier in this same batch (forward refs).
|
// may have been created earlier in this same batch (forward refs).
|
||||||
new RemoveFromGraphIndexOperation(this.graphIndex, verb, () => this.resolveVerbEndpointInts(verb), this.graphWriteGeneration)
|
new RemoveFromGraphIndexOperation(this.graphIndex, verb, () => this.resolveVerbEndpointInts(verb), this.graphWriteGeneration)
|
||||||
)
|
)
|
||||||
|
// The transact() mirror of unrelate()'s metadata-index leg
|
||||||
|
// (null-metadata-safe; see metadataIndexRetractionOp's JSDoc — a
|
||||||
|
// present `verb` here is never metadata-omitted, but the closure
|
||||||
|
// stays defensive rather than assuming).
|
||||||
|
const retractionOp = this.metadataIndexRetractionOp(id, verb, `transact unrelate(${id})`)
|
||||||
|
if (retractionOp) plan.operations.push(retractionOp)
|
||||||
}
|
}
|
||||||
plan.operations.push(new DeleteVerbMetadataOperation(this.storage, id))
|
plan.operations.push(new DeleteVerbMetadataOperation(this.storage, id))
|
||||||
plan.touchedVerbs.push(id)
|
plan.touchedVerbs.push(id)
|
||||||
|
|
@ -11564,6 +11684,31 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @description Stamp every projection's watermark with the store's
|
||||||
|
* current committed generation — the door BOTH {@link flush} and {@link
|
||||||
|
* close} open right before persisting, so EITHER path leaves a stamped,
|
||||||
|
* `'adopt'`-verdicting artifact on disk (stamp-after-data still holds
|
||||||
|
* inside each owner: this only hands the generation over — the owner's
|
||||||
|
* OWN flush is what durably writes the stamp, LAST). Before this method
|
||||||
|
* existed, `close()` had its own separate flush fan-out that never
|
||||||
|
* stamped, so a `close()` without a preceding explicit `flush()` left
|
||||||
|
* every projection unstamped — a real, closed store that legitimately
|
||||||
|
* verdicts `'rescan'` on its very next open (not a bug in the verdict,
|
||||||
|
* a gap in `close()`'s persistence completeness that this closes).
|
||||||
|
* No `committedGeneration` capability, or a replacement provider that
|
||||||
|
* doesn't carry the stamp method (a native pair swaps these managers) =
|
||||||
|
* no stamp = the owner's verdict machinery treats the artifact as
|
||||||
|
* legacy — never a flush/close crash either way.
|
||||||
|
*/
|
||||||
|
private stampProjectionWatermarks(): void {
|
||||||
|
const wmGen = this.storage?.committedGeneration?.() ?? null
|
||||||
|
if (wmGen === null) return
|
||||||
|
;(this.metadataIndex as { stampWatermark?: (g: number) => void }).stampWatermark?.(wmGen)
|
||||||
|
;(this.index as { stampWatermark?: (g: number) => void }).stampWatermark?.(wmGen)
|
||||||
|
;(this.graphIndex as { stampWatermark?: (g: number) => void }).stampWatermark?.(wmGen)
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Flush all indexes and caches to persistent storage
|
* Flush all indexes and caches to persistent storage
|
||||||
* CRITICAL FIX: Ensures data survives server restarts
|
* CRITICAL FIX: Ensures data survives server restarts
|
||||||
|
|
@ -11603,22 +11748,8 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
await this.generationStore.flushPendingSingleOps()
|
await this.generationStore.flushPendingSingleOps()
|
||||||
|
|
||||||
// Flush all components in parallel for performance
|
// Flush all components in parallel for performance
|
||||||
// Watermark stamps ride every flush fan-out: stamp each projection with
|
// Watermark stamps ride every flush fan-out — see stampProjectionWatermarks().
|
||||||
// the committed generation BEFORE its flush persists (stamp-after-data
|
this.stampProjectionWatermarks()
|
||||||
// holds inside each owner — the stamp is its LAST write; here we only
|
|
||||||
// hand the generation over). No committedGeneration capability = no
|
|
||||||
// stamp = the owner's verdict machinery treats the artifact as legacy.
|
|
||||||
{
|
|
||||||
const wmGen = this.storage?.committedGeneration?.() ?? null
|
|
||||||
if (wmGen !== null) {
|
|
||||||
// ALL THREE optional-chained: a replacement provider (the native
|
|
||||||
// pair swaps these managers) may not carry the stamp method — a
|
|
||||||
// missing stamp is a verdict-side rescan, never a flush crash.
|
|
||||||
;(this.metadataIndex as { stampWatermark?: (g: number) => void }).stampWatermark?.(wmGen)
|
|
||||||
;(this.index as { stampWatermark?: (g: number) => void }).stampWatermark?.(wmGen)
|
|
||||||
;(this.graphIndex as { stampWatermark?: (g: number) => void }).stampWatermark?.(wmGen)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
await Promise.all([
|
await Promise.all([
|
||||||
// 1. Flush storage adapter counts (entity/verb counts by type)
|
// 1. Flush storage adapter counts (entity/verb counts by type)
|
||||||
(async () => {
|
(async () => {
|
||||||
|
|
@ -16316,6 +16447,179 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @description Consume the JS metadata index's watermark verdict (see
|
||||||
|
* {@link MetadataIndexManager.watermarkVerdict}) at open — the coordinator
|
||||||
|
* half of the catchup wiring; {@link MetadataIndexManager.applyWatermarkCatchup}
|
||||||
|
* is the mechanism half. Feature-detected to the JS manager only: a native
|
||||||
|
* metadata-index provider consumes the same verdict door in its own train
|
||||||
|
* (this method never touches the native-provider wrapper contract).
|
||||||
|
*
|
||||||
|
* Ordering: called from `performInit()` immediately after
|
||||||
|
* `metadataIndex.init()` has computed the verdict against the generation
|
||||||
|
* store's now-FINAL committed generation, and BEFORE `rebuildIndexesIfNeeded()`
|
||||||
|
* (the open-time rebuild gate) or any read serves — so a caller can never
|
||||||
|
* observe the pre-catchup state.
|
||||||
|
*
|
||||||
|
* @param alreadyRebuilt - `true` when crash recovery just rebuilt every
|
||||||
|
* index from canonical (rolled-back uncommitted transactions) — the
|
||||||
|
* verdict's prescribed action is redundant with what already ran (a
|
||||||
|
* fresh canonical walk supersedes any catchup fold or rescan), so it is
|
||||||
|
* skipped, narrated, rather than duplicating the work.
|
||||||
|
*/
|
||||||
|
private async consumeMetadataWatermarkVerdict(alreadyRebuilt: boolean): Promise<void> {
|
||||||
|
if (!(this.metadataIndex instanceof MetadataIndexManager)) return
|
||||||
|
const verdict = this.metadataIndex.watermarkVerdict()
|
||||||
|
if (verdict === null || verdict === 'adopt') return
|
||||||
|
|
||||||
|
if (alreadyRebuilt) {
|
||||||
|
prodLog.info(
|
||||||
|
`[Brainy] metadata index watermark verdict '${verdict}' at open — skipped: crash ` +
|
||||||
|
`recovery already rebuilt every index from canonical this open.`
|
||||||
|
)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
const window = this.metadataIndex.watermarkGap()
|
||||||
|
// A genuine first boot (no persisted artifact at all) verdicts 'rescan'
|
||||||
|
// too — same as a real unverifiable artifact — but it is routine, not
|
||||||
|
// alarming: narrate it at info level instead of warn (mirrors the
|
||||||
|
// manager's own internal distinction in loadWatermarkVerdict()).
|
||||||
|
const firstBoot = verdict === 'rescan' && !this.metadataIndex.watermarkArtifactPresent()
|
||||||
|
const preNarrate = firstBoot ? prodLog.info.bind(prodLog) : prodLog.warn.bind(prodLog)
|
||||||
|
preNarrate(
|
||||||
|
verdict === 'catchup' && window
|
||||||
|
? `[Brainy] metadata index watermark verdict: CATCHUP — folding generations ` +
|
||||||
|
`(${window.from}, ${window.to}] from the fact log before this open serves reads.`
|
||||||
|
: firstBoot
|
||||||
|
? `[Brainy] metadata index watermark verdict: rescan (no persisted artifact — first ` +
|
||||||
|
`boot; the rebuild below is a trivial no-op walk).`
|
||||||
|
: `[Brainy] metadata index watermark verdict: RESCAN — the persisted artifact is ` +
|
||||||
|
`unverifiable (unstamped, or ahead of the store's committed generation); ` +
|
||||||
|
`forcing a full rebuild from canonical at open.`
|
||||||
|
)
|
||||||
|
|
||||||
|
const scan = window
|
||||||
|
? this.scanFacts({ fromGeneration: window.from + 1, toGeneration: window.to })
|
||||||
|
: null
|
||||||
|
const result = await this.metadataIndex.applyWatermarkCatchup(scan)
|
||||||
|
|
||||||
|
if (result.action === 'rescan') {
|
||||||
|
const postNarrate = firstBoot ? prodLog.debug.bind(prodLog) : prodLog.warn.bind(prodLog)
|
||||||
|
postNarrate(
|
||||||
|
`[Brainy] metadata index catchup demoted to a full rebuild` +
|
||||||
|
`${result.reason ? ` — ${result.reason}` : ''}.`
|
||||||
|
)
|
||||||
|
} else if (result.action === 'caught-up') {
|
||||||
|
prodLog.warn(
|
||||||
|
`[Brainy] metadata index catchup complete: ${result.factsApplied} fact(s) folded ` +
|
||||||
|
`(${result.nounsApplied} noun op(s), ${result.verbsApplied} verb op(s)) — index now ` +
|
||||||
|
`reflects generation ${result.window?.to}.`
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @description B3 Deliverable 3 — THE ONLINE METADATA REBUILD.
|
||||||
|
* `repairIndex()`'s ceremony door for the `'metadata'` family routes here
|
||||||
|
* instead of calling `MetadataIndexManager.rebuild()` directly: build a
|
||||||
|
* FRESH replacement manager BESIDE the live one (same storage, same
|
||||||
|
* idMapper — identity is shared, never a second mapper), walk canonical
|
||||||
|
* into it while every live write during the build ALSO mirrors there
|
||||||
|
* (`MetadataIndexManager.beginShadow`), fold the generation window the
|
||||||
|
* walk may have read stale, then atomically swap this brain's reference —
|
||||||
|
* `this.metadataIndex` points at the OLD manager for the ENTIRE build, so
|
||||||
|
* every read in progress (and every read that starts before the swap
|
||||||
|
* line executes) keeps serving its full, unbuilt-adjacent population;
|
||||||
|
* nothing ever observes a half-built index.
|
||||||
|
*
|
||||||
|
* PERSISTENCE CHOICE (named per the B3 brief): the JS manager's persisted
|
||||||
|
* keys (field-index chunks, column-store segments, the watermark stamp,
|
||||||
|
* the id-mapper record) are GLOBAL per storage — not namespaced per
|
||||||
|
* manager instance — so two managers cannot safely persist independently
|
||||||
|
* mid-build (a segment-number race, a stamp race, an id-mapper reload
|
||||||
|
* that would discard the live manager's not-yet-flushed assignments —
|
||||||
|
* see `MetadataIndexManager.initForShadowBuild`'s JSDoc for the id-mapper
|
||||||
|
* hazard specifically). This build therefore PERSISTS ONLY AT SWAP: the
|
||||||
|
* shadow builds entirely in memory (`rebuild({ inMemoryOnly: true })` +
|
||||||
|
* a fact-log fold — neither touches storage) and flushes exactly once,
|
||||||
|
* after the swap, as the sole owner of the shared keys.
|
||||||
|
*
|
||||||
|
* FALLBACK: a store with no fact log (or a non-JS/native metadata
|
||||||
|
* provider — its own train owns its online-rebuild strategy) cannot
|
||||||
|
* safely bound "what landed during the walk"; this method falls back to
|
||||||
|
* the ORIGINAL blocking clear-then-walk `rebuild()`, narrated.
|
||||||
|
*/
|
||||||
|
private async rebuildMetadataIndexOnline(): Promise<void> {
|
||||||
|
if (!(this.metadataIndex instanceof MetadataIndexManager)) {
|
||||||
|
// A registered provider (e.g. a native accelerator) may replace
|
||||||
|
// `this.metadataIndex` with a non-MetadataIndexManager object at
|
||||||
|
// runtime even though the field's declared type is the JS class —
|
||||||
|
// the cast mirrors the same reach-in used elsewhere in this file
|
||||||
|
// (e.g. checkHealth()'s `metadataProvider` locals) for exactly this.
|
||||||
|
const provider = this.metadataIndex as unknown as MetadataIndexProvider
|
||||||
|
await provider.rebuild()
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
const committedAtStart = this.storage.committedGeneration?.() ?? null
|
||||||
|
const factLogAvailable = committedAtStart !== null && this.scanFacts() !== null
|
||||||
|
if (!factLogAvailable) {
|
||||||
|
prodLog.warn(
|
||||||
|
`[Brainy] repairIndex(): metadata rebuild — no fact log on this store, build-beside ` +
|
||||||
|
`is unavailable; falling back to the blocking rebuild (reads may serve a ` +
|
||||||
|
`partially-built index for its duration).`
|
||||||
|
)
|
||||||
|
await this.metadataIndex.rebuild()
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
prodLog.warn(
|
||||||
|
`[Brainy] repairIndex(): metadata rebuild — building a fresh replacement index BESIDE ` +
|
||||||
|
`the live one (reads keep serving the current index throughout); swapping in ` +
|
||||||
|
`atomically once it is caught up.`
|
||||||
|
)
|
||||||
|
const startedAt = Date.now()
|
||||||
|
const oldManager = this.metadataIndex
|
||||||
|
const shadow = new MetadataIndexManager(this.storage, {}, {
|
||||||
|
entityIdMapper: oldManager.getIdMapper()
|
||||||
|
})
|
||||||
|
|
||||||
|
oldManager.beginShadow(shadow)
|
||||||
|
let committedAtSwap: number
|
||||||
|
try {
|
||||||
|
await shadow.buildBeside(committedAtStart!)
|
||||||
|
// Capture the true final generation right before the swap — a
|
||||||
|
// synchronous read, no `await` between here and the reference
|
||||||
|
// assignment below, so nothing can land ungoverned in the gap: the
|
||||||
|
// shadow has been live-mirroring every write since beginShadow()
|
||||||
|
// above, and this generation is the floor a FUTURE open's watermark
|
||||||
|
// verdict will trust once stamped.
|
||||||
|
committedAtSwap = this.storage.committedGeneration?.() ?? committedAtStart!
|
||||||
|
} catch (err) {
|
||||||
|
oldManager.endShadow()
|
||||||
|
prodLog.error(
|
||||||
|
`[Brainy] repairIndex(): online metadata rebuild FAILED during the walk/fold — the ` +
|
||||||
|
`live index is UNCHANGED (never swapped); reads keep serving the current ` +
|
||||||
|
`(pre-rebuild) metadata index. Error: ${(err as Error).message}`
|
||||||
|
)
|
||||||
|
throw err
|
||||||
|
}
|
||||||
|
|
||||||
|
oldManager.endShadow()
|
||||||
|
this.metadataIndex = shadow
|
||||||
|
|
||||||
|
// NOW persist — the shadow is the SOLE owner of the shared storage keys
|
||||||
|
// (nothing references `oldManager` any more; it never flushes again).
|
||||||
|
shadow.stampWatermark(committedAtSwap)
|
||||||
|
await shadow.flush()
|
||||||
|
|
||||||
|
prodLog.warn(
|
||||||
|
`[Brainy] repairIndex(): online metadata rebuild complete in ${Date.now() - startedAt}ms — ` +
|
||||||
|
`swapped in a fresh index reflecting generation ${committedAtSwap}, zero read downtime.`
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @description Rebuild indexes from persisted data if needed — THE OPEN-TIME
|
* @description Rebuild indexes from persisted data if needed — THE OPEN-TIME
|
||||||
* BUILD. Called once per open (init calls it; `repairIndex()`'s
|
* BUILD. Called once per open (init calls it; `repairIndex()`'s
|
||||||
|
|
@ -17189,7 +17493,15 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
`[Brainy] repairIndex(): explicit rebuild requested for '${familyName}' — ` +
|
`[Brainy] repairIndex(): explicit rebuild requested for '${familyName}' — ` +
|
||||||
`rebuilding unconditionally (no invariant consulted).`
|
`rebuilding unconditionally (no invariant consulted).`
|
||||||
)
|
)
|
||||||
|
// The metadata family routes through the online build-beside
|
||||||
|
// orchestrator (B3 D3) instead of the provider's own rebuild() —
|
||||||
|
// zero read downtime when a fact log is available, narrated
|
||||||
|
// fallback to the blocking rebuild() otherwise.
|
||||||
|
if (familyName === 'metadata') {
|
||||||
|
await this.rebuildMetadataIndexOnline()
|
||||||
|
} else {
|
||||||
await p.rebuild()
|
await p.rebuild()
|
||||||
|
}
|
||||||
record(`provider:${familyName}`, {
|
record(`provider:${familyName}`, {
|
||||||
checked: true,
|
checked: true,
|
||||||
healed: 1,
|
healed: 1,
|
||||||
|
|
@ -17230,7 +17542,13 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
`[Brainy] repairIndex(): provider '${report.provider}' has a failing invariant ` +
|
`[Brainy] repairIndex(): provider '${report.provider}' has a failing invariant ` +
|
||||||
`requiring a rebuild — reconciling its derived state from canonical.`
|
`requiring a rebuild — reconciling its derived state from canonical.`
|
||||||
)
|
)
|
||||||
|
// See the explicit-rebuild branch above: 'metadata' routes through
|
||||||
|
// the online build-beside orchestrator (B3 D3).
|
||||||
|
if (familyName === 'metadata') {
|
||||||
|
await this.rebuildMetadataIndexOnline()
|
||||||
|
} else {
|
||||||
await p.rebuild()
|
await p.rebuild()
|
||||||
|
}
|
||||||
} else {
|
} else {
|
||||||
record(`provider:${report.provider}`, {
|
record(`provider:${report.provider}`, {
|
||||||
checked: true, healed: 0,
|
checked: true, healed: 0,
|
||||||
|
|
@ -17821,6 +18139,14 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
}
|
}
|
||||||
await this.autoCompactHistory()
|
await this.autoCompactHistory()
|
||||||
|
|
||||||
|
// Watermark stamps ride this flush too — see stampProjectionWatermarks().
|
||||||
|
// Read-only instances skip it (no writes, no committed-generation drift
|
||||||
|
// to certify; ensureInitialized()'s guard below never runs for them
|
||||||
|
// either, so this must not assume a writer's invariants).
|
||||||
|
if (!this.isReadOnly) {
|
||||||
|
this.stampProjectionWatermarks()
|
||||||
|
}
|
||||||
|
|
||||||
// Phase 1: Flush ALL components in parallel to persist buffered data
|
// Phase 1: Flush ALL components in parallel to persist buffered data
|
||||||
// This is critical when cor native providers buffer data in Rust memory
|
// This is critical when cor native providers buffer data in Rust memory
|
||||||
await Promise.all([
|
await Promise.all([
|
||||||
|
|
|
||||||
|
|
@ -20,6 +20,7 @@ import {
|
||||||
type WatermarkVerdict,
|
type WatermarkVerdict,
|
||||||
type WatermarkVerdictResult
|
type WatermarkVerdictResult
|
||||||
} from './projectionWatermark.js'
|
} from './projectionWatermark.js'
|
||||||
|
import type { FactScanHandle } from '../db/factLog.js'
|
||||||
import {
|
import {
|
||||||
NounType,
|
NounType,
|
||||||
VerbType,
|
VerbType,
|
||||||
|
|
@ -77,6 +78,31 @@ export interface MetadataIndexStats {
|
||||||
indexSize: number // in bytes
|
indexSize: number // in bytes
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @description What {@link MetadataIndexManager.applyWatermarkCatchup} did,
|
||||||
|
* for the caller's narration.
|
||||||
|
* - `'noop'` — the verdict was `null`/`'adopt'`: the artifact already
|
||||||
|
* reflects committed truth. Zero index writes.
|
||||||
|
* - `'rescan'` — the verdict was `'rescan'`, OR a `'catchup'` verdict was
|
||||||
|
* demoted (no window, or no fact log to scan) — either way a full
|
||||||
|
* {@link MetadataIndexManager.rebuild} already ran; `reason` names why.
|
||||||
|
* - `'caught-up'` — the `(from, to]` window folded successfully; the
|
||||||
|
* artifact is stamped and flushed at `to`.
|
||||||
|
*/
|
||||||
|
export interface CatchupApplyResult {
|
||||||
|
action: 'noop' | 'rescan' | 'caught-up'
|
||||||
|
/** Present on `'rescan'` — why the fold could not proceed as a catchup. */
|
||||||
|
reason?: string
|
||||||
|
/** Present on `'caught-up'` — the fact-log window that was folded. */
|
||||||
|
window?: { from: number; to: number }
|
||||||
|
/** Present on `'caught-up'` — noun ops applied (add/update/delete). */
|
||||||
|
nounsApplied?: number
|
||||||
|
/** Present on `'caught-up'` — verb ops applied (add/update/delete). */
|
||||||
|
verbsApplied?: number
|
||||||
|
/** Present on `'caught-up'` — distinct committed generations folded. */
|
||||||
|
factsApplied?: number
|
||||||
|
}
|
||||||
|
|
||||||
export interface MetadataIndexConfig {
|
export interface MetadataIndexConfig {
|
||||||
maxIndexSize?: number // Max number of entries per field value (default: 10000)
|
maxIndexSize?: number // Max number of entries per field value (default: 10000)
|
||||||
rebuildThreshold?: number // Rebuild if index is this % stale (default: 0.1)
|
rebuildThreshold?: number // Rebuild if index is this % stale (default: 0.1)
|
||||||
|
|
@ -147,6 +173,52 @@ export class MetadataIndexManager implements MetadataIndexProvider {
|
||||||
private stampedWatermark: number | null = null
|
private stampedWatermark: number | null = null
|
||||||
/** The three-way verdict computed at init; null until init runs. */
|
/** The three-way verdict computed at init; null until init runs. */
|
||||||
private loadVerdict: WatermarkVerdictResult | null = null
|
private loadVerdict: WatermarkVerdictResult | null = null
|
||||||
|
/**
|
||||||
|
* Set only when {@link loadVerdict}.verdict is `'rescan'`: whether a
|
||||||
|
* persisted artifact existed at load (even an unstamped/unverifiable
|
||||||
|
* one) — distinguishes genuine first boot (nothing here yet, routine)
|
||||||
|
* from an artifact whose watermark is unverifiable (the loud case). The
|
||||||
|
* verdict value alone doesn't carry this distinction; see {@link
|
||||||
|
* watermarkArtifactPresent}.
|
||||||
|
*/
|
||||||
|
private rescanArtifactPresent = false
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @description THE BUILD-BESIDE SEAM (B3 Deliverable 3): when set (via
|
||||||
|
* {@link beginShadow}), every live `addToIndex`/`removeFromIndex` call on
|
||||||
|
* THIS instance also applies to the shadow instance — so a caller building
|
||||||
|
* a fresh replacement manager beside this one (walking canonical into it)
|
||||||
|
* never misses a write that lands during the build. This is the ONE seam
|
||||||
|
* that makes build-beside possible without touching every call site: every
|
||||||
|
* existing `AddToMetadataIndexOperation`/`RemoveFromMetadataIndexOperation`
|
||||||
|
* (and the JS manager's own `rebuild()`/catchup fold) keep calling the SAME
|
||||||
|
* serving instance exactly as before; only THIS instance knows it is also
|
||||||
|
* mirroring to a shadow. Null = no build in flight (the overwhelmingly
|
||||||
|
* common case; the check costs one property read per write).
|
||||||
|
*/
|
||||||
|
private shadow: MetadataIndexManager | null = null
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @description Start mirroring every `addToIndex`/`removeFromIndex` call on
|
||||||
|
* this instance to `shadow` too — see {@link shadow}'s JSDoc. The caller
|
||||||
|
* owns sequencing: writes mirrored WHILE a canonical walk is populating
|
||||||
|
* `shadow` may be clobbered by the walk's own (possibly stale) reads for
|
||||||
|
* the same id; the caller closes that window with a bounded fact-log fold
|
||||||
|
* AFTER the walk (the same mechanism {@link applyWatermarkCatchup} uses)
|
||||||
|
* before treating `shadow` as authoritative.
|
||||||
|
* @param shadow - The manager to mirror writes to.
|
||||||
|
*/
|
||||||
|
beginShadow(shadow: MetadataIndexManager): void {
|
||||||
|
this.shadow = shadow
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @description Stop mirroring writes to a shadow (see {@link beginShadow}).
|
||||||
|
* Idempotent; a no-op when no shadow is attached.
|
||||||
|
*/
|
||||||
|
endShadow(): void {
|
||||||
|
this.shadow = null
|
||||||
|
}
|
||||||
|
|
||||||
// Cardinality and field statistics tracking
|
// Cardinality and field statistics tracking
|
||||||
private fieldStats = new Map<string, FieldStats>()
|
private fieldStats = new Map<string, FieldStats>()
|
||||||
|
|
@ -1604,6 +1676,15 @@ export class MetadataIndexManager implements MetadataIndexProvider {
|
||||||
for (const { field } of fields) {
|
for (const { field } of fields) {
|
||||||
this.metadataCache.invalidatePattern(`field_values_${field}`)
|
this.metadataCache.invalidatePattern(`field_values_${field}`)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// THE BUILD-BESIDE SEAM — see `shadow`'s JSDoc. Mirrors this write to a
|
||||||
|
// shadow manager under construction, if one is attached. `skipFlush:
|
||||||
|
// true` always: the shadow's own persistence is the build orchestrator's
|
||||||
|
// job (it flushes once, after the swap — never mid-build, to avoid
|
||||||
|
// colliding with this instance's own persisted keys).
|
||||||
|
if (this.shadow) {
|
||||||
|
await this.shadow.addToIndex(id, entityOrMetadata, true, false, generation)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
@ -1676,6 +1757,11 @@ export class MetadataIndexManager implements MetadataIndexProvider {
|
||||||
// the real commit watermark (the JS mapper ignores it).
|
// the real commit watermark (the JS mapper ignores it).
|
||||||
this.idMapper.remove(id, generation)
|
this.idMapper.remove(id, generation)
|
||||||
await this.idMapper.flush()
|
await this.idMapper.flush()
|
||||||
|
|
||||||
|
// THE BUILD-BESIDE SEAM — see `shadow`'s JSDoc.
|
||||||
|
if (this.shadow) {
|
||||||
|
await this.shadow.removeFromIndex(id, metadata, generation)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
@ -2759,8 +2845,8 @@ export class MetadataIndexManager implements MetadataIndexProvider {
|
||||||
* `'adopt'` (stamped == committed, zero work), `'catchup'` (stamped <
|
* `'adopt'` (stamped == committed, zero work), `'catchup'` (stamped <
|
||||||
* committed; the gap from {@link watermarkGap} awaits an incremental
|
* committed; the gap from {@link watermarkGap} awaits an incremental
|
||||||
* fold), `'rescan'` (unstamped or stamped above committed — never
|
* fold), `'rescan'` (unstamped or stamped above committed — never
|
||||||
* trusted). Null until init() has run. Computed and exposed only; no
|
* trusted). Null until init() has run. The coordinator (`Brainy.open()`)
|
||||||
* load behavior changes ride on it yet.
|
* consumes this via {@link applyWatermarkCatchup} right after init.
|
||||||
*/
|
*/
|
||||||
watermarkVerdict(): WatermarkVerdict | null {
|
watermarkVerdict(): WatermarkVerdict | null {
|
||||||
return this.loadVerdict?.verdict ?? null
|
return this.loadVerdict?.verdict ?? null
|
||||||
|
|
@ -2774,6 +2860,223 @@ export class MetadataIndexManager implements MetadataIndexProvider {
|
||||||
return this.loadVerdict?.gap ?? null
|
return this.loadVerdict?.gap ?? null
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @description Meaningful only when {@link watermarkVerdict} is
|
||||||
|
* `'rescan'`: `true` when a persisted artifact existed at load (even an
|
||||||
|
* unstamped/unverifiable one — real prior state, worth narrating loudly);
|
||||||
|
* `false` for a genuine first boot (nothing persisted yet — a caller
|
||||||
|
* should narrate this at a routine log level, not as an alarm, even
|
||||||
|
* though the verdict value is the same `'rescan'` either way).
|
||||||
|
*/
|
||||||
|
watermarkArtifactPresent(): boolean {
|
||||||
|
return this.rescanArtifactPresent
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @description Consume the three-way watermark verdict {@link
|
||||||
|
* watermarkVerdict} computed at init — the cure for a crash-recovered
|
||||||
|
* store whose canonical reads/counts recover every acked write but whose
|
||||||
|
* metadata projection (flushed only periodically, not per-commit) keeps
|
||||||
|
* serving the pre-crash state. Call once, right after `init()`, before
|
||||||
|
* anything reads from this projection.
|
||||||
|
*
|
||||||
|
* - `null`/`'adopt'` → the artifact already reflects the store's
|
||||||
|
* committed generation. Zero index writes.
|
||||||
|
* - `'catchup'` → the caller-supplied `scan` (expected already opened
|
||||||
|
* over `(watermarkGap().from, watermarkGap().to]`) is folded in, ONE
|
||||||
|
* op at a time, through the SAME two legs {@link rebuild} uses (ADR-007
|
||||||
|
* A4 — one mechanism, never a second hand-rolled add/update shape): a
|
||||||
|
* tombstone (`op.record === null`) retracts id-keyed (this projection
|
||||||
|
* keeps no per-record delta log, so the pre-crash metadata for that id
|
||||||
|
* — if any — is what a value-precise removal would need, and it isn't
|
||||||
|
* available; the same tradeoff `remove()`'s null-metadata closure
|
||||||
|
* already accepts elsewhere); an after-image retracts-then-reposts, so
|
||||||
|
* an update never leaves stale postings under the old field values. A
|
||||||
|
* fact outside the window is skipped defensively (belt: the scan is
|
||||||
|
* already opened to the window; suspenders: this loop never trusts an
|
||||||
|
* over-run). On success the artifact is stamped at `to` and flushed —
|
||||||
|
* the same STAMP-AFTER-DATA door {@link flush} always writes through.
|
||||||
|
* - `'rescan'` (or a `'catchup'` verdict with no window, or no `scan` to
|
||||||
|
* fold — the store hosts no fact log) → the persisted artifact is
|
||||||
|
* unverifiable; this method runs the existing {@link rebuild} itself
|
||||||
|
* rather than leave the caller to notice and trigger it separately.
|
||||||
|
*
|
||||||
|
* @param scan - An open fact scan covering the catchup window (see
|
||||||
|
* {@link Brainy.scanFacts}), or `null` when none is available/needed.
|
||||||
|
* Ignored when the verdict is not `'catchup'`.
|
||||||
|
* @returns What happened — see {@link CatchupApplyResult}.
|
||||||
|
*/
|
||||||
|
async applyWatermarkCatchup(scan: FactScanHandle | null): Promise<CatchupApplyResult> {
|
||||||
|
const verdict = this.watermarkVerdict()
|
||||||
|
if (verdict === null || verdict === 'adopt') return { action: 'noop' }
|
||||||
|
|
||||||
|
if (verdict === 'rescan') {
|
||||||
|
await this.rebuild()
|
||||||
|
return {
|
||||||
|
action: 'rescan',
|
||||||
|
reason: 'persisted artifact is unverifiable (unstamped, or stamped ABOVE the ' +
|
||||||
|
"store's committed generation) — never adopting unverifiable state"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// verdict === 'catchup'
|
||||||
|
const window = this.watermarkGap()
|
||||||
|
if (window === null) {
|
||||||
|
await this.rebuild()
|
||||||
|
return { action: 'rescan', reason: "'catchup' verdict exposed no window — cannot bound a fold" }
|
||||||
|
}
|
||||||
|
if (scan === null) {
|
||||||
|
await this.rebuild()
|
||||||
|
return {
|
||||||
|
action: 'rescan',
|
||||||
|
reason: `no fact log available to fold the (${window.from}, ${window.to}] catchup window`
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
const { nounsApplied, verbsApplied, factsApplied } = await this.foldFactWindow(scan, window.from, window.to)
|
||||||
|
|
||||||
|
this.stampWatermark(window.to)
|
||||||
|
await this.flush()
|
||||||
|
return { action: 'caught-up', window, nounsApplied, verbsApplied, factsApplied }
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @description Fold an open fact scan's `(fromGeneration, toGeneration]`
|
||||||
|
* window into this projection, ONE op at a time, through the SAME two legs
|
||||||
|
* {@link rebuild} uses (ADR-007 A4 — one mechanism, never a second
|
||||||
|
* hand-rolled add/update shape): a tombstone retracts id-keyed; an
|
||||||
|
* after-image retracts-then-reposts. THE CORE LOOP shared by {@link
|
||||||
|
* applyWatermarkCatchup} (which stamps + flushes after) and {@link
|
||||||
|
* buildBeside} (which does neither — persistence is the caller's job,
|
||||||
|
* exactly once, after a swap). Never stamps, never flushes, never touches
|
||||||
|
* storage beyond what `addToIndex`/`removeFromIndex` do internally
|
||||||
|
* (skipFlush is always forced true).
|
||||||
|
* @param scan - An open fact scan.
|
||||||
|
* @param fromGeneration - Window lower bound (exclusive).
|
||||||
|
* @param toGeneration - Window upper bound (inclusive).
|
||||||
|
* @returns Counts for the caller's narration.
|
||||||
|
*/
|
||||||
|
private async foldFactWindow(
|
||||||
|
scan: FactScanHandle,
|
||||||
|
fromGeneration: number,
|
||||||
|
toGeneration: number
|
||||||
|
): Promise<{ nounsApplied: number; verbsApplied: number; factsApplied: number }> {
|
||||||
|
let nounsApplied = 0
|
||||||
|
let verbsApplied = 0
|
||||||
|
let factsApplied = 0
|
||||||
|
for await (const batch of scan.batches()) {
|
||||||
|
for (const fact of batch.facts) {
|
||||||
|
// Defensive containment: the scan is already opened to the window,
|
||||||
|
// but a fact outside it is never applied regardless.
|
||||||
|
if (fact.generation <= fromGeneration || fact.generation > toGeneration) continue
|
||||||
|
const generation = BigInt(fact.generation)
|
||||||
|
for (const op of fact.ops) {
|
||||||
|
if (op.record === null) {
|
||||||
|
// TOMBSTONE — the id-keyed removal path (no per-record delta
|
||||||
|
// log to recover the old field values from).
|
||||||
|
await this.removeFromIndex(op.id, undefined, generation)
|
||||||
|
} else {
|
||||||
|
// AFTER-IMAGE — retract any stale posting for this id, then
|
||||||
|
// repost the new shape. Covers both a fresh add (nothing to
|
||||||
|
// retract; a no-op-ish remove) and an update, through the same
|
||||||
|
// two calls.
|
||||||
|
await this.removeFromIndex(op.id, undefined, generation)
|
||||||
|
await this.indexStoredRecord(op.id, op.record.metadata, {
|
||||||
|
skipFlush: true,
|
||||||
|
deferWrites: false,
|
||||||
|
generation
|
||||||
|
})
|
||||||
|
}
|
||||||
|
if (op.kind === 'noun') nounsApplied++
|
||||||
|
else verbsApplied++
|
||||||
|
}
|
||||||
|
factsApplied++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return { nounsApplied, verbsApplied, factsApplied }
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @description B3 Deliverable 3 — the shadow-build lifecycle's init: the
|
||||||
|
* MINIMUM setup {@link buildBeside} needs, deliberately NOT the general
|
||||||
|
* {@link init} sequence. Two reasons general `init()` is unsafe for a
|
||||||
|
* build-beside shadow:
|
||||||
|
* 1. `init()` unconditionally re-initializes the id mapper from storage
|
||||||
|
* (`idMapper.init()`) — safe for a FRESH mapper, but this instance is
|
||||||
|
* constructed with the CURRENTLY-SERVING manager's SHARED, already-live
|
||||||
|
* mapper (identity is shared, never a second mapper — this train's own
|
||||||
|
* law). Re-running its init() would DISCARD every not-yet-flushed
|
||||||
|
* UUID↔int assignment sitting in memory, breaking the live manager's
|
||||||
|
* own serving mid-build.
|
||||||
|
* 2. `init()` loads the field registry and, on a registry that's
|
||||||
|
* missing/empty while canonical has entities (exactly the shape a
|
||||||
|
* rebuild is often invoked to FIX), triggers `rebuild()` itself —
|
||||||
|
* WITHOUT `inMemoryOnly`, which would touch the shared storage keys
|
||||||
|
* the live manager depends on.
|
||||||
|
* What this DOES run: the WASM roaring-bitmap library init (idempotent;
|
||||||
|
* needed before any column-store write) and the column store's OWN
|
||||||
|
* segment-manifest discovery (read-only against shared storage; needed so
|
||||||
|
* THIS instance's eventual post-swap flush continues segment numbering
|
||||||
|
* correctly instead of colliding with the retiring manager's segments).
|
||||||
|
*/
|
||||||
|
private async initForShadowBuild(): Promise<void> {
|
||||||
|
await roaringLibraryInitialize()
|
||||||
|
try {
|
||||||
|
await this.columnStore.init(this.storage, this.idMapper)
|
||||||
|
} catch (err) {
|
||||||
|
prodLog.warn('[MetadataIndex] shadow build: column store storage discovery failed:', err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @description B3 Deliverable 3 — THE ONLINE REBUILD's manager-side half:
|
||||||
|
* populate THIS instance (expected fresh/empty, constructed with the SAME
|
||||||
|
* storage + idMapper as the manager it will replace — see {@link
|
||||||
|
* initForShadowBuild}) from canonical storage without ever touching the
|
||||||
|
* shared storage keys the currently-serving manager depends on — no chunk
|
||||||
|
* deletion, no flush, anywhere in this call. The caller (the brain's
|
||||||
|
* rebuild-beside orchestrator) is responsible for:
|
||||||
|
* 1. Attaching this instance as a {@link beginShadow} target on the OLD
|
||||||
|
* manager BEFORE calling this, so live writes during the walk mirror
|
||||||
|
* here too (best-effort — the walk below may still clobber a mirrored
|
||||||
|
* write with a stale read for the same id; the fold after the walk is
|
||||||
|
* what makes the final state authoritative, not the mirror).
|
||||||
|
* 2. Swapping its own reference to this instance once this resolves.
|
||||||
|
* 3. Calling {@link stampWatermark} + {@link flush} EXACTLY ONCE, after
|
||||||
|
* the swap — this instance never persists itself.
|
||||||
|
* @param committedGenerationAtStart - The store's committed generation
|
||||||
|
* captured by the caller BEFORE this call — the fold's lower bound.
|
||||||
|
* @returns The generation this instance's canonical data reflects once the
|
||||||
|
* walk + fold settle — the fold's upper bound (writes committed after
|
||||||
|
* this point but before the swap only reach this instance via the live
|
||||||
|
* {@link beginShadow} mirror, so the caller re-reads the store's
|
||||||
|
* committed generation right before stamping, rather than trusting this
|
||||||
|
* return value as final).
|
||||||
|
* @throws If canonical advanced during the walk but no fact log is
|
||||||
|
* available to fold the gap — never a silently incomplete shadow.
|
||||||
|
*/
|
||||||
|
async buildBeside(committedGenerationAtStart: number): Promise<number> {
|
||||||
|
await this.initForShadowBuild()
|
||||||
|
await this.rebuild({ inMemoryOnly: true })
|
||||||
|
|
||||||
|
const committedAfterWalk = this.storage.committedGeneration?.() ?? committedGenerationAtStart
|
||||||
|
if (committedAfterWalk > committedGenerationAtStart) {
|
||||||
|
const scan = this.storage.scanFacts?.({
|
||||||
|
fromGeneration: committedGenerationAtStart + 1,
|
||||||
|
toGeneration: committedAfterWalk
|
||||||
|
}) ?? null
|
||||||
|
if (scan === null) {
|
||||||
|
throw new Error(
|
||||||
|
`MetadataIndexManager.buildBeside: canonical advanced from generation ` +
|
||||||
|
`${committedGenerationAtStart} to ${committedAfterWalk} during the walk, but this ` +
|
||||||
|
`store hosts no fact log to fold the gap — refusing a silently incomplete shadow`
|
||||||
|
)
|
||||||
|
}
|
||||||
|
await this.foldFactWindow(scan, committedGenerationAtStart, committedAfterWalk)
|
||||||
|
}
|
||||||
|
return committedAfterWalk
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @description Write the pending watermark stamp as a sidecar record —
|
* @description Write the pending watermark stamp as a sidecar record —
|
||||||
* always called AFTER the data it certifies is durable. A stamp-write
|
* always called AFTER the data it certifies is durable. A stamp-write
|
||||||
|
|
@ -2826,6 +3129,7 @@ export class MetadataIndexManager implements MetadataIndexProvider {
|
||||||
|
|
||||||
if (result.verdict === 'rescan') {
|
if (result.verdict === 'rescan') {
|
||||||
const artifactPresent = this.fieldIndexes.size > 0 || stamped !== null
|
const artifactPresent = this.fieldIndexes.size > 0 || stamped !== null
|
||||||
|
this.rescanArtifactPresent = artifactPresent
|
||||||
if (artifactPresent) {
|
if (artifactPresent) {
|
||||||
prodLog.warn(
|
prodLog.warn(
|
||||||
`[MetadataIndex] watermark verdict: RESCAN — persisted index is ` +
|
`[MetadataIndex] watermark verdict: RESCAN — persisted index is ` +
|
||||||
|
|
@ -3484,13 +3788,48 @@ export class MetadataIndexManager implements MetadataIndexProvider {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @description Index one raw stored noun/verb record — THE ONE add leg
|
||||||
|
* shared by {@link rebuild}'s canonical walk and {@link
|
||||||
|
* applyWatermarkCatchup}'s fact-log fold (ADR-007 A4: one mechanism,
|
||||||
|
* never a second hand-rolled shape). No conversion step is needed here:
|
||||||
|
* a raw stored record (`storage.getNounMetadata`/`getVerbMetadata`, or a
|
||||||
|
* fact's after-image `record.metadata`) is byte-identical — both read the
|
||||||
|
* exact same canonical path — and already the v2 nested-bag
|
||||||
|
* ("entity-record") shape {@link extractIndexableFields} expects.
|
||||||
|
* @param id - Entity/relationship id.
|
||||||
|
* @param storedMetadata - The raw stored metadata record.
|
||||||
|
* @param opts.skipFlush - Forwarded to {@link addToIndex}.
|
||||||
|
* @param opts.deferWrites - Forwarded to {@link addToIndex}.
|
||||||
|
* @param opts.generation - Forwarded to {@link addToIndex}.
|
||||||
|
*/
|
||||||
|
private async indexStoredRecord(
|
||||||
|
id: string,
|
||||||
|
storedMetadata: unknown,
|
||||||
|
opts: { skipFlush: boolean; deferWrites: boolean; generation?: bigint }
|
||||||
|
): Promise<void> {
|
||||||
|
await this.addToIndex(id, storedMetadata, opts.skipFlush, opts.deferWrites, opts.generation)
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Rebuild entire index from scratch using pagination
|
* Rebuild entire index from scratch using pagination
|
||||||
* Non-blocking version that yields control back to event loop
|
* Non-blocking version that yields control back to event loop
|
||||||
* Sparse indices now lazy-loaded via UnifiedCache (no need to clear Map)
|
* Sparse indices now lazy-loaded via UnifiedCache (no need to clear Map)
|
||||||
|
*
|
||||||
|
* @param options.inMemoryOnly - B3 Deliverable 3 (build-beside): when
|
||||||
|
* `true`, this call never touches the shared storage keys another,
|
||||||
|
* currently-serving `MetadataIndexManager` over the SAME storage may
|
||||||
|
* depend on — it skips deleting persisted legacy chunk files AND skips
|
||||||
|
* the final `flush()` (which would otherwise write field indexes AND
|
||||||
|
* flush the column store's tail buffers to shared segment keys,
|
||||||
|
* colliding with a live manager's own writes). The caller ({@link
|
||||||
|
* buildBeside}) owns persistence entirely — exactly once, after this
|
||||||
|
* instance becomes the sole owner via an atomic swap. Default `false`
|
||||||
|
* (every other caller keeps today's clear-then-persist behavior).
|
||||||
*/
|
*/
|
||||||
async rebuild(): Promise<void> {
|
async rebuild(options?: { inMemoryOnly?: boolean }): Promise<void> {
|
||||||
if (this.isRebuilding) return
|
if (this.isRebuilding) return
|
||||||
|
const inMemoryOnly = options?.inMemoryOnly ?? false
|
||||||
|
|
||||||
this.isRebuilding = true
|
this.isRebuilding = true
|
||||||
try {
|
try {
|
||||||
|
|
@ -3519,6 +3858,12 @@ export class MetadataIndexManager implements MetadataIndexProvider {
|
||||||
// here — it's always saved at the end of rebuild via flush(). This ensures
|
// here — it's always saved at the end of rebuild via flush(). This ensures
|
||||||
// that if rebuild fails partway, the next init() can still discover fields
|
// that if rebuild fails partway, the next init() can still discover fields
|
||||||
// and trigger another rebuild attempt.
|
// and trigger another rebuild attempt.
|
||||||
|
//
|
||||||
|
// SKIPPED for inMemoryOnly: these are the SHARED storage keys a live
|
||||||
|
// manager over the same storage may still be reading (see this
|
||||||
|
// method's JSDoc) — deleting them before the swap is a live-read
|
||||||
|
// hazard, not a cleanup.
|
||||||
|
if (!inMemoryOnly) {
|
||||||
prodLog.info('Clearing existing metadata index chunks from storage...')
|
prodLog.info('Clearing existing metadata index chunks from storage...')
|
||||||
const existingFields = await this.getPersistedFieldList()
|
const existingFields = await this.getPersistedFieldList()
|
||||||
|
|
||||||
|
|
@ -3529,6 +3874,7 @@ export class MetadataIndexManager implements MetadataIndexProvider {
|
||||||
|
|
||||||
prodLog.info(`Cleared ${existingFields.length} field indexes from storage`)
|
prodLog.info(`Cleared ${existingFields.length} field indexes from storage`)
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// EntityIdMapper is intentionally NOT cleared here. Rebuild re-iterates
|
// EntityIdMapper is intentionally NOT cleared here. Rebuild re-iterates
|
||||||
// every entity in storage and calls idMapper.getOrAssign(uuid), which
|
// every entity in storage and calls idMapper.getOrAssign(uuid), which
|
||||||
|
|
@ -3582,7 +3928,7 @@ export class MetadataIndexManager implements MetadataIndexProvider {
|
||||||
for (const noun of result.items) {
|
for (const noun of result.items) {
|
||||||
const metadata = metadataBatch.get(noun.id)
|
const metadata = metadataBatch.get(noun.id)
|
||||||
if (metadata) {
|
if (metadata) {
|
||||||
await this.addToIndex(noun.id, metadata, true, true)
|
await this.indexStoredRecord(noun.id, metadata, { skipFlush: true, deferWrites: true })
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -3627,7 +3973,7 @@ export class MetadataIndexManager implements MetadataIndexProvider {
|
||||||
for (const verb of result.items) {
|
for (const verb of result.items) {
|
||||||
const metadata = verbMetadataBatch.get(verb.id)
|
const metadata = verbMetadataBatch.get(verb.id)
|
||||||
if (metadata) {
|
if (metadata) {
|
||||||
await this.addToIndex(verb.id, metadata, true, true)
|
await this.indexStoredRecord(verb.id, metadata, { skipFlush: true, deferWrites: true })
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -3637,8 +3983,16 @@ export class MetadataIndexManager implements MetadataIndexProvider {
|
||||||
|
|
||||||
// Flush to storage. The column store's flush() handles tail-buffer-to-
|
// Flush to storage. The column store's flush() handles tail-buffer-to-
|
||||||
// segment promotion + manifest persistence.
|
// segment promotion + manifest persistence.
|
||||||
|
//
|
||||||
|
// SKIPPED for inMemoryOnly — see this method's JSDoc: flush() writes
|
||||||
|
// the shared field-index keys AND flushes the column store's tail
|
||||||
|
// buffers to shared segment keys, which would race a live manager's
|
||||||
|
// own flushes over the SAME storage. The caller flushes exactly once,
|
||||||
|
// after the swap.
|
||||||
|
if (!inMemoryOnly) {
|
||||||
prodLog.debug('💾 Flushing metadata index to storage...')
|
prodLog.debug('💾 Flushing metadata index to storage...')
|
||||||
await this.flush()
|
await this.flush()
|
||||||
|
}
|
||||||
|
|
||||||
prodLog.info(`✅ Metadata index rebuild completed! Processed ${totalNounsProcessed} nouns and ${totalVerbsProcessed} verbs`)
|
prodLog.info(`✅ Metadata index rebuild completed! Processed ${totalNounsProcessed} nouns and ${totalVerbsProcessed} verbs`)
|
||||||
|
|
||||||
|
|
|
||||||
167
tests/integration/metadata-online-rebuild.test.ts
Normal file
167
tests/integration/metadata-online-rebuild.test.ts
Normal file
|
|
@ -0,0 +1,167 @@
|
||||||
|
/**
|
||||||
|
* @module tests/integration/metadata-online-rebuild
|
||||||
|
* @description THE ONLINE JS METADATA REBUILD (B3 Deliverable 3) pins.
|
||||||
|
* `MetadataIndexManager.rebuild()` used to be clear-then-walk — reads went
|
||||||
|
* dark for the duration. `repairIndex({ rebuild: ['metadata'] })` now builds
|
||||||
|
* a fresh replacement index BESIDE the live one (walk canonical + mirror
|
||||||
|
* every live write via `beginShadow`/`endShadow` + a bounded fact-log fold),
|
||||||
|
* then atomically swaps the brain's reference — `find()` never observes a
|
||||||
|
* half-built index, and a write landing DURING the build is never lost.
|
||||||
|
*/
|
||||||
|
process.env.BRAINY_DETERMINISTIC_EMBEDDINGS = 'true'
|
||||||
|
|
||||||
|
import { describe, it, expect, afterEach } from 'vitest'
|
||||||
|
import { mkdtempSync, rmSync } from 'node:fs'
|
||||||
|
import { tmpdir } from 'node:os'
|
||||||
|
import { join } from 'node:path'
|
||||||
|
import { Brainy } from '../../src/brainy.js'
|
||||||
|
import { NounType, VerbType } from '../../src/types/graphTypes.js'
|
||||||
|
import type { MetadataIndexManager } from '../../src/utils/metadataIndex.js'
|
||||||
|
|
||||||
|
const dirs: string[] = []
|
||||||
|
const brains: Brainy<any>[] = []
|
||||||
|
afterEach(async () => {
|
||||||
|
for (const b of brains.splice(0)) await b.close().catch(() => {})
|
||||||
|
for (const d of dirs.splice(0)) rmSync(d, { recursive: true, force: true })
|
||||||
|
})
|
||||||
|
|
||||||
|
function metadataIndexOf(brain: Brainy<any>): MetadataIndexManager {
|
||||||
|
return (brain as unknown as { metadataIndex: MetadataIndexManager }).metadataIndex
|
||||||
|
}
|
||||||
|
|
||||||
|
async function openBrain(): Promise<{ brain: Brainy<any>; dir: string }> {
|
||||||
|
const dir = mkdtempSync(join(tmpdir(), 'brainy-online-rebuild-'))
|
||||||
|
dirs.push(dir)
|
||||||
|
const brain = new Brainy<any>({
|
||||||
|
requireSubtype: false,
|
||||||
|
storage: { type: 'filesystem', path: dir },
|
||||||
|
silent: true,
|
||||||
|
persistence: { policy: 'manual' },
|
||||||
|
logAuthority: 'adopt'
|
||||||
|
})
|
||||||
|
await brain.init()
|
||||||
|
brains.push(brain)
|
||||||
|
return { brain, dir }
|
||||||
|
}
|
||||||
|
|
||||||
|
describe('repairIndex({ rebuild: ["metadata"] }) — the online build-beside rebuild', () => {
|
||||||
|
it(
|
||||||
|
'a find() polled throughout the rebuild of a 2k-noun store never returns fewer rows than ' +
|
||||||
|
'before the build started, and a write landing DURING the build is never lost',
|
||||||
|
async () => {
|
||||||
|
const { brain, dir } = await openBrain()
|
||||||
|
void dir
|
||||||
|
|
||||||
|
const N = 2000
|
||||||
|
const ids: string[] = []
|
||||||
|
for (let i = 0; i < N; i++) {
|
||||||
|
ids.push(
|
||||||
|
await brain.add({
|
||||||
|
data: `entity ${i}`,
|
||||||
|
type: NounType.Person,
|
||||||
|
metadata: { status: i % 2 === 0 ? 'active' : 'inactive' }
|
||||||
|
})
|
||||||
|
)
|
||||||
|
}
|
||||||
|
for (let i = 0; i < 20; i++) {
|
||||||
|
await brain.relate({
|
||||||
|
from: ids[i], to: ids[i + 1], type: VerbType.WorksWith, metadata: { tag: 'orig' }
|
||||||
|
})
|
||||||
|
}
|
||||||
|
await brain.flush()
|
||||||
|
|
||||||
|
const baseline = await brain.find({ where: { status: 'active' }, limit: 10000 })
|
||||||
|
expect(baseline.length).toBe(N / 2)
|
||||||
|
|
||||||
|
// Kick off the online rebuild WITHOUT awaiting — poll reads and
|
||||||
|
// perform a live write concurrently with it.
|
||||||
|
const repairPromise = brain.repairIndex({ rebuild: ['metadata'] })
|
||||||
|
|
||||||
|
let minObserved = Infinity
|
||||||
|
let polls = 0
|
||||||
|
const pollPromise = (async () => {
|
||||||
|
// Poll until the rebuild settles — bounded so a slow CI box can't
|
||||||
|
// spin forever, generous enough to actually overlap the walk.
|
||||||
|
while (polls < 200) {
|
||||||
|
const rows = await brain.find({ where: { status: 'active' }, limit: 10000 })
|
||||||
|
minObserved = Math.min(minObserved, rows.length)
|
||||||
|
polls++
|
||||||
|
await new Promise((resolve) => setTimeout(resolve, 1))
|
||||||
|
}
|
||||||
|
})()
|
||||||
|
|
||||||
|
const newId = await brain.add({
|
||||||
|
data: 'added during the rebuild',
|
||||||
|
type: NounType.Person,
|
||||||
|
metadata: { status: 'active' }
|
||||||
|
})
|
||||||
|
const newRelId = await brain.relate({
|
||||||
|
from: newId, to: ids[0], type: VerbType.WorksWith, metadata: { tag: 'during-build' }
|
||||||
|
})
|
||||||
|
|
||||||
|
const [report] = await Promise.all([repairPromise, pollPromise])
|
||||||
|
|
||||||
|
// THE PIN: never fewer rows than the pre-build baseline, at any polled
|
||||||
|
// instant — reads served the OLD (fully-populated) manager throughout.
|
||||||
|
expect(polls).toBeGreaterThan(0)
|
||||||
|
expect(minObserved).toBeGreaterThanOrEqual(baseline.length)
|
||||||
|
|
||||||
|
// The repair report still accounts for the family (same receipt shape
|
||||||
|
// regardless of which rebuild mechanism actually ran underneath).
|
||||||
|
const metadataFamily = report.families.find((f) => f.family === 'provider:metadata')
|
||||||
|
expect(metadataFamily?.checked).toBe(true)
|
||||||
|
expect(metadataFamily?.rebuilt).toBe(true)
|
||||||
|
|
||||||
|
// Post-swap correctness: the live write during the build was never
|
||||||
|
// lost (the beginShadow mirror + post-walk fold caught it).
|
||||||
|
const afterActive = await brain.find({ where: { status: 'active' }, limit: 10000 })
|
||||||
|
expect(afterActive.length).toBe(baseline.length + 1)
|
||||||
|
expect(afterActive.some((r) => r.id === newId)).toBe(true)
|
||||||
|
|
||||||
|
const index = metadataIndexOf(brain)
|
||||||
|
expect(await index.getIds('tag', 'during-build')).toEqual([newRelId])
|
||||||
|
expect((await index.getIds('tag', 'orig')).length).toBe(20)
|
||||||
|
|
||||||
|
// The swap stamped the watermark — a reopen adopts, zero rebuild.
|
||||||
|
await brain.close()
|
||||||
|
brains.length = 0 // already closed above; afterEach must not double-close
|
||||||
|
const reopened = new Brainy<any>({
|
||||||
|
requireSubtype: false,
|
||||||
|
storage: { type: 'filesystem', path: dir },
|
||||||
|
silent: true,
|
||||||
|
persistence: { policy: 'manual' },
|
||||||
|
logAuthority: 'adopt'
|
||||||
|
})
|
||||||
|
await reopened.init()
|
||||||
|
brains.push(reopened)
|
||||||
|
const reopenedIndex = metadataIndexOf(reopened)
|
||||||
|
expect(reopenedIndex.watermarkVerdict()).toBe('adopt')
|
||||||
|
const reopenedActive = await reopened.find({ where: { status: 'active' }, limit: 10000 })
|
||||||
|
expect(reopenedActive.length).toBe(afterActive.length)
|
||||||
|
},
|
||||||
|
60000
|
||||||
|
)
|
||||||
|
|
||||||
|
it('repairIndex({ rebuild: ["metadata"] }) on an empty store is a trivial no-op walk', async () => {
|
||||||
|
const { brain } = await openBrain()
|
||||||
|
const report = await brain.repairIndex({ rebuild: ['metadata'] })
|
||||||
|
const metadataFamily = report.families.find((f) => f.family === 'provider:metadata')
|
||||||
|
expect(metadataFamily?.checked).toBe(true)
|
||||||
|
expect(await brain.getNounCount()).toBe(0)
|
||||||
|
})
|
||||||
|
|
||||||
|
it('two consecutive online rebuilds both leave the index correct (idempotent)', async () => {
|
||||||
|
const { brain } = await openBrain()
|
||||||
|
const a = await brain.add({ data: 'a', type: NounType.Person, metadata: { status: 'active' } })
|
||||||
|
await brain.add({ data: 'b', type: NounType.Person, metadata: { status: 'inactive' } })
|
||||||
|
await brain.flush()
|
||||||
|
|
||||||
|
await brain.repairIndex({ rebuild: ['metadata'] })
|
||||||
|
const first = await brain.find({ where: { status: 'active' } })
|
||||||
|
expect(first.map((r) => r.id)).toEqual([a])
|
||||||
|
|
||||||
|
await brain.repairIndex({ rebuild: ['metadata'] })
|
||||||
|
const second = await brain.find({ where: { status: 'active' } })
|
||||||
|
expect(second.map((r) => r.id)).toEqual([a])
|
||||||
|
})
|
||||||
|
})
|
||||||
190
tests/integration/verb-metadata-rows.test.ts
Normal file
190
tests/integration/verb-metadata-rows.test.ts
Normal file
|
|
@ -0,0 +1,190 @@
|
||||||
|
/**
|
||||||
|
* @module tests/integration/verb-metadata-rows
|
||||||
|
* @description THE LIVE VERB PATH pins. Before this train, verb rows entered
|
||||||
|
* the metadata index ONLY via `MetadataIndexManager.rebuild()`'s canonical
|
||||||
|
* walk — every relate()/unrelate()/updateRelation() call, and every
|
||||||
|
* remove()-cascaded relationship, left the metadata index blind to verb
|
||||||
|
* writes until the next rebuild. This file pins that `relate()`,
|
||||||
|
* `unrelate()`, `updateRelation()`, `remove()`'s cascade, and their
|
||||||
|
* `transact()` mirrors now post/retract the SAME verb rows a rebuild would
|
||||||
|
* derive from canonical (ADR-007 A4: one mechanism for add/update, live and
|
||||||
|
* rebuilt).
|
||||||
|
*/
|
||||||
|
process.env.BRAINY_DETERMINISTIC_EMBEDDINGS = 'true'
|
||||||
|
|
||||||
|
import { describe, it, expect, beforeEach, afterEach } from 'vitest'
|
||||||
|
import { Brainy } from '../../src/brainy.js'
|
||||||
|
import { NounType, VerbType } from '../../src/types/graphTypes.js'
|
||||||
|
import type { MetadataIndexManager } from '../../src/utils/metadataIndex.js'
|
||||||
|
|
||||||
|
/** The JS metadata-index manager backing a memory-storage brain in these
|
||||||
|
* tests (feature-detected in production code via `instanceof
|
||||||
|
* MetadataIndexManager`; a narrow test-only reach-in here, matching the
|
||||||
|
* existing idiom in tests/integration/find-where-zero.test.ts and
|
||||||
|
* tests/integration/level-field-shadow.test.ts). */
|
||||||
|
function metadataIndexOf(brain: Brainy<any>): MetadataIndexManager {
|
||||||
|
return (brain as unknown as { metadataIndex: MetadataIndexManager }).metadataIndex
|
||||||
|
}
|
||||||
|
|
||||||
|
describe('verb metadata rows — the live path matches the rebuild walk', () => {
|
||||||
|
let brain: Brainy<any>
|
||||||
|
|
||||||
|
beforeEach(async () => {
|
||||||
|
brain = new Brainy({ requireSubtype: false, storage: { type: 'memory' }, silent: true })
|
||||||
|
await brain.init()
|
||||||
|
})
|
||||||
|
|
||||||
|
afterEach(async () => {
|
||||||
|
await brain.close()
|
||||||
|
})
|
||||||
|
|
||||||
|
async function addPerson(label: string): Promise<string> {
|
||||||
|
return brain.add({
|
||||||
|
data: `person ${label}`,
|
||||||
|
type: NounType.Person,
|
||||||
|
metadata: { label }
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
it('(a) relate() posts a metadata-index-backed verb row a query can find', async () => {
|
||||||
|
const a = await addPerson('a')
|
||||||
|
const b = await addPerson('b')
|
||||||
|
const relId = await brain.relate({
|
||||||
|
from: a, to: b, type: VerbType.WorksWith, metadata: { role: 'lead' }
|
||||||
|
})
|
||||||
|
|
||||||
|
// Read it back the SAME way a rebuild-sourced row is queried — the
|
||||||
|
// manager's own posting lookup, keyed on the custom field the caller wrote.
|
||||||
|
const index = metadataIndexOf(brain)
|
||||||
|
expect(await index.getIds('role', 'lead')).toEqual([relId])
|
||||||
|
})
|
||||||
|
|
||||||
|
it('(b) unrelate() retracts the row', async () => {
|
||||||
|
const a = await addPerson('a')
|
||||||
|
const b = await addPerson('b')
|
||||||
|
const relId = await brain.relate({
|
||||||
|
from: a, to: b, type: VerbType.WorksWith, metadata: { role: 'lead' }
|
||||||
|
})
|
||||||
|
|
||||||
|
const index = metadataIndexOf(brain)
|
||||||
|
expect(await index.getIds('role', 'lead')).toEqual([relId])
|
||||||
|
|
||||||
|
// Flush BEFORE retracting the field's only occurrence: this durably
|
||||||
|
// persists the 'role' column (a segment on disk/in the store), so the
|
||||||
|
// post-retraction query below reads "this field exists, zero live
|
||||||
|
// postings" (→ []) rather than "this field has never been written"
|
||||||
|
// (→ FIELD_NOT_INDEXED) — an orthogonal column-store characteristic
|
||||||
|
// (an unflushed field with its last live posting removed reverts to
|
||||||
|
// unknown), not a D2 behavior.
|
||||||
|
await brain.flush()
|
||||||
|
|
||||||
|
await brain.unrelate(relId)
|
||||||
|
|
||||||
|
expect(await index.getIds('role', 'lead')).toEqual([])
|
||||||
|
})
|
||||||
|
|
||||||
|
it('(c) updateRelation({ metadata }) leaves exactly the new values', async () => {
|
||||||
|
const a = await addPerson('a')
|
||||||
|
const b = await addPerson('b')
|
||||||
|
const relId = await brain.relate({
|
||||||
|
from: a, to: b, type: VerbType.WorksWith, metadata: { role: 'lead', team: 'core' }
|
||||||
|
})
|
||||||
|
|
||||||
|
const index = metadataIndexOf(brain)
|
||||||
|
expect(await index.getIds('role', 'lead')).toEqual([relId])
|
||||||
|
|
||||||
|
// Flush first — see (b)'s note: 'role'/'team' must be durably known
|
||||||
|
// fields before their only value is retracted, or the post-update
|
||||||
|
// "gone" checks below throw FIELD_NOT_INDEXED instead of returning [].
|
||||||
|
await brain.flush()
|
||||||
|
|
||||||
|
await brain.updateRelation({ id: relId, metadata: { role: 'reviewer' }, merge: false })
|
||||||
|
|
||||||
|
// Stale values gone (the old shape AND the merge:false-dropped field)…
|
||||||
|
expect(await index.getIds('role', 'lead')).toEqual([])
|
||||||
|
expect(await index.getIds('team', 'core')).toEqual([])
|
||||||
|
// …only the new value serves.
|
||||||
|
expect(await index.getIds('role', 'reviewer')).toEqual([relId])
|
||||||
|
})
|
||||||
|
|
||||||
|
it("(d) remove(entity) cascade retracts every incident relation's metadata row", async () => {
|
||||||
|
const a = await addPerson('a')
|
||||||
|
const b = await addPerson('b')
|
||||||
|
const c = await addPerson('c')
|
||||||
|
const rel1 = await brain.relate({
|
||||||
|
from: a, to: b, type: VerbType.WorksWith, metadata: { tag: 'cascade-test' }
|
||||||
|
})
|
||||||
|
const rel2 = await brain.relate({
|
||||||
|
from: c, to: a, type: VerbType.WorksWith, metadata: { tag: 'cascade-test' }
|
||||||
|
})
|
||||||
|
|
||||||
|
const index = metadataIndexOf(brain)
|
||||||
|
expect((await index.getIds('tag', 'cascade-test')).sort()).toEqual([rel1, rel2].sort())
|
||||||
|
|
||||||
|
// Flush first — see (b)'s note.
|
||||||
|
await brain.flush()
|
||||||
|
|
||||||
|
await brain.remove(a) // a is source of rel1, target of rel2 — both cascade
|
||||||
|
|
||||||
|
expect(await index.getIds('tag', 'cascade-test')).toEqual([])
|
||||||
|
})
|
||||||
|
|
||||||
|
it('(e) a rebuild() reproduces exactly the verb-row population the live path built', async () => {
|
||||||
|
const a = await addPerson('a')
|
||||||
|
const b = await addPerson('b')
|
||||||
|
const c = await addPerson('c')
|
||||||
|
await brain.relate({ from: a, to: b, type: VerbType.WorksWith, metadata: { tag: 'parity', label: 'ab' } })
|
||||||
|
await brain.relate({ from: b, to: c, type: VerbType.RelatedTo, metadata: { tag: 'parity', label: 'bc' } })
|
||||||
|
const relId3 = await brain.relate({
|
||||||
|
from: c, to: a, type: VerbType.WorksWith, metadata: { tag: 'parity', label: 'ca' }
|
||||||
|
})
|
||||||
|
await brain.unrelate(relId3) // exercise retraction too — the rebuild must NOT resurrect it
|
||||||
|
|
||||||
|
const index = metadataIndexOf(brain)
|
||||||
|
const beforeIds = (await index.getIds('tag', 'parity')).slice().sort()
|
||||||
|
expect(beforeIds.length).toBe(2)
|
||||||
|
const beforeAb = await index.getIds('label', 'ab')
|
||||||
|
const beforeBc = await index.getIds('label', 'bc')
|
||||||
|
|
||||||
|
await index.rebuild()
|
||||||
|
|
||||||
|
const afterIds = (await index.getIds('tag', 'parity')).slice().sort()
|
||||||
|
expect(afterIds).toEqual(beforeIds)
|
||||||
|
expect(await index.getIds('label', 'ab')).toEqual(beforeAb)
|
||||||
|
expect(await index.getIds('label', 'bc')).toEqual(beforeBc)
|
||||||
|
expect(await index.getIds('label', 'ca')).toEqual([]) // the unrelated edge stays gone
|
||||||
|
})
|
||||||
|
|
||||||
|
it('(f) transact() relate/unrelate posts/retracts the same metadata-index rows as single-op', async () => {
|
||||||
|
const a = await addPerson('a')
|
||||||
|
const b = await addPerson('b')
|
||||||
|
const c = await addPerson('c')
|
||||||
|
const d = await addPerson('d')
|
||||||
|
|
||||||
|
// Single-op baseline.
|
||||||
|
const singleOpId = await brain.relate({
|
||||||
|
from: a, to: b, type: VerbType.WorksWith, metadata: { tag: 'parity-f' }
|
||||||
|
})
|
||||||
|
|
||||||
|
// transact() mirror.
|
||||||
|
const relateDb = await brain.transact([
|
||||||
|
{ op: 'relate', from: c, to: d, type: VerbType.WorksWith, metadata: { tag: 'parity-f' } }
|
||||||
|
])
|
||||||
|
const transactId = relateDb.receipt!.ids[0]
|
||||||
|
await relateDb.release()
|
||||||
|
|
||||||
|
const index = metadataIndexOf(brain)
|
||||||
|
expect((await index.getIds('tag', 'parity-f')).sort()).toEqual([singleOpId, transactId].sort())
|
||||||
|
|
||||||
|
// Flush first — see (b)'s note: 'tag' must be durably known before its
|
||||||
|
// last live posting is retracted below.
|
||||||
|
await brain.flush()
|
||||||
|
|
||||||
|
// Retract both ways — single-op unrelate() and transact() unrelate.
|
||||||
|
await brain.unrelate(singleOpId)
|
||||||
|
const unrelateDb = await brain.transact([{ op: 'unrelate', id: transactId }])
|
||||||
|
await unrelateDb.release()
|
||||||
|
|
||||||
|
expect(await index.getIds('tag', 'parity-f')).toEqual([])
|
||||||
|
})
|
||||||
|
})
|
||||||
|
|
@ -286,32 +286,7 @@ describe.sequential('lifecycle — the working store', () => {
|
||||||
300000
|
300000
|
||||||
)
|
)
|
||||||
|
|
||||||
/**
|
it(
|
||||||
* Ch4 CRASH is a LIVE ENGINE FINDING, not a defect in this lane (see
|
|
||||||
* README.md and the project report this lane's build produced): after a
|
|
||||||
* crash (writes acked at commit but never flushed, the process abandoned
|
|
||||||
* exactly as `abandonAsCrashed` models, then reopened), canonical storage
|
|
||||||
* (`get()`), the vector index, and `getNounCount()`/`getCanonicalCounts()`
|
|
||||||
* all correctly recover every acked write — but the METADATA INDEX behind
|
|
||||||
* `find({ where })` recovers NONE of the crash-window's acked writes
|
|
||||||
* (neither new adds nor metadata updates to pre-existing entities), even
|
|
||||||
* though `getIndexStatus()` reports `projections.metadata.synchronous:
|
|
||||||
* true`. `repairIndex()` cannot close the gap either: its own report names
|
|
||||||
* `provider:metadata` as `checked: false, skipped: "no
|
|
||||||
* validateInvariants/rebuild contract"`. The assertion below states the
|
|
||||||
* TRUE contract (find() must agree with get()) and is expected to fail
|
|
||||||
* against the current engine — it must never be loosened to paper over
|
|
||||||
* this. Ch5/Ch6 are written in full below it and will start running the
|
|
||||||
* moment this gap is closed; they are not dead code, they are blocked code.
|
|
||||||
*/
|
|
||||||
// RELEASE-BLOCKING FINDING (the kill-matrix convention: assert the CONTRACT,
|
|
||||||
// mark `.fails`, never weaken): after a crash + adopt reopen, the JS metadata
|
|
||||||
// index computes its watermark verdict but nothing consumes 'catchup'
|
|
||||||
// (metadataIndex.ts loadWatermarkVerdict) — find() serves the pre-crash
|
|
||||||
// index while get()/counts recover. The catchup wiring is the cure; when it
|
|
||||||
// lands this `.fails` marker MUST be removed (vitest will force it: a
|
|
||||||
// passing `.fails` test is itself a failure).
|
|
||||||
it.fails(
|
|
||||||
'Ch4 CRASH -> Ch5 REPAIR -> Ch6 SECOND LIFE: continues the Ch3 store',
|
'Ch4 CRASH -> Ch5 REPAIR -> Ch6 SECOND LIFE: continues the Ch3 store',
|
||||||
async () => {
|
async () => {
|
||||||
try {
|
try {
|
||||||
|
|
|
||||||
|
|
@ -11,8 +11,11 @@
|
||||||
* Same rule, same verdict names as the shipped aggregation machinery
|
* Same rule, same verdict names as the shipped aggregation machinery
|
||||||
* (AggregationIndex.stateAdoptionVerdict).
|
* (AggregationIndex.stateAdoptionVerdict).
|
||||||
*
|
*
|
||||||
* The verdict is COMPUTED AND EXPOSED only — these pins assert no rebuild
|
* The verdict is computed at init and consumed via
|
||||||
* trigger changed; acting on 'catchup' lands with the coordinator's wiring.
|
* {@link MetadataIndexManager.applyWatermarkCatchup} — the coordinator
|
||||||
|
* (`Brainy.performInit`) calls it right after `init()`, with an open fact
|
||||||
|
* scan when the verdict is `'catchup'`. This file pins both halves: the
|
||||||
|
* verdict computation (above) and the fold/no-op/demotion behavior below.
|
||||||
*/
|
*/
|
||||||
import { describe, it, expect, vi, afterEach } from 'vitest'
|
import { describe, it, expect, vi, afterEach } from 'vitest'
|
||||||
import { v4 as uuidv4 } from 'uuid'
|
import { v4 as uuidv4 } from 'uuid'
|
||||||
|
|
@ -22,6 +25,46 @@ import {
|
||||||
} from '../../../src/utils/metadataIndex.js'
|
} from '../../../src/utils/metadataIndex.js'
|
||||||
import { MemoryStorage } from '../../../src/storage/adapters/memoryStorage.js'
|
import { MemoryStorage } from '../../../src/storage/adapters/memoryStorage.js'
|
||||||
import { prodLog } from '../../../src/utils/logger.js'
|
import { prodLog } from '../../../src/utils/logger.js'
|
||||||
|
import type { CommitFact, FactScanBatch, FactScanHandle } from '../../../src/db/factLog.js'
|
||||||
|
|
||||||
|
/** A fact scan handle over an in-memory list of facts — batches them one
|
||||||
|
* fact at a time (batch size is irrelevant to the fold, which reads
|
||||||
|
* `batch.facts` only). */
|
||||||
|
function fakeScan(facts: CommitFact[]): FactScanHandle {
|
||||||
|
return {
|
||||||
|
headGeneration: facts.length > 0 ? facts[facts.length - 1].generation : 0,
|
||||||
|
segmentCount: 1,
|
||||||
|
approxFactCount: facts.length,
|
||||||
|
async *batches(): AsyncGenerator<FactScanBatch> {
|
||||||
|
for (const fact of facts) {
|
||||||
|
yield {
|
||||||
|
facts: [fact],
|
||||||
|
firstGeneration: fact.generation,
|
||||||
|
lastGeneration: fact.generation,
|
||||||
|
factCount: 1,
|
||||||
|
byteSize: 0,
|
||||||
|
segmentId: 'fake'
|
||||||
|
}
|
||||||
|
}
|
||||||
|
},
|
||||||
|
summary: () => ({ factsYielded: facts.length, segmentsRead: 1 })
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/** One noun after-image fact — the flat-record shape (no nested `metadata`
|
||||||
|
* key), matching this file's existing `writeArtifact` convention. */
|
||||||
|
function nounAdd(generation: number, id: string, metadata: Record<string, unknown>): CommitFact {
|
||||||
|
return {
|
||||||
|
generation,
|
||||||
|
timestamp: Date.now(),
|
||||||
|
ops: [{ kind: 'noun', id, record: { metadata, vector: null } }]
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/** One noun tombstone fact. */
|
||||||
|
function nounDelete(generation: number, id: string): CommitFact {
|
||||||
|
return { generation, timestamp: Date.now(), ops: [{ kind: 'noun', id, record: null }] }
|
||||||
|
}
|
||||||
|
|
||||||
/** Fresh storage with a controllable committed generation. */
|
/** Fresh storage with a controllable committed generation. */
|
||||||
async function makeStorage(committed: number | null): Promise<MemoryStorage> {
|
async function makeStorage(committed: number | null): Promise<MemoryStorage> {
|
||||||
|
|
@ -169,3 +212,106 @@ describe('metadata index — watermark stamp + three-way load verdict', () => {
|
||||||
expect(await storage.getMetadata(METADATA_INDEX_STAMP_KEY)).toBeNull()
|
expect(await storage.getMetadata(METADATA_INDEX_STAMP_KEY)).toBeNull()
|
||||||
})
|
})
|
||||||
})
|
})
|
||||||
|
|
||||||
|
describe('metadata index — applyWatermarkCatchup (the coordinator door)', () => {
|
||||||
|
it("an 'adopt' verdict performs zero index writes", async () => {
|
||||||
|
const storage = await makeStorage(5)
|
||||||
|
await writeArtifact(storage, 5)
|
||||||
|
const index = await reopen(storage)
|
||||||
|
expect(index.watermarkVerdict()).toBe('adopt')
|
||||||
|
|
||||||
|
const addSpy = vi.spyOn(index, 'addToIndex')
|
||||||
|
const removeSpy = vi.spyOn(index, 'removeFromIndex')
|
||||||
|
|
||||||
|
const result = await index.applyWatermarkCatchup(null)
|
||||||
|
|
||||||
|
expect(result).toEqual({ action: 'noop' })
|
||||||
|
expect(addSpy).not.toHaveBeenCalled()
|
||||||
|
expect(removeSpy).not.toHaveBeenCalled()
|
||||||
|
})
|
||||||
|
|
||||||
|
it('a catchup window folding an add, an update (same id twice), and a delete → the index serves exactly the final state', async () => {
|
||||||
|
const storage = await makeStorage(5)
|
||||||
|
|
||||||
|
// Session 1: two pre-existing entities, stamped at generation 5.
|
||||||
|
const survivorId = uuidv4()
|
||||||
|
const deletedId = uuidv4()
|
||||||
|
{
|
||||||
|
const index = new MetadataIndexManager(storage)
|
||||||
|
await index.init()
|
||||||
|
await index.addToIndex(survivorId, { status: 'active' })
|
||||||
|
await index.addToIndex(deletedId, { status: 'active' })
|
||||||
|
index.stampWatermark(5)
|
||||||
|
await index.flush()
|
||||||
|
}
|
||||||
|
|
||||||
|
// The store advanced to generation 8 without another metadata flush —
|
||||||
|
// the exact shape a crash-then-adopt-reopen leaves behind.
|
||||||
|
setCommitted(storage, 8)
|
||||||
|
|
||||||
|
const index = await reopen(storage)
|
||||||
|
expect(index.watermarkVerdict()).toBe('catchup')
|
||||||
|
expect(index.watermarkGap()).toEqual({ from: 5, to: 8 })
|
||||||
|
|
||||||
|
const addedId = uuidv4()
|
||||||
|
const scan = fakeScan([
|
||||||
|
nounAdd(6, addedId, { status: 'new' }), // add
|
||||||
|
nounAdd(7, addedId, { status: 'updated' }), // update — same id twice
|
||||||
|
nounDelete(8, deletedId) // delete
|
||||||
|
])
|
||||||
|
|
||||||
|
const result = await index.applyWatermarkCatchup(scan)
|
||||||
|
|
||||||
|
expect(result.action).toBe('caught-up')
|
||||||
|
expect(result.window).toEqual({ from: 5, to: 8 })
|
||||||
|
expect(result.factsApplied).toBe(3)
|
||||||
|
expect(result.nounsApplied).toBe(3)
|
||||||
|
expect(result.verbsApplied).toBe(0)
|
||||||
|
|
||||||
|
// Final state: the added/updated id serves ONLY its final value...
|
||||||
|
expect(await index.getIds('status', 'updated')).toEqual([addedId])
|
||||||
|
expect(await index.getIds('status', 'new')).toEqual([]) // stale value gone
|
||||||
|
// ...the deleted id is gone...
|
||||||
|
expect(await index.getIds('status', 'active')).toEqual([survivorId])
|
||||||
|
// ...and the untouched survivor is unaffected.
|
||||||
|
expect(await index.getIds('status', 'active')).toContain(survivorId)
|
||||||
|
|
||||||
|
// The window is certified: watermark stamped at `to`, and a fresh
|
||||||
|
// reopen now verdicts 'adopt'.
|
||||||
|
expect(index.watermark()).toBe(8)
|
||||||
|
const reopened = await reopen(storage)
|
||||||
|
expect(reopened.watermarkVerdict()).toBe('adopt')
|
||||||
|
})
|
||||||
|
|
||||||
|
it("a 'rescan' verdict runs the existing rebuild path instead of folding", async () => {
|
||||||
|
const storage = await makeStorage(9)
|
||||||
|
await writeArtifact(storage, 9)
|
||||||
|
setCommitted(storage, 4) // a truncated log pulled the watermark back — stamp ABOVE committed → rescan
|
||||||
|
|
||||||
|
const index = await reopen(storage)
|
||||||
|
expect(index.watermarkVerdict()).toBe('rescan')
|
||||||
|
|
||||||
|
const rebuildSpy = vi.spyOn(index, 'rebuild')
|
||||||
|
const result = await index.applyWatermarkCatchup(null)
|
||||||
|
|
||||||
|
expect(result.action).toBe('rescan')
|
||||||
|
expect(result.reason).toBeTruthy()
|
||||||
|
expect(rebuildSpy).toHaveBeenCalledTimes(1)
|
||||||
|
})
|
||||||
|
|
||||||
|
it("a 'catchup' verdict with no fact log available demotes to rebuild, narrated", async () => {
|
||||||
|
const storage = await makeStorage(5)
|
||||||
|
await writeArtifact(storage, 5)
|
||||||
|
setCommitted(storage, 8)
|
||||||
|
|
||||||
|
const index = await reopen(storage)
|
||||||
|
expect(index.watermarkVerdict()).toBe('catchup')
|
||||||
|
|
||||||
|
const rebuildSpy = vi.spyOn(index, 'rebuild')
|
||||||
|
const result = await index.applyWatermarkCatchup(null) // no scan — no fact log
|
||||||
|
|
||||||
|
expect(result.action).toBe('rescan')
|
||||||
|
expect(result.reason).toContain('no fact log')
|
||||||
|
expect(rebuildSpy).toHaveBeenCalledTimes(1)
|
||||||
|
})
|
||||||
|
})
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue