feat(embedding): deferred-embed markers become log records — the sidecar recovery path is deleted
The private recovery discipline, applied to its own machinery: pending- embed markers stop being sidecar files and become first-class log records riding the write's OWN commit fact — embed.pending lands in the same atomic append as its after-image (a marker can never be orphaned from its write, or vice versa; in durable-at-ack mode it shares the write's covering fsync — zero extra syncs), and the worker's landing commit rides embed.landed with the inline vector. Crash recovery is now a FOLD of the log (pending without a matching landed = recovered), skipped wholesale on brains with no v2 history; the one-time legacy bridge folds existing sidecar files in, migrates them as one fact, and deletes them — idempotent under a crash mid-bridge. No code path writes the sidecar again. Plus the ENTITY-TRUTH digest law, found by this train's own pins: canonical vector wrappers denormalize HNSW residue (connections + the randomly-assigned node level) that the log deliberately does not carry — the verification oracle digested it and would have reported false state-differs on ~any nonzero-level node (a ~15% flake in the cutover pin was the symptom). Both sides of every oracle comparison now normalize to entity truth (nounEntityTruth); index residue has its own rebuild path and is not entity state. Pins: embed-markers-in-log 5/5 (same-generation marker, landed+fold-to- zero, crash recovery via the log with the sidecar prefix EMPTY on disk, legacy bridge, VFS hung-embedder ack) · deferred-embedding 5/5 unchanged (the contract outlived its mechanism) · kill-matrix 11/11 · cutover 5/5 ×10 runs (flake dead) · unit 2031/2031.
This commit is contained in:
parent
c95bea8887
commit
b47787bbf7
7 changed files with 637 additions and 76 deletions
294
src/brainy.ts
294
src/brainy.ts
|
|
@ -178,7 +178,7 @@ import {
|
||||||
type ImportResult
|
type ImportResult
|
||||||
} from './db/portableGraph.js'
|
} from './db/portableGraph.js'
|
||||||
import { GenerationStore, type CommitBeforeImages } from './db/generationStore.js'
|
import { GenerationStore, type CommitBeforeImages } from './db/generationStore.js'
|
||||||
import type { FactScanHandle } from './db/factLog.js'
|
import type { FactScanHandle, FactMarkerRecord } from './db/factLog.js'
|
||||||
import {
|
import {
|
||||||
ENTITY_TREE_STAMP_PATH,
|
ENTITY_TREE_STAMP_PATH,
|
||||||
readFamilyStamp,
|
readFamilyStamp,
|
||||||
|
|
@ -201,6 +201,7 @@ import {
|
||||||
runLogCompletenessOracle,
|
runLogCompletenessOracle,
|
||||||
flipToLogAuthority,
|
flipToLogAuthority,
|
||||||
recordDigest,
|
recordDigest,
|
||||||
|
nounEntityTruth,
|
||||||
type LogAuthorityRecord,
|
type LogAuthorityRecord,
|
||||||
type LogAuthorityStorage,
|
type LogAuthorityStorage,
|
||||||
type OracleReport
|
type OracleReport
|
||||||
|
|
@ -382,6 +383,13 @@ interface PlannedTransact {
|
||||||
* rejected batch (CAS conflict, failed apply) emits nothing.
|
* rejected batch (CAS conflict, failed apply) emits nothing.
|
||||||
*/
|
*/
|
||||||
changeEvents: PendingChangeEvent[]
|
changeEvents: PendingChangeEvent[]
|
||||||
|
/**
|
||||||
|
* V2 marker records riding the batch's ONE commit fact (e.g. the
|
||||||
|
* deferred-embedding pending markers) — same generation, same atomic
|
||||||
|
* append as the batch itself. A rejected batch appends no fact, so no
|
||||||
|
* marker outlives its write.
|
||||||
|
*/
|
||||||
|
markerRecords: FactMarkerRecord[]
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
@ -722,9 +730,12 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
private _persistIdleTimer: ReturnType<typeof setTimeout> | null = null
|
private _persistIdleTimer: ReturnType<typeof setTimeout> | null = null
|
||||||
private _persistBackgroundFlight: Promise<void> | null = null
|
private _persistBackgroundFlight: Promise<void> | null = null
|
||||||
|
|
||||||
// DEFERRED EMBEDDING (MT5): durable pending markers under
|
// DEFERRED EMBEDDING (MT5): pending markers are LOG RECORDS — an
|
||||||
// _system/pending_embeds/<id>, mirrored in-memory, drained by ONE
|
// embed.pending record rides the deferred write's own commit fact and
|
||||||
// background worker. A crash can delay a vector, never lose one.
|
// embed.landed rides the landing commit; this set is the in-memory
|
||||||
|
// fast-path index, rebuilt at open by folding the log's marker records.
|
||||||
|
// ONE background worker drains it. A crash can delay a vector, never
|
||||||
|
// lose one.
|
||||||
private _pendingEmbedIds = new Set<string>()
|
private _pendingEmbedIds = new Set<string>()
|
||||||
private _embedWorkerFlight: Promise<void> | null = null
|
private _embedWorkerFlight: Promise<void> | null = null
|
||||||
|
|
||||||
|
|
@ -1494,17 +1505,18 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// MT5 crash recovery: reload the durable pending-embed markers (a
|
// MT5 crash recovery — REPLAY, NOT LISTING: the pending-embed markers
|
||||||
// BOUNDED prefix listing — never a store walk) and resume the worker
|
// live IN the generation log (embed.pending rides the deferred write's
|
||||||
// in the background. A crash between a deferred write's ack and its
|
// own fact; embed.landed rides the landing commit), so recovery folds
|
||||||
// background embed DELAYED a vector; this is where it lands.
|
// the log's marker records back into the in-memory set — after the
|
||||||
|
// one-time bridge migrates any sidecar files a pre-log build left
|
||||||
|
// behind — and resumes 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) {
|
if (!this.isReadOnly) {
|
||||||
try {
|
try {
|
||||||
const markerPaths = await this.storage.listRawObjects(Brainy.PENDING_EMBED_PREFIX)
|
await this.bridgeLegacyPendingEmbedSidecars()
|
||||||
for (const path of markerPaths) {
|
await this.recoverPendingEmbedsFromLog()
|
||||||
const id = path.slice(path.lastIndexOf('/') + 1)
|
|
||||||
if (id) this._pendingEmbedIds.add(id)
|
|
||||||
}
|
|
||||||
if (this._pendingEmbedIds.size > 0) {
|
if (this._pendingEmbedIds.size > 0) {
|
||||||
prodLog.info(
|
prodLog.info(
|
||||||
`[Brainy] ${this._pendingEmbedIds.size} deferred embed(s) pending from a previous ` +
|
`[Brainy] ${this._pendingEmbedIds.size} deferred embed(s) pending from a previous ` +
|
||||||
|
|
@ -1515,8 +1527,8 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
}
|
}
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
prodLog.warn(
|
prodLog.warn(
|
||||||
`[Brainy] pending-embed recovery listing failed: ${(err as Error).message} — ` +
|
`[Brainy] pending-embed recovery failed: ${(err as Error).message} — ` +
|
||||||
`markers remain durable; recovery retries next open`
|
`the log's markers remain durable; recovery retries next open`
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -1942,30 +1954,144 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
* deletes — the before-image + per-id-chain set.
|
* deletes — the before-image + per-id-chain set.
|
||||||
* @param run - The single-op's existing operation batch builder (the
|
* @param run - The single-op's existing operation batch builder (the
|
||||||
* `tx => {…}` body previously passed straight to `executeTransaction`).
|
* `tx => {…}` body previously passed straight to `executeTransaction`).
|
||||||
|
* @param precommit - Optional CAS precondition, run under the commit mutex.
|
||||||
|
* @param pendingEvents - Change-feed events to stamp and emit post-commit.
|
||||||
|
* @param records - Optional v2 marker records (e.g. the deferred-embedding
|
||||||
|
* lifecycle markers) riding this write's commit fact — same generation,
|
||||||
|
* one atomic append. Refused on generation-less bootstrap writes.
|
||||||
|
*/
|
||||||
|
/**
|
||||||
|
* Storage-root-relative prefix of the RETIRED sidecar pending-embed marker
|
||||||
|
* files (pre-log builds persisted one raw object per pending embed here).
|
||||||
|
* The markers live IN the generation log now (`embed.pending` /
|
||||||
|
* `embed.landed` records); this prefix survives ONLY for the one-time
|
||||||
|
* migration bridge ({@link bridgeLegacyPendingEmbedSidecars}) — no other
|
||||||
|
* code path writes, lists, or deletes it.
|
||||||
*/
|
*/
|
||||||
/** Storage-root-relative prefix of the durable pending-embed markers. */
|
|
||||||
private static readonly PENDING_EMBED_PREFIX = '_system/pending_embeds/'
|
private static readonly PENDING_EMBED_PREFIX = '_system/pending_embeds/'
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @description Persist the durable pending-embed marker (MT5) and mirror
|
* @description Mark a deferred embed pending (MT5): the id joins the
|
||||||
* it in memory. Written BEFORE the write it belongs to commits — an
|
* in-memory fast-path set and the returned `embed.pending` record is
|
||||||
* orphaned marker (commit failed) is harmless and reaped by the worker;
|
* threaded onto the deferred write's OWN commit fact — same generation,
|
||||||
* the reverse ordering could lose an embed silently on a crash.
|
* same atomic append, and (in at-ack log durability) the same covering
|
||||||
|
* fsync as the write itself. The marker can never be orphaned from its
|
||||||
|
* write nor the write from its marker: a failed commit appends no fact,
|
||||||
|
* so no durable marker exists either (the in-memory entry is harmless
|
||||||
|
* and reaped by the worker). Recovery folds the marker back out of the
|
||||||
|
* log at open ({@link recoverPendingEmbedsFromLog}).
|
||||||
*/
|
*/
|
||||||
private async enqueuePendingEmbed(id: string): Promise<void> {
|
private enqueuePendingEmbed(id: string): FactMarkerRecord {
|
||||||
this._pendingEmbedIds.add(id)
|
this._pendingEmbedIds.add(id)
|
||||||
await this.storage.writeRawObject(`${Brainy.PENDING_EMBED_PREFIX}${id}`, {
|
return { type: 'embed.pending', id, enqueuedAt: Date.now() }
|
||||||
id,
|
|
||||||
enqueuedAt: Date.now()
|
|
||||||
})
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Remove a pending-embed marker (memory + durable), tolerating races. */
|
/**
|
||||||
private async clearPendingEmbed(id: string): Promise<void> {
|
* @description Clear a pending embed from the in-memory set. The DURABLE
|
||||||
|
* clear is the `embed.landed` record riding the landing commit's own fact
|
||||||
|
* (or, for a row deleted before its embed landed, the row's tombstone
|
||||||
|
* fact) — the recovery fold consumes those; nothing here touches storage.
|
||||||
|
* One honest residue: a pending row whose entity still exists but carries
|
||||||
|
* no data is reaped in memory only, so it re-folds at the next open and
|
||||||
|
* is re-reaped there — a bounded no-op, never a lost vector.
|
||||||
|
*/
|
||||||
|
private clearPendingEmbed(id: string): void {
|
||||||
this._pendingEmbedIds.delete(id)
|
this._pendingEmbedIds.delete(id)
|
||||||
await this.storage
|
}
|
||||||
.deleteRawObject(`${Brainy.PENDING_EMBED_PREFIX}${id}`)
|
|
||||||
.catch(() => {})
|
/**
|
||||||
|
* @description Rebuild the pending-embed set by REPLAYING the generation
|
||||||
|
* log's marker records (recovery = replay, not listing): `embed.pending`
|
||||||
|
* arms an id, `embed.landed` disarms it, and a noun tombstone disarms it
|
||||||
|
* too (a row deleted before its embed landed owes no vector). What
|
||||||
|
* survives the fold is exactly the set of acknowledged deferred writes
|
||||||
|
* whose vectors have not landed.
|
||||||
|
*
|
||||||
|
* BOUND (honest): no durable low-water mark exists for the earliest
|
||||||
|
* unconsumed pending, so the fold scans the log's committed facts from
|
||||||
|
* generation 1 — a sequential read of the log at open, O(log bytes).
|
||||||
|
* It is SKIPPED WHOLESALE when the log has never had a v2 tail
|
||||||
|
* ({@link FactLog.hasV2History} — v1 facts cannot carry marker records),
|
||||||
|
* so pre-cutover brains pay nothing; on a mixed log the scan still reads
|
||||||
|
* the v1 segments (a segment's format is only known from its bytes) but
|
||||||
|
* they fold to nothing, so the DECODE cost is bounded by v2 history.
|
||||||
|
* Storage without a fact log hosts no durable markers at all — the
|
||||||
|
* pending set is session-local there, matching that storage's overall
|
||||||
|
* durability posture.
|
||||||
|
*/
|
||||||
|
private async recoverPendingEmbedsFromLog(): Promise<void> {
|
||||||
|
const log = this.generationStore.getFactLog()
|
||||||
|
if (!log || !log.hasV2History()) return
|
||||||
|
const scan = log.scanFacts({ fromGeneration: 1 })
|
||||||
|
for await (const batch of scan.batches()) {
|
||||||
|
for (const fact of batch.facts) {
|
||||||
|
for (const record of fact.records ?? []) {
|
||||||
|
if (record.type === 'embed.pending') {
|
||||||
|
this._pendingEmbedIds.add(record.id)
|
||||||
|
} else if (record.type === 'embed.landed') {
|
||||||
|
this._pendingEmbedIds.delete(record.id)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for (const op of fact.ops) {
|
||||||
|
if (op.kind === 'noun' && op.record === null) {
|
||||||
|
this._pendingEmbedIds.delete(op.id)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @description ONE-TIME LEGACY BRIDGE: a brain that deferred embeds under
|
||||||
|
* a pre-log build persisted one sidecar marker file per pending embed
|
||||||
|
* under {@link PENDING_EMBED_PREFIX}. At open, fold those ids into the
|
||||||
|
* pending set AND migrate them: commit ONE fact carrying their
|
||||||
|
* `embed.pending` records (the log is the markers' durable home now),
|
||||||
|
* then delete the sidecar files — in that order, so a crash between the
|
||||||
|
* two re-runs the bridge instead of losing a marker (a re-migrated
|
||||||
|
* duplicate folds idempotently; at worst an already-landed embed re-runs
|
||||||
|
* once — idempotent, never lost). Narrated loudly. Storage without a
|
||||||
|
* fact log keeps its sidecars in place (there is no log to migrate into)
|
||||||
|
* and folds them into memory only, exactly as loud.
|
||||||
|
*/
|
||||||
|
private async bridgeLegacyPendingEmbedSidecars(): Promise<void> {
|
||||||
|
const markerPaths = await this.storage.listRawObjects(Brainy.PENDING_EMBED_PREFIX)
|
||||||
|
if (markerPaths.length === 0) return
|
||||||
|
const ids: string[] = []
|
||||||
|
for (const path of markerPaths) {
|
||||||
|
const id = path.slice(path.lastIndexOf('/') + 1)
|
||||||
|
if (id) ids.push(id)
|
||||||
|
}
|
||||||
|
if (ids.length === 0) return
|
||||||
|
for (const id of ids) this._pendingEmbedIds.add(id)
|
||||||
|
if (!this.generationStore.getFactLog()) {
|
||||||
|
prodLog.warn(
|
||||||
|
`[Brainy] ${ids.length} legacy pending-embed sidecar marker(s) found, but this ` +
|
||||||
|
`storage hosts no fact log to migrate them into — folded into memory; the ` +
|
||||||
|
`sidecar files remain the durable recovery source on this configuration`
|
||||||
|
)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
const enqueuedAt = Date.now()
|
||||||
|
const markers: FactMarkerRecord[] = ids.map((id) => ({
|
||||||
|
type: 'embed.pending',
|
||||||
|
id,
|
||||||
|
enqueuedAt
|
||||||
|
}))
|
||||||
|
// One migration commit: a zero-op fact carrying every legacy marker
|
||||||
|
// (empty-ops facts are legal; the records leg makes this one visible).
|
||||||
|
await this.generationStore.commitSingleOp({
|
||||||
|
touched: {},
|
||||||
|
records: markers,
|
||||||
|
execute: async () => {}
|
||||||
|
})
|
||||||
|
for (const id of ids) {
|
||||||
|
await this.storage.deleteRawObject(`${Brainy.PENDING_EMBED_PREFIX}${id}`).catch(() => {})
|
||||||
|
}
|
||||||
|
prodLog.info(
|
||||||
|
`[Brainy] migrated ${ids.length} legacy pending-embed sidecar marker(s) into the ` +
|
||||||
|
`generation log and removed the sidecar files (one-time bridge)`
|
||||||
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
@ -2005,7 +2131,10 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
try {
|
try {
|
||||||
const entity = await this.get(id, { includeVectors: true })
|
const entity = await this.get(id, { includeVectors: true })
|
||||||
if (!entity || entity.data === undefined || entity.data === null) {
|
if (!entity || entity.data === undefined || entity.data === null) {
|
||||||
await this.clearPendingEmbed(id)
|
// Orphan reap: a deleted row's tombstone fact durably disarms the
|
||||||
|
// marker at the next recovery fold; a data-less-but-present row
|
||||||
|
// (edge case) re-folds and re-reaps — bounded, never a lost vector.
|
||||||
|
this.clearPendingEmbed(id)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
// Hang guard: a wedged embedder must not block every later pending
|
// Hang guard: a wedged embedder must not block every later pending
|
||||||
|
|
@ -2030,20 +2159,29 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
const oldVector = (entity.vector as number[] | undefined) ?? []
|
const oldVector = (entity.vector as number[] | undefined) ?? []
|
||||||
await this.persistSingleOp({ nouns: [id] }, async (tx) => {
|
// The landing commit's fact carries the embed.landed record (vector
|
||||||
tx.addOperation(
|
// inline, per the v2 format) alongside the row's after-image — the
|
||||||
new SaveNounOperation(this.storage, {
|
// durable "this pending is consumed" that recovery's fold reads.
|
||||||
id,
|
await this.persistSingleOp(
|
||||||
vector: newVector,
|
{ nouns: [id] },
|
||||||
connections: new Map(),
|
async (tx) => {
|
||||||
level: 0
|
tx.addOperation(
|
||||||
})
|
new SaveNounOperation(this.storage, {
|
||||||
)
|
id,
|
||||||
tx.addOperation(
|
vector: newVector,
|
||||||
new ReplaceInVectorIndexOperation(this.index, id, oldVector, newVector, this.indexWriteGeneration)
|
connections: new Map(),
|
||||||
)
|
level: 0
|
||||||
})
|
})
|
||||||
await this.clearPendingEmbed(id)
|
)
|
||||||
|
tx.addOperation(
|
||||||
|
new ReplaceInVectorIndexOperation(this.index, id, oldVector, newVector, this.indexWriteGeneration)
|
||||||
|
)
|
||||||
|
},
|
||||||
|
undefined,
|
||||||
|
undefined,
|
||||||
|
[{ type: 'embed.landed', id, vector: newVector }]
|
||||||
|
)
|
||||||
|
this.clearPendingEmbed(id)
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
prodLog.warn(
|
prodLog.warn(
|
||||||
`[Brainy] deferred embed for ${id} failed: ${(err as Error).message} — marker retained for retry`
|
`[Brainy] deferred embed for ${id} failed: ${(err as Error).message} — marker retained for retry`
|
||||||
|
|
@ -2242,7 +2380,8 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
touched: { nouns?: string[]; verbs?: string[] },
|
touched: { nouns?: string[]; verbs?: string[] },
|
||||||
run: TransactionFunction<void>,
|
run: TransactionFunction<void>,
|
||||||
precommit?: (before: CommitBeforeImages) => void,
|
precommit?: (before: CommitBeforeImages) => void,
|
||||||
pendingEvents?: PendingChangeEvent[]
|
pendingEvents?: PendingChangeEvent[],
|
||||||
|
records?: FactMarkerRecord[]
|
||||||
): Promise<{ generation?: number; timestamp: number; degraded?: string[] }> {
|
): Promise<{ generation?: number; timestamp: number; degraded?: string[] }> {
|
||||||
// Change-feed capture: when this write will emit, hold a reference to the
|
// Change-feed capture: when this write will emit, hold a reference to the
|
||||||
// commit's before-images so `remove` events can carry the record's last
|
// commit's before-images so `remove` events can carry the record's last
|
||||||
|
|
@ -2257,6 +2396,15 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
: precommit
|
: precommit
|
||||||
|
|
||||||
if (!this._generationStampingActive) {
|
if (!this._generationStampingActive) {
|
||||||
|
// Marker records ride a commit FACT — a generation-less bootstrap
|
||||||
|
// write has none to ride. No bootstrap path defers embeds today;
|
||||||
|
// refuse loudly rather than silently dropping a durable marker.
|
||||||
|
if (records && records.length > 0) {
|
||||||
|
throw new Error(
|
||||||
|
'persistSingleOp: marker records require a generation-stamped commit — ' +
|
||||||
|
'a bootstrap (generation-0) write cannot carry them'
|
||||||
|
)
|
||||||
|
}
|
||||||
// Init-time / infrastructure baseline write (e.g. the VFS root): apply
|
// Init-time / infrastructure baseline write (e.g. the VFS root): apply
|
||||||
// WITHOUT creating a generation. Generation 0 is the freshly-materialized
|
// WITHOUT creating a generation. Generation 0 is the freshly-materialized
|
||||||
// brain (bootstrap included); the first USER write is generation 1.
|
// brain (bootstrap included); the first USER write is generation 1.
|
||||||
|
|
@ -2295,6 +2443,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
receipt = await this.generationStore.commitSingleOp({
|
receipt = await this.generationStore.commitSingleOp({
|
||||||
touched,
|
touched,
|
||||||
precommit: captureAndCheck,
|
precommit: captureAndCheck,
|
||||||
|
...(records && records.length > 0 ? { records } : {}),
|
||||||
execute: () =>
|
execute: () =>
|
||||||
this.transactionManager.executeTransaction(run, {
|
this.transactionManager.executeTransaction(run, {
|
||||||
timeout: transactTimeoutBudget(
|
timeout: transactTimeoutBudget(
|
||||||
|
|
@ -2507,10 +2656,10 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
|
|
||||||
// Get or compute vector
|
// Get or compute vector
|
||||||
// MT5 deferred embedding: ack at durability with a stub vector and a
|
// MT5 deferred embedding: ack at durability with a stub vector and a
|
||||||
// DURABLE pending marker (written BEFORE the commit — an orphaned marker
|
// pending marker riding the insert's OWN commit fact (same generation,
|
||||||
// from a failed commit is harmless and reaped by the worker; a
|
// one atomic append — a marker-less committed row, the silently-missing-
|
||||||
// marker-less committed row would be a silently missing vector, which is
|
// vector shape, is structurally impossible). The background worker
|
||||||
// the disallowed direction). The background worker embeds + inserts.
|
// embeds + inserts.
|
||||||
const deferringEmbed = params.deferEmbedding === true && !params.vector
|
const deferringEmbed = params.deferEmbedding === true && !params.vector
|
||||||
const vector = deferringEmbed
|
const vector = deferringEmbed
|
||||||
? []
|
? []
|
||||||
|
|
@ -2605,11 +2754,13 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
}
|
}
|
||||||
: undefined
|
: undefined
|
||||||
|
|
||||||
// MT5: the durable marker lands BEFORE the commit (orphan-safe; the
|
// MT5: the pending marker RIDES the insert's own commit fact (same
|
||||||
// reverse order could lose an embed silently on a crash).
|
// generation, one atomic append) — threaded to persistSingleOp below.
|
||||||
if (deferringEmbed) {
|
// A failed commit appends nothing, so no orphaned durable marker can
|
||||||
await this.enqueuePendingEmbed(id)
|
// exist; the in-memory entry is harmless and reaped by the worker.
|
||||||
}
|
const embedMarkers: FactMarkerRecord[] | undefined = deferringEmbed
|
||||||
|
? [this.enqueuePendingEmbed(id)]
|
||||||
|
: undefined
|
||||||
|
|
||||||
const runInsert: TransactionFunction<void> = async (tx) => {
|
const runInsert: TransactionFunction<void> = async (tx) => {
|
||||||
// Operation 1: Save metadata FIRST (TypeAwareStorage caching)
|
// Operation 1: Save metadata FIRST (TypeAwareStorage caching)
|
||||||
|
|
@ -2670,7 +2821,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
const MAX_UPSERT_ATTEMPTS = 10
|
const MAX_UPSERT_ATTEMPTS = 10
|
||||||
for (let attempt = 0; ; attempt++) {
|
for (let attempt = 0; ; attempt++) {
|
||||||
try {
|
try {
|
||||||
await this.persistSingleOp({ nouns: [id] }, runInsert, insertPrecommit, addEvents)
|
await this.persistSingleOp({ nouns: [id] }, runInsert, insertPrecommit, addEvents, embedMarkers)
|
||||||
break
|
break
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
if (!(err instanceof InsertPreconditionExistsSignal)) {
|
if (!(err instanceof InsertPreconditionExistsSignal)) {
|
||||||
|
|
@ -3296,10 +3447,11 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
updatedMetadata._rev = authoritativeRev + 1
|
updatedMetadata._rev = authoritativeRev + 1
|
||||||
}
|
}
|
||||||
|
|
||||||
// MT5: durable marker BEFORE the commit (orphan-safe direction).
|
// MT5: the pending marker rides the update's own commit fact (same
|
||||||
if (deferringEmbed) {
|
// generation, one atomic append) — threaded to persistSingleOp below.
|
||||||
await this.enqueuePendingEmbed(params.id)
|
const embedMarkers: FactMarkerRecord[] | undefined = deferringEmbed
|
||||||
}
|
? [this.enqueuePendingEmbed(params.id)]
|
||||||
|
: undefined
|
||||||
|
|
||||||
// Execute atomically with transaction system, generation-stamped as one
|
// Execute atomically with transaction system, generation-stamped as one
|
||||||
// immutable Model-B generation (before-image = the entity's prior state).
|
// immutable Model-B generation (before-image = the entity's prior state).
|
||||||
|
|
@ -3389,7 +3541,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
]
|
]
|
||||||
: undefined)
|
: undefined, embedMarkers)
|
||||||
|
|
||||||
// Aggregation hook (outside transaction — derived data). `existing` is
|
// Aggregation hook (outside transaction — derived data). `existing` is
|
||||||
// the full get() view — every reserved field top-level — and must be
|
// the full get() view — every reserved field top-level — and must be
|
||||||
|
|
@ -7962,12 +8114,17 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
return runLogCompletenessOracle({
|
return runLogCompletenessOracle({
|
||||||
storage: this.storage as unknown as LogAuthorityStorage,
|
storage: this.storage as unknown as LogAuthorityStorage,
|
||||||
scanFacts: () => this.scanFacts(),
|
scanFacts: () => this.scanFacts(),
|
||||||
|
// Both sides normalize to ENTITY TRUTH before digesting: canonical
|
||||||
|
// wrappers denormalize HNSW residue (connections/level) the log never
|
||||||
|
// carries — digesting it would fake state-differs on any nonzero-level
|
||||||
|
// node (the residue has its own rebuild path; it is not entity state).
|
||||||
canonicalNounDigest: async (id: string) => {
|
canonicalNounDigest: async (id: string) => {
|
||||||
const raw = await this.storage.readNounRaw(id)
|
const raw = await this.storage.readNounRaw(id)
|
||||||
if (raw.metadata === null && raw.vector === null) return null
|
if (raw.metadata === null && raw.vector === null) return null
|
||||||
return recordDigest({ metadata: raw.metadata, vector: raw.vector })
|
return recordDigest(nounEntityTruth({ metadata: raw.metadata, vector: raw.vector }))
|
||||||
},
|
},
|
||||||
factRecordDigest: (record: unknown) => recordDigest(record)
|
factRecordDigest: (record: unknown) =>
|
||||||
|
recordDigest(nounEntityTruth(record as { metadata: unknown; vector: unknown }))
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -8304,6 +8461,7 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
meta: options?.meta,
|
meta: options?.meta,
|
||||||
ifAtGeneration: options?.ifAtGeneration,
|
ifAtGeneration: options?.ifAtGeneration,
|
||||||
precommit: casPrecommit,
|
precommit: casPrecommit,
|
||||||
|
...(plan.markerRecords.length > 0 ? { records: plan.markerRecords } : {}),
|
||||||
execute: async () => {
|
execute: async () => {
|
||||||
await this.transactionManager.executeTransaction(
|
await this.transactionManager.executeTransaction(
|
||||||
async (tx) => {
|
async (tx) => {
|
||||||
|
|
@ -9561,7 +9719,8 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
postCommit: [],
|
postCommit: [],
|
||||||
casUpdates: [],
|
casUpdates: [],
|
||||||
createdNouns: new Set(),
|
createdNouns: new Set(),
|
||||||
changeEvents: []
|
changeEvents: [],
|
||||||
|
markerRecords: []
|
||||||
}
|
}
|
||||||
|
|
||||||
for (const op of ops) {
|
for (const op of ops) {
|
||||||
|
|
@ -9757,9 +9916,10 @@ export class Brainy<T = any> implements BrainyInterface<T> {
|
||||||
}
|
}
|
||||||
|
|
||||||
if (deferringEmbed) {
|
if (deferringEmbed) {
|
||||||
// Durable marker BEFORE the batch commits (orphan-safe direction);
|
// The pending marker rides the batch's ONE commit fact (same
|
||||||
// the worker kicks post-commit via the plan hook.
|
// generation, one atomic append); the worker kicks post-commit via
|
||||||
await this.enqueuePendingEmbed(id)
|
// the plan hook.
|
||||||
|
plan.markerRecords.push(this.enqueuePendingEmbed(id))
|
||||||
plan.postCommit.push(() => this.kickEmbedWorker())
|
plan.postCommit.push(() => this.kickEmbedWorker())
|
||||||
}
|
}
|
||||||
plan.operations.push(
|
plan.operations.push(
|
||||||
|
|
|
||||||
|
|
@ -138,9 +138,11 @@ export interface FactOp {
|
||||||
/**
|
/**
|
||||||
* V2-native records beyond noun/verb ops that a fact may carry through the
|
* V2-native records beyond noun/verb ops that a fact may carry through the
|
||||||
* ENCODER (types 6/7/8/9/10 of the v2 registry: embed markers, blob
|
* ENCODER (types 6/7/8/9/10 of the v2 registry: embed markers, blob
|
||||||
* manifests, projection notes, bootstrap baselines). Encoder-ready by
|
* manifests, projection notes, bootstrap baselines). The deferred-embedding
|
||||||
* design; nothing produces them yet — the deferred-embed sidecar and blob
|
* lifecycle PRODUCES types 6/7 today: `embed.pending` rides the deferred
|
||||||
* lifecycle remodel onto these records in a later leg.
|
* write's own commit fact and `embed.landed` rides the background worker's
|
||||||
|
* landing commit (recovery folds the pair back out of the log at open). The
|
||||||
|
* blob lifecycle remodels onto type 8 in a later leg.
|
||||||
*/
|
*/
|
||||||
export type FactMarkerRecord =
|
export type FactMarkerRecord =
|
||||||
| EmbedPendingRecord
|
| EmbedPendingRecord
|
||||||
|
|
@ -741,6 +743,17 @@ export class FactLog {
|
||||||
return this.head
|
return this.head
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* True when this log has EVER had a v2 tail — the manifest's `brainId` is
|
||||||
|
* minted at every v2 tail creation seam and never removed (the tail-version
|
||||||
|
* check is a belt-and-braces second signal). Only v2 facts can carry marker
|
||||||
|
* records, so marker folds (e.g. the deferred-embed recovery scan) skip
|
||||||
|
* v1-only logs WHOLESALE on this one cheap check — no segment is read.
|
||||||
|
*/
|
||||||
|
hasV2History(): boolean {
|
||||||
|
return this.manifest.brainId !== undefined || this.tailVersion === FACT_LOG_FORMAT_V2
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Open the log and reconcile it to committed truth: read the manifest,
|
* Open the log and reconcile it to committed truth: read the manifest,
|
||||||
* establish the tail's intact content (torn-tail scan), then TRUNCATE any
|
* establish the tail's intact content (torn-tail scan), then TRUNCATE any
|
||||||
|
|
|
||||||
|
|
@ -51,7 +51,8 @@ import {
|
||||||
storageSupportsFactLog,
|
storageSupportsFactLog,
|
||||||
type CommitFact,
|
type CommitFact,
|
||||||
type FactOp,
|
type FactOp,
|
||||||
type FactIntMinter
|
type FactIntMinter,
|
||||||
|
type FactMarkerRecord
|
||||||
} from './factLog.js'
|
} from './factLog.js'
|
||||||
import { GenerationSegmentStore, type FoldGeneration } from './generationSegments.js'
|
import { GenerationSegmentStore, type FoldGeneration } from './generationSegments.js'
|
||||||
import { crc32c } from '../utils/crc32c.js'
|
import { crc32c } from '../utils/crc32c.js'
|
||||||
|
|
@ -903,6 +904,8 @@ export class GenerationStore {
|
||||||
nouns: string[]
|
nouns: string[]
|
||||||
verbs: string[]
|
verbs: string[]
|
||||||
meta?: Record<string, unknown>
|
meta?: Record<string, unknown>
|
||||||
|
/** V2 marker records riding this fact (same generation, same append). */
|
||||||
|
records?: FactMarkerRecord[]
|
||||||
}): Promise<CommitFact> {
|
}): Promise<CommitFact> {
|
||||||
const ops: FactOp[] = []
|
const ops: FactOp[] = []
|
||||||
const afterRecords: GenerationRecord[] = []
|
const afterRecords: GenerationRecord[] = []
|
||||||
|
|
@ -926,7 +929,8 @@ export class GenerationStore {
|
||||||
timestamp: args.timestamp,
|
timestamp: args.timestamp,
|
||||||
ops,
|
ops,
|
||||||
...(args.meta ? { meta: args.meta } : {}),
|
...(args.meta ? { meta: args.meta } : {}),
|
||||||
...(blobHashes.length > 0 ? { blobHashes } : {})
|
...(blobHashes.length > 0 ? { blobHashes } : {}),
|
||||||
|
...(args.records && args.records.length > 0 ? { records: args.records } : {})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -939,6 +943,12 @@ export class GenerationStore {
|
||||||
* per-record analogue of `ifAtGeneration`. A throw aborts the whole batch:
|
* per-record analogue of `ifAtGeneration`. A throw aborts the whole batch:
|
||||||
* the generation reservation is returned and no staging I/O has happened. */
|
* the generation reservation is returned and no staging I/O has happened. */
|
||||||
precommit?: (before: CommitBeforeImages) => void
|
precommit?: (before: CommitBeforeImages) => void
|
||||||
|
/** Optional v2 marker records riding this batch's ONE commit fact (e.g.
|
||||||
|
* the deferred-embedding lifecycle markers) — same generation, same
|
||||||
|
* atomic append, same durability barrier as the batch itself, so a
|
||||||
|
* marker can never be orphaned from its write nor the write from its
|
||||||
|
* marker. Additive: omitted on every markerless path. */
|
||||||
|
records?: FactMarkerRecord[]
|
||||||
execute: () => Promise<void>
|
execute: () => Promise<void>
|
||||||
}): Promise<{ generation: number; timestamp: number }> {
|
}): Promise<{ generation: number; timestamp: number }> {
|
||||||
return this.withMutex(async () => {
|
return this.withMutex(async () => {
|
||||||
|
|
@ -1075,7 +1085,8 @@ export class GenerationStore {
|
||||||
timestamp,
|
timestamp,
|
||||||
nouns,
|
nouns,
|
||||||
verbs,
|
verbs,
|
||||||
...(args.meta ? { meta: args.meta } : {})
|
...(args.meta ? { meta: args.meta } : {}),
|
||||||
|
...(args.records && args.records.length > 0 ? { records: args.records } : {})
|
||||||
})
|
})
|
||||||
await this.factLog.append(fact)
|
await this.factLog.append(fact)
|
||||||
await this.factLog.sync()
|
await this.factLog.sync()
|
||||||
|
|
@ -1288,6 +1299,18 @@ export class GenerationStore {
|
||||||
touched: { nouns?: string[]; verbs?: string[] }
|
touched: { nouns?: string[]; verbs?: string[] }
|
||||||
execute: () => Promise<void>
|
execute: () => Promise<void>
|
||||||
precommit?: (before: CommitBeforeImages) => void
|
precommit?: (before: CommitBeforeImages) => void
|
||||||
|
/**
|
||||||
|
* Optional v2 marker records riding this write's commit fact (e.g. the
|
||||||
|
* deferred-embedding lifecycle markers) — same generation, same atomic
|
||||||
|
* append, and in 'at-ack' log durability the SAME covering fsync as the
|
||||||
|
* write itself (zero extra sync). A marker can never be orphaned from
|
||||||
|
* its write nor the write from its marker. Additive: omitted on every
|
||||||
|
* markerless path. When the storage hosts no fact log the markers have
|
||||||
|
* no durable home — matching that storage's overall durability posture
|
||||||
|
* (it cannot host the log's crash guarantees either); callers own
|
||||||
|
* surfacing that honestly.
|
||||||
|
*/
|
||||||
|
records?: FactMarkerRecord[]
|
||||||
}): Promise<{ generation: number; timestamp: number; degraded?: string[] }> {
|
}): Promise<{ generation: number; timestamp: number; degraded?: string[] }> {
|
||||||
return this.withMutex(async () => {
|
return this.withMutex(async () => {
|
||||||
// Refuse to accept a write whose history we cannot make durable: if the
|
// Refuse to accept a write whose history we cannot make durable: if the
|
||||||
|
|
@ -1357,7 +1380,13 @@ export class GenerationStore {
|
||||||
// buffered history).
|
// buffered history).
|
||||||
if (this.factLog) {
|
if (this.factLog) {
|
||||||
await this.factLog.append(
|
await this.factLog.append(
|
||||||
await this.buildCommitFact({ generation: gen, timestamp, nouns, verbs })
|
await this.buildCommitFact({
|
||||||
|
generation: gen,
|
||||||
|
timestamp,
|
||||||
|
nouns,
|
||||||
|
verbs,
|
||||||
|
...(args.records && args.records.length > 0 ? { records: args.records } : {})
|
||||||
|
})
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
prodLog.warn(
|
prodLog.warn(
|
||||||
|
|
@ -1411,7 +1440,13 @@ export class GenerationStore {
|
||||||
if (this.factLog) {
|
if (this.factLog) {
|
||||||
try {
|
try {
|
||||||
await this.factLog.append(
|
await this.factLog.append(
|
||||||
await this.buildCommitFact({ generation: gen, timestamp, nouns, verbs })
|
await this.buildCommitFact({
|
||||||
|
generation: gen,
|
||||||
|
timestamp,
|
||||||
|
nouns,
|
||||||
|
verbs,
|
||||||
|
...(args.records && args.records.length > 0 ? { records: args.records } : {})
|
||||||
|
})
|
||||||
)
|
)
|
||||||
if (this.logDurability === 'at-ack') {
|
if (this.logDurability === 'at-ack') {
|
||||||
await this.factLog.ensureSynced()
|
await this.factLog.ensureSynced()
|
||||||
|
|
|
||||||
|
|
@ -92,6 +92,27 @@ export async function readLogAuthority(
|
||||||
return { authority: 'tree' }
|
return { authority: 'tree' }
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Normalize a canonical noun record to its ENTITY TRUTH before diffing:
|
||||||
|
* the canonical vector-file wrapper denormalizes derived index residue
|
||||||
|
* (`connections` — HNSW graph edges; `level` — the node's random skip-list
|
||||||
|
* level) that the generation log deliberately does NOT carry (projections
|
||||||
|
* own their own rebuild paths). Digesting the residue would report false
|
||||||
|
* `state-differs` on ~any brain whose HNSW assigned a nonzero level. Both
|
||||||
|
* sides of every oracle comparison pass through this normalizer.
|
||||||
|
*/
|
||||||
|
export function nounEntityTruth(record: {
|
||||||
|
metadata: unknown
|
||||||
|
vector: unknown
|
||||||
|
}): { metadata: unknown; vector: unknown } {
|
||||||
|
const v = record.vector
|
||||||
|
if (v && typeof v === 'object' && !Array.isArray(v)) {
|
||||||
|
const { connections: _c, level: _l, ...entity } = v as Record<string, unknown>
|
||||||
|
return { metadata: record.metadata, vector: entity }
|
||||||
|
}
|
||||||
|
return { metadata: record.metadata, vector: v }
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Stable content hash of a stored record for diffing — key-sorted JSON so
|
* Stable content hash of a stored record for diffing — key-sorted JSON so
|
||||||
* property order can never fake a divergence.
|
* property order can never fake a divergence.
|
||||||
|
|
|
||||||
320
tests/integration/embed-markers-in-log.test.ts
Normal file
320
tests/integration/embed-markers-in-log.test.ts
Normal file
|
|
@ -0,0 +1,320 @@
|
||||||
|
/**
|
||||||
|
* @module tests/integration/embed-markers-in-log
|
||||||
|
* @description DEFERRED-EMBED MARKERS ARE LOG RECORDS — the sidecar is dead.
|
||||||
|
* The pending-embed lifecycle lives IN the generation log as first-class v2
|
||||||
|
* records: `embed.pending` rides the deferred write's OWN commit fact (same
|
||||||
|
* generation, one atomic append — a marker can never be orphaned from its
|
||||||
|
* write nor the write from its marker) and `embed.landed` rides the
|
||||||
|
* background worker's landing commit. Recovery is REPLAY, NOT LISTING: the
|
||||||
|
* open-time fold arms every pending without a matching landed (minus rows
|
||||||
|
* the log later tombstoned). The pins:
|
||||||
|
*
|
||||||
|
* (a) SAME-FACT ATOMICITY: a deferred add's commit fact carries the
|
||||||
|
* embed.pending record BESIDE its noun after-image — one generation,
|
||||||
|
* one frame — and no sidecar file is ever written.
|
||||||
|
* (b) LANDING: after the barrier, the log carries embed.landed (inline
|
||||||
|
* vector, per the v2 format) riding the landing commit's own fact, and
|
||||||
|
* a fresh fold of the whole log nets ZERO pending.
|
||||||
|
* (c) CRASH RECOVERY VIA THE LOG: kill mid-defer (hung embedder, flushed
|
||||||
|
* durability, crash-style abandon), reopen — the fold re-arms exactly
|
||||||
|
* one pending with NO sidecar file existing anywhere, and the vector
|
||||||
|
* then lands.
|
||||||
|
* (d) LEGACY BRIDGE: a sidecar marker file left by a pre-log build is
|
||||||
|
* folded in at open, migrated into the log as an embed.pending record,
|
||||||
|
* and the file is deleted — one-time, durable, idempotent.
|
||||||
|
* (e) VFS ACK LAW (unchanged contract, new mechanism): writeFile acks
|
||||||
|
* under a forever-hung embedder while its pending marker sits durably
|
||||||
|
* in the log.
|
||||||
|
*/
|
||||||
|
import { describe, it, expect, afterEach, vi } from 'vitest'
|
||||||
|
import * as fs from 'node:fs'
|
||||||
|
import * as path from 'node:path'
|
||||||
|
import * as zlib from 'node:zlib'
|
||||||
|
import { Brainy } from '../../src/brainy.js'
|
||||||
|
import { NounType } from '../../src/types/graphTypes.js'
|
||||||
|
import type { CommitFact } from '../../src/db/factLog.js'
|
||||||
|
import {
|
||||||
|
makeTempDir,
|
||||||
|
openBrain,
|
||||||
|
abandonAsCrashed,
|
||||||
|
vec,
|
||||||
|
uid
|
||||||
|
} from '../helpers/durabilityKillMatrix.js'
|
||||||
|
|
||||||
|
/** The retired sidecar prefix — asserted ABSENT (or bridged away) on disk. */
|
||||||
|
const SIDECAR_DIR = ['_system', 'pending_embeds'] as const
|
||||||
|
|
||||||
|
const sidecarDir = (dir: string): string => path.join(dir, ...SIDECAR_DIR)
|
||||||
|
|
||||||
|
/** Every committed fact in the brain's log, generation-ascending. */
|
||||||
|
async function allFacts(brain: Brainy): Promise<CommitFact[]> {
|
||||||
|
const scan = (
|
||||||
|
brain as unknown as {
|
||||||
|
scanFacts(o?: { fromGeneration?: number }): {
|
||||||
|
batches(): AsyncGenerator<{ facts: CommitFact[] }>
|
||||||
|
} | null
|
||||||
|
}
|
||||||
|
).scanFacts({ fromGeneration: 1 })
|
||||||
|
expect(scan, 'filesystem storage hosts a fact log').not.toBeNull()
|
||||||
|
const facts: CommitFact[] = []
|
||||||
|
for await (const batch of scan!.batches()) facts.push(...batch.facts)
|
||||||
|
return facts
|
||||||
|
}
|
||||||
|
|
||||||
|
/** The recovery fold, reimplemented independently: pending arms, landed
|
||||||
|
* disarms, a noun tombstone disarms (a deleted row owes no vector). */
|
||||||
|
function foldPending(facts: CommitFact[]): Set<string> {
|
||||||
|
const pending = new Set<string>()
|
||||||
|
for (const fact of facts) {
|
||||||
|
for (const record of fact.records ?? []) {
|
||||||
|
if (record.type === 'embed.pending') pending.add(record.id)
|
||||||
|
else if (record.type === 'embed.landed') pending.delete(record.id)
|
||||||
|
}
|
||||||
|
for (const op of fact.ops) {
|
||||||
|
if (op.kind === 'noun' && op.record === null) pending.delete(op.id)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return pending
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Hang the embedder forever (the ack-law adversary). */
|
||||||
|
function hangEmbedder(brain: Brainy): ReturnType<typeof vi.spyOn> {
|
||||||
|
return vi
|
||||||
|
.spyOn(brain as unknown as { embed(d: unknown): Promise<number[]> }, 'embed')
|
||||||
|
.mockImplementation(() => new Promise<number[]>(() => {}))
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Abandon a hung worker pass (its embed promise never resolves; production
|
||||||
|
* is covered by the worker's 60s hang guard — the test takes the white-box
|
||||||
|
* shortcut for speed, same idiom as the deferred-embedding suite). */
|
||||||
|
function abandonHungWorker(brain: Brainy): void {
|
||||||
|
;(brain as unknown as { _embedWorkerFlight: Promise<void> | null })._embedWorkerFlight = null
|
||||||
|
}
|
||||||
|
|
||||||
|
describe('deferred-embed markers in the log — the sidecar is dead', () => {
|
||||||
|
const dirs: string[] = []
|
||||||
|
const brains: Brainy[] = []
|
||||||
|
|
||||||
|
const trackDir = (): string => {
|
||||||
|
const dir = makeTempDir()
|
||||||
|
dirs.push(dir)
|
||||||
|
return dir
|
||||||
|
}
|
||||||
|
const track = (brain: Brainy): Brainy => {
|
||||||
|
brains.push(brain)
|
||||||
|
return brain
|
||||||
|
}
|
||||||
|
|
||||||
|
afterEach(async () => {
|
||||||
|
vi.restoreAllMocks()
|
||||||
|
for (const b of brains.splice(0)) {
|
||||||
|
abandonHungWorker(b)
|
||||||
|
await b.close().catch(() => {})
|
||||||
|
}
|
||||||
|
for (const d of dirs.splice(0)) fs.rmSync(d, { recursive: true, force: true })
|
||||||
|
})
|
||||||
|
|
||||||
|
it('(a) SAME-FACT ATOMICITY: the deferred add\'s ONE commit fact carries embed.pending beside its after-image; no sidecar file exists', async () => {
|
||||||
|
const dir = trackDir()
|
||||||
|
const brain = track(await openBrain(dir))
|
||||||
|
hangEmbedder(brain) // hold the pending state open for the scan
|
||||||
|
|
||||||
|
const id = await brain.add({
|
||||||
|
data: 'deferred content whose marker rides the fact',
|
||||||
|
type: NounType.Document,
|
||||||
|
deferEmbedding: true,
|
||||||
|
metadata: { pin: 'a' }
|
||||||
|
})
|
||||||
|
expect(brain.pendingEmbedCount()).toBe(1)
|
||||||
|
|
||||||
|
const facts = await allFacts(brain)
|
||||||
|
const carrying = facts.filter((f) =>
|
||||||
|
(f.records ?? []).some((r) => r.type === 'embed.pending' && r.id === id)
|
||||||
|
)
|
||||||
|
expect(carrying, 'exactly ONE fact carries the pending marker').toHaveLength(1)
|
||||||
|
const fact = carrying[0]
|
||||||
|
// The SAME fact (same generation, one atomic append) carries the write's
|
||||||
|
// own after-image — marker and write are inseparable by construction.
|
||||||
|
const afterImage = fact.ops.find((op) => op.kind === 'noun' && op.id === id)
|
||||||
|
expect(afterImage, 'the marker rides the write\'s own fact').toBeDefined()
|
||||||
|
expect(afterImage!.record, 'an after-image, not a tombstone').not.toBeNull()
|
||||||
|
const marker = (fact.records ?? []).find((r) => r.type === 'embed.pending' && r.id === id)
|
||||||
|
expect(marker && marker.type === 'embed.pending' && marker.enqueuedAt).toBeGreaterThan(0)
|
||||||
|
|
||||||
|
// The sidecar is dead: nothing under the retired prefix, ever.
|
||||||
|
expect(fs.existsSync(sidecarDir(dir)), 'no sidecar directory is created').toBe(false)
|
||||||
|
})
|
||||||
|
|
||||||
|
it('(b) LANDING: after the barrier the log carries embed.landed (inline vector) on the landing commit\'s own fact, and a fresh fold nets zero pending', async () => {
|
||||||
|
const dir = trackDir()
|
||||||
|
const brain = track(await openBrain(dir))
|
||||||
|
|
||||||
|
const id = await brain.add({
|
||||||
|
data: 'content that lands in the background',
|
||||||
|
type: NounType.Document,
|
||||||
|
deferEmbedding: true,
|
||||||
|
metadata: { pin: 'b' }
|
||||||
|
})
|
||||||
|
await brain.awaitPendingEmbeds()
|
||||||
|
expect(brain.pendingEmbedCount()).toBe(0)
|
||||||
|
|
||||||
|
const facts = await allFacts(brain)
|
||||||
|
const landingFacts = facts.filter((f) =>
|
||||||
|
(f.records ?? []).some((r) => r.type === 'embed.landed' && r.id === id)
|
||||||
|
)
|
||||||
|
expect(landingFacts, 'exactly ONE landing fact').toHaveLength(1)
|
||||||
|
const landed = (landingFacts[0].records ?? []).find(
|
||||||
|
(r) => r.type === 'embed.landed' && r.id === id
|
||||||
|
)
|
||||||
|
expect(landed && landed.type === 'embed.landed' && landed.vector.length).toBeGreaterThan(0)
|
||||||
|
// The landing commit's own after-image rides the same fact — the worker's
|
||||||
|
// vector swap and its durable "pending consumed" are one atomic append.
|
||||||
|
const landingAfterImage = landingFacts[0].ops.find((op) => op.kind === 'noun' && op.id === id)
|
||||||
|
expect(landingAfterImage, 'the landed marker rides the swap\'s own fact').toBeDefined()
|
||||||
|
expect(landingAfterImage!.record).not.toBeNull()
|
||||||
|
|
||||||
|
// A fresh fold of the WHOLE log — the exact recovery computation — nets zero.
|
||||||
|
expect(foldPending(facts).size).toBe(0)
|
||||||
|
expect(fs.existsSync(sidecarDir(dir))).toBe(false)
|
||||||
|
})
|
||||||
|
|
||||||
|
it('(c) CRASH RECOVERY VIA THE LOG: kill mid-defer, reopen — one pending re-armed from the fold, NO sidecar file anywhere, and the vector then lands', async () => {
|
||||||
|
const dir = trackDir()
|
||||||
|
|
||||||
|
// Session 1: embedder hung, deferred add acked, durability flushed, then
|
||||||
|
// a crash-style abandon (RAM gone, no close, no background machinery).
|
||||||
|
const first = await openBrain(dir)
|
||||||
|
brains.push(first)
|
||||||
|
hangEmbedder(first)
|
||||||
|
const id = await first.add({
|
||||||
|
data: 'survives the kill through the log',
|
||||||
|
type: NounType.Document,
|
||||||
|
deferEmbedding: true,
|
||||||
|
metadata: { pin: 'c' }
|
||||||
|
})
|
||||||
|
expect(first.pendingEmbedCount()).toBe(1)
|
||||||
|
await first.flush() // the durability barrier: fact (with marker) + manifest
|
||||||
|
expect(fs.existsSync(sidecarDir(dir)), 'no sidecar before the kill').toBe(false)
|
||||||
|
await abandonAsCrashed(first)
|
||||||
|
brains.splice(brains.indexOf(first), 1)
|
||||||
|
vi.restoreAllMocks()
|
||||||
|
|
||||||
|
// Session 2: recovery folds the log — embedder hung BEFORE init so the
|
||||||
|
// re-armed pending is observable, not raced away by the fast worker.
|
||||||
|
const second = new Brainy({
|
||||||
|
requireSubtype: false,
|
||||||
|
storage: { type: 'filesystem', path: dir },
|
||||||
|
silent: true,
|
||||||
|
persistence: { policy: 'manual' }
|
||||||
|
})
|
||||||
|
const hang = hangEmbedder(second)
|
||||||
|
await second.init()
|
||||||
|
track(second)
|
||||||
|
expect(second.pendingEmbedCount(), 'the fold re-armed the pending').toBe(1)
|
||||||
|
expect(fs.existsSync(sidecarDir(dir)), 'recovery used the LOG, not files').toBe(false)
|
||||||
|
|
||||||
|
// Un-hang and drain: a crash DELAYED the vector, never lost it.
|
||||||
|
hang.mockRestore()
|
||||||
|
abandonHungWorker(second)
|
||||||
|
await second.awaitPendingEmbeds()
|
||||||
|
expect(second.pendingEmbedCount()).toBe(0)
|
||||||
|
const after = await second.get(id, { includeVectors: true })
|
||||||
|
expect(after, 'the deferred row survived the crash').toBeTruthy()
|
||||||
|
expect((after!.vector as number[]).length, 'the delayed vector landed').toBeGreaterThan(0)
|
||||||
|
expect(foldPending(await allFacts(second)).size, 'the landing is durable in the log').toBe(0)
|
||||||
|
})
|
||||||
|
|
||||||
|
it('(d) LEGACY BRIDGE: a pre-log sidecar marker folds in at open, migrates into the log, and the file dies — one-time and durable', async () => {
|
||||||
|
const dir = trackDir()
|
||||||
|
|
||||||
|
// Session 1: a normal committed row (the entity the legacy marker names).
|
||||||
|
const first = await openBrain(dir)
|
||||||
|
brains.push(first)
|
||||||
|
const id = uid('legacy-defer')
|
||||||
|
await first.add({
|
||||||
|
id,
|
||||||
|
data: 'legacy deferred content',
|
||||||
|
type: NounType.Document,
|
||||||
|
vector: vec(9),
|
||||||
|
metadata: { pin: 'd' }
|
||||||
|
})
|
||||||
|
await first.flush()
|
||||||
|
await first.close()
|
||||||
|
brains.splice(brains.indexOf(first), 1)
|
||||||
|
|
||||||
|
// A pre-log build's sidecar marker, hand-written exactly as the old
|
||||||
|
// writeRawObject persisted it (the filesystem adapter compresses raw
|
||||||
|
// objects by default: gzipped JSON at `<path>.gz`).
|
||||||
|
fs.mkdirSync(sidecarDir(dir), { recursive: true })
|
||||||
|
const sidecarFile = path.join(sidecarDir(dir), id)
|
||||||
|
fs.writeFileSync(
|
||||||
|
`${sidecarFile}.gz`,
|
||||||
|
zlib.gzipSync(JSON.stringify({ id, enqueuedAt: 1234567890 }, null, 2))
|
||||||
|
)
|
||||||
|
|
||||||
|
// Session 2: the bridge fires at open. Embedder hung BEFORE init so the
|
||||||
|
// folded pending is observable.
|
||||||
|
const second = new Brainy({
|
||||||
|
requireSubtype: false,
|
||||||
|
storage: { type: 'filesystem', path: dir },
|
||||||
|
silent: true,
|
||||||
|
persistence: { policy: 'manual' }
|
||||||
|
})
|
||||||
|
const hang = hangEmbedder(second)
|
||||||
|
await second.init()
|
||||||
|
track(second)
|
||||||
|
expect(second.pendingEmbedCount(), 'the legacy marker folded in').toBe(1)
|
||||||
|
expect(fs.existsSync(sidecarFile), 'the sidecar file was deleted').toBe(false)
|
||||||
|
expect(fs.existsSync(`${sidecarFile}.gz`), 'the compressed variant too').toBe(false)
|
||||||
|
const migrated = await allFacts(second)
|
||||||
|
expect(
|
||||||
|
migrated.some((f) => (f.records ?? []).some((r) => r.type === 'embed.pending' && r.id === id)),
|
||||||
|
'the marker now lives IN the log'
|
||||||
|
).toBe(true)
|
||||||
|
|
||||||
|
// Drain: the bridged pending embeds and lands like any other.
|
||||||
|
hang.mockRestore()
|
||||||
|
abandonHungWorker(second)
|
||||||
|
await second.awaitPendingEmbeds()
|
||||||
|
expect(second.pendingEmbedCount()).toBe(0)
|
||||||
|
const facts = await allFacts(second)
|
||||||
|
expect(
|
||||||
|
facts.some((f) => (f.records ?? []).some((r) => r.type === 'embed.landed' && r.id === id)),
|
||||||
|
'the bridged pending landed durably'
|
||||||
|
).toBe(true)
|
||||||
|
expect(foldPending(facts).size).toBe(0)
|
||||||
|
await second.flush()
|
||||||
|
await second.close()
|
||||||
|
brains.splice(brains.indexOf(second), 1)
|
||||||
|
|
||||||
|
// Session 3: nothing resurrects — the bridge was one-time, the clear durable.
|
||||||
|
const third = track(await openBrain(dir))
|
||||||
|
expect(third.pendingEmbedCount(), 'no zombie pending on the next open').toBe(0)
|
||||||
|
expect(fs.existsSync(sidecarDir(dir)) && fs.readdirSync(sidecarDir(dir)).length > 0).toBe(false)
|
||||||
|
})
|
||||||
|
|
||||||
|
it('(e) VFS ACK LAW: writeFile acks under a forever-hung embedder while its pending marker sits durably in the log', async () => {
|
||||||
|
const dir = trackDir()
|
||||||
|
const brain = track(await openBrain(dir))
|
||||||
|
const hang = hangEmbedder(brain)
|
||||||
|
|
||||||
|
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)
|
||||||
|
|
||||||
|
// The marker is already durable IN the log while the embedder hangs —
|
||||||
|
// the exact state a crash here would recover from.
|
||||||
|
expect(foldPending(await allFacts(brain)).size).toBeGreaterThanOrEqual(1)
|
||||||
|
expect(fs.existsSync(sidecarDir(dir))).toBe(false)
|
||||||
|
|
||||||
|
// Un-hang, abandon the poisoned pass, drain, verify.
|
||||||
|
hang.mockRestore()
|
||||||
|
abandonHungWorker(brain)
|
||||||
|
await brain.awaitPendingEmbeds()
|
||||||
|
expect(brain.pendingEmbedCount()).toBe(0)
|
||||||
|
expect(foldPending(await allFacts(brain)).size).toBe(0)
|
||||||
|
})
|
||||||
|
})
|
||||||
|
|
@ -174,7 +174,15 @@ describe('fact log v2 cutover — live writes land in the v2 segment format', ()
|
||||||
expect(op.kind).toBe('noun')
|
expect(op.kind).toBe('noun')
|
||||||
const canonical = await internals(reopened).storage.readNounRaw(id)
|
const canonical = await internals(reopened).storage.readNounRaw(id)
|
||||||
expect(op.record!.metadata).toStrictEqual(canonical.metadata)
|
expect(op.record!.metadata).toStrictEqual(canonical.metadata)
|
||||||
expect(op.record!.vector).toStrictEqual(canonical.vector)
|
// ENTITY TRUTH comparison: canonical wrappers denormalize HNSW residue
|
||||||
|
// (connections + the randomly-assigned level) that the log record
|
||||||
|
// deliberately reconstructs empty — strip both sides (the oracle's
|
||||||
|
// normalizer law) so a nonzero random level can't fake a divergence.
|
||||||
|
const strip = (w: unknown) => {
|
||||||
|
const { connections: _c, level: _l, ...rest } = w as Record<string, unknown>
|
||||||
|
return rest
|
||||||
|
}
|
||||||
|
expect(strip(op.record!.vector)).toStrictEqual(strip(canonical.vector))
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -33,6 +33,10 @@ const MANUAL_ONLY = new Set<string>([
|
||||||
// Conformance suites run as an explicit gate stage (both engines run them
|
// Conformance suites run as an explicit gate stage (both engines run them
|
||||||
// by direct invocation), never swept into the unit/integration configs.
|
// by direct invocation), never swept into the unit/integration configs.
|
||||||
'tests/conformance/collider-fidelity.test.ts',
|
'tests/conformance/collider-fidelity.test.ts',
|
||||||
|
// Golden-log fold-conformance oracle: the two-implementation contract pin
|
||||||
|
// (byte + fold digests) — runs in the explicit conformance gate stage,
|
||||||
|
// same invocation family as the other conformance suites.
|
||||||
|
'tests/conformance/golden-log-fold.test.ts',
|
||||||
'tests/api/performance-benchmarks.test.ts',
|
'tests/api/performance-benchmarks.test.ts',
|
||||||
'tests/critical-neural-validation.test.ts',
|
'tests/critical-neural-validation.test.ts',
|
||||||
'tests/critical-performance-benchmark.test.ts',
|
'tests/critical-performance-benchmark.test.ts',
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue