feat: generation fact log — after-image commit records, dual-written at every commit point
Every committed generation now also appends a FACT — an after-image commit record (what each touched entity/relationship became, or a body-less tombstone for a removal) — to an append-only, crc32c-framed segment log under _generations/facts/. The before-image history and the canonical tree remain authoritative; the fact log gives consumers ONE sequential, self-verifying stream (index heals, incremental replays) in place of a per-entity directory walk. - Wire format: positional msgpack facts [generation, timestamp, ops, meta, blobHashes]; op = [kind u8, id bin16, record | nil tombstone]; 32-byte segment header (magic, formatVersion, firstGeneration, zeroed+verified reserved); length+crc32c frame per fact; zero-padded segment names so lexicographic order == generation order; JSON manifest with an atomic rename flip, manifest-first rotation. - Commit protocol: facts append+fsync BEFORE the commit point inside the existing durability window, so a crash can only leave the log AHEAD of committed truth — open() truncates back (torn tails detected by CRC). Absent generation = never committed; a scan can never see an uncommitted fact. transact() facts are durable-on-return; single-op facts ride the group-commit flush exactly like buffered history. A fact-append failure fails the write, loudly — a silent gap would be a lie a later replay discovers. - New public surface: brain.scanFacts() (sequential batches with heal telemetry: head/segments/approx up front, per-batch generation range + bytes + segment id, loud abort on gaps, summary cross-check) and brain.factSegmentPaths() (immutable sealed segments for zero-copy consumers; the mutable tail excluded). Exported types CommitFact, FactOp, FactScanBatch, FactScanHandle. - Storage: optional binary raw-byte primitives (appendRawBytes, readRawBytes, writeRawBytes, rawByteSize) on StorageAdapter — feature-detected; filesystem + memory adapters implement them; an adapter without them hosts no fact log. Fact segments are byte-copied (never hard-linked) into snapshots. The _generations/facts/ namespace is registered as a protected family (rebuildable: false): no sweeper or GC may delete under it. - New crc32c (Castagnoli) utility with RFC known-answer tests.
This commit is contained in:
parent
92299f27be
commit
38b0041464
13 changed files with 1493 additions and 4 deletions
|
|
@ -515,7 +515,15 @@ describe('8.0 Db API — generational MVCC', () => {
|
|||
await brain.transact([{ op: 'update', id: uid('compact-e'), metadata: { v: 4 } }])
|
||||
).release()
|
||||
|
||||
const recordsBefore = (await storage.listRawObjects('_generations')).length
|
||||
// History record-sets only — the generation FACT LOG also lives under
|
||||
// `_generations/` (at `facts/`) and is deliberately NOT reclaimed by
|
||||
// history compaction (facts are the future canonical, not undo history).
|
||||
const historyRecords = async (): Promise<number> =>
|
||||
(await storage.listRawObjects('_generations')).filter(
|
||||
(p: string) => !p.startsWith('_generations/facts/')
|
||||
).length
|
||||
|
||||
const recordsBefore = await historyRecords()
|
||||
expect(recordsBefore).toBeGreaterThan(0)
|
||||
|
||||
// Compact while pinned: record-sets above the pin survive, pinned reads stay correct.
|
||||
|
|
@ -528,7 +536,7 @@ describe('8.0 Db API — generational MVCC', () => {
|
|||
const second = await brain.compactHistory()
|
||||
expect(first.removedGenerations + second.removedGenerations).toBeGreaterThan(0)
|
||||
|
||||
const recordsAfter = (await storage.listRawObjects('_generations')).length
|
||||
const recordsAfter = await historyRecords()
|
||||
expect(recordsAfter).toBeLessThan(recordsBefore)
|
||||
expect(recordsAfter).toBe(0)
|
||||
|
||||
|
|
|
|||
200
tests/integration/fact-log-dual-write.test.ts
Normal file
200
tests/integration/fact-log-dual-write.test.ts
Normal file
|
|
@ -0,0 +1,200 @@
|
|||
/**
|
||||
* @module tests/integration/fact-log-dual-write
|
||||
* @description The generation fact log end-to-end through real commits: every
|
||||
* committed generation (single-op AND transact) appends its AFTER-IMAGE fact
|
||||
* at the commit point; removals append body-less tombstones; an aborted
|
||||
* transaction leaves no fact; facts survive reopen and continue monotonically;
|
||||
* the scan surface (brain.scanFacts) carries the frozen telemetry shape; and
|
||||
* the fact-log namespace is protected against prefix-nuking.
|
||||
*/
|
||||
import { describe, it, expect, beforeEach, afterEach } from 'vitest'
|
||||
import * as fs from 'node:fs'
|
||||
import * as os from 'node:os'
|
||||
import * as path from 'node:path'
|
||||
import { Brainy, ProtectedArtifactError, type CommitFact } from '../../src/index.js'
|
||||
|
||||
async function allFacts(brain: any): Promise<CommitFact[]> {
|
||||
const scan = brain.scanFacts()
|
||||
expect(scan).not.toBeNull()
|
||||
const facts: CommitFact[] = []
|
||||
for await (const batch of scan!.batches()) facts.push(...batch.facts)
|
||||
return facts
|
||||
}
|
||||
|
||||
describe('fact log dual-write (memory adapter)', () => {
|
||||
let brain: any
|
||||
|
||||
beforeEach(async () => {
|
||||
process.env.BRAINY_DETERMINISTIC_EMBEDDINGS = 'true'
|
||||
brain = new Brainy({ requireSubtype: false, storage: { type: 'memory' }, silent: true, dimensions: 384 })
|
||||
await brain.init()
|
||||
})
|
||||
afterEach(async () => {
|
||||
await brain.close?.().catch(() => {})
|
||||
})
|
||||
|
||||
it('every single-op write appends its after-image fact; a remove appends a tombstone', async () => {
|
||||
const id = await brain.add({ data: 'first', type: 'document', metadata: { rev: 1 } })
|
||||
await brain.update({ id, metadata: { rev: 2 } })
|
||||
await brain.remove(id)
|
||||
|
||||
const facts = await allFacts(brain)
|
||||
// add + update + remove each committed a generation (the remove may span
|
||||
// cascade ops but is ONE generation). Facts are monotonic.
|
||||
const gens = facts.map((f) => f.generation)
|
||||
expect([...gens].sort((a, b) => a - b)).toEqual(gens)
|
||||
expect(facts.length).toBeGreaterThanOrEqual(3)
|
||||
|
||||
// The add fact carries the after-image of the new entity.
|
||||
const addFact = facts.find((f) => f.ops.some((op) => op.id === id && op.record !== null))
|
||||
expect(addFact).toBeDefined()
|
||||
|
||||
// The remove fact carries a body-less tombstone for the id.
|
||||
const removeFact = facts[facts.length - 1]
|
||||
const tombstone = removeFact.ops.find((op) => op.id === id)
|
||||
expect(tombstone).toBeDefined()
|
||||
expect(tombstone!.record).toBeNull()
|
||||
expect(tombstone!.kind).toBe('noun')
|
||||
})
|
||||
|
||||
it('the update fact holds the NEW state (after-image, not before)', async () => {
|
||||
const id = await brain.add({ data: 'versioned', type: 'document', metadata: { v: 'old' } })
|
||||
await brain.update({ id, metadata: { v: 'new' } })
|
||||
|
||||
const facts = await allFacts(brain)
|
||||
const updateFact = facts[facts.length - 1]
|
||||
const op = updateFact.ops.find((o) => o.id === id)!
|
||||
expect(op.record).not.toBeNull()
|
||||
expect((op.record!.metadata as any).v).toBe('new')
|
||||
})
|
||||
|
||||
it('a transact commits ONE fact carrying all its ops, with meta', async () => {
|
||||
const receipt = await brain.transact(
|
||||
[
|
||||
{ op: 'add', type: 'document', metadata: { part: 1 }, data: 'a' },
|
||||
{ op: 'add', type: 'document', metadata: { part: 2 }, data: 'b' }
|
||||
],
|
||||
{ meta: { source: 'batch-import' } }
|
||||
)
|
||||
|
||||
const facts = await allFacts(brain)
|
||||
const txFact = facts.find((f) => f.generation === receipt.generation)
|
||||
expect(txFact).toBeDefined()
|
||||
expect(txFact!.ops.filter((op) => op.kind === 'noun').length).toBeGreaterThanOrEqual(2)
|
||||
expect(txFact!.meta).toEqual({ source: 'batch-import' })
|
||||
})
|
||||
|
||||
it('an aborted transact leaves NO fact (absent = never committed)', async () => {
|
||||
const id = await brain.add({ data: 'cas target', type: 'document', metadata: { n: 1 } })
|
||||
const before = (await allFacts(brain)).length
|
||||
|
||||
await expect(
|
||||
brain.transact([{ op: 'update', id, ifRev: 999, metadata: { n: 2 } }])
|
||||
).rejects.toThrow()
|
||||
|
||||
const after = await allFacts(brain)
|
||||
expect(after.length).toBe(before)
|
||||
})
|
||||
|
||||
it('fact generations line up with the transaction log', async () => {
|
||||
await brain.add({ data: 'x', type: 'document', metadata: {} })
|
||||
await brain.add({ data: 'y', type: 'document', metadata: {} })
|
||||
await brain.flush()
|
||||
|
||||
const facts = await allFacts(brain)
|
||||
const logGens = new Set((await brain.transactionLog()).map((e: any) => e.generation))
|
||||
for (const f of facts) {
|
||||
expect(logGens.has(f.generation)).toBe(true)
|
||||
}
|
||||
})
|
||||
|
||||
it('scan telemetry carries the frozen shape end-to-end', async () => {
|
||||
for (let i = 0; i < 5; i++) await brain.add({ data: `t${i}`, type: 'document', metadata: { i } })
|
||||
|
||||
const scan = brain.scanFacts({ batchSize: 2 })!
|
||||
expect(scan.headGeneration).toBeGreaterThanOrEqual(5)
|
||||
expect(scan.approxFactCount).toBeGreaterThanOrEqual(5)
|
||||
let batches = 0
|
||||
for await (const b of scan.batches()) {
|
||||
batches++
|
||||
expect(b.factCount).toBe(b.facts.length)
|
||||
expect(b.firstGeneration).toBe(b.facts[0].generation)
|
||||
expect(b.lastGeneration).toBe(b.facts[b.facts.length - 1].generation)
|
||||
expect(b.byteSize).toBeGreaterThan(0)
|
||||
expect(typeof b.segmentId).toBe('string')
|
||||
}
|
||||
expect(batches).toBeGreaterThan(1)
|
||||
expect(scan.summary().factsYielded).toBe(scan.approxFactCount)
|
||||
})
|
||||
})
|
||||
|
||||
describe('fact log dual-write (filesystem adapter — durability + protection)', () => {
|
||||
let dir: string
|
||||
let brain: any
|
||||
|
||||
const open = async () => {
|
||||
const b: any = new Brainy({
|
||||
requireSubtype: false,
|
||||
storage: { type: 'filesystem', path: dir },
|
||||
silent: true,
|
||||
dimensions: 384
|
||||
})
|
||||
await b.init()
|
||||
return b
|
||||
}
|
||||
|
||||
beforeEach(async () => {
|
||||
process.env.BRAINY_DETERMINISTIC_EMBEDDINGS = 'true'
|
||||
dir = fs.mkdtempSync(path.join(os.tmpdir(), 'brainy-factlog-'))
|
||||
brain = await open()
|
||||
})
|
||||
afterEach(async () => {
|
||||
await brain.close?.().catch(() => {})
|
||||
fs.rmSync(dir, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
it('facts survive close + reopen and appends continue monotonically', async () => {
|
||||
const id = await brain.add({ data: 'persist me', type: 'document', metadata: { k: 1 } })
|
||||
await brain.remove(id)
|
||||
await brain.close()
|
||||
|
||||
brain = await open()
|
||||
const facts = await allFacts(brain)
|
||||
expect(facts.length).toBeGreaterThanOrEqual(2)
|
||||
const headBefore = facts[facts.length - 1].generation
|
||||
|
||||
await brain.add({ data: 'after reopen', type: 'document', metadata: { k: 2 } })
|
||||
const facts2 = await allFacts(brain)
|
||||
expect(facts2[facts2.length - 1].generation).toBeGreaterThan(headBefore)
|
||||
})
|
||||
|
||||
it('the fact segments exist on disk under _generations/facts/ with zero-padded names', async () => {
|
||||
await brain.add({ data: 'on disk', type: 'document', metadata: {} })
|
||||
await brain.flush()
|
||||
const factsDir = path.join(dir, '_generations', 'facts')
|
||||
const files = fs.readdirSync(factsDir)
|
||||
// The manifest rides the store's JSON object discipline (gzip on disk).
|
||||
expect(files.some((f) => f.startsWith('manifest.json'))).toBe(true)
|
||||
const segs = files.filter((f) => /^seg-\d{20}\.bfl$/.test(f))
|
||||
expect(segs.length).toBeGreaterThanOrEqual(1)
|
||||
})
|
||||
|
||||
it('the fact-log namespace is PROTECTED: a prefix-nuke is refused', async () => {
|
||||
await brain.add({ data: 'protected', type: 'document', metadata: {} })
|
||||
await expect(brain.storage.removeRawPrefix('_generations/facts')).rejects.toBeInstanceOf(
|
||||
ProtectedArtifactError
|
||||
)
|
||||
// Per-generation history cleanup remains unaffected (no false intersect).
|
||||
await expect(brain.storage.removeRawPrefix('_generations/999999')).resolves.toBeUndefined()
|
||||
})
|
||||
|
||||
it('transact facts are durable-on-return (no flush needed before reopen)', async () => {
|
||||
const receipt = await brain.transact([
|
||||
{ op: 'add', type: 'document', metadata: { durable: true }, data: 'tx' }
|
||||
])
|
||||
// Simulate an abrupt end: no flush(), no close() — reopen from disk.
|
||||
brain = await open()
|
||||
const facts = await allFacts(brain)
|
||||
expect(facts.some((f) => f.generation === receipt.generation)).toBe(true)
|
||||
})
|
||||
})
|
||||
Loading…
Add table
Add a link
Reference in a new issue