feat(embedding): MT5 — deferred embedding with durable markers; write acks never wait on a neural net
Some checks failed
CI / Node 22 (push) Has been cancelled
CI / Node 24 (push) Has been cancelled
CI / Bun (latest) (push) Has been cancelled

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:
David Snelling 2026-08-05 16:26:43 -07:00
parent ebe06cdf33
commit 287384cf1e
5 changed files with 477 additions and 22 deletions

View file

@ -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,9 +2326,19 @@ 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
// 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) {
@ -2176,6 +2346,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
`Vector dimension mismatch: expected ${this.dimensions}, got ${vector.length}`
)
}
}
// Prepare metadata for storage: a v2 nested-bag record — engine fields
// top-level, the user's bag nested VERBATIM (any name, including engine
@ -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)
// 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,7 +9321,16 @@ export class Brainy<T = any> implements BrainyInterface<T> {
}
}
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))
if (!deferringEmbed) {
if (!this.dimensions) {
this.dimensions = vector.length
} else if (vector.length !== this.dimensions) {
@ -9131,6 +9338,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
`Vector dimension mismatch: expected ${this.dimensions}, got ${vector.length}`
)
}
}
// isNew controls the operation's rollback strategy: a custom id may
// collide with an existing entity (add() overwrite semantics), and a
@ -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

View file

