/** * @module tests/unit/utils/metadataIndex-watermark * @description Watermark-stamp pins for the metadata projection. * * THE LAW under test: every persisted projection artifact carries a stamp * asserting "this state reflects every committed generation ≤ W and nothing * above W, atomically" — written AFTER every byte it certifies is durable — * and at load the owner computes the three-way verdict: * stamped==committed → 'adopt' · stampedcommitted OR unstamped → 'rescan', LOUDLY. * Same rule, same verdict names as the shipped aggregation machinery * (AggregationIndex.stateAdoptionVerdict). * * The verdict is computed at init and consumed via * {@link MetadataIndexManager.applyWatermarkCatchup} — the coordinator * (`Brainy.performInit`) calls it right after `init()`, with an open fact * scan when the verdict is `'catchup'`. This file pins both halves: the * verdict computation (above) and the fold/no-op/demotion behavior below. */ import { describe, it, expect, vi, afterEach } from 'vitest' import { v4 as uuidv4 } from 'uuid' import { MetadataIndexManager, METADATA_INDEX_STAMP_KEY } from '../../../src/utils/metadataIndex.js' import { MemoryStorage } from '../../../src/storage/adapters/memoryStorage.js' import { prodLog } from '../../../src/utils/logger.js' import type { CommitFact, FactScanBatch, FactScanHandle } from '../../../src/db/factLog.js' /** A fact scan handle over an in-memory list of facts — batches them one * fact at a time (batch size is irrelevant to the fold, which reads * `batch.facts` only). */ function fakeScan(facts: CommitFact[]): FactScanHandle { return { headGeneration: facts.length > 0 ? facts[facts.length - 1].generation : 0, segmentCount: 1, approxFactCount: facts.length, async *batches(): AsyncGenerator { for (const fact of facts) { yield { facts: [fact], firstGeneration: fact.generation, lastGeneration: fact.generation, factCount: 1, byteSize: 0, segmentId: 'fake' } } }, summary: () => ({ factsYielded: facts.length, segmentsRead: 1 }) } } /** One noun after-image fact — the flat-record shape (no nested `metadata` * key), matching this file's existing `writeArtifact` convention. */ function nounAdd(generation: number, id: string, metadata: Record): CommitFact { return { generation, timestamp: Date.now(), ops: [{ kind: 'noun', id, record: { metadata, vector: null } }] } } /** One noun tombstone fact. */ function nounDelete(generation: number, id: string): CommitFact { return { generation, timestamp: Date.now(), ops: [{ kind: 'noun', id, record: null }] } } /** Fresh storage with a controllable committed generation. */ async function makeStorage(committed: number | null): Promise { const storage = new MemoryStorage() await storage.init() if (committed !== null) { vi.spyOn(storage, 'committedGeneration').mockReturnValue(committed) } return storage } /** Set (or reset) the mocked committed generation on an existing storage. */ function setCommitted(storage: MemoryStorage, committed: number): void { vi.spyOn(storage, 'committedGeneration').mockReturnValue(committed) } /** Session 1: index a field, optionally stamp, flush — the durable artifact. */ async function writeArtifact( storage: MemoryStorage, stamp: number | null ): Promise { const index = new MetadataIndexManager(storage) await index.init() await index.addToIndex(uuidv4(), { status: 'active', role: 'admin' }) if (stamp !== null) index.stampWatermark(stamp) await index.flush() } /** Session 2: reopen on the same storage and return the loaded manager. */ async function reopen(storage: MemoryStorage): Promise { const index = new MetadataIndexManager(storage) await index.init() return index } afterEach(() => { vi.restoreAllMocks() }) describe('metadata index — watermark stamp + three-way load verdict', () => { it("save-with-stamp then reopen at the same committed generation → 'adopt', zero-work verdict", async () => { const storage = await makeStorage(5) await writeArtifact(storage, 5) const index = await reopen(storage) expect(index.watermarkVerdict()).toBe('adopt') expect(index.watermark()).toBe(5) expect(index.watermarkGap()).toBeNull() }) it("stamp BEHIND the committed generation → 'catchup' with the exact gap reported", async () => { const storage = await makeStorage(5) await writeArtifact(storage, 5) // Later commits landed after the last stamped flush (unclean exit shape). setCommitted(storage, 8) const index = await reopen(storage) expect(index.watermarkVerdict()).toBe('catchup') expect(index.watermark()).toBe(5) expect(index.watermarkGap()).toEqual({ from: 5, to: 8 }) }) it("stamp ABOVE the committed generation → 'rescan', said out loud", async () => { const storage = await makeStorage(9) await writeArtifact(storage, 9) // A truncated log on a copied store pulled the watermark back. setCommitted(storage, 4) const warnSpy = vi.spyOn(prodLog, 'warn') const index = await reopen(storage) expect(index.watermarkVerdict()).toBe('rescan') expect(index.watermarkGap()).toBeNull() const said = warnSpy.mock.calls.map(c => String(c[0])).join('\n') expect(said).toContain('RESCAN') expect(said).toContain('ABOVE') }) it("legacy unstamped artifact on a stamped store → 'rescan', LOUD — never a silent adopt", async () => { const storage = await makeStorage(3) await writeArtifact(storage, null) // pre-stamp brain: data flushed, no stamp expect(await storage.getMetadata(METADATA_INDEX_STAMP_KEY)).toBeNull() const warnSpy = vi.spyOn(prodLog, 'warn') const index = await reopen(storage) expect(index.watermarkVerdict()).toBe('rescan') expect(index.watermark()).toBeNull() const said = warnSpy.mock.calls.map(c => String(c[0])).join('\n') expect(said).toContain('RESCAN') expect(said).toContain('unstamped') }) it("a store with no committed-generation capability keeps pre-stamp behavior → 'adopt'", async () => { const storage = await makeStorage(null) // committedGeneration() → null await writeArtifact(storage, null) const index = await reopen(storage) expect(index.watermarkVerdict()).toBe('adopt') expect(index.watermark()).toBeNull() }) it('STAMP-AFTER-DATA: the stamp is the last saveMetadata of the flush, after registry and field indexes', async () => { const storage = await makeStorage(2) const index = new MetadataIndexManager(storage) await index.init() await index.addToIndex(uuidv4(), { status: 'active' }) const keys: string[] = [] const originalSave = storage.saveMetadata.bind(storage) vi.spyOn(storage, 'saveMetadata').mockImplementation(async (id, metadata) => { keys.push(id) return originalSave(id, metadata) }) index.stampWatermark(2) await index.flush() const stampAt = keys.indexOf(METADATA_INDEX_STAMP_KEY) expect(stampAt, 'stamp record was written').toBeGreaterThanOrEqual(0) expect(stampAt, 'stamp is the FINAL metadata write of the flush').toBe(keys.length - 1) const registryAt = keys.indexOf('__metadata_field_registry__') expect(registryAt, 'field registry written during this flush').toBeGreaterThanOrEqual(0) expect(registryAt).toBeLessThan(stampAt) // The persisted stamp record carries the required shape. const record = (await storage.getMetadata(METADATA_INDEX_STAMP_KEY)) as { watermark: number formatVersion: number stampedAt: number } expect(record.watermark).toBe(2) expect(record.formatVersion).toBe(1) expect(typeof record.stampedAt).toBe('number') }) it('a flush WITHOUT a pending stamp writes no stamp record (no phantom certification)', async () => { const storage = await makeStorage(2) const index = new MetadataIndexManager(storage) await index.init() await index.addToIndex(uuidv4(), { status: 'active' }) await index.flush() expect(await storage.getMetadata(METADATA_INDEX_STAMP_KEY)).toBeNull() }) }) describe('metadata index — applyWatermarkCatchup (the coordinator door)', () => { it("an 'adopt' verdict performs zero index writes", async () => { const storage = await makeStorage(5) await writeArtifact(storage, 5) const index = await reopen(storage) expect(index.watermarkVerdict()).toBe('adopt') const addSpy = vi.spyOn(index, 'addToIndex') const removeSpy = vi.spyOn(index, 'removeFromIndex') const result = await index.applyWatermarkCatchup(null) expect(result).toEqual({ action: 'noop' }) expect(addSpy).not.toHaveBeenCalled() expect(removeSpy).not.toHaveBeenCalled() }) it('a catchup window folding an add, an update (same id twice), and a delete → the index serves exactly the final state', async () => { const storage = await makeStorage(5) // Session 1: two pre-existing entities, stamped at generation 5. const survivorId = uuidv4() const deletedId = uuidv4() { const index = new MetadataIndexManager(storage) await index.init() await index.addToIndex(survivorId, { status: 'active' }) await index.addToIndex(deletedId, { status: 'active' }) index.stampWatermark(5) await index.flush() } // The store advanced to generation 8 without another metadata flush — // the exact shape a crash-then-adopt-reopen leaves behind. setCommitted(storage, 8) const index = await reopen(storage) expect(index.watermarkVerdict()).toBe('catchup') expect(index.watermarkGap()).toEqual({ from: 5, to: 8 }) const addedId = uuidv4() const scan = fakeScan([ nounAdd(6, addedId, { status: 'new' }), // add nounAdd(7, addedId, { status: 'updated' }), // update — same id twice nounDelete(8, deletedId) // delete ]) const result = await index.applyWatermarkCatchup(scan) expect(result.action).toBe('caught-up') expect(result.window).toEqual({ from: 5, to: 8 }) expect(result.factsApplied).toBe(3) expect(result.nounsApplied).toBe(3) expect(result.verbsApplied).toBe(0) // Final state: the added/updated id serves ONLY its final value... expect(await index.getIds('status', 'updated')).toEqual([addedId]) expect(await index.getIds('status', 'new')).toEqual([]) // stale value gone // ...the deleted id is gone... expect(await index.getIds('status', 'active')).toEqual([survivorId]) // ...and the untouched survivor is unaffected. expect(await index.getIds('status', 'active')).toContain(survivorId) // The window is certified: watermark stamped at `to`, and a fresh // reopen now verdicts 'adopt'. expect(index.watermark()).toBe(8) const reopened = await reopen(storage) expect(reopened.watermarkVerdict()).toBe('adopt') }) it("a 'rescan' verdict runs the existing rebuild path instead of folding", async () => { const storage = await makeStorage(9) await writeArtifact(storage, 9) setCommitted(storage, 4) // a truncated log pulled the watermark back — stamp ABOVE committed → rescan const index = await reopen(storage) expect(index.watermarkVerdict()).toBe('rescan') const rebuildSpy = vi.spyOn(index, 'rebuild') const result = await index.applyWatermarkCatchup(null) expect(result.action).toBe('rescan') expect(result.reason).toBeTruthy() expect(rebuildSpy).toHaveBeenCalledTimes(1) }) it("a 'catchup' verdict with no fact log available demotes to rebuild, narrated", async () => { const storage = await makeStorage(5) await writeArtifact(storage, 5) setCommitted(storage, 8) const index = await reopen(storage) expect(index.watermarkVerdict()).toBe('catchup') const rebuildSpy = vi.spyOn(index, 'rebuild') const result = await index.applyWatermarkCatchup(null) // no scan — no fact log expect(result.action).toBe('rescan') expect(result.reason).toContain('no fact log') expect(rebuildSpy).toHaveBeenCalledTimes(1) }) })