/** * @module tests/unit/db/generationStore * @description Unit tests for the generational MVCC record layer * (`src/db/generationStore.ts`) against a real `MemoryStorage` adapter — * counter monotonicity, the commit protocol (staging → execute → manifest * rename → tx-log), point-in-time resolution from before-images, pin * refcounting, compaction retention rules, CAS conflicts, timestamp * resolution, and crash rollback of uncommitted generations. */ import { describe, it, expect, beforeEach, vi } from 'vitest' import { MemoryStorage } from '../../../src/storage/adapters/memoryStorage.js' import { GenerationStore, GENERATIONS_PREFIX, MANIFEST_PATH } from '../../../src/db/generationStore.js' import { GenerationCompactedError, GenerationConflictError } from '../../../src/db/errors.js' import { NounType } from '../../../src/types/graphTypes.js' /** Stored-metadata fixture in the canonical shape the live write paths use. */ function metadataFixture(version: number): Record { return { noun: NounType.Document, subtype: 'note', data: `payload-v${version}`, version, createdAt: 1000, updatedAt: 1000 + version, _rev: version } } // Entity ids must be UUID-shaped (the sharded storage layout derives the // shard from the UUID hex). Fixed values keep assertions deterministic; // suffix ordering (…aa < …bb < …cc) matches `changedBetween`'s sorted output. const ID_A = '00000000-0000-4000-8000-0000000000aa' const ID_B = '00000000-0000-4000-8000-0000000000bb' const ID_C = '00000000-0000-4000-8000-0000000000cc' const ID_SOLO = '00000000-0000-4000-8000-0000000000d0' const ID_LATER = '00000000-0000-4000-8000-0000000000e0' describe('db/GenerationStore', () => { let storage: MemoryStorage let store: GenerationStore beforeEach(async () => { storage = new MemoryStorage() await storage.init() store = new GenerationStore(storage) await store.open() }) /** Commit one transaction that writes `metadata` for `id`. */ async function commitWrite( id: string, version: number, options?: { meta?: Record; ifAtGeneration?: number } ): Promise<{ generation: number; timestamp: number }> { return store.commitTransaction({ touched: { nouns: [id], verbs: [] }, meta: options?.meta, ifAtGeneration: options?.ifAtGeneration, execute: async () => { await storage.saveNounMetadata(id, metadataFixture(version)) } }) } describe('generation counter', () => { it('starts at 0 on a fresh store', () => { expect(store.generation()).toBe(0) expect(store.committedGeneration()).toBe(0) }) it('advances by exactly one per committed transaction', async () => { await commitWrite(ID_A, 1) expect(store.generation()).toBe(1) await commitWrite(ID_A, 2) expect(store.generation()).toBe(2) expect(store.committedGeneration()).toBe(2) }) it('bumps on single-operation writes via the storage hook (suppressed inside transact)', async () => { // Single-op write outside any transaction → hook bumps. await storage.saveNounMetadata(ID_SOLO, metadataFixture(1)) expect(store.generation()).toBe(1) expect(store.committedGeneration()).toBe(0) // no transact committed // A transact batch writing twice bumps ONCE (the batch is one generation). await store.commitTransaction({ touched: { nouns: [ID_A, ID_B], verbs: [] }, execute: async () => { await storage.saveNounMetadata(ID_A, metadataFixture(1)) await storage.saveNounMetadata(ID_B, metadataFixture(1)) } }) expect(store.generation()).toBe(2) }) it('survives close + reopen monotonically', async () => { await commitWrite(ID_A, 1) await storage.saveNounMetadata(ID_SOLO, metadataFixture(1)) // single-op bump const before = store.generation() await store.close() const reopened = new GenerationStore(storage) await reopened.open() expect(reopened.generation()).toBeGreaterThanOrEqual(before) expect(reopened.committedGeneration()).toBe(1) }) }) describe('CAS (ifAtGeneration)', () => { it('commits when the expectation matches and rejects with expected/actual when stale', async () => { await commitWrite(ID_A, 1) const pinned = store.generation() await commitWrite(ID_A, 2, { ifAtGeneration: pinned }) // fresh → commits expect(store.generation()).toBe(pinned + 1) try { await commitWrite(ID_A, 3, { ifAtGeneration: pinned }) // stale → rejects expect.unreachable('should have thrown GenerationConflictError') } catch (err) { expect(err).toBeInstanceOf(GenerationConflictError) expect((err as GenerationConflictError).expected).toBe(pinned) expect((err as GenerationConflictError).actual).toBe(pinned + 1) } expect(store.generation()).toBe(pinned + 1) // nothing moved }) }) describe('failed transactions', () => { it('removes the staging directory and returns the generation reservation', async () => { await commitWrite(ID_A, 1) const before = store.generation() await expect( store.commitTransaction({ touched: { nouns: [ID_A], verbs: [] }, execute: async () => { throw new Error('boom mid-batch') } }) ).rejects.toThrow('boom mid-batch') expect(store.generation()).toBe(before) const leftovers = await storage.listRawObjects(`${GENERATIONS_PREFIX}/${before + 1}`) expect(leftovers).toEqual([]) }) }) describe('point-in-time resolution', () => { it('resolves untouched ids to the live fast path', async () => { await commitWrite(ID_A, 1) const pinned = store.generation() await commitWrite(ID_B, 1) // later commit touches a DIFFERENT id expect(await store.resolveAt('noun', ID_A, pinned)).toEqual({ source: 'current' }) }) it('resolves changed ids from the first before-image after the pin', async () => { await commitWrite(ID_A, 1) const pinned = store.generation() await commitWrite(ID_A, 2) await commitWrite(ID_A, 3) const resolved = await store.resolveAt('noun', ID_A, pinned) expect(resolved.source).toBe('record') if (resolved.source === 'record') { expect(resolved.metadata.version).toBe(1) // state as of the pin } }) it('resolves ids created after the pin as absent', async () => { await commitWrite(ID_A, 1) const pinned = store.generation() await commitWrite(ID_LATER, 1) expect(await store.resolveAt('noun', ID_LATER, pinned)).toEqual({ source: 'absent' }) }) it('changedBetween reports the ids touched in the interval', async () => { await commitWrite(ID_A, 1) const from = store.generation() await commitWrite(ID_B, 1) await commitWrite(ID_C, 1) const to = store.generation() const changed = await store.changedBetween(from, to) expect(changed.nouns).toEqual([ID_B, ID_C]) expect(changed.verbs).toEqual([]) expect(changed.fromGeneration).toBe(from) expect(changed.toGeneration).toBe(to) }) }) describe('timestamp resolution + tx-log', () => { it('records meta in the tx-log and resolves timestamps to generations', async () => { const t0 = Date.now() - 10 const first = await commitWrite(ID_A, 1, { meta: { author: 'unit-test' } }) // Distinct commit timestamps: resolution picks the NEWEST generation // committed at-or-before the instant, so same-millisecond commits tie. await new Promise((resolve) => setTimeout(resolve, 5)) const second = await commitWrite(ID_A, 2) const lines = await storage.readTxLogLines() expect(lines.length).toBe(2) const firstEntry = JSON.parse(lines[0]) expect(firstEntry.generation).toBe(first.generation) expect(firstEntry.meta).toEqual({ author: 'unit-test' }) expect((await store.resolveTimestamp(t0)).generation).toBe(0) expect((await store.resolveTimestamp(first.timestamp)).generation).toBe(first.generation) expect((await store.resolveTimestamp(Date.now() + 1000)).generation).toBe( second.generation ) }) }) describe('pins + compaction', () => { it('refcounts pins and never reclaims records a live pin needs', async () => { await commitWrite(ID_A, 1) const pinned = store.generation() store.pin(pinned) store.pin(pinned) // refcount 2 await commitWrite(ID_A, 2) await commitWrite(ID_A, 3) // Pin at G protects every record-set ABOVE G (resolution reads // before-images from generations strictly greater than the pin). await store.compact() const resolved = await store.resolveAt('noun', ID_A, pinned) expect(resolved.source).toBe('record') if (resolved.source === 'record') { expect(resolved.metadata.version).toBe(1) } store.release(pinned) await store.compact() // Still one live pin — records above it still resolve correctly. const stillResolved = await store.resolveAt('noun', ID_A, pinned) expect(stillResolved.source).toBe('record') store.release(pinned) const result = await store.compact() expect(result.removedGenerations).toBeGreaterThan(0) // History record-sets only — the fact log (at `_generations/facts/`) is // deliberately NOT reclaimed by history compaction. const remaining = (await storage.listRawObjects(GENERATIONS_PREFIX)).filter( (p: string) => !p.startsWith(`${GENERATIONS_PREFIX}/facts/`) ) expect(remaining).toEqual([]) }) it('maxGenerations cap keeps the most recent record-sets', async () => { await commitWrite(ID_A, 1) await commitWrite(ID_A, 2) await commitWrite(ID_A, 3) const result = await store.compact({ maxGenerations: 2 }) expect(result.removedGenerations).toBe(1) expect(result.horizon).toBe(1) // Removing record-set N breaks pins BELOW N (their reads need N's // before-images) — the horizon itself stays reachable: state at the // horizon resolves from the retained record-sets above it. expect(() => store.assertReachable(0)).toThrow(GenerationCompactedError) expect(store.assertReachable(1)).toBe(1) expect(store.assertReachable(2)).toBe(2) }) it('rejects future and negative generations', async () => { await commitWrite(ID_A, 1) expect(() => store.assertReachable(99)).toThrow(RangeError) expect(() => store.assertReachable(-1)).toThrow(RangeError) expect(() => store.assertReachable(1.5)).toThrow(RangeError) }) }) describe('crash recovery', () => { it('rolls back an uncommitted generation on open (manifest rename is the commit point)', async () => { await commitWrite(ID_A, 1) // Simulate a crash between record staging and the manifest rename: a // throwing fault injector skips the abort cleanup, exactly as a dead // process would, leaving canonical files mutated and the staging // directory present — but the manifest still at the prior generation. store.setCommitFaultInjector((phase) => { if (phase === 'before-manifest-rename') { throw new Error('simulated crash') } }) await expect(commitWrite(ID_A, 2)).rejects.toThrow('simulated crash') store.setCommitFaultInjector(undefined) // The canonical file currently holds the EXECUTED (uncommitted) state. const dirty = await storage.readNounRaw(ID_A) expect((dirty.metadata as { version: number }).version).toBe(2) // Recovery on the next open restores the before-image byte-identically. const reopened = new GenerationStore(storage) const openResult = await reopened.open() expect(openResult.rolledBackGenerations).toBe(1) const repaired = await storage.readNounRaw(ID_A) expect((repaired.metadata as { version: number }).version).toBe(1) expect(reopened.committedGeneration()).toBe(1) // The crashed generation number is never reused. expect(reopened.generation()).toBeGreaterThanOrEqual(2) }) it('keeps a transaction whose manifest rename landed (tx-log append is advisory)', async () => { await commitWrite(ID_A, 1) store.setCommitFaultInjector((phase) => { if (phase === 'after-manifest-rename') { throw new Error('simulated crash after commit point') } }) await expect(commitWrite(ID_A, 2)).rejects.toThrow('simulated crash after commit point') store.setCommitFaultInjector(undefined) const reopened = new GenerationStore(storage) const openResult = await reopened.open() expect(openResult.rolledBackGenerations).toBe(0) expect(reopened.committedGeneration()).toBe(2) const kept = await storage.readNounRaw(ID_A) expect((kept.metadata as { version: number }).version).toBe(2) const manifest = await storage.readRawObject(MANIFEST_PATH) expect(manifest.generation).toBe(2) }) }) // ========================================================================== // Model-B single-op generation-stamping (per-write history + group-commit) // ========================================================================== describe('Model-B single-op generations', () => { /** Apply one single-op write as its own generation (live write + buffered history). */ async function singleOpWrite(id: string, version: number): Promise { const { generation } = await store.commitSingleOp({ touched: { nouns: [id] }, execute: async () => { await storage.saveNounMetadata(id, metadataFixture(version)) } }) return generation } it('each single-op write is its own generation, resolvable BEFORE any flush', async () => { const g1 = await singleOpWrite(ID_A, 1) const g2 = await singleOpWrite(ID_A, 2) expect(g1).toBe(1) expect(g2).toBe(2) // Counter advanced per write; nothing persisted to disk yet. expect(store.generation()).toBe(2) expect(store.committedGeneration()).toBe(0) // Reads resolve through the in-memory pending tier with NO forced flush: // the state as-of g1 is v1 (the before-image buffered by g2). const atG1 = await store.resolveAt('noun', ID_A, g1) expect(atG1.source).toBe('record') expect(((atG1 as { metadata: { version: number } }).metadata).version).toBe(1) // hasCommittedAfter counts pending → a now()-style pin at g1 is historical. expect(store.hasCommittedAfter(g1)).toBe(true) }) it('range queries (changedBetween / generationsTouching) include un-flushed pending generations', async () => { // Two single-op writes to different ids, NEITHER flushed (still in the // pending tier). diff()/since()/history() are built on these, so they must // see un-flushed single-ops with NO forced flush. const g1 = await singleOpWrite(ID_A, 1) const g2 = await singleOpWrite(ID_B, 1) expect(store.committedGeneration()).toBe(0) // nothing on disk yet const changed = await store.changedBetween(0, store.generation()) expect(changed.nouns).toEqual([ID_A, ID_B]) // resolves over committed ∪ pending expect(await store.generationsTouching('noun', ID_A, 0, store.generation())).toEqual([g1]) expect(await store.generationsTouching('noun', ID_B, 0, store.generation())).toEqual([g2]) // …and the range stays identical across the flush (now reading from disk). await store.flushPendingSingleOps() expect((await store.changedBetween(0, store.generation())).nouns).toEqual([ID_A, ID_B]) }) it('flush persists pending single-op generations (groupCommit-marked) and advances committed', async () => { await singleOpWrite(ID_A, 1) await singleOpWrite(ID_A, 2) await store.flushPendingSingleOps() expect(store.committedGeneration()).toBe(2) const delta = await storage.readRawObject(`${GENERATIONS_PREFIX}/2/tx.json`) expect(delta.groupCommit).toBe(true) expect(typeof delta.bytes).toBe('number') // asOf resolution survives the flush (now reading before-images from disk). const atG1 = await store.resolveAt('noun', ID_A, 1) expect(((atG1 as { metadata: { version: number } }).metadata).version).toBe(1) }) it('🛑 a crash mid group-commit flush DROPS the generation WITHOUT reverting the live write', async () => { // gen 1: X = v1, flushed (durable). await singleOpWrite(ID_A, 1) await store.flushPendingSingleOps() expect(store.committedGeneration()).toBe(1) // gen 2: X = v2 acknowledged to canonical storage; its history is buffered. await singleOpWrite(ID_A, 2) expect((await storage.readNounRaw(ID_A)).metadata).toMatchObject({ version: 2 }) // Crash before the manifest rename of the group-commit flush: the gen-2 // dir (before-image v1) lands on disk; the manifest stays at gen 1. store.setCommitFaultInjector((phase) => { if (phase === 'before-manifest-rename') throw new Error('simulated crash mid-flush') }) await expect(store.flushPendingSingleOps()).rejects.toThrow('simulated crash mid-flush') store.setCommitFaultInjector(undefined) expect(await storage.readRawObject(`${GENERATIONS_PREFIX}/2/tx.json`)).not.toBeNull() expect((await storage.readRawObject(MANIFEST_PATH)).generation).toBe(1) // Recover on a FRESH store (process restart: in-memory pending is gone). const reopened = new GenerationStore(storage) const result = await reopened.open() expect(result.rolledBackGenerations).toBe(1) expect(reopened.committedGeneration()).toBe(1) // THE CORRUPTION TRAP: drop-without-restore — the acknowledged live write // (v2) is NOT reverted to the before-image (v1). Only the crashed flush // window's HISTORY is lost; live data stays correct. expect((await storage.readNounRaw(ID_A)).metadata).toMatchObject({ version: 2 }) // The orphan dir is gone and the dropped generation number is never reissued. expect(await storage.readRawObject(`${GENERATIONS_PREFIX}/2/tx.json`)).toBeNull() expect(reopened.generation()).toBeGreaterThanOrEqual(2) }) it('a clean reopen replays committed single-op history (asOf survives restart)', async () => { await singleOpWrite(ID_A, 1) await singleOpWrite(ID_A, 2) await store.flushPendingSingleOps() const reopened = new GenerationStore(storage) await reopened.open() expect(reopened.committedGeneration()).toBe(2) const atG1 = await reopened.resolveAt('noun', ID_A, 1) expect(((atG1 as { metadata: { version: number } }).metadata).version).toBe(1) }) }) // ========================================================================== // Retention CAPS (Model-B `retention` knob) // ========================================================================== describe('retention caps', () => { async function manyGens(n: number): Promise { for (let v = 1; v <= n; v++) await commitWrite(ID_A, v) } it('maxBytes reclaims the oldest generations until total history fits the budget', async () => { await manyGens(5) const total = await store.historyBytes() expect(total).toBeGreaterThan(0) const perGen = total / 5 const budget = Math.floor(perGen * 2.5) const result = await store.compact({ maxBytes: budget }) expect(result.removedGenerations).toBeGreaterThan(0) expect(await store.historyBytes()).toBeLessThanOrEqual(budget) }) it('maxAge keeps generations within the window and reclaims older ones', async () => { await manyGens(3) // A wide window keeps everything. expect((await store.compact({ maxAge: 60_000 })).removedGenerations).toBe(0) // Advance the clock 10s past the commits, then a 1s window reclaims all // three (fake timers make the age comparison deterministic). vi.useFakeTimers() try { vi.setSystemTime(Date.now() + 10_000) expect((await store.compact({ maxAge: 1_000 })).removedGenerations).toBe(3) } finally { vi.useRealTimers() } }) it('no caps reclaims every unpinned generation; a pin protects newer ones', async () => { await manyGens(3) store.pin(2) // protects generations > 2 (i.e. gen 3) from reclamation const result = await store.compact() // gens 1 and 2 are ≤ the pin → reclaimable; gen 3 is pin-protected. expect(result.removedGenerations).toBe(2) store.release(2) }) it('timeBudgetMs bounds a pass; the next pass resumes the same prefix', async () => { await manyGens(4) // A spent budget (0ms) stops before reclaiming anything — an early stop // is a consistent prefix, never a partial generation. const bounded = await store.compact({ timeBudgetMs: 0 }) expect(bounded.removedGenerations).toBe(0) expect(bounded.horizon).toBe(0) // The next (unbounded) pass picks up exactly where the bounded one // stopped and completes the same work. const resumed = await store.compact() expect(resumed.removedGenerations).toBe(4) expect(resumed.horizon).toBe(4) }) }) // ========================================================================== describe('history-bytes running total (the O(1) retention check)', () => { /** A fresh walk with the cache dropped — ground truth for the invariant. */ async function groundTruthBytes(): Promise { ;(store as any).historyBytesTotal = null return store.historyBytes() } it('is seeded once, then maintained through commits WITHOUT re-walks', async () => { await commitWrite(ID_A, 1) await commitWrite(ID_A, 2) const seeded = await store.historyBytes() expect(seeded).toBe(await groundTruthBytes()) // From here every read must come from the running total, not a walk: // getDelta re-reads are the walk's cost — commits must not trigger any. const getDeltaSpy = vi.spyOn(store as any, 'getDelta') await commitWrite(ID_B, 1) const afterCommit = await store.historyBytes() expect(getDeltaSpy).not.toHaveBeenCalled() getDeltaSpy.mockRestore() expect(afterCommit).toBe(await groundTruthBytes()) }) it('stays exact through single-op group commits and compaction', async () => { await commitWrite(ID_A, 1) await store.historyBytes() // seed // Single-op path: buffered generations flushed as one group commit. await store.commitSingleOp({ touched: { nouns: [ID_B] }, execute: async () => { await storage.saveNounMetadata(ID_B, metadataFixture(1)) } }) await store.flushPendingSingleOps() expect(await store.historyBytes()).toBe(await groundTruthBytes()) await store.historyBytes() // re-seed after ground-truth reset await store.compact({ maxGenerations: 1 }) expect(await store.historyBytes()).toBe(await groundTruthBytes()) }) it('historyStats reports counts, bytes, range, and horizon read-only', async () => { await commitWrite(ID_A, 1) await commitWrite(ID_B, 1) const stats = await store.historyStats() expect(stats.generations).toBe(2) expect(stats.bytes).toBe(await store.historyBytes()) expect(stats.oldestGeneration).toBe(1) expect(stats.newestGeneration).toBe(2) expect(stats.oldestTimestamp).toBeLessThanOrEqual(stats.newestTimestamp!) expect(stats.horizon).toBe(0) // Read-only: nothing was reclaimed by asking. expect(store.committedGeneration()).toBe(2) await store.compact({ maxGenerations: 1 }) const after = await store.historyStats() expect(after.generations).toBe(1) expect(after.oldestGeneration).toBe(2) expect(after.horizon).toBe(1) }) it('empty history reports null range and zero bytes', async () => { const stats = await store.historyStats() expect(stats).toMatchObject({ generations: 0, bytes: 0, oldestGeneration: null, newestGeneration: null, oldestTimestamp: null, newestTimestamp: null, horizon: 0 }) }) }) })