@ -338,6 +338,20 @@ export interface AddParams<T = any> {
id?: string
/** Pre-computed embedding vector (skips auto-embedding when provided) */
vector?: Vector
/**
* DEFER THE EMBEDDING (MT5, the deferred-embedding worker): the write
* acknowledges at durability data + metadata persisted, a durable
* pending-embed marker written and the embedding + vector-index insert
* run on the engine's single-flight background worker. HONEST SEMANTICS:
* the row is findable by id/metadata/path IMMEDIATELY; vector/semantic
* search sees it when the background embed completes (eventual vector
* index `getIndexStatus().pendingEmbeds` counts the backlog, and
* `awaitPendingEmbeds()` is the barrier). CRASH-SAFE: markers persist
* before the ack and are recovered at the next open a crash can DELAY
* a vector, never lose one. Refused (typed) together with `vector`
* a supplied vector has nothing to defer.
*/
deferEmbedding?: boolean
/** Multi-tenancy service identifier */
service?: string
/** Type classification confidence (0-1) */
@ -379,6 +393,15 @@ export interface AddParams<T = any> {
export interface UpdateParams<T = any> {
id: string // Entity to update
data?: any // New content to re-embed
/**
* Defer the re-embedding of new `data` (see `AddParams.deferEmbedding`).
* The write acks at durability; the OLD vector keeps serving semantic
* search stale-but-present, never absent (the flicker law) until the
* background worker embeds the new content and swaps it in atomically.
* `data` reads return the NEW content immediately. Refused (typed) with
* an explicit `vector`.
*/
deferEmbedding?: boolean
type?: NounType // Change type
subtype?: string // Change subtype (set to '' or null-equivalent via dedicated unset is future work)
/**

View file

@ -540,6 +540,22 @@ function rejectForgedSystemKeys(metadata: Record<string, unknown> | undefined, s
export function validateAddParams(params: AddParams): void {
rejectForgedSystemKeys(params.metadata as Record<string, unknown> | undefined, 'add()')
// MT5 deferred embedding: an explicit vector has nothing to defer, and a
// deferral without data has nothing to embed — both are caller bugs that
// must refuse with the fix, never be silently reinterpreted.
if ((params as AddParams & { deferEmbedding?: boolean }).deferEmbedding === true) {
if (params.vector) {
throw new Error(
`add(): deferEmbedding cannot be combined with an explicit 'vector' — ` +
`the vector is already computed; drop one of the two.`
)
}
if (!params.data) {
throw new Error(
`add(): deferEmbedding requires 'data' (the content the background worker will embed).`
)
}
}
// Universal truth: must have data or vector
if (!params.data && !params.vector) {
throw new Error(
@ -581,6 +597,19 @@ export function validateAddParams(params: AddParams): void {
*/
export function validateUpdateParams(params: UpdateParams): void {
rejectForgedSystemKeys(params.metadata as Record<string, unknown> | undefined, 'update()')
if ((params as UpdateParams & { deferEmbedding?: boolean }).deferEmbedding === true) {
if (params.vector) {
throw new Error(
`update(): deferEmbedding cannot be combined with an explicit 'vector' — ` +
`the vector is already computed; drop one of the two.`
)
}
if (!params.data) {
throw new Error(
`update(): deferEmbedding requires new 'data' — without a data change there is nothing to re-embed.`
)
}
}
// Universal truth: must have an ID
if (!params.id) {
throw new Error('id is required for update')

View file

@ -694,6 +694,12 @@ export class VirtualFileSystem implements IVirtualFileSystem {
await this.brain.update({
id: existingId,
data: embeddingData,
// MT5: the caller's write acks at durability; the re-embed (a neural
// net — it dominated the measured 5.6s p50 per file write) runs on
// the background worker and swaps in atomically. Content is readable
// and metadata-findable immediately; semantic search converges when
// the embed lands (eventual vector index, the documented contract).
deferEmbedding: true,
metadata
})
@ -729,6 +735,9 @@ export class VirtualFileSystem implements IVirtualFileSystem {
data: embeddingData, // Always provide string for embeddings
type: this.getFileNounType(mimeType),
subtype: 'vfs-file', // Standard subtype for VFS file entities (7.30+)
// MT5: ack at durability; embedding backgrounds (see the overwrite
// branch note above).
deferEmbedding: true,
metadata
})
@ -1117,6 +1126,9 @@ export class VirtualFileSystem implements IVirtualFileSystem {
data: path, // Directory path as string content
type: NounType.Collection,
subtype: 'vfs-directory', // Standard subtype for VFS directory entities (7.30+)
// MT5: a directory creation on a write path must not wait on the
// embedder either — same ack-at-durability contract as file writes.
deferEmbedding: true,
metadata
})

View file

@ -0,0 +1,171 @@
/**
* @module tests/integration/deferred-embedding
* @description MT5 THE DEFERRED-EMBEDDING CONTRACT (A3 of the service-class
* pair, BRAINY-PROD-LATENCY-TRIAD). The production disease: a VFS file write
* ran a neural network synchronously while the caller waited (5.6s p50 per
* small file). The contract pinned here:
*
* 1. ACK AT DURABILITY: a deferred write never calls the embedder on the
* caller's path the row is id/metadata-findable immediately, with a
* durable pending marker and an honest `pendingEmbeds` gauge.
* 2. EVENTUAL VECTOR INDEX: `awaitPendingEmbeds()` is the barrier after
* it, the vector is real, indexed, and the marker is reaped.
* 3. STALE-BEATS-ABSENT on deferred updates: the OLD vector keeps serving
* until the atomic swap (the flicker law, never a dark window).
* 4. CRASH-SAFE: markers survive a session that dies mid-defer; the next
* open recovers and lands the vector. A crash DELAYS a vector, never
* loses one.
* 5. TYPED REFUSALS: deferEmbedding + vector, and deferEmbedding without
* data, are caller bugs that refuse with the fix in the message.
*/
import { describe, it, expect, afterEach, vi } from 'vitest'
import { mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { Brainy } from '../../src/index.js'
import { NounType } from '../../src/types/graphTypes.js'
const dirs: string[] = []
const brains: Brainy[] = []
async function memBrain(): Promise<Brainy> {
const b = new Brainy({ storage: { type: 'memory' }, requireSubtype: false })
await b.init()
brains.push(b)
return b
}
afterEach(async () => {
vi.restoreAllMocks()
for (const b of brains.splice(0)) await b.close().catch(() => {})
for (const d of dirs.splice(0)) rmSync(d, { recursive: true, force: true })
})
describe('MT5 — deferred embedding', () => {
it('ACK LAW: add({deferEmbedding}) never embeds on the caller path; row findable immediately; barrier lands the vector and reaps the marker', async () => {
const brain = await memBrain()
const embedSpy = vi.spyOn(brain, 'embed')
const id = await brain.add({
data: 'deferred content',
type: NounType.Document,
deferEmbedding: true,
metadata: { tag: 'deferred' }
})
// The caller's path never ran the embedder.
expect(embedSpy, 'no embed on the ack path').not.toHaveBeenCalled()
// Immediately findable by metadata; vector is the stub; gauge honest.
const found = await brain.find({ where: { tag: 'deferred' }, limit: 5 })
expect(found.map((r) => r.id)).toContain(id)
expect((await brain.getIndexStatus()).pendingEmbeds).toBeGreaterThanOrEqual(1)
// The barrier: vector lands, marker reaped, index carries the row.
await brain.awaitPendingEmbeds()
expect(embedSpy).toHaveBeenCalled()
const after = await brain.get(id, { includeVectors: true })
expect((after!.vector as number[]).length, 'real vector after the barrier').toBeGreaterThan(0)
expect(brain.pendingEmbedCount()).toBe(0)
expect((await brain.getIndexStatus()).pendingEmbeds).toBe(0)
})
it('STALE-BEATS-ABSENT: a deferred update serves the OLD vector until the atomic swap; data reads NEW immediately', async () => {
const brain = await memBrain()
const id = await brain.add({ data: 'original content', type: NounType.Document, metadata: {} })
const before = await brain.get(id, { includeVectors: true })
const oldVector = [...(before!.vector as number[])]
expect(oldVector.length).toBeGreaterThan(0)
await brain.update({ id, data: 'completely different content', deferEmbedding: true })
// Data is new IMMEDIATELY; the vector is still the old one (present,
// never absent) until the worker swaps it.
const mid = await brain.get(id, { includeVectors: true })
expect(mid!.data).toBe('completely different content')
expect(mid!.vector as number[], 'old vector keeps serving').toEqual(oldVector)
await brain.awaitPendingEmbeds()
const after = await brain.get(id, { includeVectors: true })
expect((after!.vector as number[]).length).toBeGreaterThan(0)
expect(after!.vector as number[], 'vector swapped after the barrier').not.toEqual(oldVector)
})
it('CRASH-SAFE: a session dying mid-defer leaves the durable marker; the next open recovers and lands the vector', async () => {
const dir = mkdtempSync(join(tmpdir(), 'brainy-defer-crash-'))
dirs.push(dir)
// Session 1: the embedder hangs → the worker can never complete; close()
// does not wait for it (crash-equivalent for the embed leg).
let brain = new Brainy({ storage: { type: 'filesystem', path: dir }, requireSubtype: false })
await brain.init()
brains.push(brain)
vi.spyOn(brain, 'embed').mockImplementation(() => new Promise(() => {}))
const id = await brain.add({
data: 'survives the crash',
type: NounType.Document,
deferEmbedding: true,
metadata: { k: 1 }
})
expect(brain.pendingEmbedCount()).toBe(1)
await brain.close()
brains.pop()
vi.restoreAllMocks()
// Session 2: recovery lists the marker and resumes in the background.
brain = new Brainy({ storage: { type: 'filesystem', path: dir }, requireSubtype: false })
await brain.init()
brains.push(brain)
expect(brain.pendingEmbedCount(), 'marker recovered at open').toBe(1)
await brain.awaitPendingEmbeds()
const after = await brain.get(id, { includeVectors: true })
expect((after!.vector as number[]).length, 'the delayed vector landed').toBeGreaterThan(0)
expect(brain.pendingEmbedCount()).toBe(0)
}, 120000)
it('VFS ACK LAW: writeFile resolves even when the embedder HANGS forever — the ack never depends on a neural net', async () => {
const brain = await memBrain()
// The strongest form of the pin: an embedder that never resolves. If any
// part of the writeFile ack path awaited an embed, this test would hang.
// (The background worker legitimately picks the deferred embeds up later
// — it may even interleave on the event loop during writeFile's other
// awaits — but the CALLER'S promise must never depend on it.)
const hang = vi
.spyOn(brain, 'embed')
.mockImplementation(() => new Promise<number[]>(() => {}))
await brain.vfs.writeFile('/notes/today.md', '# The day\nA deferred capture.')
// Acked with the embedder hung: content + metadata fully readable.
const content = await brain.vfs.readFile('/notes/today.md')
expect(content.toString()).toContain('A deferred capture.')
expect(brain.pendingEmbedCount()).toBeGreaterThanOrEqual(1)
// Un-hang, abandon the poisoned in-flight run (its embed promise never
// resolves — production is covered by the worker's 60s hang guard; the
// test takes the white-box shortcut for speed), drain, verify.
hang.mockRestore()
;(brain as unknown as { _embedWorkerFlight: Promise<void> | null })._embedWorkerFlight = null
await brain.awaitPendingEmbeds()
expect(brain.pendingEmbedCount()).toBe(0)
})
it('TYPED REFUSALS: defer+vector and defer-without-data both refuse with the fix', async () => {
const brain = await memBrain()
await expect(
brain.add({
data: 'x',
vector: new Array(384).fill(0.1),
type: NounType.Document,
deferEmbedding: true,
metadata: {}
})
).rejects.toThrow(/deferEmbedding cannot be combined/)
const id = await brain.add({ data: 'y', type: NounType.Document, metadata: {} })
await expect(
brain.update({ id, deferEmbedding: true, metadata: { z: 1 } })
).rejects.toThrow(/requires new 'data'/)
})
})