/** * @module db/factLog * @description The generation FACT LOG — an append-only, CRC-framed record of * every committed generation as an AFTER-IMAGE "fact": what each touched * entity/relationship BECAME (or a body-less tombstone when it was removed). * This is the dual-write half of the log-canonical transition: today the * before-image history + canonical tree remain authoritative; the fact log is * appended at the same commit points and reconciled to committed truth at * open, so consumers (index heals, replays, scans) can read one sequential, * self-verifying stream instead of walking the entity tree. * * ## Wire format (frozen; additive-only within a major) * * Fact (msgpack, POSITIONAL array — the segment header's formatVersion * governs the schema): * * fact := [ generation:u64, timestamp:u64, ops, meta|nil, blobHashes|nil ] * op := [ kind:u8 (0=noun, 1=verb), id:bin16 (raw uuid bytes), * record:[metaLeg, vecLeg] | nil ] // nil = TOMBSTONE * * Segment file (`_generations/facts/seg-.bfl`): * * header := magic "BFACTS\0\0" (8B) | formatVersion:u32 LE | * firstGeneration:u64 LE | reserved 12B (ZEROED, verified) * frame := length:u32 LE | crc32c:u32 LE (of payload) | payload * * A fact is never split across segments; a torn tail (length overrun or CRC * mismatch) terminates that segment's scan — everything before it is intact. * Zero-padded names make lexicographic order == generation order. * * ## Invariant * * After {@link FactLog.open}, the log contains EXACTLY the committed prefix: * facts are appended BEFORE the commit point (inside the same durability * window), so a crash can only leave the log AHEAD of committed truth — open * truncates any fact beyond the committed generation. Absent generation = * never committed; present = committed. A scan can never see an uncommitted * fact. * * The manifest (`_generations/facts/manifest.json`, JSON — forensics stay * terminal-readable) is the single source of truth for the segment SET; * rotation flips it atomically (write-new → fsync → rename) BEFORE the new * tail's first byte exists, so no segment file is ever unaccounted for. */ import { encode as defaultEncode, decode as defaultDecode } from '@msgpack/msgpack' import { crc32c } from '../utils/crc32c.js' import { prodLog } from '../utils/logger.js' // Swappable msgpack implementation — defaults to the JS codec; a native // provider (registered via the plugin registry's 'msgpack' key) may replace // it. Byte-compatibility is the contract (positional arrays, bin16 ids). let msgpackEncode: (value: unknown) => Uint8Array = defaultEncode let msgpackDecode: (bytes: Uint8Array) => unknown = defaultDecode /** Replace the msgpack encode/decode implementation at runtime. */ export function setFactCodec(impl: { encode: (value: unknown) => Uint8Array decode: (bytes: Uint8Array) => unknown }): void { msgpackEncode = impl.encode msgpackDecode = impl.decode } /** Storage-root-relative home of the fact log. */ export const FACTS_PREFIX = '_generations/facts' /** The facts manifest path (JSON). */ export const FACTS_MANIFEST_PATH = `${FACTS_PREFIX}/manifest.json` /** Current segment format version (header field; additive-only within a major). */ export const FACTS_FORMAT_VERSION = 1 /** Rotation threshold: seal the tail segment once it exceeds this many bytes. */ const SEGMENT_ROTATE_BYTES = 8 * 1024 * 1024 /** Segment header: magic(8) + formatVersion(4) + firstGeneration(8) + reserved(12). */ const HEADER_BYTES = 32 const MAGIC = new Uint8Array([0x42, 0x46, 0x41, 0x43, 0x54, 0x53, 0x00, 0x00]) // "BFACTS\0\0" /** Frame prefix: length(4) + crc32c(4). */ const FRAME_PREFIX_BYTES = 8 /** One write inside a fact: what the id became (or a tombstone). */ export interface FactOp { kind: 'noun' | 'verb' id: string /** The AFTER-IMAGE legs, or `null` for a tombstone (the id was removed). */ record: { metadata: unknown | null; vector: unknown | null } | null } /** One committed generation, as scanned back out of the log. */ export interface CommitFact { generation: number timestamp: number ops: FactOp[] meta?: Record blobHashes?: string[] } /** The telemetry a scan batch carries (frozen shape). */ export interface FactScanBatch { facts: CommitFact[] firstGeneration: number lastGeneration: number factCount: number byteSize: number segmentId: string } /** The telemetry a scan OPEN returns (frozen shape). */ export interface FactScanHandle { headGeneration: number segmentCount: number approxFactCount: number /** Ordered batches; a detected gap aborts LOUDLY, never a silent skip. */ batches: () => AsyncGenerator /** Close telemetry — the invariant cross-check, valid after iteration ends. */ summary: () => { factsYielded: number; segmentsRead: number } } /** Manifest entry for a sealed segment. */ interface SegmentEntry { file: string firstGeneration: number lastGeneration: number facts: number bytes: number } /** The facts manifest (JSON on disk). */ interface FactsManifest { formatVersion: number segments: SegmentEntry[] /** The append target. Its true content is established by scanning (crash tolerance). */ tailSegment: string | null updatedAt: string } /** The narrow byte-level storage surface the fact log rides. */ export interface FactLogStorage { appendRawBytes(path: string, bytes: Uint8Array): Promise readRawBytes(path: string): Promise writeRawBytes(path: string, bytes: Uint8Array): Promise rawByteSize(path: string): Promise readRawObject(path: string): Promise writeRawObject(path: string, data: any): Promise syncRawObjects(paths: string[]): Promise deleteRawObject(path: string): Promise } /** True when the storage adapter exposes every primitive the fact log needs. */ export function storageSupportsFactLog(storage: unknown): storage is FactLogStorage { const s = storage as Record return ( typeof s.appendRawBytes === 'function' && typeof s.readRawBytes === 'function' && typeof s.writeRawBytes === 'function' && typeof s.rawByteSize === 'function' ) } /** uuid string → 16 raw bytes (bin16 on the wire). */ function uuidToBytes(id: string): Uint8Array { const hex = id.replace(/-/g, '') if (hex.length !== 32) { // Non-uuid ids (legacy/natural keys) ride as UTF-8 with a length prefix // marker impossible for uuids: we refuse instead — the write API has // guaranteed uuid ids since 8.0, so anything else is a corruption signal. throw new Error(`fact log: id is not a uuid: ${id}`) } const bytes = new Uint8Array(16) for (let i = 0; i < 16; i++) { bytes[i] = parseInt(hex.slice(i * 2, i * 2 + 2), 16) } return bytes } /** 16 raw bytes → canonical lowercase uuid string. */ function bytesToUuid(bytes: Uint8Array): string { let hex = '' for (let i = 0; i < 16; i++) hex += bytes[i].toString(16).padStart(2, '0') return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20)}` } /** Zero-padded segment filename: lexicographic order == generation order. */ function segmentFileName(firstGeneration: number): string { return `seg-${String(firstGeneration).padStart(20, '0')}.bfl` } /** Build a segment header. Reserved bytes are ZEROED (and verified on open). */ function buildHeader(firstGeneration: number): Uint8Array { const header = new Uint8Array(HEADER_BYTES) header.set(MAGIC, 0) const view = new DataView(header.buffer) view.setUint32(8, FACTS_FORMAT_VERSION, true) view.setBigUint64(12, BigInt(firstGeneration), true) // bytes 20..31 stay zero (reserved) return header } /** Encode one fact into a framed record (length + crc32c + msgpack payload). */ function encodeFrame(fact: CommitFact): Uint8Array { const payload = msgpackEncode([ fact.generation, fact.timestamp, fact.ops.map((op) => [ op.kind === 'noun' ? 0 : 1, uuidToBytes(op.id), op.record === null ? null : [op.record.metadata, op.record.vector] ]), fact.meta ?? null, fact.blobHashes && fact.blobHashes.length > 0 ? fact.blobHashes : null ]) const frame = new Uint8Array(FRAME_PREFIX_BYTES + payload.length) const view = new DataView(frame.buffer) view.setUint32(0, payload.length, true) view.setUint32(4, crc32c(payload), true) frame.set(payload, FRAME_PREFIX_BYTES) return frame } /** Decode one msgpack payload back into a CommitFact. */ function decodeFact(payload: Uint8Array): CommitFact { const raw = msgpackDecode(payload) as unknown[] const [generation, timestamp, ops, meta, blobHashes] = raw as [ number, number, Array<[number, Uint8Array, [unknown, unknown] | null]>, Record | null, string[] | null ] return { generation: Number(generation), timestamp: Number(timestamp), ops: ops.map(([kind, idBytes, record]) => ({ kind: kind === 0 ? ('noun' as const) : ('verb' as const), id: bytesToUuid(idBytes), record: record === null ? null : { metadata: record[0] ?? null, vector: record[1] ?? null } })), ...(meta ? { meta } : {}), ...(blobHashes && blobHashes.length > 0 ? { blobHashes } : {}) } } /** * Parse a segment's bytes: verify the header, then walk frames until the end * or a torn tail (length overrun / CRC mismatch), which terminates the walk — * everything before it is intact. Returns the decoded facts plus the byte * length of the VALID prefix (header + intact frames), which reconciliation * uses to cut a torn tail without re-encoding. */ function parseSegment( file: string, bytes: Uint8Array ): { facts: CommitFact[]; validBytes: number } { if (bytes.length < HEADER_BYTES) { prodLog.warn(`[FactLog] segment ${file} shorter than its header — treating as empty`) return { facts: [], validBytes: 0 } } for (let i = 0; i < MAGIC.length; i++) { if (bytes[i] !== MAGIC[i]) { throw new Error(`fact log: segment ${file} has a bad magic — not a fact segment`) } } const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength) const version = view.getUint32(8, true) if (version !== FACTS_FORMAT_VERSION) { throw new Error( `fact log: segment ${file} has formatVersion ${version}; this build reads ${FACTS_FORMAT_VERSION}` ) } for (let i = 20; i < HEADER_BYTES; i++) { if (bytes[i] !== 0) { // Non-zero reserved bytes = a future format this build cannot verify. throw new Error(`fact log: segment ${file} has non-zero reserved header bytes — unverifiable`) } } const facts: CommitFact[] = [] let offset = HEADER_BYTES while (offset + FRAME_PREFIX_BYTES <= bytes.length) { const length = view.getUint32(offset, true) const expectedCrc = view.getUint32(offset + 4, true) const start = offset + FRAME_PREFIX_BYTES const end = start + length if (end > bytes.length) break // torn tail: frame length overruns the file const payload = bytes.subarray(start, end) if (crc32c(payload) !== expectedCrc) break // torn tail: payload CRC mismatch facts.push(decodeFact(payload)) offset = end } return { facts, validBytes: offset } } /** * The generation fact log. One instance per open store; every method assumes * the single-writer discipline the generation store already enforces (calls * arrive under its commit mutex). */ export class FactLog { private readonly storage: FactLogStorage /** Rotation threshold (bytes); tests may lower it to exercise rotation. */ private readonly rotateBytes: number private manifest: FactsManifest = { formatVersion: FACTS_FORMAT_VERSION, segments: [], tailSegment: null, updatedAt: new Date(0).toISOString() } /** Decoded facts of the TAIL segment (bounded by the rotation threshold). */ private tailFacts: CommitFact[] = [] /** Byte size of the tail segment file (valid prefix). */ private tailBytes = 0 /** Highest generation in the log (0 = empty). */ private head = 0 /** Segment paths appended since the last sync (the fsync batch). */ private readonly dirtySegments = new Set() constructor(storage: FactLogStorage, options?: { rotateBytes?: number }) { this.storage = storage this.rotateBytes = options?.rotateBytes ?? SEGMENT_ROTATE_BYTES } /** The highest committed generation the log holds (0 = empty). */ headGeneration(): number { return this.head } /** * Open the log and reconcile it to committed truth: read the manifest, * establish the tail's intact content (torn-tail scan), then TRUNCATE any * fact with `generation > committedGeneration` — those never committed (a * crash between fact-append and the commit point). After open, the log is * exactly the committed prefix. */ 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)) { if (stored.formatVersion !== FACTS_FORMAT_VERSION) { throw new Error( `fact log: manifest formatVersion ${stored.formatVersion}; this build reads ${FACTS_FORMAT_VERSION}` ) } this.manifest = stored } // Drop sealed segments that sit ENTIRELY beyond committed truth (a crash // right after a rotation whose facts never committed), newest first. while (this.manifest.segments.length > 0) { const last = this.manifest.segments[this.manifest.segments.length - 1] if (last.firstGeneration > committedGeneration) { prodLog.warn( `[FactLog] dropping sealed segment ${last.file} (generations ${last.firstGeneration}..` + `${last.lastGeneration} never committed)` ) await this.storage.deleteRawObject(`${FACTS_PREFIX}/${last.file}`) this.manifest.segments.pop() await this.persistManifest() } else if (last.lastGeneration > committedGeneration) { // A sealed segment STRADDLING committed truth: cut it back. await this.truncateSegmentTo(last.file, committedGeneration) const cut = await this.reloadSegmentEntry(last.file) this.manifest.segments[this.manifest.segments.length - 1] = cut await this.persistManifest() break } else { break } } // Establish the tail: scan its intact prefix, then truncate beyond // committed truth (the common crash shape: buffered single-op facts whose // counter never went durable). if (this.manifest.tailSegment) { const tailPath = `${FACTS_PREFIX}/${this.manifest.tailSegment}` const bytes = await this.storage.readRawBytes(tailPath) if (bytes === null) { // Manifest named a tail whose first byte never landed — an empty tail. this.tailFacts = [] this.tailBytes = 0 } else { const { facts, validBytes } = parseSegment(this.manifest.tailSegment, bytes) const kept = facts.filter((f) => f.generation <= committedGeneration) if (kept.length !== facts.length || validBytes !== bytes.length) { const dropped = facts.length - kept.length if (dropped > 0) { prodLog.warn( `[FactLog] truncating ${dropped} uncommitted fact(s) beyond generation ` + `${committedGeneration} from the tail (never committed)` ) } await this.rewriteTail(kept) } else { this.tailFacts = facts this.tailBytes = validBytes } } } this.head = this.computeHead() } /** * Append one committed generation's fact. NOT durable until {@link sync} — * the caller batches durability at its commit barrier (transact syncs in * the same call; Model-B group-commit syncs at flush). */ async append(fact: CommitFact): Promise { if (fact.generation <= this.head) { throw new Error( `fact log: non-monotonic append (generation ${fact.generation} ≤ head ${this.head})` ) } if (this.manifest.tailSegment === null) { await this.startTail(fact.generation) } else if (this.tailBytes >= this.rotateBytes) { await this.rotate(fact.generation) } const frame = encodeFrame(fact) const tailPath = `${FACTS_PREFIX}/${this.manifest.tailSegment}` await this.storage.appendRawBytes(tailPath, frame) this.tailFacts.push(fact) this.tailBytes += frame.length this.head = fact.generation this.dirtySegments.add(tailPath) } /** Fsync every segment appended since the last sync. */ async sync(): Promise { if (this.dirtySegments.size === 0) return const paths = [...this.dirtySegments] this.dirtySegments.clear() await this.storage.syncRawObjects(paths) } /** * Open a scan over committed facts. The scan runs against a MANIFEST * SNAPSHOT (sealed segments + the tail's decoded facts at open) — exactly- * once per fact, inclusive bounds, stable under concurrent appends. Gaps * abort LOUDLY: a missing generation inside a segment's declared range is * corruption, never silently skipped. */ scanFacts(options?: { fromGeneration?: number toGeneration?: number kinds?: Array<'noun' | 'verb'> batchSize?: number }): FactScanHandle { const from = options?.fromGeneration ?? 1 const to = options?.toGeneration ?? this.head const kinds = options?.kinds const batchSize = Math.max(1, options?.batchSize ?? 256) // Snapshot: the segment list + tail content as of NOW. const segments = this.manifest.segments.filter( (s) => s.lastGeneration >= from && s.firstGeneration <= to ) const tailSnapshot = this.tailFacts.filter((f) => f.generation >= from && f.generation <= to) const tailId = this.manifest.tailSegment ?? 'tail' const approxFactCount = segments.reduce((sum, s) => sum + s.facts, 0) + tailSnapshot.length let factsYielded = 0 let segmentsRead = 0 const storage = this.storage async function* batches(this: void): AsyncGenerator { let expectedNext = 0 // gap detection: generations are monotonic, not necessarily dense const emit = (facts: CommitFact[], segmentId: string, byteSize: number): FactScanBatch => ({ facts, firstGeneration: facts[0].generation, lastGeneration: facts[facts.length - 1].generation, factCount: facts.length, byteSize, segmentId }) const filterOps = (fact: CommitFact): CommitFact => kinds ? { ...fact, ops: fact.ops.filter((op) => kinds.includes(op.kind)) } : fact for (const entry of segments) { const bytes = await storage.readRawBytes(`${FACTS_PREFIX}/${entry.file}`) if (bytes === null) { throw new Error( `fact log: sealed segment ${entry.file} is MISSING — the log is damaged; aborting scan` ) } const { facts } = parseSegment(entry.file, bytes) segmentsRead++ const inRange = facts.filter((f) => f.generation >= from && f.generation <= to) for (const f of inRange) { if (f.generation <= expectedNext - 1) { throw new Error(`fact log: out-of-order fact ${f.generation} in ${entry.file} — aborting scan`) } expectedNext = f.generation + 1 } for (let i = 0; i < inRange.length; i += batchSize) { const slice = inRange.slice(i, i + batchSize).map(filterOps) if (slice.length === 0) continue factsYielded += slice.length yield emit(slice, entry.file, slice.reduce((n, f) => n + encodeFrame(f).length, 0)) } } if (tailSnapshot.length > 0) { segmentsRead++ for (const f of tailSnapshot) { if (f.generation <= expectedNext - 1) { throw new Error(`fact log: out-of-order fact ${f.generation} in the tail — aborting scan`) } expectedNext = f.generation + 1 } for (let i = 0; i < tailSnapshot.length; i += batchSize) { const slice = tailSnapshot.slice(i, i + batchSize).map(filterOps) factsYielded += slice.length yield emit(slice, tailId, slice.reduce((n, f) => n + encodeFrame(f).length, 0)) } } } return { headGeneration: this.head, segmentCount: segments.length + (tailSnapshot.length > 0 ? 1 : 0), approxFactCount, batches, summary: () => ({ factsYielded, segmentsRead }) } } /** * The mmap fast path (capability handoff): the immutable sealed segment * files covering `fromGeneration`, in order. The TAIL is deliberately NOT * included — it is append-mutable; consumers read it via {@link scanFacts}. */ segmentPaths(options?: { fromGeneration?: number }): string[] { const from = options?.fromGeneration ?? 1 return this.manifest.segments .filter((s) => s.lastGeneration >= from) .map((s) => `${FACTS_PREFIX}/${s.file}`) } /** * Drop every fact with `generation > keepThrough` — the in-session abort * compensation: a transact appends its fact BEFORE the commit point, so a * real (non-crash) abort after the append must take the fact back out. The * dropped facts can only live in the TAIL (they were just appended); the * rewrite is atomic and bounded by the rotation threshold. */ async dropAbove(keepThrough: number): Promise { if (this.head <= keepThrough) return const kept = this.tailFacts.filter((f) => f.generation <= keepThrough) if (kept.length === this.tailFacts.length) { throw new Error( `fact log: dropAbove(${keepThrough}) found no droppable facts in the tail ` + `(head ${this.head}) — the fact to drop was already sealed; the log needs reopen` ) } await this.rewriteTail(kept) this.head = this.computeHead() } // -- internals ------------------------------------------------------------- private computeHead(): number { if (this.tailFacts.length > 0) return this.tailFacts[this.tailFacts.length - 1].generation const sealed = this.manifest.segments if (sealed.length > 0) return sealed[sealed.length - 1].lastGeneration return 0 } /** Create the very first tail segment (manifest-first, then header bytes). */ private async startTail(firstGeneration: number): Promise { const file = segmentFileName(firstGeneration) this.manifest.tailSegment = file await this.persistManifest() await this.storage.appendRawBytes(`${FACTS_PREFIX}/${file}`, buildHeader(firstGeneration)) this.tailFacts = [] this.tailBytes = HEADER_BYTES } /** * Seal the tail into the manifest and start a new one. Manifest-first: the * flip both seals the old tail AND names the new one atomically, so no * segment file ever exists unaccounted for. */ private async rotate(nextGeneration: number): Promise { const sealedFile = this.manifest.tailSegment if (!sealedFile) return // Seal what the tail actually holds. await this.sync() // sealed segments are always fully durable const entry: SegmentEntry = { file: sealedFile, firstGeneration: this.tailFacts[0]?.generation ?? nextGeneration, lastGeneration: this.tailFacts[this.tailFacts.length - 1]?.generation ?? nextGeneration - 1, facts: this.tailFacts.length, bytes: this.tailBytes } const newFile = segmentFileName(nextGeneration) this.manifest.segments.push(entry) this.manifest.tailSegment = newFile await this.persistManifest() await this.storage.appendRawBytes(`${FACTS_PREFIX}/${newFile}`, buildHeader(nextGeneration)) this.tailFacts = [] this.tailBytes = HEADER_BYTES } /** Atomically persist the manifest (write-new → fsync → rename downstream). */ private async persistManifest(): Promise { this.manifest.updatedAt = new Date().toISOString() await this.storage.writeRawObject(FACTS_MANIFEST_PATH, this.manifest) await this.storage.syncRawObjects([FACTS_MANIFEST_PATH]) } /** Rewrite the tail segment to hold exactly `facts` (atomic replace). */ private async rewriteTail(facts: CommitFact[]): Promise { const file = this.manifest.tailSegment if (!file) return const first = facts[0]?.generation ?? this.segmentFirstGenerationFromName(file) const parts: Uint8Array[] = [buildHeader(first)] for (const f of facts) parts.push(encodeFrame(f)) const total = parts.reduce((n, p) => n + p.length, 0) const merged = new Uint8Array(total) let offset = 0 for (const p of parts) { merged.set(p, offset) offset += p.length } await this.storage.writeRawBytes(`${FACTS_PREFIX}/${file}`, merged) this.tailFacts = facts this.tailBytes = total } /** Cut a SEALED segment back to `committedGeneration` (atomic replace). */ private async truncateSegmentTo(file: string, committedGeneration: number): Promise { const path = `${FACTS_PREFIX}/${file}` const bytes = await this.storage.readRawBytes(path) if (bytes === null) return const { facts } = parseSegment(file, bytes) const kept = facts.filter((f) => f.generation <= committedGeneration) prodLog.warn( `[FactLog] truncating sealed segment ${file} to generation ${committedGeneration} ` + `(${facts.length - kept.length} uncommitted fact(s) dropped)` ) const first = kept[0]?.generation ?? this.segmentFirstGenerationFromName(file) const parts: Uint8Array[] = [buildHeader(first)] for (const f of kept) parts.push(encodeFrame(f)) const total = parts.reduce((n, p) => n + p.length, 0) const merged = new Uint8Array(total) let offset = 0 for (const p of parts) { merged.set(p, offset) offset += p.length } await this.storage.writeRawBytes(path, merged) } /** Re-derive a sealed segment's manifest entry from its actual bytes. */ private async reloadSegmentEntry(file: string): Promise { const bytes = await this.storage.readRawBytes(`${FACTS_PREFIX}/${file}`) const { facts, validBytes } = bytes ? parseSegment(file, bytes) : { facts: [] as CommitFact[], validBytes: 0 } return { file, firstGeneration: facts[0]?.generation ?? this.segmentFirstGenerationFromName(file), lastGeneration: facts[facts.length - 1]?.generation ?? 0, facts: facts.length, bytes: validBytes } } /** Parse the zero-padded firstGeneration back out of a segment filename. */ private segmentFirstGenerationFromName(file: string): number { const match = /^seg-(\d{20})\.bfl$/.exec(file) return match ? Number(match[1]) : 0 } }