diff --git a/src/brainy.ts b/src/brainy.ts index fff176fd..20dccbfa 100644 --- a/src/brainy.ts +++ b/src/brainy.ts @@ -178,7 +178,7 @@ import { type ImportResult } from './db/portableGraph.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 { ENTITY_TREE_STAMP_PATH, readFamilyStamp, @@ -201,6 +201,7 @@ import { runLogCompletenessOracle, flipToLogAuthority, recordDigest, + nounEntityTruth, type LogAuthorityRecord, type LogAuthorityStorage, type OracleReport @@ -382,6 +383,13 @@ interface PlannedTransact { * rejected batch (CAS conflict, failed apply) emits nothing. */ 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 implements BrainyInterface { private _persistIdleTimer: ReturnType | null = null private _persistBackgroundFlight: Promise | null = null - // DEFERRED EMBEDDING (MT5): durable pending markers under - // _system/pending_embeds/, mirrored in-memory, drained by ONE - // background worker. A crash can delay a vector, never lose one. + // DEFERRED EMBEDDING (MT5): pending markers are LOG RECORDS — an + // embed.pending record rides the deferred write's own commit fact and + // 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() private _embedWorkerFlight: Promise | null = null @@ -1494,17 +1505,18 @@ export class Brainy implements BrainyInterface { } } - // 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. + // MT5 crash recovery — REPLAY, NOT LISTING: the pending-embed markers + // live IN the generation log (embed.pending rides the deferred write's + // own fact; embed.landed rides the landing commit), so recovery folds + // 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) { 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) - } + await this.bridgeLegacyPendingEmbedSidecars() + await this.recoverPendingEmbedsFromLog() if (this._pendingEmbedIds.size > 0) { prodLog.info( `[Brainy] ${this._pendingEmbedIds.size} deferred embed(s) pending from a previous ` + @@ -1515,8 +1527,8 @@ export class Brainy implements BrainyInterface { } } catch (err) { prodLog.warn( - `[Brainy] pending-embed recovery listing failed: ${(err as Error).message} — ` + - `markers remain durable; recovery retries next open` + `[Brainy] pending-embed recovery failed: ${(err as Error).message} — ` + + `the log's markers remain durable; recovery retries next open` ) } } @@ -1942,30 +1954,144 @@ export class Brainy implements BrainyInterface { * deletes — the before-image + per-id-chain set. * @param run - The single-op's existing operation batch builder (the * `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/' /** - * @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. + * @description Mark a deferred embed pending (MT5): the id joins the + * in-memory fast-path set and the returned `embed.pending` record is + * threaded onto the deferred write's OWN commit fact — same generation, + * 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 { + private enqueuePendingEmbed(id: string): FactMarkerRecord { this._pendingEmbedIds.add(id) - await this.storage.writeRawObject(`${Brainy.PENDING_EMBED_PREFIX}${id}`, { - id, - enqueuedAt: Date.now() - }) + return { type: 'embed.pending', id, enqueuedAt: Date.now() } } - /** Remove a pending-embed marker (memory + durable), tolerating races. */ - private async clearPendingEmbed(id: string): Promise { + /** + * @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) - 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 { + 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 { + 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 implements BrainyInterface { try { const entity = await this.get(id, { includeVectors: true }) 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 } // Hang guard: a wedged embedder must not block every later pending @@ -2030,20 +2159,29 @@ export class Brainy implements BrainyInterface { ) } 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, this.indexWriteGeneration) - ) - }) - await this.clearPendingEmbed(id) + // The landing commit's fact carries the embed.landed record (vector + // inline, per the v2 format) alongside the row's after-image — the + // durable "this pending is consumed" that recovery's fold reads. + 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, this.indexWriteGeneration) + ) + }, + undefined, + undefined, + [{ type: 'embed.landed', id, vector: newVector }] + ) + this.clearPendingEmbed(id) } catch (err) { prodLog.warn( `[Brainy] deferred embed for ${id} failed: ${(err as Error).message} — marker retained for retry` @@ -2242,7 +2380,8 @@ export class Brainy implements BrainyInterface { touched: { nouns?: string[]; verbs?: string[] }, run: TransactionFunction, precommit?: (before: CommitBeforeImages) => void, - pendingEvents?: PendingChangeEvent[] + pendingEvents?: PendingChangeEvent[], + records?: FactMarkerRecord[] ): Promise<{ generation?: number; timestamp: number; degraded?: string[] }> { // 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 @@ -2257,6 +2396,15 @@ export class Brainy implements BrainyInterface { : precommit 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 // WITHOUT creating a generation. Generation 0 is the freshly-materialized // brain (bootstrap included); the first USER write is generation 1. @@ -2295,6 +2443,7 @@ export class Brainy implements BrainyInterface { receipt = await this.generationStore.commitSingleOp({ touched, precommit: captureAndCheck, + ...(records && records.length > 0 ? { records } : {}), execute: () => this.transactionManager.executeTransaction(run, { timeout: transactTimeoutBudget( @@ -2507,10 +2656,10 @@ export class Brainy implements BrainyInterface { // Get or compute vector // 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. + // pending marker riding the insert's OWN commit fact (same generation, + // one atomic append — a marker-less committed row, the silently-missing- + // vector shape, is structurally impossible). The background worker + // embeds + inserts. const deferringEmbed = params.deferEmbedding === true && !params.vector const vector = deferringEmbed ? [] @@ -2605,11 +2754,13 @@ export class Brainy implements BrainyInterface { } : 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) - } + // MT5: the pending marker RIDES the insert's own commit fact (same + // generation, one atomic append) — threaded to persistSingleOp below. + // A failed commit appends nothing, so no orphaned durable marker can + // exist; the in-memory entry is harmless and reaped by the worker. + const embedMarkers: FactMarkerRecord[] | undefined = deferringEmbed + ? [this.enqueuePendingEmbed(id)] + : undefined const runInsert: TransactionFunction = async (tx) => { // Operation 1: Save metadata FIRST (TypeAwareStorage caching) @@ -2670,7 +2821,7 @@ export class Brainy implements BrainyInterface { const MAX_UPSERT_ATTEMPTS = 10 for (let attempt = 0; ; attempt++) { try { - await this.persistSingleOp({ nouns: [id] }, runInsert, insertPrecommit, addEvents) + await this.persistSingleOp({ nouns: [id] }, runInsert, insertPrecommit, addEvents, embedMarkers) break } catch (err) { if (!(err instanceof InsertPreconditionExistsSignal)) { @@ -3296,10 +3447,11 @@ export class Brainy implements BrainyInterface { updatedMetadata._rev = authoritativeRev + 1 } - // MT5: durable marker BEFORE the commit (orphan-safe direction). - if (deferringEmbed) { - await this.enqueuePendingEmbed(params.id) - } + // MT5: the pending marker rides the update's own commit fact (same + // generation, one atomic append) — threaded to persistSingleOp below. + const embedMarkers: FactMarkerRecord[] | undefined = deferringEmbed + ? [this.enqueuePendingEmbed(params.id)] + : undefined // Execute atomically with transaction system, generation-stamped as one // immutable Model-B generation (before-image = the entity's prior state). @@ -3389,7 +3541,7 @@ export class Brainy implements BrainyInterface { } } ] - : undefined) + : undefined, embedMarkers) // Aggregation hook (outside transaction — derived data). `existing` is // the full get() view — every reserved field top-level — and must be @@ -7962,12 +8114,17 @@ export class Brainy implements BrainyInterface { return runLogCompletenessOracle({ storage: this.storage as unknown as LogAuthorityStorage, 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) => { const raw = await this.storage.readNounRaw(id) 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 implements BrainyInterface { meta: options?.meta, ifAtGeneration: options?.ifAtGeneration, precommit: casPrecommit, + ...(plan.markerRecords.length > 0 ? { records: plan.markerRecords } : {}), execute: async () => { await this.transactionManager.executeTransaction( async (tx) => { @@ -9561,7 +9719,8 @@ export class Brainy implements BrainyInterface { postCommit: [], casUpdates: [], createdNouns: new Set(), - changeEvents: [] + changeEvents: [], + markerRecords: [] } for (const op of ops) { @@ -9757,9 +9916,10 @@ export class Brainy implements BrainyInterface { } if (deferringEmbed) { - // Durable marker BEFORE the batch commits (orphan-safe direction); - // the worker kicks post-commit via the plan hook. - await this.enqueuePendingEmbed(id) + // The pending marker rides the batch's ONE commit fact (same + // generation, one atomic append); the worker kicks post-commit via + // the plan hook. + plan.markerRecords.push(this.enqueuePendingEmbed(id)) plan.postCommit.push(() => this.kickEmbedWorker()) } plan.operations.push( diff --git a/src/db/factLog.ts b/src/db/factLog.ts index c005d74e..9583365a 100644 --- a/src/db/factLog.ts +++ b/src/db/factLog.ts @@ -138,9 +138,11 @@ export interface FactOp { /** * 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 - * manifests, projection notes, bootstrap baselines). Encoder-ready by - * design; nothing produces them yet — the deferred-embed sidecar and blob - * lifecycle remodel onto these records in a later leg. + * manifests, projection notes, bootstrap baselines). The deferred-embedding + * lifecycle PRODUCES types 6/7 today: `embed.pending` rides the deferred + * 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 = | EmbedPendingRecord @@ -741,6 +743,17 @@ export class FactLog { 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, * establish the tail's intact content (torn-tail scan), then TRUNCATE any diff --git a/src/db/generationStore.ts b/src/db/generationStore.ts index 2f623e3b..93a221ce 100644 --- a/src/db/generationStore.ts +++ b/src/db/generationStore.ts @@ -51,7 +51,8 @@ import { storageSupportsFactLog, type CommitFact, type FactOp, - type FactIntMinter + type FactIntMinter, + type FactMarkerRecord } from './factLog.js' import { GenerationSegmentStore, type FoldGeneration } from './generationSegments.js' import { crc32c } from '../utils/crc32c.js' @@ -903,6 +904,8 @@ export class GenerationStore { nouns: string[] verbs: string[] meta?: Record + /** V2 marker records riding this fact (same generation, same append). */ + records?: FactMarkerRecord[] }): Promise { const ops: FactOp[] = [] const afterRecords: GenerationRecord[] = [] @@ -926,7 +929,8 @@ export class GenerationStore { timestamp: args.timestamp, ops, ...(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: * the generation reservation is returned and no staging I/O has happened. */ 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 }): Promise<{ generation: number; timestamp: number }> { return this.withMutex(async () => { @@ -1075,7 +1085,8 @@ export class GenerationStore { timestamp, nouns, 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.sync() @@ -1288,6 +1299,18 @@ export class GenerationStore { touched: { nouns?: string[]; verbs?: string[] } execute: () => Promise 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[] }> { return this.withMutex(async () => { // Refuse to accept a write whose history we cannot make durable: if the @@ -1357,7 +1380,13 @@ export class GenerationStore { // buffered history). if (this.factLog) { 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( @@ -1411,7 +1440,13 @@ export class GenerationStore { if (this.factLog) { try { 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') { await this.factLog.ensureSynced() diff --git a/src/db/logAuthority.ts b/src/db/logAuthority.ts index a148f04e..36cf4880 100644 --- a/src/db/logAuthority.ts +++ b/src/db/logAuthority.ts @@ -92,6 +92,27 @@ export async function readLogAuthority( 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 + 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 * property order can never fake a divergence. diff --git a/tests/integration/embed-markers-in-log.test.ts b/tests/integration/embed-markers-in-log.test.ts new file mode 100644 index 00000000..2dcad2f1 --- /dev/null +++ b/tests/integration/embed-markers-in-log.test.ts @@ -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 { + 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 { + const pending = new Set() + 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 { + return vi + .spyOn(brain as unknown as { embed(d: unknown): Promise }, 'embed') + .mockImplementation(() => new Promise(() => {})) +} + +/** 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 | 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 `.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) + }) +}) diff --git a/tests/integration/fact-log-v2-cutover.test.ts b/tests/integration/fact-log-v2-cutover.test.ts index 6c05ef42..6e8d9fb6 100644 --- a/tests/integration/fact-log-v2-cutover.test.ts +++ b/tests/integration/fact-log-v2-cutover.test.ts @@ -174,7 +174,15 @@ describe('fact log v2 cutover — live writes land in the v2 segment format', () expect(op.kind).toBe('noun') const canonical = await internals(reopened).storage.readNounRaw(id) 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 + return rest + } + expect(strip(op.record!.vector)).toStrictEqual(strip(canonical.vector)) } }) diff --git a/tests/unit/test-suite-coverage-guard.test.ts b/tests/unit/test-suite-coverage-guard.test.ts index 21f918f1..4b078146 100644 --- a/tests/unit/test-suite-coverage-guard.test.ts +++ b/tests/unit/test-suite-coverage-guard.test.ts @@ -33,6 +33,10 @@ const MANUAL_ONLY = new Set([ // Conformance suites run as an explicit gate stage (both engines run them // by direct invocation), never swept into the unit/integration configs. '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/critical-neural-validation.test.ts', 'tests/critical-performance-benchmark.test.ts',