diff --git a/src/db/factLog.ts b/src/db/factLog.ts index 04f466ed..19bbb10e 100644 --- a/src/db/factLog.ts +++ b/src/db/factLog.ts @@ -342,6 +342,34 @@ export class FactLog { * crash between fact-append and the commit point). After open, the log is * exactly the committed prefix. */ + /** + * Read (without truncating) every intact fact ABOVE a generation — the + * log-authority recovery surface: after a crash, facts beyond the + * manifest watermark that survived with valid CRCs are ACKED writes in + * durable-at-ack mode, and the owner REPLAYS them instead of letting + * open() truncate them. Must be called BEFORE open() (it reads the raw + * segments directly; the torn tail's invalid suffix is ignored exactly + * like open() would). + */ + async peekFactsAbove(committedGeneration: number): Promise { + const stored = (await this.storage.readRawObject(FACTS_MANIFEST_PATH)) as FactsManifest | null + if (!stored || typeof stored !== 'object' || !Array.isArray(stored.segments)) return [] + if (stored.formatVersion !== FACTS_FORMAT_VERSION) return [] + const out: CommitFact[] = [] + const files = [...stored.segments.map((s) => s.file)] + if (stored.tailSegment) files.push(stored.tailSegment) + for (const file of files) { + const bytes = await this.storage.readRawBytes(`${FACTS_PREFIX}/${file}`) + if (bytes === null) continue + const { facts } = parseSegment(file, bytes) + for (const f of facts) { + if (f.generation > committedGeneration) out.push(f) + } + } + out.sort((a, b) => a.generation - b.generation) + return out + } + async open(committedGeneration: number): Promise { const stored = (await this.storage.readRawObject(FACTS_MANIFEST_PATH)) as FactsManifest | null if (stored && typeof stored === 'object' && Array.isArray(stored.segments)) { diff --git a/src/db/generationStore.ts b/src/db/generationStore.ts index 5db274b6..663784c6 100644 --- a/src/db/generationStore.ts +++ b/src/db/generationStore.ts @@ -45,6 +45,7 @@ import type { GenerationStorage, TxLogEntry } from './types.js' +import { readLogAuthority } from './logAuthority.js' import { FactLog, storageSupportsFactLog, type CommitFact, type FactOp } from './factLog.js' import { GenerationSegmentStore, type FoldGeneration } from './generationSegments.js' import { crc32c } from '../utils/crc32c.js' @@ -88,12 +89,43 @@ export const GENERATIONS_PREFIX = '_generations' * IS committed); the tx-log append has NOT happened yet. A crash here must * keep the transaction (the tx-log is advisory metadata, not the source of * commit truth). + * - `'transact-after-fact-sync'` — the batch's fact is appended AND fsynced, + * but neither the counter nor the manifest advanced. A crash here must cost + * the whole batch: recovery restores the before-images and open() truncates + * the synced fact back to the manifest watermark. + * + * Single-op (Model-B group-commit) phases — `commitSingleOp`: + * + * - `'singleop-after-execute'` — the live canonical write has applied (tmp+ + * rename, not individually fsynced); no history, fact, or generation record + * exists yet. A crash here must cost only the never-returned ack — the + * baseline stays intact and the log stays at the committed watermark. + * - `'singleop-after-fact-append'` — the fact is appended (and, in at-ack + * mode, fsynced); the manifest never saw the generation. A crash here must + * cost the buffered history + the fact (open() truncates it back), never + * the baseline. + * + * Pending-tier flush phases — `flushPendingSingleOps`: + * + * - `'flush-after-staging'` — the window's record-set dirs are written but not + * fsynced and the manifest never advanced. A crash here must cost only the + * window's HISTORY (drop-without-restore) — the acked live writes stay. + * - `'flush-before-manifest'` — staging is fsynced and the facts are fsynced, + * but the manifest never advanced. A crash here must cost only the window's + * history and its facts (truncated at open) — the acked live writes stay. + * - `'before-manifest-rename'` is ALSO fired by the flush path just before its + * commit point (see `flushPendingSingleOpsUnlocked`). */ export type CommitFaultPhase = | 'after-staging' | 'after-execute' | 'before-manifest-rename' | 'after-manifest-rename' + | 'transact-after-fact-sync' + | 'singleop-after-execute' + | 'singleop-after-fact-append' + | 'flush-after-staging' + | 'flush-before-manifest' /** * @description Identifies which ids a transaction touches, split by kind. @@ -461,6 +493,54 @@ export class GenerationStore { // hosts no fact log (readers fall back to canonical enumeration). if (storageSupportsFactLog(this.storage)) { this.factLog = new FactLog(this.storage) + // LOG-AUTHORITY REPLAY (durable-at-ack's recovery half): when this + // brain's stored authority is the log, an intact fact ABOVE the + // manifest is an ACKED write whose canonical bytes may not have + // survived the crash — its fsynced fact is the ONLY durable copy. + // Truncating it would lose an acked write; instead REPLAY it into + // canonical and advance the manifest to cover it. Tree-authority + // brains keep the truncate contract (their acks never promised the + // fact was durable). Derived indexes reconcile through the normal + // drift machinery at open — same as group-commit recovery. + const authority = await readLogAuthority(this.storage) + if (authority.authority === 'log') { + const orphans = await this.factLog.peekFactsAbove(this.committed) + if (orphans.length > 0) { + for (const fact of orphans) { + for (const op of fact.ops) { + const image = + op.record === null + ? { metadata: null, vector: null } + : { metadata: op.record.metadata, vector: op.record.vector } + if (op.kind === 'verb') await this.storage.writeVerbRaw(op.id, image) + else await this.storage.writeNounRaw(op.id, image) + } + this.committed = fact.generation + this.appendCommittedGen(fact.generation) + this.setDelta(fact.generation, { + nouns: new Set(fact.ops.filter((o) => o.kind === 'noun').map((o) => o.id)), + verbs: new Set(fact.ops.filter((o) => o.kind === 'verb').map((o) => o.id)), + timestamp: fact.timestamp, + bytes: 0 + }) + } + if (this.counter < this.committed) this.counter = this.committed + await this.persistCounterUnlocked() + const manifest: GenerationManifest = { + version: 1, + generation: this.committed, + committedAt: new Date().toISOString(), + horizon: this.horizonGen + } + await this.storage.writeRawObject(MANIFEST_PATH, manifest) + await this.storage.syncRawObjects([MANIFEST_PATH]) + prodLog.warn( + `[GenerationStore] log-authority recovery REPLAYED ${orphans.length} acked ` + + `fact(s) beyond the manifest into canonical (now committed at ${this.committed}) — ` + + `an acked write is never lost` + ) + } + } await this.factLog.open(this.committed) } else { this.factLog = null @@ -977,6 +1057,9 @@ export class GenerationStore { await this.factLog.append(fact) await this.factLog.sync() } + // A crash here must cost the whole batch: the synced fact is truncated + // back at open() and the before-images are restored byte-identically. + faultPoint('transact-after-fact-sync') // -- 5. Counter + manifest rename (COMMIT POINT) ---------------------- await this.persistCounterUnlocked() @@ -1278,6 +1361,12 @@ export class GenerationStore { throw err } this.inTransact = false + // Test-only crash simulation (direct call — a throw propagates with no + // cleanup, exactly like a process death; recovery-on-open restores the + // contract). A crash here must cost only the never-returned ack: the + // live canonical write applied, but no history, fact, or generation + // record exists for it yet. + if (this.commitFaultInjector) this.commitFaultInjector('singleop-after-execute') // Buffer the pending generation + make it instantly visible to reads. this.pendingBuffer.set(gen, { nouns: nounBefore, verbs: verbBefore, timestamp }) @@ -1297,13 +1386,35 @@ export class GenerationStore { // the log's group-commit (many concurrent writers share ONE sync) — // an acked write's fact survives power loss, by contract. if (this.factLog) { - await this.factLog.append( - await this.buildCommitFact({ generation: gen, timestamp, nouns, verbs }) - ) - if (this.logDurability === 'at-ack') { - await this.factLog.ensureSynced() + try { + await this.factLog.append( + await this.buildCommitFact({ generation: gen, timestamp, nouns, verbs }) + ) + if (this.logDurability === 'at-ack') { + await this.factLog.ensureSynced() + } + } catch (err) { + // A rejected write must NOT commit: the generation was buffered + // before the append, so un-buffer it and return the counter + // reservation — otherwise the next flush would durably commit a + // generation with NO fact, a silent log gap a later replay would + // turn into loss. Canonical bytes from execute() remain as an + // uncommitted orphan — identical to a crash at this point; never + // a torn committed state. + this.pendingBuffer.delete(gen) + const idx = this.pendingGens.lastIndexOf(gen) + if (idx !== -1) this.pendingGens.splice(idx, 1) + this.invalidateChains() + if (this.counter === gen) this.counter = gen - 1 + throw err } } + // Test-only crash simulation. A crash here must cost the buffered + // history + the appended fact in 'deferred' mode (open() truncates it + // back to the manifest watermark) — while under 'log' authority the + // intact fact is REPLAYED at open, never the baseline or the applied + // live write. + if (this.commitFaultInjector) this.commitFaultInjector('singleop-after-fact-append') this.schedulePendingFlush() return { generation: gen, timestamp } }) @@ -1422,6 +1533,11 @@ export class GenerationStore { logEntries.push({ generation: gen, timestamp: buf.timestamp }) } + // Test-only crash simulation. A crash here must cost only the window's + // HISTORY: un-fsynced record-set dirs may sit above the manifest, and + // recovery drops them WITHOUT restore — the acked live writes stay. + if (this.commitFaultInjector) this.commitFaultInjector('flush-after-staging') + // ONE fsync for the whole window — the durability-batching win. await this.storage.syncRawObjects(stagedPaths) @@ -1431,6 +1547,12 @@ export class GenerationStore { // generation without its durable fact. await this.factLog?.sync() + // Test-only crash simulation. A crash here must cost only the window's + // history and its (already fsynced) facts — open() truncates the facts + // back to the manifest watermark and drops the staged group-commit dirs + // without restore; the acked live writes stay. + if (this.commitFaultInjector) this.commitFaultInjector('flush-before-manifest') + // Test-only crash simulation: a throwing injector here leaves the staged // group-commit generation dirs on disk with NO manifest advance — the // exact "crashed mid-flush" state recovery must DROP-WITHOUT-RESTORE diff --git a/tests/helpers/durabilityKillMatrix.ts b/tests/helpers/durabilityKillMatrix.ts new file mode 100644 index 00000000..219c9084 --- /dev/null +++ b/tests/helpers/durabilityKillMatrix.ts @@ -0,0 +1,200 @@ +/** + * @module tests/helpers/durabilityKillMatrix + * @description Shared machinery for the durability kill-matrix suite + * (tests/integration/durability-kill-matrix.test.ts): open filesystem brains + * with fully explicit durability (no background cadence, no embedder), arm + * the generation store's test-only commit fault injector at one exact phase, + * abandon a "crashed" brain the way a dead process would (its RAM is gone, + * nothing flushes, nothing closes), and read the fact log / on-disk state the + * recovery assertions pin. + * + * The crash model is PROCESS DEATH: in-memory state is lost, file bytes the + * process already handed to the OS survive. One helper additionally models + * POWER LOSS for a chosen entity by removing its canonical files — legal, + * because single-op canonical writes are tmp+rename WITHOUT fsync, and a + * rename that was never fsynced may surface as "no directory entry" after + * power loss. + */ +import * as fs from 'node:fs' +import * as os from 'node:os' +import * as path from 'node:path' +import { Brainy } from '../../src/brainy.js' +import type { CommitFaultPhase, GenerationStore } from '../../src/db/generationStore.js' + +/** The error a throwing fault injector uses to simulate a process crash. */ +export class SimulatedCrash extends Error { + constructor(phase: CommitFaultPhase) { + super(`simulated process crash at ${phase}`) + this.name = 'SimulatedCrash' + } +} + +/** Deterministic 384-dim vector so no test ever invokes the embedder. */ +export function vec(seed: number): number[] { + return Array.from({ length: 384 }, (_, i) => ((seed * 31 + i * 7) % 100) / 100) +} + +/** + * Map a readable label to a deterministic UUID-shaped id (entity ids must be + * UUIDs — the sharded storage layout derives the shard from the UUID hex). + */ +export function uid(label: string): string { + let h1 = 0x811c9dc5 + for (let i = 0; i < label.length; i++) { + h1 = Math.imul(h1 ^ label.charCodeAt(i), 0x01000193) >>> 0 + } + let h2 = 0xdeadbeef + for (let i = label.length - 1; i >= 0; i--) { + h2 = Math.imul(h2 ^ label.charCodeAt(i), 0x85ebca6b) >>> 0 + } + const hex = h1.toString(16).padStart(8, '0') + h2.toString(16).padStart(8, '0') + return `00000000-0000-4000-8000-${hex.slice(0, 12)}` +} + +/** Create a fresh temp directory for one brain's storage root. */ +export function makeTempDir(): string { + return fs.mkdtempSync(path.join(os.tmpdir(), 'brainy-kill-matrix-')) +} + +/** + * Open a writer brain over `dir` with every implicit durability knob off: + * persistence policy 'manual' (the engine never flushes on its own, so every + * durable transition in a test is an explicit `flush()`/commit), deterministic + * embeddings (tests always pass explicit vectors anyway), silent logs. + */ +export async function openBrain(dir: string): Promise { + process.env.BRAINY_DETERMINISTIC_EMBEDDINGS = 'true' + const brain = new Brainy({ + requireSubtype: false, + storage: { type: 'filesystem', path: dir }, + silent: true, + persistence: { policy: 'manual' } + }) + await brain.init() + return brain +} + +/** Typed access to the brain's private generation store (test injection point). */ +export function storeOf(brain: Brainy): GenerationStore { + return (brain as unknown as { generationStore: GenerationStore }).generationStore +} + +/** + * Arm the commit fault injector to simulate a process crash at EXACTLY one + * phase (all other phases pass through untouched). Returns the list of phases + * observed before (and including) the trip, so a test can assert the fault + * actually fired where intended. + */ +export function armCrash(brain: Brainy, phase: CommitFaultPhase): { fired: CommitFaultPhase[] } { + const fired: CommitFaultPhase[] = [] + storeOf(brain).setCommitFaultInjector((p) => { + fired.push(p) + if (p === phase) { + throw new SimulatedCrash(p) + } + }) + return { fired } +} + +/** + * Abandon a crashed brain the way process death would: its buffered RAM state + * is discarded and no background machinery may ever touch the storage + * directory again (a dead process cannot flush). The fault injector stays + * installed so any in-flight commit path still "crashes". Serialized behind + * the store's commit mutex so an interleaved background flush cannot be + * severed mid-section. + * + * NEVER calls close() — graceful close is exactly what a crash denies. + */ +export async function abandonAsCrashed(brain: Brainy): Promise { + const store = storeOf(brain) as unknown as { + withMutex(fn: () => Promise): Promise + clearPendingFlushTimer(): void + pendingGens: number[] + pendingBuffer: Map + } + await store.withMutex(async () => { + store.clearPendingFlushTimer() + store.pendingGens = [] + store.pendingBuffer.clear() + }) +} + +/** + * Every generation present in the brain's fact log, ascending — the suite's + * "what does the log claim is committed" probe. Empty when no fact log exists. + * A scan abort (gap detection) propagates — callers that PIN gap behavior + * catch it themselves. + */ +export async function factGenerations(brain: Brainy): Promise { + const scan = brain.scanFacts({ fromGeneration: 1 }) + if (!scan) return [] + const gens: number[] = [] + for await (const batch of scan.batches()) { + for (const fact of batch.facts) gens.push(fact.generation) + } + return gens.sort((a, b) => a - b) +} + +/** An ENOSPC-shaped error, matching what a full disk surfaces from node:fs. */ +export function enospcError(): NodeJS.ErrnoException { + const err = new Error("ENOSPC: no space left on device, write") as NodeJS.ErrnoException + err.code = 'ENOSPC' + err.errno = -28 + err.syscall = 'write' + return err +} + +/** + * Make the storage adapter's next raw-byte append (the fact-log append path) + * fail once with ENOSPC, then restore the original — "the disk filled for one + * append, then space was freed". Returns a probe telling how many appends + * were failed. + */ +export function failNextAppendWithEnospc(brain: Brainy): { failed: () => number } { + const storage = (brain as unknown as { + storage: { appendRawBytes(p: string, b: Uint8Array): Promise } + }).storage + const original = storage.appendRawBytes.bind(storage) + let failures = 0 + storage.appendRawBytes = async (p: string, b: Uint8Array): Promise => { + storage.appendRawBytes = original + failures++ + throw enospcError() + } + return { failed: () => failures } +} + +/** + * POWER-LOSS MODEL for one entity: remove its canonical noun files from the + * storage root. Legal disk state — a single-op write's canonical bytes are + * tmp+rename WITHOUT fsync (only `transact()` runs the write barrier), and an + * un-fsynced rename may resolve to "no directory entry" after power loss. + * Throws when nothing was removed (the caller's premise would be wrong). + */ +export function dropCanonicalNoun(dir: string, id: string): void { + const removed: string[] = [] + const walk = (p: string): void => { + for (const entry of fs.readdirSync(p, { withFileTypes: true })) { + const full = path.join(p, entry.name) + if (entry.isDirectory()) { + if (entry.name === id) { + fs.rmSync(full, { recursive: true, force: true }) + removed.push(full) + } else { + walk(full) + } + } + } + } + const nounsRoot = path.join(dir, 'entities', 'nouns') + if (fs.existsSync(nounsRoot)) walk(nounsRoot) + if (removed.length === 0) { + throw new Error(`power-loss model: no canonical files found for noun ${id} under ${nounsRoot}`) + } +} + +/** True when the staged record-set directory for `gen` exists on disk. */ +export function generationDirExists(dir: string, gen: number): boolean { + return fs.existsSync(path.join(dir, '_generations', String(gen))) +} diff --git a/tests/integration/durability-kill-matrix.test.ts b/tests/integration/durability-kill-matrix.test.ts new file mode 100644 index 00000000..1e543bc1 --- /dev/null +++ b/tests/integration/durability-kill-matrix.test.ts @@ -0,0 +1,633 @@ +/** + * @module tests/integration/durability-kill-matrix + * @description THE DURABILITY KILL MATRIX — for every step of the commit + * path, inject a crash AT that step (the generation store's test-only fault + * injector), then reopen the same storage directory with a brand-new Brainy + * and assert the recovery contract BY CONSTRUCTION, not by timing: + * + * - an ACKED write survives the crash (never a lost ack), and + * - an UN-ACKED write leaves no torn state (fully present or fully absent, + * never half). + * + * The crash simulation is honest process death: the crashed brain is NEVER + * closed — `abandonAsCrashed` discards its buffered RAM state exactly as a + * dead process would, and recovery on the next open is the only repair that + * runs. File bytes already handed to the OS survive (process-crash model); + * one row additionally models POWER LOSS by removing an entity's un-fsynced + * canonical files (legal: single-op canonical writes are tmp+rename without + * fsync). + * + * Matrix rows (fault point → durability barrier position): + * + * BEFORE the barrier (nothing durable records the write): + * singleop-after-execute · singleop-after-fact-append · flush-after-staging + * AFTER partial durability (staged/synced bytes exist, manifest did not advance): + * flush-before-manifest · before-manifest-rename (transact) · + * transact-after-fact-sync + * AFTER the commit point: + * after-manifest-rename (transact) + * MODE VARIANTS: singleop-after-fact-append under durable-at-ack. + * DISK FULL: one ENOSPC'd append — loud typed rejection, reads keep + * serving, a later write succeeds. + * + * Where the observed recovery contract differs from the ideal, the pin states + * the OBSERVED behavior with a comment; where the observed behavior violates + * "never a torn state / never a lost ack", the pin asserts the CONTRACT and + * is marked `.fails` — a release-blocking finding, deliberately not weakened. + */ +import { describe, it, expect, afterEach } from 'vitest' +import * as fs from 'node:fs' +import { Brainy } from '../../src/brainy.js' +import { NounType } from '../../src/types/graphTypes.js' +import { + abandonAsCrashed, + armCrash, + dropCanonicalNoun, + factGenerations, + failNextAppendWithEnospc, + generationDirExists, + makeTempDir, + openBrain, + storeOf, + uid, + vec +} from '../helpers/durabilityKillMatrix.js' + +describe('durability kill matrix — crash at every commit-path step, recover by reopen', () => { + const dirs: string[] = [] + const liveBrains: Brainy[] = [] + // Crashed brains are deliberately NEVER closed (a dead process cannot + // close); they are severed by abandonAsCrashed inside each test. + + function trackDir(): string { + const dir = makeTempDir() + dirs.push(dir) + return dir + } + + async function openLive(dir: string): Promise { + const brain = await openBrain(dir) + liveBrains.push(brain) + return brain + } + + afterEach(async () => { + for (const brain of liveBrains.splice(0)) { + try { + await brain.close() + } catch { + // already closed / crashed mid-close — teardown only + } + } + for (const dir of dirs.splice(0)) { + await fs.promises.rm(dir, { recursive: true, force: true }) + } + }) + + /** Baseline arrangement: one durable row + explicit flush = the durable floor. */ + async function arrangeBaseline(label: string): Promise<{ + dir: string + brain: Brainy + baselineId: string + floor: number + }> { + const dir = trackDir() + const brain = await openBrain(dir) // NOT tracked live — most rows crash it + const baselineId = uid(`${label}-baseline`) + await brain.add({ + id: baselineId, + data: 'baseline row', + type: NounType.Document, + vector: vec(1), + metadata: { v: 1 } + }) + await brain.flush() + return { dir, brain, baselineId, floor: storeOf(brain).committedGeneration() } + } + + /** + * Flip a brain to durable-at-ack (log-authority) mode. + * + * NOT via `adoptLogAuthority()`: the sanctioned flip REFUSES on a freshly + * materialized brain — its verification oracle reports the generation-0 + * VFS-root baseline as a divergence (`state-differs` even after an + * identity-update backfill; verified 2026-08-10). This helper flips the + * SAME switch the sanctioned path flips (`setLogDurability('at-ack')`) and + * persists the SAME authority artifact, so a reopened brain also runs in + * log-authority mode. The durability semantics under test are governed + * entirely by that switch. + */ + async function flipToAtAck(brain: Brainy): Promise { + const storage = ( + brain as unknown as { + storage: { + writeRawObject(p: string, d: unknown): Promise + syncRawObjects(p: string[]): Promise + } + } + ).storage + await storage.writeRawObject('_system/log-authority.json', { + authority: 'log', + flippedAt: Date.now() + }) + await storage.syncRawObjects(['_system/log-authority.json']) + storeOf(brain).setLogDurability('at-ack') + } + + // ========================================================================== + // Rows BEFORE the durability barrier — the write never became durable-acked + // ========================================================================== + + it('singleop-after-execute — un-acked write is atomic (present-whole), baseline and log stay at the floor', async () => { + const { dir, brain, baselineId, floor } = await arrangeBaseline('sae') + const crashedId = uid('sae-crashed') + const arm = armCrash(brain, 'singleop-after-execute') + await expect( + brain.add({ + id: crashedId, + data: 'never acked', + type: NounType.Document, + vector: vec(2), + metadata: { v: 2 } + }) + ).rejects.toThrow('simulated process crash at singleop-after-execute') + expect(arm.fired).toContain('singleop-after-execute') + await abandonAsCrashed(brain) + + const reopened = await openLive(dir) + // Baseline intact. + expect(((await reopened.get(baselineId)) as { metadata: { v: number } }).metadata.v).toBe(1) + // The log holds nothing beyond the committed watermark (no fact was ever + // appended for the crashed write). + expect(await factGenerations(reopened)).toEqual([floor]) + expect(storeOf(reopened).committedGeneration()).toBe(floor) + // The un-acked write: Model-B applies the live canonical write BEFORE the + // ack, so under process death its bytes survive — the row is PRESENT and + // WHOLE by id (atomic, not torn). Under power loss the same un-fsynced + // bytes may instead vanish entirely; both end states are atomic. NOTE the + // divergence: the row is get()-visible but find()-invisible (no index + // entry survived, no generation/fact records it, and no repair is pending + // — a permanent canonical orphan; see the suite report). + const orphan = (await reopened.get(crashedId)) as { metadata: { v: number } } | null + expect(orphan).not.toBeNull() + expect(orphan!.metadata.v).toBe(2) // whole, byte-consistent — never torn + const found = (await reopened.find({ type: NounType.Document, limit: 10 })) as Array<{ id: string }> + expect(found.map((f) => f.id)).toContain(baselineId) + expect(found.map((f) => f.id)).not.toContain(crashedId) + // A fresh write succeeds with a monotonic generation. The crashed + // generation number is REUSED (nothing durable references it): the + // counter reopened at the floor. + expect(reopened.generation()).toBe(floor) + const freshId = uid('sae-fresh') + await reopened.add({ + id: freshId, + data: 'fresh after recovery', + type: NounType.Document, + vector: vec(3), + metadata: { v: 3 } + }) + await reopened.flush() + expect(storeOf(reopened).committedGeneration()).toBe(floor + 1) + expect(((await reopened.get(freshId)) as { metadata: { v: number } }).metadata.v).toBe(3) + }) + + it('singleop-after-fact-append (deferred mode) — the appended fact is truncated back at reopen', async () => { + const { dir, brain, baselineId, floor } = await arrangeBaseline('sfa') + const crashedId = uid('sfa-crashed') + const arm = armCrash(brain, 'singleop-after-fact-append') + await expect( + brain.add({ + id: crashedId, + data: 'never acked', + type: NounType.Document, + vector: vec(2), + metadata: { v: 2 } + }) + ).rejects.toThrow('simulated process crash at singleop-after-fact-append') + expect(arm.fired).toContain('singleop-after-fact-append') + await abandonAsCrashed(brain) + + const reopened = await openLive(dir) + // The fact WAS appended to the log file before the crash (process death + // keeps file bytes) — open() must truncate it back to the manifest + // watermark, and does. + expect(await factGenerations(reopened)).toEqual([floor]) + expect(storeOf(reopened).committedGeneration()).toBe(floor) + // Baseline intact; un-acked row atomic (present-whole via canonical, as + // in the singleop-after-execute row). + expect(((await reopened.get(baselineId)) as { metadata: { v: number } }).metadata.v).toBe(1) + const orphan = (await reopened.get(crashedId)) as { metadata: { v: number } } | null + expect(orphan).not.toBeNull() + expect(orphan!.metadata.v).toBe(2) + // Fresh write with a monotonic generation (crashed number reused — the + // truncated fact freed it). + expect(reopened.generation()).toBe(floor) + const freshId = uid('sfa-fresh') + await reopened.add({ + id: freshId, + data: 'fresh', + type: NounType.Document, + vector: vec(3), + metadata: { v: 3 } + }) + await reopened.flush() + expect(storeOf(reopened).committedGeneration()).toBe(floor + 1) + expect(await factGenerations(reopened)).toEqual([floor, floor + 1]) + }) + + it('flush-after-staging — the ACKED write survives (drop-without-restore); only the window history is lost', async () => { + const { dir, brain, baselineId, floor } = await arrangeBaseline('fas') + const ackedId = uid('fas-acked') + await brain.add({ + id: ackedId, + data: 'acked before flush', + type: NounType.Document, + vector: vec(2), + metadata: { v: 2 } + }) + const ackedGen = storeOf(brain).generation() + const arm = armCrash(brain, 'flush-after-staging') + await expect(brain.flush()).rejects.toThrow('simulated process crash at flush-after-staging') + expect(arm.fired).toContain('flush-after-staging') + // The crashed flush left the staged record-set dir on disk, above the manifest. + expect(generationDirExists(dir, ackedGen)).toBe(true) + await abandonAsCrashed(brain) + + const reopened = await openLive(dir) + // Recovery DROPPED the staged group-commit dir WITHOUT restoring its + // before-images — restoring would silently revert an acknowledged write. + expect(generationDirExists(dir, ackedGen)).toBe(false) + expect(storeOf(reopened).committedGeneration()).toBe(floor) + // NEVER A LOST ACK: the acknowledged write is present and whole. + const acked = (await reopened.get(ackedId)) as { metadata: { v: number } } | null + expect(acked).not.toBeNull() + expect(acked!.metadata.v).toBe(2) + expect(((await reopened.get(baselineId)) as { metadata: { v: number } }).metadata.v).toBe(1) + // Recovery rolled generations back → index reconciliation ran → the acked + // row is find()-visible too. + const found = (await reopened.find({ type: NounType.Document, limit: 10 })) as Array<{ id: string }> + expect(found.map((f) => f.id)).toEqual(expect.arrayContaining([baselineId, ackedId])) + // The window's HISTORY is the documented cost: its fact is truncated back + // (the acked row now lives only in canonical bytes, not the log). + expect(await factGenerations(reopened)).toEqual([floor]) + // The crashed generation number is NOT reused (its dropped dir was seen + // at open): fresh writes continue above it. + expect(reopened.generation()).toBe(ackedGen) + const freshId = uid('fas-fresh') + await reopened.add({ + id: freshId, + data: 'fresh', + type: NounType.Document, + vector: vec(3), + metadata: { v: 3 } + }) + await reopened.flush() + expect(storeOf(reopened).committedGeneration()).toBe(ackedGen + 1) + }) + + // ========================================================================== + // Rows AFTER partial durability — staged/synced bytes exist, no manifest + // ========================================================================== + + it('flush-before-manifest — staged bytes + synced facts above the manifest are dropped/truncated; the acked write stays', async () => { + const { dir, brain, baselineId, floor } = await arrangeBaseline('fbm') + const ackedId = uid('fbm-acked') + await brain.add({ + id: ackedId, + data: 'acked before flush', + type: NounType.Document, + vector: vec(2), + metadata: { v: 2 } + }) + const ackedGen = storeOf(brain).generation() + const arm = armCrash(brain, 'flush-before-manifest') + await expect(brain.flush()).rejects.toThrow('simulated process crash at flush-before-manifest') + // The earlier flush phase passed through untripped before the target fired. + expect(arm.fired).toContain('flush-after-staging') + expect(arm.fired).toContain('flush-before-manifest') + expect(generationDirExists(dir, ackedGen)).toBe(true) + await abandonAsCrashed(brain) + + const reopened = await openLive(dir) + // Per the recovery contract in open(): groupCommit record-sets above the + // manifest are dropped WITHOUT restore, and the (fsynced!) facts above + // the manifest are truncated back. The acked live write stays. + expect(generationDirExists(dir, ackedGen)).toBe(false) + expect(storeOf(reopened).committedGeneration()).toBe(floor) + expect(await factGenerations(reopened)).toEqual([floor]) + const acked = (await reopened.get(ackedId)) as { metadata: { v: number } } | null + expect(acked).not.toBeNull() // never a lost ack + expect(acked!.metadata.v).toBe(2) + expect(((await reopened.get(baselineId)) as { metadata: { v: number } }).metadata.v).toBe(1) + // Fresh write above the crashed generation (number not reused). + expect(reopened.generation()).toBe(ackedGen) + const freshId = uid('fbm-fresh') + await reopened.add({ + id: freshId, + data: 'fresh', + type: NounType.Document, + vector: vec(3), + metadata: { v: 3 } + }) + await reopened.flush() + expect(storeOf(reopened).committedGeneration()).toBe(ackedGen + 1) + }) + + it('before-manifest-rename (transact) — fully staged, never committed: rolled back byte-identically', async () => { + const { dir, brain, baselineId, floor } = await arrangeBaseline('bmr') + const newId = uid('bmr-new') + const arm = armCrash(brain, 'before-manifest-rename') + await expect( + brain.transact([ + { op: 'update', id: baselineId, metadata: { v: 2 } }, + { + op: 'add', + id: newId, + type: NounType.Document, + data: 'uncommitted', + vector: vec(2), + metadata: { v: 2 } + } + ]) + ).rejects.toThrow('simulated process crash at before-manifest-rename') + expect(arm.fired).toContain('before-manifest-rename') + const txGen = storeOf(brain).generation() + expect(generationDirExists(dir, txGen)).toBe(true) + await abandonAsCrashed(brain) + + const reopened = await openLive(dir) + // Rolled back cleanly: the update is undone, the add is ABSENT everywhere. + expect(((await reopened.get(baselineId)) as { metadata: { v: number } }).metadata.v).toBe(1) + expect(await reopened.get(newId)).toBeNull() + const found = (await reopened.find({ type: NounType.Document, limit: 10 })) as Array<{ id: string }> + expect(found.map((f) => f.id)).not.toContain(newId) + expect(generationDirExists(dir, txGen)).toBe(false) + expect(storeOf(reopened).committedGeneration()).toBe(floor) + expect(await factGenerations(reopened)).toEqual([floor]) + // The crashed generation number is never reissued (counter persisted + // before the crash point). + expect(reopened.generation()).toBe(txGen) + const freshId = uid('bmr-fresh') + await reopened.add({ + id: freshId, + data: 'fresh', + type: NounType.Document, + vector: vec(3), + metadata: { v: 3 } + }) + await reopened.flush() + expect(storeOf(reopened).committedGeneration()).toBe(txGen + 1) + }) + + it('transact-after-fact-sync — the fsynced fact of an uncommitted transact is truncated back; rollback is clean', async () => { + const { dir, brain, baselineId, floor } = await arrangeBaseline('tfs') + const newId = uid('tfs-new') + const arm = armCrash(brain, 'transact-after-fact-sync') + await expect( + brain.transact([ + { op: 'update', id: baselineId, metadata: { v: 2 } }, + { + op: 'add', + id: newId, + type: NounType.Document, + data: 'uncommitted', + vector: vec(2), + metadata: { v: 2 } + } + ]) + ).rejects.toThrow('simulated process crash at transact-after-fact-sync') + expect(arm.fired).toContain('transact-after-fact-sync') + const txGen = storeOf(brain).generation() + await abandonAsCrashed(brain) + + const reopened = await openLive(dir) + // The batch's fact was appended AND fsynced before the crash — open() + // must truncate it back to the manifest watermark (the generation never + // committed), and the before-images must restore byte-identically. + expect(await factGenerations(reopened)).toEqual([floor]) + expect(((await reopened.get(baselineId)) as { metadata: { v: number } }).metadata.v).toBe(1) + expect(await reopened.get(newId)).toBeNull() + expect(storeOf(reopened).committedGeneration()).toBe(floor) + expect(generationDirExists(dir, txGen)).toBe(false) + // Counter: the staged dir was seen at open, so the number is not reused. + expect(reopened.generation()).toBe(txGen) + const freshId = uid('tfs-fresh') + await reopened.add({ + id: freshId, + data: 'fresh', + type: NounType.Document, + vector: vec(3), + metadata: { v: 3 } + }) + await reopened.flush() + expect(storeOf(reopened).committedGeneration()).toBe(txGen + 1) + }) + + // ========================================================================== + // Row AFTER the commit point — the transaction must be kept + // ========================================================================== + + it('after-manifest-rename (transact) — the manifest rename landed: the transaction is COMMITTED and fully present', async () => { + const { dir, brain, baselineId, floor } = await arrangeBaseline('amr') + const newId = uid('amr-new') + const arm = armCrash(brain, 'after-manifest-rename') + await expect( + brain.transact([ + { op: 'update', id: baselineId, metadata: { v: 2 } }, + { + op: 'add', + id: newId, + type: NounType.Document, + data: 'committed by the rename', + vector: vec(2), + metadata: { v: 2 } + } + ]) + ).rejects.toThrow('simulated process crash at after-manifest-rename') + expect(arm.fired).toContain('after-manifest-rename') + const txGen = storeOf(brain).generation() + await abandonAsCrashed(brain) + + const reopened = await openLive(dir) + // COMMITTED: both operations present, atomically. + expect(((await reopened.get(baselineId)) as { metadata: { v: number } }).metadata.v).toBe(2) + const added = (await reopened.get(newId)) as { metadata: { v: number } } | null + expect(added).not.toBeNull() + expect(added!.metadata.v).toBe(2) + expect(storeOf(reopened).committedGeneration()).toBe(txGen) + // The fact was synced before the commit point and sits at/below the + // manifest — it is KEPT. + expect(await factGenerations(reopened)).toEqual([floor, txGen]) + // Fresh writes continue above the committed generation. + const freshId = uid('amr-fresh') + await reopened.add({ + id: freshId, + data: 'fresh', + type: NounType.Document, + vector: vec(3), + metadata: { v: 3 } + }) + await reopened.flush() + expect(storeOf(reopened).committedGeneration()).toBe(txGen + 1) + }) + + // ========================================================================== + // Durable-at-ack (log-authority) mode variants + // ========================================================================== + + it('singleop-after-fact-append (at-ack mode) — the intact fact is REPLAYED at reopen; the write commits', async () => { + const { dir, brain, baselineId, floor } = await arrangeBaseline('aaf') + await flipToAtAck(brain) + const crashedId = uid('aaf-crashed') + const arm = armCrash(brain, 'singleop-after-fact-append') + await expect( + brain.add({ + id: crashedId, + data: 'fact fsynced, never acked', + type: NounType.Document, + vector: vec(2), + metadata: { v: 2 } + }) + ).rejects.toThrow('simulated process crash at singleop-after-fact-append') + expect(arm.fired).toContain('singleop-after-fact-append') + await abandonAsCrashed(brain) + + const reopened = await openLive(dir) + // LOG-AUTHORITY RECOVERY CONTRACT: under 'log' authority, an intact + // fact above the manifest is adopted at open — REPLAYED into canonical + // and committed — never truncated. (At-least-once at the fact layer: a + // crashed-pre-ack write whose fact survived intact becomes committed; + // that is a valid write landing, never a torn or lost state.) + expect(await factGenerations(reopened)).toEqual([floor, floor + 1]) + expect(storeOf(reopened).committedGeneration()).toBe(floor + 1) + expect(((await reopened.get(baselineId)) as { metadata: { v: number } }).metadata.v).toBe(1) + const replayed = (await reopened.get(crashedId)) as { metadata: { v: number } } | null + expect(replayed).not.toBeNull() + expect(replayed!.metadata.v).toBe(2) + // Fresh write lands monotonically ABOVE the replayed generation. + const freshId = uid('aaf-fresh') + await reopened.add({ + id: freshId, + data: 'fresh', + type: NounType.Document, + vector: vec(3), + metadata: { v: 3 } + }) + await reopened.flush() + expect(storeOf(reopened).committedGeneration()).toBe(floor + 2) + }) + + // THE AT-ACK CONTRACT, END TO END (was a release-blocking finding; fixed + // by log-authority replay-at-open): under power loss the un-fsynced + // tmp+rename canonical bytes legally vanish while the fsynced fact + // survives — recovery REPLAYS that fact into canonical, so the acked + // write lives. This is the sentence 'durable-at-ack' actually promises. + it( + 'at-ack POWER LOSS — an ACKED write whose fact is fsynced SURVIVES reopen via log replay', + async () => { + const { dir, brain, baselineId } = await arrangeBaseline('apl') + await flipToAtAck(brain) + const ackedId = uid('apl-acked') + // No fault injector: this write ACKS normally — in at-ack mode the ack + // returned only after a covering log fsync. + await brain.add({ + id: ackedId, + data: 'acked, fact fsynced', + type: NounType.Document, + vector: vec(2), + metadata: { v: 2 } + }) + // Crash before any flush: RAM is gone… + await abandonAsCrashed(brain) + // …and power loss takes the un-fsynced canonical rename with it. The + // fsynced fact log survives — it is the write's only durable copy. + dropCanonicalNoun(dir, ackedId) + + const reopened = await openLive(dir) + expect(((await reopened.get(baselineId)) as { metadata: { v: number } }).metadata.v).toBe(1) + // THE AT-ACK CONTRACT: the acknowledged write survives the crash. + // Observed today: open() truncates its fact back to the manifest + // watermark and the write is gone everywhere. + const acked = (await reopened.get(ackedId)) as { metadata: { v: number } } | null + expect(acked).not.toBeNull() + expect(acked!.metadata.v).toBe(2) + } + ) + + // ========================================================================== + // Disk full — one ENOSPC'd append + // ========================================================================== + + it('disk full — an ENOSPC append rejects loudly and typed; reads keep serving; a later write succeeds', async () => { + const { dir, brain, baselineId, floor } = await arrangeBaseline('nospc') + liveBrains.push(brain) // this row never crashes the brain + void dir + const failedId = uid('nospc-failed') + const probe = failNextAppendWithEnospc(brain) + // LOUD, TYPED, never a silent success: the raw ENOSPC surfaces to the + // caller with its errno code intact. + await expect( + brain.add({ + id: failedId, + data: 'no space', + type: NounType.Document, + vector: vec(2), + metadata: { v: 2 } + }) + ).rejects.toMatchObject({ code: 'ENOSPC' }) + expect(probe.failed()).toBe(1) + // The store still serves reads. + expect(((await brain.get(baselineId)) as { metadata: { v: number } }).metadata.v).toBe(1) + // Space "restored" (the failing patch self-cleared): a later write succeeds + // end to end, including its fact and an explicit durability barrier. + const laterId = uid('nospc-later') + await brain.add({ + id: laterId, + data: 'space restored', + type: NounType.Document, + vector: vec(3), + metadata: { v: 3 } + }) + await brain.flush() + expect(((await brain.get(laterId)) as { metadata: { v: number } }).metadata.v).toBe(3) + expect(storeOf(brain).committedGeneration()).toBeGreaterThan(floor) + // FIXED BEHAVIOR (was: the rejected generation stayed buffered and the + // next flush committed it with NO fact — a silent log gap): the failure + // path un-buffers the generation and returns the counter reservation, + // so the later write takes floor+1 and the log is gap-free. + expect(storeOf(brain).committedGeneration()).toBe(floor + 1) + expect(await factGenerations(brain)).toEqual([floor, floor + 1]) + // Canonical residue of the rejected write (execute ran before the + // append failed) is the documented Model-B crash-equivalent orphan — + // uncommitted, absent from the log, same shape as a crash at execute. + expect(((await brain.get(failedId)) as { metadata: { v: number } } | null)?.metadata.v).toBe(2) + }) + + // THE NO-SILENT-COMMIT CONTRACT (was a release-blocking finding; fixed by + // un-buffering on append failure): a loudly-rejected write never becomes + // durably committed and the log never carries a gap. Canonical residue + // (the execute-before-commit orphan) is the documented Model-B + // crash-equivalent, pinned in the row above — NOT a commit. + it('disk full — a write rejected for a failed fact append is NOT silently committed', async () => { + const { brain, floor } = await arrangeBaseline('nogap') + liveBrains.push(brain) + const failedId = uid('nogap-failed') + failNextAppendWithEnospc(brain) + await expect( + brain.add({ + id: failedId, + data: 'no space', + type: NounType.Document, + vector: vec(2), + metadata: { v: 2 } + }) + ).rejects.toMatchObject({ code: 'ENOSPC' }) + await brain.flush() + // THE CONTRACT: nothing was committed behind the caller's back — the + // log carries no gap and no generation for the rejected write. (get() + // still serves the canonical execute-residue orphan — the documented + // Model-B crash-equivalent, pinned in the row above.) + expect(storeOf(brain).committedGeneration()).toBe(floor) + expect(await factGenerations(brain)).toEqual([floor]) + }) +}) diff --git a/tests/integration/log-authority.test.ts b/tests/integration/log-authority.test.ts new file mode 100644 index 00000000..14278cd1 --- /dev/null +++ b/tests/integration/log-authority.test.ts @@ -0,0 +1,340 @@ +/** + * @module tests/integration/log-authority + * @description The guarded log-authority core, end-to-end: the per-brain + * authority switch (default 'tree', stored artifact, checked at open only), + * the verification oracle (replay the fact log, diff latest per-id state + * against the canonical tree, NAME every divergence by class), the guarded + * flip (refuses on red with the cure in the message; lands on green and + * engages durable-at-ack immediately), and the switch surviving reopen. + * + * KNOWN GAPS PINNED WITH `.fails` (real findings, not test bugs — see the + * comments on each): a fresh brain is NOT log-complete by construction + * today, because the VFS root is written at init as a baseline + * (generation-less) write that never gets a fact, so the oracle reports it + * as a `pre-log-record` and no fresh brain can flip without a manual + * baseline backfill. The tests that need a green oracle perform that + * backfill explicitly (an identity update of the root as the FINAL write — + * final, because derived-index maintenance rewrites canonical noun records + * outside generations, so an earlier fact's after-image goes stale; see the + * module tail comment on `backfillBaseline`). + */ +import { describe, it, expect, afterEach } from 'vitest' +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { Brainy } from '../../src/index.js' +import type { OracleReport } from '../../src/db/logAuthority.js' + +/** The VFS root — created at init by a baseline (generation-less) write. */ +const VFS_ROOT = '00000000-0000-0000-0000-000000000000' +const AUTHORITY_ARTIFACT = '_system/log-authority.json' + +/** White-box view of the internals this suite instruments (read-only spies + * plus the sanctioned direct-storage writes for aging/drifting a brain). */ +type BrainInternals = { + generationStore: { + getFactLog(): { ensureSynced(): Promise } | null + logDurability: 'deferred' | 'at-ack' + } + storage: { + readRawObject(path: string): Promise + saveNoun(n: unknown): Promise + saveNounMetadata(id: string, m: Record): Promise + getNounMetadata(id: string): Promise | null> + } +} + +const internals = (brain: Brainy): BrainInternals => + brain as unknown as BrainInternals + +/** Count calls to the fact log's ensureSynced without changing behavior. */ +function spyEnsureSynced(brain: Brainy): { calls: () => number } { + const factLog = internals(brain).generationStore.getFactLog() + expect(factLog, 'filesystem storage hosts a fact log').not.toBeNull() + let calls = 0 + const original = factLog!.ensureSynced.bind(factLog) + factLog!.ensureSynced = async () => { + calls++ + return original() + } + return { calls: () => calls } +} + +/** + * The minimal baseline backfill: an identity update of the VFS root, so the + * one canonical record the log never saw (the init-time baseline write) gets + * a fact carrying its current state. MUST be the final write of the setup — + * derived-index maintenance (HNSW/enumeration denormalization) rewrites the + * root's canonical noun record outside any generation, so a root fact taken + * before later writes digests stale and reports `state-differs`. + */ +async function backfillBaseline(brain: Brainy): Promise { + const root = await brain.get(VFS_ROOT) + expect(root, 'the VFS root exists on a fresh brain').toBeTruthy() + await brain.update({ id: VFS_ROOT, metadata: root!.metadata }) +} + +/** Seed a brain with the standard write mix: 2 adds, an update, a remove. */ +async function seedWrites(brain: Brainy): Promise<{ kept: string; removed: string }> { + const kept = await brain.add({ data: 'alpha document', type: 'document', metadata: { n: 1 } }) + const removed = await brain.add({ data: 'beta document', type: 'document', metadata: { n: 2 } }) + await brain.update({ id: kept, metadata: { n: 10 } }) + await brain.remove(removed) + return { kept, removed } +} + +describe('log authority — the switch, the oracle, the guarded flip', () => { + const dirs: string[] = [] + const brains: Brainy[] = [] + + const openBrain = async (dir?: string): Promise<{ brain: Brainy; dir: string }> => { + const d = dir ?? mkdtempSync(join(tmpdir(), 'brainy-log-authority-')) + if (!dir) dirs.push(d) + const brain = new Brainy({ + storage: { type: 'filesystem', path: d }, + requireSubtype: false, + silent: true, + dimensions: 384 + }) + brains.push(brain) + await brain.init() + return { brain, dir: d } + } + + afterEach(async () => { + for (const b of brains.splice(0)) { + await (b as unknown as { close?: () => Promise }).close?.().catch(() => {}) + } + for (const d of dirs.splice(0)) rmSync(d, { recursive: true, force: true }) + }) + + it('DEFAULT IS TREE: a fresh brain reports tree authority, stores no artifact, and plain acks never await a log fsync', async () => { + const { brain } = await openBrain() + + expect(brain.logAuthority().authority).toBe('tree') + expect(brain.logAuthority().flippedAt).toBeUndefined() + + const artifact = await internals(brain) + .storage.readRawObject(AUTHORITY_ARTIFACT) + .catch(() => null) + expect(artifact, 'no switch artifact exists before any flip').toBeNull() + + // The MODE assertion (not a timing one): in tree authority a single-op + // ack must never call the log's covering-fsync path. + const spy = spyEnsureSynced(brain) + await brain.add({ data: 'tree mode write', type: 'document', metadata: { n: 1 } }) + expect(spy.calls(), 'tree mode: add() does not call ensureSynced').toBe(0) + expect(internals(brain).generationStore.logDurability).toBe('deferred') + }) + + // KNOWN GAP (marked .fails — remove the marker when fixed in src): the + // intended contract is that a fresh brain is log-complete by construction, + // because every write dual-writes a fact. Today the VFS root + // (00000000-0000-0000-0000-000000000000) is created at init by a baseline + // write with NO generation and NO fact, yet it is enumerated by the + // canonical walk — so the oracle on a fresh brain is red with exactly one + // `pre-log-record` mismatch on the root, and adoptLogAuthority() refuses + // on every fresh brain. Verified empirically on this branch. + it.fails('ORACLE INTENT: a fresh brain is log-complete by construction — verdict green with zero mismatches', async () => { + const { brain } = await openBrain() + await seedWrites(brain) + await brain.flush() + + const report = await brain.verifyLogAuthority() + expect(report.verdict).toBe('green') + expect(report.mismatches).toEqual([]) + }) + + it('a fresh, un-backfilled brain diverges ONLY on the init-time baseline record — every user write is exactly reproduced', async () => { + const { brain } = await openBrain() + await seedWrites(brain) + await brain.flush() + + const report = await brain.verifyLogAuthority() + // Tolerant pin (stays true after the baseline gap is fixed in src): + // whatever the verdict, no USER record may ever diverge — the only + // admissible mismatch is the init-time baseline root, as pre-log-record. + expect( + report.mismatches.every( + (m) => m.id === VFS_ROOT && m.reason === 'pre-log-record' && m.kind === 'noun' + ), + 'the only divergence on a fresh brain is the baseline root record' + ).toBe(true) + expect(report.matched).toBe(report.nounsChecked - report.mismatches.length) + expect(report.mismatchListTruncated).toBe(false) + }) + + it('THE ORACLE GOES GREEN on a log-complete brain: adds + update + remove, every canonical row exactly reproduced', async () => { + const { brain } = await openBrain() + await seedWrites(brain) + await backfillBaseline(brain) // final write — see the helper's contract + await brain.flush() + + const report = await brain.verifyLogAuthority() + expect(report.verdict).toBe('green') + expect(report.mismatches).toEqual([]) + expect(report.mismatchListTruncated).toBe(false) + // Live count: the kept document + the VFS root (the removed one is a + // tombstone in the log and absent from canonical — checked, not counted). + expect(report.nounsChecked).toBe(2) + expect(report.matched).toBe(2) + // 5 committed generations: add, add, update, remove, root backfill. + expect(report.generationsScanned).toBe(5) + }) + + it('THE ORACLE NAMES pre-log records: a canonical row no fact ever recorded reports pre-log-record, by id', async () => { + const { brain } = await openBrain() + await seedWrites(brain) + await backfillBaseline(brain) + await brain.flush() + expect((await brain.verifyLogAuthority()).verdict, 'sanity: green before aging').toBe('green') + + // Simulate an aged brain: write one canonical record DIRECTLY at the + // storage layer (the write path never sees it, so no fact exists) — + // the pre-log shape: flat metadata, no _fmt stamp, 384-dim vector. + const legacyId = '00000000-0000-4000-8000-00000000a6ed' + const storage = internals(brain).storage + await storage.saveNoun({ + id: legacyId, + vector: new Array(384).fill(0.01), + connections: new Map(), + level: 0 + }) + await storage.saveNounMetadata(legacyId, { + noun: 'document', + confidence: 0.75, + createdAt: 1700000000000, + updatedAt: 1700000000000, + _rev: 1, + legacyField: 'legacy-value' + }) + + const report = await brain.verifyLogAuthority() + expect(report.verdict).toBe('red') + expect(report.mismatches).toHaveLength(1) + expect(report.mismatches[0]).toEqual({ + id: legacyId, + kind: 'noun', + reason: 'pre-log-record' + }) + }) + + it('THE FLIP REFUSES ON RED: names the oracle verdict and the cure, writes nothing, changes nothing', async () => { + const { brain } = await openBrain() + await seedWrites(brain) + await backfillBaseline(brain) + await brain.flush() + + // Age the brain: one canonical record the log never saw. + const legacyId = '00000000-0000-4000-8000-00000000a6ed' + const storage = internals(brain).storage + await storage.saveNoun({ + id: legacyId, + vector: new Array(384).fill(0.01), + connections: new Map(), + level: 0 + }) + await storage.saveNounMetadata(legacyId, { + noun: 'document', + confidence: 0.5, + createdAt: 1700000000000, + updatedAt: 1700000000000, + _rev: 1 + }) + + let error: Error | null = null + try { + await brain.adoptLogAuthority() + } catch (err) { + error = err as Error + } + expect(error, 'the flip rejects on a red oracle').not.toBeNull() + expect(error!.message).toMatch(/oracle is RED/) + expect(error!.message).toMatch(/baseline backfill/) + + // Nothing changed: authority still tree, no artifact, deferred durability. + expect(brain.logAuthority().authority).toBe('tree') + const artifact = await storage.readRawObject(AUTHORITY_ARTIFACT).catch(() => null) + expect(artifact, 'a refused flip writes no artifact').toBeNull() + expect(internals(brain).generationStore.logDurability).toBe('deferred') + }) + + it('THE FLIP LANDS ON GREEN: the report is the receipt, the artifact is on disk, and durable-at-ack engages immediately', async () => { + const { brain } = await openBrain() + await seedWrites(brain) + await backfillBaseline(brain) + await brain.flush() + + const report: OracleReport = await brain.adoptLogAuthority() + expect(report.verdict).toBe('green') + + const authority = brain.logAuthority() + expect(authority.authority).toBe('log') + expect(typeof authority.flippedAt).toBe('number') + expect(authority.oracle).toBeDefined() + expect(authority.oracle!.nounsChecked).toBe(report.nounsChecked) + expect(authority.oracle!.generationsScanned).toBe(report.generationsScanned) + + const artifact = (await internals(brain) + .storage.readRawObject(AUTHORITY_ARTIFACT) + .catch(() => null)) as { authority?: string } | null + expect(artifact, 'the switch artifact exists on disk').not.toBeNull() + expect(artifact!.authority).toBe('log') + + // Durable-at-ack engaged in THIS session: the next single-op ack awaits + // a covering log fsync. + expect(internals(brain).generationStore.logDurability).toBe('at-ack') + const spy = spyEnsureSynced(brain) + await brain.add({ data: 'post-flip write', type: 'document', metadata: { n: 3 } }) + expect(spy.calls(), 'log mode: add() awaits the covering fsync').toBeGreaterThanOrEqual(1) + }) + + it('THE SWITCH SURVIVES REOPEN: authority restored at open with no re-verification, durable-at-ack active in the new session', async () => { + const { brain, dir } = await openBrain() + await seedWrites(brain) + await backfillBaseline(brain) + await brain.flush() + await brain.adoptLogAuthority() + const flipReceipt = brain.logAuthority() + await (brain as unknown as { close: () => Promise }).close() + + const { brain: reopened } = await openBrain(dir) + const restored = reopened.logAuthority() + expect(restored.authority).toBe('log') + // No re-verification happened at open: the restored record IS the stored + // flip receipt, oracle summary and timestamp intact. + expect(restored.flippedAt).toBe(flipReceipt.flippedAt) + expect(restored.oracle).toEqual(flipReceipt.oracle) + + // Mode restored at open: an ack in the new session awaits the log fsync. + expect(internals(reopened).generationStore.logDurability).toBe('at-ack') + const spy = spyEnsureSynced(reopened) + await reopened.add({ data: 'new session write', type: 'document', metadata: { n: 4 } }) + expect(spy.calls(), 'reopened log mode: add() awaits the covering fsync').toBeGreaterThanOrEqual(1) + }) + + it('STATE-DIFFERS: canonical drift the write path never saw is named, by id', async () => { + const { brain } = await openBrain() + const { kept } = await seedWrites(brain) + await backfillBaseline(brain) + await brain.flush() + expect((await brain.verifyLogAuthority()).verdict, 'sanity: green before drift').toBe('green') + + // Drift one canonical metadata record DIRECTLY at the storage layer — + // the log never hears about it. This is the witness-drift case the + // oracle exists to catch. + const storage = internals(brain).storage + const current = await storage.getNounMetadata(kept) + expect(current, 'the seeded record has stored metadata').toBeTruthy() + await storage.saveNounMetadata(kept, { ...current!, driftedByTest: true }) + + const report = await brain.verifyLogAuthority() + expect(report.verdict).toBe('red') + expect(report.mismatches).toHaveLength(1) + expect(report.mismatches[0]).toEqual({ + id: kept, + kind: 'noun', + reason: 'state-differs' + }) + }) +}) diff --git a/tests/unit/db/fact-log-group-sync.test.ts b/tests/unit/db/fact-log-group-sync.test.ts new file mode 100644 index 00000000..3f4b1f42 --- /dev/null +++ b/tests/unit/db/fact-log-group-sync.test.ts @@ -0,0 +1,271 @@ +/** + * @module tests/unit/db/fact-log-group-sync + * @description Group commit on the fact log — the covering guarantee behind + * durable-at-ack: concurrent callers of ensureSynced() share ONE covering + * fsync (running + queued slots), a caller appending during a running sync + * joins a sync that STARTS after its append (never the possibly-stale running + * one), a solo writer syncs immediately, and at the brain level an at-ack + * ack resolving means the write's fact is on disk. + * + * One pin is marked `.fails` (real finding, not a test bug): the at-ack + * durability contract says an acked write's fact survives power loss, but + * FactLog.open() truncates every fact beyond the store's committed + * generation watermark — which only advances at the pending-tier flush. A + * crash-shaped reopen (acks landed, flush never ran) therefore DISCARDS the + * fsynced facts at open. See the test comment for the exact mechanism. + */ +import { describe, it, expect, beforeEach, afterEach } from 'vitest' +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { Brainy } from '../../../src/index.js' +import { FileSystemStorage } from '../../../src/storage/adapters/fileSystemStorage.js' +import { + FactLog, + storageSupportsFactLog, + type CommitFact, + type FactLogStorage +} from '../../../src/db/factLog.js' + +const UUID = (n: number): string => + `00000000-0000-4000-8000-${String(n).padStart(12, '0')}` + +const fact = (generation: number): CommitFact => ({ + generation, + timestamp: 1_700_000_000_000 + generation, + ops: [ + { + kind: 'noun', + id: UUID(generation), + record: { metadata: { noun: 'document', title: `doc ${generation}` }, vector: { v: [1, 2] } } + } + ] +}) + +/** Scan every fact from a FRESH reader log over the same directory. */ +async function readBack(dir: string, committedHead: number): Promise { + const storage: any = new FileSystemStorage(dir) + await storage.init() + const reader = new FactLog(storage as FactLogStorage) + await reader.open(committedHead) + const facts: CommitFact[] = [] + const scan = reader.scanFacts() + for await (const batch of scan.batches()) facts.push(...batch.facts) + return facts +} + +describe('fact log group commit — the covering fsync', () => { + let dir: string + let storage: any + let log: FactLog + + beforeEach(async () => { + dir = mkdtempSync(join(tmpdir(), 'brainy-group-sync-')) + storage = new FileSystemStorage(dir) + await storage.init() + expect(storageSupportsFactLog(storage)).toBe(true) + log = new FactLog(storage as FactLogStorage) + await log.open(0) + }) + + afterEach(() => { + rmSync(dir, { recursive: true, force: true }) + }) + + it('many concurrent ensureSynced() callers share one covering fsync — every caller resolves, batching happened', async () => { + for (let g = 1; g <= 10; g++) await log.append(fact(g)) + + // Count REAL fsync batches at the storage boundary, with a small delay so + // the concurrent callers genuinely overlap the running sync. + let fsyncBatches = 0 + const origSync = storage.syncRawObjects.bind(storage) + storage.syncRawObjects = async (paths: string[]) => { + fsyncBatches++ + await new Promise((r) => setTimeout(r, 15)) + return origSync(paths) + } + + const callers = Array.from({ length: 10 }, () => log.ensureSynced()) + await Promise.all(callers) // every caller resolves — no lost writer + + expect(fsyncBatches, 'callers shared a covering fsync').toBeLessThan(10) + expect(fsyncBatches).toBeGreaterThanOrEqual(1) + + // Durable: a fresh reader over the same directory sees all 10 facts. + const facts = await readBack(dir, 10) + expect(facts.map((f) => f.generation)).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]) + }) + + it('an append during a RUNNING sync is covered by a sync that starts after it — never the stale running one', async () => { + for (let g = 1; g <= 3; g++) await log.append(fact(g)) + + // Gate the FIRST fsync so a sync is provably in flight. + let fsyncBatches = 0 + let releaseGate!: () => void + const gate = new Promise((r) => { + releaseGate = r + }) + let gated = true + const origSync = storage.syncRawObjects.bind(storage) + storage.syncRawObjects = async (paths: string[]) => { + fsyncBatches++ + if (gated) { + gated = false + await gate + } + return origSync(paths) + } + + const p1 = log.ensureSynced() // sync A: snapshots gens 1..3, blocks in fsync + await new Promise((r) => setTimeout(r, 10)) + expect(fsyncBatches, 'sync A is in flight').toBe(1) + + await log.append(fact(4)) // lands AFTER sync A snapshotted + let p2Resolved = false + const p2 = log.ensureSynced().then(() => { + p2Resolved = true + }) + + // The covering guarantee: p2 must NOT resolve off the running sync (it + // may have snapshotted before the append) — it waits for the queued one. + await new Promise((r) => setTimeout(r, 25)) + expect(p2Resolved, 'p2 never joins the possibly-stale running sync').toBe(false) + + releaseGate() + await p1 + await p2 + expect(p2Resolved).toBe(true) + expect(fsyncBatches, 'the queued covering sync ran after the running one').toBe(2) + + // The late append is durable once p2 resolved. + const facts = await readBack(dir, 4) + expect(facts.map((f) => f.generation)).toEqual([1, 2, 3, 4]) + }) + + it('a solo writer syncs immediately — one fsync, and a dirty-free ensureSynced adds none', async () => { + // Count only covering syncs: the first append itself fsyncs the tail + // manifest (the manifest-first flip), so instrument AFTER it. + await log.append(fact(1)) + let fsyncBatches = 0 + const origSync = storage.syncRawObjects.bind(storage) + storage.syncRawObjects = async (paths: string[]) => { + fsyncBatches++ + return origSync(paths) + } + + await log.ensureSynced() + expect(fsyncBatches).toBe(1) + + // Nothing new appended: the covering sync finds nothing dirty. + await log.ensureSynced() + expect(fsyncBatches).toBe(1) + }) +}) + +describe('durable-at-ack through the brain (group commit end-to-end)', () => { + const dirs: string[] = [] + const brains: any[] = [] + + const openBrain = async (dir?: string): Promise<{ brain: any; dir: string }> => { + process.env.BRAINY_DETERMINISTIC_EMBEDDINGS = 'true' + const d = dir ?? mkdtempSync(join(tmpdir(), 'brainy-at-ack-')) + if (!dir) dirs.push(d) + const brain: any = new Brainy({ + storage: { type: 'filesystem', path: d }, + requireSubtype: false, + silent: true, + dimensions: 384 + }) + brains.push(brain) + await brain.init() + return { brain, dir: d } + } + + afterEach(async () => { + for (const b of brains.splice(0)) await b.close?.().catch(() => {}) + for (const d of dirs.splice(0)) rmSync(d, { recursive: true, force: true }) + }) + + it('at-ack: N concurrent add() acks all resolve, every ack was covered by a log sync, and every fact is on disk after reopen', async () => { + const { brain, dir } = await openBrain() + // White-box: engage the at-ack durability mode directly (the guarded + // authority flip that normally enables it is covered by the integration + // suite — this test pins the durability machinery itself). + brain.generationStore.setLogDurability('at-ack') + + const factLog = brain.generationStore.getFactLog() + expect(factLog).not.toBeNull() + let syncs = 0 + const origSync = factLog.sync.bind(factLog) + factLog.sync = async () => { + syncs++ + return origSync() + } + + const ids: string[] = await Promise.all( + Array.from({ length: 10 }, (_, i) => + brain.add({ data: `concurrent write ${i}`, type: 'document', metadata: { i } }) + ) + ) + expect(new Set(ids).size, 'every ack resolved with a distinct id').toBe(10) + // Honest pin: single-op acks serialize under the commit mutex (append + + // covering sync run inside it), so concurrent add() acks do not currently + // share one fsync — cross-writer batching is the FactLog-layer property + // pinned above. What must hold here: at least one covering sync ran, and + // no ack resolved without the machinery engaged. + expect(syncs).toBeGreaterThanOrEqual(1) + expect(syncs).toBeLessThanOrEqual(10) + + await brain.close() + const { brain: reopened } = await openBrain(dir) + const scan = reopened.scanFacts() + expect(scan).not.toBeNull() + const liveFactIds = new Set() + for await (const batch of scan!.batches()) { + for (const f of batch.facts) { + for (const op of f.ops) if (op.kind === 'noun' && op.record !== null) liveFactIds.add(op.id) + } + } + for (const id of ids) { + expect(liveFactIds.has(id), `fact for acked write ${id} survives reopen`).toBe(true) + } + }) + + // KNOWN GAP (marked .fails — remove the marker when fixed in src): the + // at-ack contract is that an acked write's fact survives power loss. The + // fsync at ack does put the fact's bytes on disk — but FactLog.open() + // truncates every fact with generation > the store's committed watermark, + // and that watermark only advances at the pending-tier flush + // (flushPendingSingleOps). So on a crash-shaped reopen (acks landed, flush + // never ran) the store logs "[FactLog] truncating N uncommitted fact(s)" + // and DISCARDS the acked, fsynced facts. Until recovery treats the log as + // authoritative past the tree's watermark (or the watermark goes durable + // at ack), durable-at-ack does not survive the very crash it exists for. + it.fails('at-ack CONTRACT: acked facts survive a crash-shaped reopen (no flush ever ran)', async () => { + const { brain, dir } = await openBrain() + brain.generationStore.setLogDurability('at-ack') + // Crash simulation: the pending-tier durability flush never happens + // (every trigger routes through flushPendingSingleOps), and the brain is + // abandoned without close() — exactly the power-loss shape at-ack is for. + brain.generationStore.flushPendingSingleOps = async () => {} + + const ids: string[] = [] + for (let i = 0; i < 5; i++) { + ids.push(await brain.add({ data: `acked write ${i}`, type: 'document', metadata: { i } })) + } + + // No flush, no close — reopen the directory as a new session. + const { brain: reopened } = await openBrain(dir) + const scan = reopened.scanFacts() + expect(scan).not.toBeNull() + const liveFactIds = new Set() + for await (const batch of scan!.batches()) { + for (const f of batch.facts) { + for (const op of f.ops) if (op.kind === 'noun' && op.record !== null) liveFactIds.add(op.id) + } + } + for (const id of ids) { + expect(liveFactIds.has(id), `acked fact ${id} survives the crash-shaped reopen`).toBe(true) + } + }) +})