fix(log): acked writes survive power loss; rejected writes never silently commit — the kill-matrix goes 11/11 with zero .fails debt
Some checks failed
CI / Node 22 (push) Successful in 12m17s
CI / Node 24 (push) Has been cancelled
CI / Bun (latest) (push) Has been cancelled

Two release-blocking findings from the durability kill-matrix, both fixed
in the owning layer:

1. LOG-AUTHORITY REPLAY AT OPEN: durable-at-ack fsynced the fact before
   the ack, but open() truncated every fact above the manifest — after a
   power loss that takes the un-fsynced tmp+rename canonical bytes, the
   acked write's ONLY durable copy was discarded. Now: under 'log'
   authority, open() REPLAYS intact facts above the manifest into
   canonical (FactLog.peekFactsAbove — CRC-gated, order-sorted) and
   advances the manifest to cover them; tree-authority brains keep the
   truncate contract they were promised. Pinned end to end: the power-loss
   row constructs the exact disk state (fsynced log, vanished canonical
   rename) and the acked write lives.

2. NO SILENT COMMIT: commitSingleOp buffered the generation BEFORE the
   fact append; an append failure (ENOSPC) rejected the caller but the
   next flush durably committed the generation with NO fact — a permanent
   silent log gap. Now the failure path un-buffers and returns the counter
   reservation: nothing commits, the log stays gap-free, and the canonical
   execute-residue orphan is the documented crash-equivalent.

Plus: the kill-matrix itself (11 rows — every commit-path fault point ×
reopen-as-crash recovery contract, at-ack variants, disk-full row; five
new zero-cost faultPoint sites), the log-authority pin suite (oracle
green/red/state-differs, flip refusal, switch survives reopen, 9/9), and
the group-commit covering pins (5/5).

Gates: unit 2002/2002 (152 files) · integration 785 · conformance 27/27.
This commit is contained in:
David Snelling 2026-08-10 09:29:21 -07:00
parent 2d532684b4
commit 13022c510b
6 changed files with 1599 additions and 5 deletions

View file

@ -342,6 +342,34 @@ export class FactLog {
* crash between fact-append and the commit point). After open, the log is * crash between fact-append and the commit point). After open, the log is
* exactly the committed prefix. * 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<CommitFact[]> {
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<void> { async open(committedGeneration: number): Promise<void> {
const stored = (await this.storage.readRawObject(FACTS_MANIFEST_PATH)) as FactsManifest | null const stored = (await this.storage.readRawObject(FACTS_MANIFEST_PATH)) as FactsManifest | null
if (stored && typeof stored === 'object' && Array.isArray(stored.segments)) { if (stored && typeof stored === 'object' && Array.isArray(stored.segments)) {

View file

@ -45,6 +45,7 @@ import type {
GenerationStorage, GenerationStorage,
TxLogEntry TxLogEntry
} from './types.js' } from './types.js'
import { readLogAuthority } from './logAuthority.js'
import { FactLog, storageSupportsFactLog, type CommitFact, type FactOp } from './factLog.js' import { FactLog, storageSupportsFactLog, type CommitFact, type FactOp } from './factLog.js'
import { GenerationSegmentStore, type FoldGeneration } from './generationSegments.js' import { GenerationSegmentStore, type FoldGeneration } from './generationSegments.js'
import { crc32c } from '../utils/crc32c.js' import { crc32c } from '../utils/crc32c.js'
@ -88,12 +89,43 @@ export const GENERATIONS_PREFIX = '_generations'
* IS committed); the tx-log append has NOT happened yet. A crash here must * 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 * keep the transaction (the tx-log is advisory metadata, not the source of
* commit truth). * 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 = export type CommitFaultPhase =
| 'after-staging' | 'after-staging'
| 'after-execute' | 'after-execute'
| 'before-manifest-rename' | 'before-manifest-rename'
| 'after-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. * @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). // hosts no fact log (readers fall back to canonical enumeration).
if (storageSupportsFactLog(this.storage)) { if (storageSupportsFactLog(this.storage)) {
this.factLog = new FactLog(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) await this.factLog.open(this.committed)
} else { } else {
this.factLog = null this.factLog = null
@ -977,6 +1057,9 @@ export class GenerationStore {
await this.factLog.append(fact) await this.factLog.append(fact)
await this.factLog.sync() 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) ---------------------- // -- 5. Counter + manifest rename (COMMIT POINT) ----------------------
await this.persistCounterUnlocked() await this.persistCounterUnlocked()
@ -1278,6 +1361,12 @@ export class GenerationStore {
throw err throw err
} }
this.inTransact = false 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. // Buffer the pending generation + make it instantly visible to reads.
this.pendingBuffer.set(gen, { nouns: nounBefore, verbs: verbBefore, timestamp }) 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) — // the log's group-commit (many concurrent writers share ONE sync) —
// an acked write's fact survives power loss, by contract. // an acked write's fact survives power loss, by contract.
if (this.factLog) { if (this.factLog) {
await this.factLog.append( try {
await this.buildCommitFact({ generation: gen, timestamp, nouns, verbs }) await this.factLog.append(
) await this.buildCommitFact({ generation: gen, timestamp, nouns, verbs })
if (this.logDurability === 'at-ack') { )
await this.factLog.ensureSynced() 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() this.schedulePendingFlush()
return { generation: gen, timestamp } return { generation: gen, timestamp }
}) })
@ -1422,6 +1533,11 @@ export class GenerationStore {
logEntries.push({ generation: gen, timestamp: buf.timestamp }) 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. // ONE fsync for the whole window — the durability-batching win.
await this.storage.syncRawObjects(stagedPaths) await this.storage.syncRawObjects(stagedPaths)
@ -1431,6 +1547,12 @@ export class GenerationStore {
// generation without its durable fact. // generation without its durable fact.
await this.factLog?.sync() 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 // Test-only crash simulation: a throwing injector here leaves the staged
// group-commit generation dirs on disk with NO manifest advance — the // group-commit generation dirs on disk with NO manifest advance — the
// exact "crashed mid-flush" state recovery must DROP-WITHOUT-RESTORE // exact "crashed mid-flush" state recovery must DROP-WITHOUT-RESTORE

View file

@ -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<Brainy> {
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<void> {
const store = storeOf(brain) as unknown as {
withMutex<R>(fn: () => Promise<R>): Promise<R>
clearPendingFlushTimer(): void
pendingGens: number[]
pendingBuffer: Map<number, unknown>
}
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<number[]> {
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<void> }
}).storage
const original = storage.appendRawBytes.bind(storage)
let failures = 0
storage.appendRawBytes = async (p: string, b: Uint8Array): Promise<void> => {
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)))
}

View file

@ -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<Brainy> {
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<void> {
const storage = (
brain as unknown as {
storage: {
writeRawObject(p: string, d: unknown): Promise<void>
syncRawObjects(p: string[]): Promise<void>
}
}
).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])
})
})

View file

@ -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<void> } | null
logDurability: 'deferred' | 'at-ack'
}
storage: {
readRawObject(path: string): Promise<unknown | null>
saveNoun(n: unknown): Promise<void>
saveNounMetadata(id: string, m: Record<string, unknown>): Promise<void>
getNounMetadata(id: string): Promise<Record<string, unknown> | 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<void> {
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<void> }).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<void> }).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'
})
})
})

View file

@ -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<CommitFact[]> {
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<void>((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<string>()
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<string>()
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)
}
})
})