feat(embedding): MT5 — deferred embedding with durable markers; write acks never wait on a neural net
A3 of the service-class pair (BRAINY-PROD-LATENCY-TRIAD): a VFS file write ran the embedder synchronously while the caller waited — 5.6s p50 / 21.4s p95 per small file on a production deployment, the dominant stage of every capture write. - add()/update() gain deferEmbedding: the write acks at durability (data + metadata persisted, a DURABLE pending marker under _system/pending_embeds/<id> written BEFORE the commit — orphan-safe direction); the single-flight background worker embeds the CURRENT data and swaps the vector in ATOMICALLY (ReplaceInVectorIndex — the row is never absent from search; a deferred UPDATE keeps serving the OLD vector, stale-beats-absent per the flicker law). Typed refusals: defer+vector, defer-without-data. - CRASH-SAFE: markers are recovered at open by a BOUNDED prefix listing (never a store walk) and the worker resumes in the background — a crash can delay a vector, never lose one. A wedged embedder trips a LOUD 60s hang guard and the worker moves on (marker retained for retry). - The honest gauges: getIndexStatus().pendingEmbeds + pendingEmbedCount(); awaitPendingEmbeds() is the eventual-vector-index BARRIER for callers and tests that need searchability before proceeding. - VFS adopts it everywhere a write path could wait on the embedder: writeFile (both branches) and directory creation. Pinned in the strongest form: writeFile resolves while the embedder HANGS FOREVER. Pins: deferred-embedding 5/5 (ack law · stale-beats-absent · crash recovery across sessions · VFS hung-embedder ack · typed refusals). Gates: unit 1928/1928 · integration 765 · conformance 27/27.
This commit is contained in:
parent
ebe06cdf33
commit
287384cf1e
5 changed files with 477 additions and 22 deletions
264
src/brainy.ts
264
src/brainy.ts
|
|
@ -695,6 +695,12 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
private _persistLastFlushAt = Date.now()
|
||||
private _persistIdleTimer: ReturnType<typeof setTimeout> | null = null
|
||||
private _persistBackgroundFlight: Promise<void> | null = null
|
||||
|
||||
// DEFERRED EMBEDDING (MT5): durable pending markers under
|
||||
// _system/pending_embeds/<id>, mirrored in-memory, drained by ONE
|
||||
// background worker. A crash can delay a vector, never lose one.
|
||||
private _pendingEmbedIds = new Set<string>()
|
||||
private _embedWorkerFlight: Promise<void> | null = null
|
||||
// A failed walk latches its error: retries within the cooldown rethrow it
|
||||
// instantly instead of re-walking, so a tight caller-side retry loop costs
|
||||
// one loud error per query, never a full store walk per query.
|
||||
|
|
@ -1418,6 +1424,33 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
this._generationStampingActive = true
|
||||
}
|
||||
|
||||
// MT5 crash recovery: reload the durable pending-embed markers (a
|
||||
// BOUNDED prefix listing — never a store walk) and resume the worker
|
||||
// in the background. A crash between a deferred write's ack and its
|
||||
// background embed DELAYED a vector; this is where it lands.
|
||||
if (!this.isReadOnly) {
|
||||
try {
|
||||
const markerPaths = await this.storage.listRawObjects(Brainy.PENDING_EMBED_PREFIX)
|
||||
for (const path of markerPaths) {
|
||||
const id = path.slice(path.lastIndexOf('/') + 1)
|
||||
if (id) this._pendingEmbedIds.add(id)
|
||||
}
|
||||
if (this._pendingEmbedIds.size > 0) {
|
||||
prodLog.info(
|
||||
`[Brainy] ${this._pendingEmbedIds.size} deferred embed(s) pending from a previous ` +
|
||||
`session — resuming in the background`
|
||||
)
|
||||
const t = setTimeout(() => this.kickEmbedWorker(), 0)
|
||||
;(t as { unref?: () => void }).unref?.()
|
||||
}
|
||||
} catch (err) {
|
||||
prodLog.warn(
|
||||
`[Brainy] pending-embed recovery listing failed: ${(err as Error).message} — ` +
|
||||
`markers remain durable; recovery retries next open`
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// Eager embedding initialization.
|
||||
//
|
||||
// Adaptive default (8.0): the WASM embedding engine eagerly initializes
|
||||
|
|
@ -1840,6 +1873,133 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
* @param run - The single-op's existing operation batch builder (the
|
||||
* `tx => {…}` body previously passed straight to `executeTransaction`).
|
||||
*/
|
||||
/** Storage-root-relative prefix of the durable pending-embed markers. */
|
||||
private static readonly PENDING_EMBED_PREFIX = '_system/pending_embeds/'
|
||||
|
||||
/**
|
||||
* @description Persist the durable pending-embed marker (MT5) and mirror
|
||||
* it in memory. Written BEFORE the write it belongs to commits — an
|
||||
* orphaned marker (commit failed) is harmless and reaped by the worker;
|
||||
* the reverse ordering could lose an embed silently on a crash.
|
||||
*/
|
||||
private async enqueuePendingEmbed(id: string): Promise<void> {
|
||||
this._pendingEmbedIds.add(id)
|
||||
await this.storage.writeRawObject(`${Brainy.PENDING_EMBED_PREFIX}${id}`, {
|
||||
id,
|
||||
enqueuedAt: Date.now()
|
||||
})
|
||||
}
|
||||
|
||||
/** Remove a pending-embed marker (memory + durable), tolerating races. */
|
||||
private async clearPendingEmbed(id: string): Promise<void> {
|
||||
this._pendingEmbedIds.delete(id)
|
||||
await this.storage
|
||||
.deleteRawObject(`${Brainy.PENDING_EMBED_PREFIX}${id}`)
|
||||
.catch(() => {})
|
||||
}
|
||||
|
||||
/**
|
||||
* @description Start (or skip into) the ONE deferred-embedding worker.
|
||||
* Never awaited by write paths; failures are LOUD and markers survive for
|
||||
* the next kick (next deferred write, or the next open's recovery).
|
||||
*/
|
||||
private kickEmbedWorker(): void {
|
||||
if (this._embedWorkerFlight || this._pendingEmbedIds.size === 0 || this.isReadOnly) return
|
||||
this._embedWorkerFlight = this.runEmbedWorker()
|
||||
.catch((err) => {
|
||||
prodLog.error(
|
||||
`[Brainy] deferred-embed worker failed: ${(err as Error).message} — ` +
|
||||
`markers retained; retries at the next deferred write or open`
|
||||
)
|
||||
})
|
||||
.finally(() => {
|
||||
this._embedWorkerFlight = null
|
||||
if (this._pendingEmbedIds.size > 0) {
|
||||
// New arrivals during the run: schedule (never recurse) the next pass.
|
||||
const t = setTimeout(() => this.kickEmbedWorker(), 0)
|
||||
;(t as { unref?: () => void }).unref?.()
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* @description Drain the pending-embed set: embed each row's CURRENT data
|
||||
* (a row updated again before its turn embeds the latest content — the
|
||||
* marker set is idempotent per id) and swap the vector in ATOMICALLY
|
||||
* (ReplaceInVectorIndex → the in-place update; the row is never absent
|
||||
* from search). Orphans (row deleted, or no data) reap their markers.
|
||||
*/
|
||||
private async runEmbedWorker(): Promise<void> {
|
||||
const batch = Array.from(this._pendingEmbedIds)
|
||||
for (const id of batch) {
|
||||
try {
|
||||
const entity = await this.get(id, { includeVectors: true })
|
||||
if (!entity || entity.data === undefined || entity.data === null) {
|
||||
await this.clearPendingEmbed(id)
|
||||
continue
|
||||
}
|
||||
// Hang guard: a wedged embedder must not block every later pending
|
||||
// embed forever — time out LOUDLY, keep the marker, move on. (A
|
||||
// failure is retryable; an unbounded silent wait is the outlawed
|
||||
// shape.)
|
||||
const newVector = await Promise.race([
|
||||
this.embed(entity.data),
|
||||
new Promise<never>((_, reject) => {
|
||||
const t = setTimeout(
|
||||
() => reject(new Error('deferred embed timed out after 60s')),
|
||||
60_000
|
||||
)
|
||||
;(t as { unref?: () => void }).unref?.()
|
||||
})
|
||||
])
|
||||
if (!this.dimensions) {
|
||||
this.dimensions = newVector.length
|
||||
} else if (newVector.length !== this.dimensions) {
|
||||
throw new Error(
|
||||
`deferred embed produced ${newVector.length} dimensions, store expects ${this.dimensions}`
|
||||
)
|
||||
}
|
||||
const oldVector = (entity.vector as number[] | undefined) ?? []
|
||||
await this.persistSingleOp({ nouns: [id] }, async (tx) => {
|
||||
tx.addOperation(
|
||||
new SaveNounOperation(this.storage, {
|
||||
id,
|
||||
vector: newVector,
|
||||
connections: new Map(),
|
||||
level: 0
|
||||
})
|
||||
)
|
||||
tx.addOperation(
|
||||
new ReplaceInVectorIndexOperation(this.index, id, oldVector, newVector)
|
||||
)
|
||||
})
|
||||
await this.clearPendingEmbed(id)
|
||||
} catch (err) {
|
||||
prodLog.warn(
|
||||
`[Brainy] deferred embed for ${id} failed: ${(err as Error).message} — marker retained for retry`
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @description The deferred-embedding BARRIER: resolves when every pending
|
||||
* embed has landed (vector searchable) or been reaped. The eventual-
|
||||
* vector-index contract's awaitable edge — tests and "must be searchable
|
||||
* before I proceed" callers use this; nothing else ever needs to wait.
|
||||
*/
|
||||
public async awaitPendingEmbeds(): Promise<void> {
|
||||
while (this._pendingEmbedIds.size > 0 || this._embedWorkerFlight) {
|
||||
this.kickEmbedWorker()
|
||||
await (this._embedWorkerFlight ?? Promise.resolve())
|
||||
}
|
||||
}
|
||||
|
||||
/** The deferred-embedding backlog size (also on getIndexStatus().pendingEmbeds). */
|
||||
public pendingEmbedCount(): number {
|
||||
return this._pendingEmbedIds.size
|
||||
}
|
||||
|
||||
/**
|
||||
* @description The write-side persistence trigger (policy `'auto'`): count
|
||||
* the committed write, kick a single-flight BACKGROUND flush when the
|
||||
|
|
@ -2166,15 +2326,26 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
}
|
||||
|
||||
// Get or compute vector
|
||||
const vector = params.vector || (await this.embed(params.data))
|
||||
// MT5 deferred embedding: ack at durability with a stub vector and a
|
||||
// DURABLE pending marker (written BEFORE the commit — an orphaned marker
|
||||
// from a failed commit is harmless and reaped by the worker; a
|
||||
// marker-less committed row would be a silently missing vector, which is
|
||||
// the disallowed direction). The background worker embeds + inserts.
|
||||
const deferringEmbed = params.deferEmbedding === true && !params.vector
|
||||
const vector = deferringEmbed
|
||||
? []
|
||||
: params.vector || (await this.embed(params.data))
|
||||
|
||||
// Ensure dimensions are set
|
||||
if (!this.dimensions) {
|
||||
this.dimensions = vector.length
|
||||
} else if (vector.length !== this.dimensions) {
|
||||
throw new Error(
|
||||
`Vector dimension mismatch: expected ${this.dimensions}, got ${vector.length}`
|
||||
)
|
||||
// Ensure dimensions are set (a deferred-embed stub carries no dimension
|
||||
// information — the worker's real vector goes through the same guard).
|
||||
if (!deferringEmbed) {
|
||||
if (!this.dimensions) {
|
||||
this.dimensions = vector.length
|
||||
} else if (vector.length !== this.dimensions) {
|
||||
throw new Error(
|
||||
`Vector dimension mismatch: expected ${this.dimensions}, got ${vector.length}`
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// Prepare metadata for storage: a v2 nested-bag record — engine fields
|
||||
|
|
@ -2254,6 +2425,12 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
}
|
||||
: undefined
|
||||
|
||||
// MT5: the durable marker lands BEFORE the commit (orphan-safe; the
|
||||
// reverse order could lose an embed silently on a crash).
|
||||
if (deferringEmbed) {
|
||||
await this.enqueuePendingEmbed(id)
|
||||
}
|
||||
|
||||
const runInsert: TransactionFunction<void> = async (tx) => {
|
||||
// Operation 1: Save metadata FIRST (TypeAwareStorage caching)
|
||||
// isNew=true: skip pre-read for rollback (entity doesn't exist yet)
|
||||
|
|
@ -2272,10 +2449,14 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
}, true)
|
||||
)
|
||||
|
||||
// Operation 3: Add to HNSW index (after entity saved)
|
||||
tx.addOperation(
|
||||
new AddToVectorIndexOperation(this.index, id, vector)
|
||||
)
|
||||
// Operation 3: Add to HNSW index (after entity saved). A deferred
|
||||
// embed has nothing to index yet — the worker's atomic update
|
||||
// inserts the real vector.
|
||||
if (!deferringEmbed) {
|
||||
tx.addOperation(
|
||||
new AddToVectorIndexOperation(this.index, id, vector)
|
||||
)
|
||||
}
|
||||
|
||||
// Operation 4: Add to metadata index
|
||||
tx.addOperation(
|
||||
|
|
@ -2343,6 +2524,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
this._aggregationIndex.onEntityAdded(id, entityForIndexing)
|
||||
}
|
||||
|
||||
if (deferringEmbed) this.kickEmbedWorker()
|
||||
return id
|
||||
}
|
||||
|
||||
|
|
@ -2828,6 +3010,11 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
// new `data`); otherwise new `data` re-embeds; otherwise the existing
|
||||
// vector is kept. Any vector change re-indexes HNSW below.
|
||||
let vector = existing.vector
|
||||
// MT5 deferred re-embedding: the OLD vector keeps serving semantic
|
||||
// search — stale-but-present, never absent (the flicker law) — until
|
||||
// the background worker embeds the new data and swaps it atomically.
|
||||
const deferringEmbed =
|
||||
params.deferEmbedding === true && Boolean(params.data) && !params.vector
|
||||
if (params.vector) {
|
||||
if (this.dimensions && params.vector.length !== this.dimensions) {
|
||||
throw new Error(
|
||||
|
|
@ -2835,10 +3022,14 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
)
|
||||
}
|
||||
vector = params.vector
|
||||
} else if (params.data) {
|
||||
} else if (params.data && !deferringEmbed) {
|
||||
vector = await this.embed(params.data)
|
||||
}
|
||||
const needsReindexing = Boolean(params.data || params.type || params.vector)
|
||||
// A deferred data change does NOT reindex now (the vector is unchanged;
|
||||
// the worker's atomic swap carries the real reindex later).
|
||||
const needsReindexing = Boolean(
|
||||
(params.data && !deferringEmbed) || params.type || params.vector
|
||||
)
|
||||
|
||||
// Always update the noun with new metadata
|
||||
const newMetadata = params.merge !== false
|
||||
|
|
@ -2925,6 +3116,11 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
updatedMetadata._rev = authoritativeRev + 1
|
||||
}
|
||||
|
||||
// MT5: durable marker BEFORE the commit (orphan-safe direction).
|
||||
if (deferringEmbed) {
|
||||
await this.enqueuePendingEmbed(params.id)
|
||||
}
|
||||
|
||||
// Execute atomically with transaction system, generation-stamped as one
|
||||
// immutable Model-B generation (before-image = the entity's prior state).
|
||||
await this.persistSingleOp({ nouns: [params.id] }, async (tx) => {
|
||||
|
|
@ -3026,6 +3222,8 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
existing as unknown as Record<string, unknown>
|
||||
)
|
||||
}
|
||||
|
||||
if (deferringEmbed) this.kickEmbedWorker()
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -9123,13 +9321,23 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
}
|
||||
}
|
||||
|
||||
const vector = params.vector || (await this.embed(params.data))
|
||||
if (!this.dimensions) {
|
||||
this.dimensions = vector.length
|
||||
} else if (vector.length !== this.dimensions) {
|
||||
throw new Error(
|
||||
`Vector dimension mismatch: expected ${this.dimensions}, got ${vector.length}`
|
||||
)
|
||||
// MT5 deferred embedding: ack at durability with a stub vector and a
|
||||
// DURABLE pending marker (written BEFORE the commit — an orphaned marker
|
||||
// from a failed commit is harmless and reaped by the worker; a
|
||||
// marker-less committed row would be a silently missing vector, which is
|
||||
// the disallowed direction). The background worker embeds + inserts.
|
||||
const deferringEmbed = params.deferEmbedding === true && !params.vector
|
||||
const vector = deferringEmbed
|
||||
? []
|
||||
: params.vector || (await this.embed(params.data))
|
||||
if (!deferringEmbed) {
|
||||
if (!this.dimensions) {
|
||||
this.dimensions = vector.length
|
||||
} else if (vector.length !== this.dimensions) {
|
||||
throw new Error(
|
||||
`Vector dimension mismatch: expected ${this.dimensions}, got ${vector.length}`
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// isNew controls the operation's rollback strategy: a custom id may
|
||||
|
|
@ -9192,10 +9400,18 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
}
|
||||
}
|
||||
|
||||
if (deferringEmbed) {
|
||||
// Durable marker BEFORE the batch commits (orphan-safe direction);
|
||||
// the worker kicks post-commit via the plan hook.
|
||||
await this.enqueuePendingEmbed(id)
|
||||
plan.postCommit.push(() => this.kickEmbedWorker())
|
||||
}
|
||||
plan.operations.push(
|
||||
new SaveNounMetadataOperation(this.storage, id, storageMetadata, isNew),
|
||||
new SaveNounOperation(this.storage, { id, vector, connections: new Map(), level: 0 }, isNew),
|
||||
new AddToVectorIndexOperation(this.index, id, vector),
|
||||
...(deferringEmbed
|
||||
? []
|
||||
: [new AddToVectorIndexOperation(this.index, id, vector)]),
|
||||
new AddToMetadataIndexOperation(this.metadataIndex, id, entityForIndexing)
|
||||
)
|
||||
plan.touchedNouns.push(id)
|
||||
|
|
@ -10672,6 +10888,8 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
async getIndexStatus(): Promise<{
|
||||
initialized: boolean
|
||||
lazyRebuildCompleted: boolean
|
||||
/** Deferred embeds not yet landed (MT5) — the eventual-vector-index backlog. */
|
||||
pendingEmbeds: number
|
||||
disableAutoRebuild: boolean
|
||||
/** `true` while a native provider runs the one-time 7.x → 8.0 rebuild LOCK.
|
||||
* A readiness probe should map this to HTTP 503 + Retry-After (transiently
|
||||
|
|
@ -10717,6 +10935,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
return {
|
||||
initialized: false,
|
||||
lazyRebuildCompleted: this.lazyRebuildCompleted,
|
||||
pendingEmbeds: this._pendingEmbedIds.size,
|
||||
disableAutoRebuild: this.config.disableAutoRebuild || false,
|
||||
migrating: false,
|
||||
rebuildFailed: this._indexRebuildFailed != null,
|
||||
|
|
@ -10759,6 +10978,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
|||
return {
|
||||
initialized: this.initialized,
|
||||
lazyRebuildCompleted: this.lazyRebuildCompleted,
|
||||
pendingEmbeds: this._pendingEmbedIds.size,
|
||||
disableAutoRebuild: this.config.disableAutoRebuild || false,
|
||||
// A non-fatal index-rebuild failure recorded at init(), or adopt-forward
|
||||
// degraded ids, are degraded states (queries may be incomplete) — surface
|
||||
|
|
|
|||
Reference in a new issue