/** * @module tests/unit/db/bounded-chains * @description Correctness + RAM-boundedness oracle for the bounded per-id * generation history chains in `src/db/generationStore.ts` (GA #33, Approach C: * hot-tail window + bounded cold LRU + a mutex-free bulk `resolveManyAt`). * * The old design kept ONE resident chain per id ever touched across retained * history (`Map`) → O(distinct-ids) RAM, which defeats * billion-scale time travel. The replacement bounds RAM to O(W·d̄)/O(L) — a * resident window over the newest `W` generations plus a bounded LRU of * reconstructed deep-pin chains — while staying EXACTLY correct. * * These tests are the correctness oracle for that change: * * 1. Oracle vs brute-force: for every id × every pin, `resolveAt` must equal a * brute-force scan of every reserved generation's persisted `tx.json`. * 2. Bounded RAM independent of N: the resident structures stay O(W)/O(L) no * matter how many distinct ids the corpus touched. * 3. Held-Db across cold eviction + a concurrent compaction (lock-light). * 4. Un-flushed pending single-ops resolve identically before and after flush. * 5. A deep-pin materialize is O(R) `getDelta` (not O(N·R)) and never deadlocks * inside the mutex-held reconciliation pass. * 6. `resolveManyAt` ≡ per-id `resolveAt` for random id-sets × random pins. * 7. The write path is I/O-free across window slides (the inverse-index ring * supplies slid-out ids — zero `tx.json` reads on commit). * 8. Compaction invalidates + rebuilds the window over survivors; a pin below * the new horizon throws `GenerationCompactedError`. */ import { describe, it, expect, beforeEach, afterEach } from 'vitest' import { MemoryStorage } from '../../../src/storage/adapters/memoryStorage.js' import { GenerationStore } from '../../../src/db/generationStore.js' import { GenerationCompactedError } from '../../../src/db/errors.js' import { Brainy } from '../../../src/index.js' import { NounType } from '../../../src/types/graphTypes.js' import { createTestConfig, generateTestVector } from '../../helpers/test-factory.js' /** A precomputed embedding so adds skip the (slow) embedding model — these tests * exercise the generation layer, not semantics. */ const VEC = generateTestVector() /** Deterministic UUID-shaped id (the sharded storage layout derives the shard * from the UUID hex), unique per `n`. */ function uid(n: number): string { return `00000000-0000-4000-8000-${n.toString(16).padStart(12, '0')}` } /** Stored-metadata fixture carrying a `version` the oracle compares on. */ function fixture(version: number): Record { return { noun: NounType.Document, subtype: 'note', data: `payload-v${version}`, version, createdAt: 1000, updatedAt: 1000 + version, _rev: version } } /** A small deterministic PRNG so the random oracle runs are reproducible. */ function mulberry32(seed: number): () => number { let a = seed >>> 0 return () => { a |= 0 a = (a + 0x6d2b79f5) | 0 let t = Math.imul(a ^ (a >>> 15), 1 | a) t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t return ((t ^ (t >>> 14)) >>> 0) / 4294967296 } } /** Reject after `ms` so a deadlock surfaces as a test failure, not a hang. */ function withTimeout(p: Promise, ms: number, label: string): Promise { return Promise.race([ p, new Promise((_, reject) => setTimeout(() => reject(new Error(`timed out after ${ms}ms (likely deadlock): ${label}`)), ms) ) ]) } /** Normalize a `resolveAt` result to a comparable scalar. */ function norm(r: { source: string; metadata?: any }): string { if (r.source === 'current') return 'current' if (r.source === 'absent') return 'absent' return `rec:${r.metadata?.version}` } describe('db/GenerationStore — bounded per-id chains (GA #33)', () => { let storage: MemoryStorage let store: GenerationStore beforeEach(async () => { storage = new MemoryStorage() await storage.init() store = new GenerationStore(storage) await store.open() }) /** Commit one single-op noun write (its own generation); returns the gen. */ async function writeNoun(id: string, version: number): Promise { const { generation } = await store.commitSingleOp({ touched: { nouns: [id] }, execute: async () => { await storage.saveNounMetadata(id, fixture(version)) } }) return generation } /** Commit one single-op verb write (raw, so before-images are captured). */ async function writeVerb(id: string, version: number): Promise { const { generation } = await store.commitSingleOp({ touched: { verbs: [id] }, execute: async () => { await storage.writeVerbRaw(id, { metadata: { ...fixture(version), verb: 'RelatedTo' }, vector: { sourceId: uid(900001), targetId: uid(900002), verb: 'RelatedTo' } }) } }) return generation } /** Brute-force `resolveAt`: scan EVERY reserved generation's persisted tx.json * (the gens must be flushed) for the first one after `pin` that touched `id`, * then read its before-image. Fully independent of the resident chains. */ async function bruteForce( kind: 'noun' | 'verb', id: string, pin: number, maxGen: number ): Promise { for (let g = pin + 1; g <= maxGen; g++) { const delta = (await storage.readRawObject(`_generations/${g}/tx.json`)) as | { nouns: string[]; verbs: string[] } | null if (!delta) continue const touched = kind === 'noun' ? delta.nouns : delta.verbs if (Array.isArray(touched) && touched.includes(id)) { const rec = (await storage.readRawObject(`_generations/${g}/prev/${id}.json`)) as | { metadata: any; vector: any } | null if (!rec || (rec.metadata === null && rec.vector === null)) return 'absent' return `rec:${rec.metadata?.version}` } } return 'current' } // ========================================================================== // 1. Oracle vs brute-force — every id × every pin, nouns + verbs. // ========================================================================== it('resolveAt equals a brute-force tx.json scan for every id × every pin (recent, deep, boundary)', async () => { const s = store as any s.recentWindowGenerations = 8 // W s.coldChainLruMax = 4 // L const nounIds = Array.from({ length: 5 }, (_, i) => uid(100 + i)) const verbIds = Array.from({ length: 5 }, (_, i) => uid(200 + i)) const rand = mulberry32(1234) let v = 0 // 60 overlapping single-op generations. Trigger a historical read at the // halfway point so the SECOND half exercises the live slide path // (extendChains + window advance), not just the lazy build. for (let i = 0; i < 60; i++) { if (i === 30) { // Force the window to build so the rest of the writes drive slides. await store.resolveAt('noun', nounIds[0], 1) } if (rand() < 0.5) { await writeNoun(nounIds[Math.floor(rand() * nounIds.length)], ++v) } else { await writeVerb(verbIds[Math.floor(rand() * verbIds.length)], ++v) } } // Flush so every reserved generation has a persisted tx.json for the oracle. await store.flushPendingSingleOps() const maxGen = store.generation() // Window must have actually slid below the head (proves the slide path ran). expect(s.windowLo).toBeGreaterThan(1) expect(s.windowLo).toBe(Math.max(1, maxGen - 8 + 1)) // Every id × every pin from 0..maxGen — covers recent (≥windowLo), deep // ( { const s = store as any s.recentWindowGenerations = 4 s.coldChainLruMax = 16 const X = uid(300) const filler = (i: number) => uid(400 + i) const g1 = await writeNoun(X, 1) // X = v1 // Push X out of the window with filler writes (X untouched). for (let i = 0; i < 10; i++) await writeNoun(filler(i), 100 + i) // Deep read at g1 BEFORE any later touch: X untouched after g1 → 'current'. expect(norm(await store.resolveAt('noun', X, g1))).toBe('current') expect((await store.resolveAt('noun', X, g1)).source).toBe('current') // now cached cold // Now touch X again (v2). The stale cached cold chain would still answer // 'current' → reading the NEW live value (v1 corrupted to v2). extendChains // must invalidate the cold entry so the re-read resolves the g1 before-image. await writeNoun(X, 2) await store.flushPendingSingleOps() const resolved = await store.resolveAt('noun', X, g1) expect(resolved.source).toBe('record') if (resolved.source === 'record') expect(resolved.metadata.version).toBe(1) }) // ========================================================================== // 2. Bounded RAM independent of N. // ========================================================================== it('resident window/cold structures stay O(W)/O(L), independent of distinct-id count', async () => { const s = store as any const W = 8 const L = 4 s.recentWindowGenerations = W s.coldChainLruMax = L const N = 1200 // distinct ids, one single-op each for (let i = 0; i < N; i++) await writeNoun(uid(1000 + i), i) await store.flushPendingSingleOps() const maxGen = store.generation() // Build the window (first historical read). await store.resolveAt('noun', uid(1000), maxGen) // The hot-tail window holds at most W generations' worth of ids; one id per // gen here ⇒ exactly W resident recent chains + W inverse-index entries. expect(s.recentNounChains.size).toBeLessThanOrEqual(W) expect(s.windowNounDeltas.size).toBeLessThanOrEqual(W) expect(s.recentNounChains.size).toBeLessThan(N) // Deep-read many distinct ids → the cold LRU is capped at L regardless. for (let i = 0; i < 40; i++) await store.resolveAt('noun', uid(1000 + i), 1) expect(s.coldNounChains.size).toBeLessThanOrEqual(L) }) // ========================================================================== // 3. Held-Db across cold eviction + a concurrent compaction (lock-light). // ========================================================================== it('a held pin reads correctly across cold eviction AND a concurrent compact()', async () => { const s = store as any s.recentWindowGenerations = 4 s.coldChainLruMax = 4 const heldIds = [uid(500), uid(501), uid(502)] const filler = (i: number) => uid(600 + i) // Each held id gets value 10/11/12 at g_old, then a later mutation so a // before-image exists to resolve. const gOldVals = new Map() let pOld = 0 for (let k = 0; k < heldIds.length; k++) { const val = 10 + k pOld = await writeNoun(heldIds[k], val) gOldVals.set(heldIds[k], val) } const gOld = pOld // pin: at gOld every held id holds its value above store.pin(gOld) // Mutate the held ids AFTER the pin (so resolveAt resolves a before-image), // then slide the window well past gOld with lots of filler writes. for (const id of heldIds) await writeNoun(id, 99) for (let i = 0; i < 30; i++) await writeNoun(filler(i), 700 + i) await store.flushPendingSingleOps() // Populate then evict the held ids' cold chains by cycling other deep ids. for (const id of heldIds) await store.resolveAt('noun', id, gOld) for (let i = 0; i < 12; i++) await store.resolveAt('noun', filler(i), 1) // Re-read the held ids WHILE a compaction (which reclaims everything ≤ the // pin) runs concurrently. Lock-light reconstruction must skip the reclaimed // gens (all ≤ minPinned, never an answer) and still match the at-gOld oracle. const reads = heldIds.map((id) => withTimeout(store.resolveAt('noun', id, gOld), 5000, `held read ${id}`) ) const compaction = withTimeout(store.compact(), 5000, 'concurrent compact') const [r0, r1, r2] = await Promise.all(reads) await compaction for (const [id, res] of [ [heldIds[0], r0], [heldIds[1], r1], [heldIds[2], r2] ] as const) { expect(res.source).toBe('record') if (res.source === 'record') expect(res.metadata.version).toBe(gOldVals.get(id)) } // And a fresh read after compaction still matches (reconstruct over survivors). for (const id of heldIds) { const res = await store.resolveAt('noun', id, gOld) expect(res.source).toBe('record') if (res.source === 'record') expect(res.metadata.version).toBe(gOldVals.get(id)) } store.release(gOld) }) // ========================================================================== // 4. Un-flushed pending single-ops — flush-agnostic before-images. // ========================================================================== it('a now()-style pin reads pre-mutation before-images of un-flushed single-ops, identical after flush', async () => { const X = uid(800) const g1 = await writeNoun(X, 1) // flushed below to make g1 durable history // More UN-FLUSHED single-ops mutating the same id. await writeNoun(X, 2) await writeNoun(X, 3) expect(store.committedGeneration()).toBe(0) // nothing on disk yet const before = await store.resolveAt('noun', X, g1) expect(before.source).toBe('record') if (before.source === 'record') expect(before.metadata.version).toBe(1) // Flush → the answer is identical (resolution is flush-agnostic). await store.flushPendingSingleOps() const after = await store.resolveAt('noun', X, g1) expect(norm(after)).toBe(norm(before)) }) // ========================================================================== // 6. resolveManyAt ≡ per-id resolveAt (random id-sets × random pins). // ========================================================================== it('resolveManyAt equals per-id resolveAt for random id-sets across recent and deep pins', async () => { const s = store as any s.recentWindowGenerations = 6 s.coldChainLruMax = 8 const ids = Array.from({ length: 8 }, (_, i) => uid(2000 + i)) const rand = mulberry32(99) let v = 0 for (let i = 0; i < 50; i++) await writeNoun(ids[Math.floor(rand() * ids.length)], ++v) await store.flushPendingSingleOps() const maxGen = store.generation() /** Map a resolveManyAt firstAfter entry to the same scalar resolveAt yields. */ const derived = async (id: string, firstAfter: Map): Promise => { const g = firstAfter.get(id) if (g === undefined) return 'current' const rec = await store.readGenerationRecord('noun', g, id) if (!rec || (rec.metadata === null && rec.vector === null)) return 'absent' return `rec:${(rec.metadata as any)?.version}` } for (let pin = 0; pin <= maxGen; pin++) { // A random subset of ids for this pin. const subset = new Set(ids.filter(() => rand() < 0.6)) if (subset.size === 0) subset.add(ids[0]) const many = await store.resolveManyAt('noun', subset, pin) for (const id of subset) { expect(await derived(id, many)).toBe(norm(await store.resolveAt('noun', id, pin))) } } }) // ========================================================================== // 7. Write path is I/O-free across window slides (the ring supplies slid-out // ids — no tx.json reads on commit, even with W > deltaCacheMax). // ========================================================================== it('committing across window slides reads ZERO tx.json from disk (W > deltaCacheMax)', async () => { const s = store as any s.recentWindowGenerations = 6 // W s.deltaCacheMax = 2 // W > deltaCacheMax: a naive slide would re-read tx.json // Seed + build the window so subsequent commits drive slides. for (let i = 0; i < 8; i++) await writeNoun(uid(3000 + i), i) await store.flushPendingSingleOps() await store.resolveAt('noun', uid(3000), store.generation()) // Count tx.json reads from this point on. let txReads = 0 const realRead = storage.readRawObject.bind(storage) ;(storage as any).readRawObject = async (path: string) => { if (typeof path === 'string' && path.endsWith('/tx.json')) txReads++ return realRead(path) } try { // Many more commits, each sliding the window past evicted-from-cache gens. for (let i = 0; i < 20; i++) await writeNoun(uid(3100 + i), 1000 + i) } finally { ;(storage as any).readRawObject = realRead } expect(txReads).toBe(0) }) // ========================================================================== // 8. Compaction invalidates + rebuilds the window over survivors; a pin below // the new horizon throws GenerationCompactedError. // ========================================================================== it('rebuilds the window over survivors after compaction and rejects pins below the horizon', async () => { const s = store as any s.recentWindowGenerations = 4 const X = uid(900) const g0 = await writeNoun(X, 0) for (let v = 1; v <= 6; v++) await writeNoun(X, v) await store.flushPendingSingleOps() const maxGen = store.generation() // Build the window, then compact to keep only the 2 most recent gens. await store.resolveAt('noun', X, maxGen) const res = await store.compact({ maxGenerations: 2 }) expect(res.removedGenerations).toBeGreaterThan(0) expect(s.windowReady).toBe(false) // invalidated // A pin ABOVE the horizon rebuilds the window from survivors and resolves. const aboveHorizon = res.horizon + 1 const resolved = await store.resolveAt('noun', X, aboveHorizon) expect(['record', 'current']).toContain(resolved.source) // A pin below the horizon is unreachable. expect(() => store.assertReachable(g0)).toThrow(GenerationCompactedError) }) }) // ============================================================================ // 5. Deep-pin materialize — O(R) getDelta (not O(N·R)) + no mutex re-entrancy. // ============================================================================ describe('materializeAtGeneration — bounded & deadlock-free (GA #33)', () => { let brain: Brainy beforeEach(async () => { brain = new Brainy(createTestConfig()) await brain.init() }) afterEach(async () => { await brain.close() }) it('resolveManyAt + readGenerationRecord do NOT re-enter the commit mutex (safe inside snapshotWith)', async () => { const store = (brain as any).generationStore const id = await brain.add({ data: 'a', type: NounType.Document, subtype: 'note', vector: VEC }) await brain.add({ data: 'b', type: NounType.Document, subtype: 'note', vector: VEC }) const pin = 1 // snapshotWith HOLDS the commit mutex. If resolveManyAt/readGenerationRecord // took the mutex, this section would deadlock — assert it completes promptly. const out = await withTimeout( store.snapshotWith(async () => { const many = await store.resolveManyAt('noun', new Set([id]), pin) const rec = many.has(id) ? await store.readGenerationRecord('noun', many.get(id), id) : null return { many, rec } }), 3000, 'snapshotWith + resolveManyAt/readGenerationRecord' ) expect(out.many).toBeInstanceOf(Map) }) it('a deep-pin materialize over many ids invokes getDelta O(R) (not O(N·R)) and matches a per-id oracle', async () => { const store = (brain as any).generationStore const N = 400 for (let i = 0; i < N; i++) { await brain.add({ data: `doc ${i}`, type: NounType.Document, subtype: 'note', metadata: { i }, vector: VEC }) } const R = brain.generation() // ≈ N (each add is its own generation) expect(R).toBeGreaterThanOrEqual(N) const deepGen = 1 // Count getDelta invocations during the materialize. const realGetDelta = store.getDelta.bind(store) let getDeltaCalls = 0 store.getDelta = async (g: number) => { getDeltaCalls++ return realGetDelta(g) } let handle: any try { handle = await withTimeout( (brain as any).materializeAtGeneration(deepGen), 15000, 'deep materialize' ) } finally { store.getDelta = realGetDelta } // O(R): a small constant number of ascending passes (changedBetween + // resolveManyAt per kind), NOT a per-id rescan (which would be ~N·R). expect(getDeltaCalls).toBeLessThan(R * 5) expect(getDeltaCalls).toBeLessThan(N * N) // the regression guard // The materialized at-gen-1 brain holds exactly the one entity that existed. const atGen1 = await handle.find({ limit: N + 10 }) expect(atGen1.length).toBe(1) await handle.close() }) it('a materialize completes (no deadlock) under a forced concurrent commit', async () => { const N = 120 for (let i = 0; i < N; i++) { await brain.add({ data: `x ${i}`, type: NounType.Document, subtype: 'note', vector: VEC }) } const deepGen = 1 // Fire the materialize and several commits together: the reconciliation pass // (under snapshotWith's mutex) must resolve the raced ids via the mutex-free // bulk path without deadlocking. const matPromise = withTimeout( (brain as any).materializeAtGeneration(deepGen), 15000, 'materialize under concurrent commits' ) const commits = Promise.all( Array.from({ length: 5 }, (_, i) => brain.add({ data: `concurrent ${i}`, type: NounType.Document, subtype: 'note', vector: VEC }) ) ) const [handle] = await Promise.all([matPromise, commits]) expect(handle).toBeTruthy() await handle.close() }) })