/** * @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) }) })