/** * @module tests/unit/graph/graph-adjacency-watermark * @description Watermark-stamp pins for the graph-adjacency projection. * * THE LAW under test: the persisted adjacency artifact (the two verb-id LSM * trees' SSTables + manifests) carries a stamp asserting "this state * reflects every committed generation ≤ W and nothing above W" — written * AFTER both trees' flushes complete — and init() computes the three-way * verdict: stamped==committed → 'adopt' · stampedcommitted OR unstamped → 'rescan', LOUDLY. * * The verdict is COMPUTED AND EXPOSED only — cold-load recovery and rebuild * triggers are unchanged. */ import { describe, it, expect, vi, afterEach } from 'vitest' import { v4 as uuidv4 } from 'uuid' import { GraphAdjacencyIndex, GRAPH_ADJACENCY_STAMP_KEY } from '../../../src/graph/graphAdjacencyIndex.js' import { EntityIdMapper } from '../../../src/utils/entityIdMapper.js' import { MemoryStorage } from '../../../src/storage/adapters/memoryStorage.js' import { VerbType } from '../../../src/types/graphTypes.js' import type { GraphVerb } from '../../../src/coreTypes.js' import { prodLog } from '../../../src/utils/logger.js' function makeVerb(id: string, sourceId: string, targetId: string): GraphVerb { return { id, sourceId, targetId, vector: [], type: VerbType.RelatedTo, verb: VerbType.RelatedTo } } 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 } function setCommitted(storage: MemoryStorage, committed: number): void { vi.spyOn(storage, 'committedGeneration').mockReturnValue(committed) } /** Session 1: index verbs, optionally stamp, flush + close — the artifact. */ async function writeArtifact(storage: MemoryStorage, stamp: number | null): Promise { const idMapper = new EntityIdMapper({ storage, storageKey: 'test:graph:idMapper' }) await idMapper.init() const index = new GraphAdjacencyIndex(storage, {}, idMapper) const a = uuidv4() const b = uuidv4() const aInt = BigInt(idMapper.getOrAssign(a)) const bInt = BigInt(idMapper.getOrAssign(b)) await index.addVerb(makeVerb(uuidv4(), a, b), aInt, bInt, 1n) if (stamp !== null) index.stampWatermark(stamp) await index.flush() await index.close() } /** Session 2: reopen on the same storage via the cold-load path. */ async function reopen(storage: MemoryStorage): Promise { const index = new GraphAdjacencyIndex(storage) await index.init() return index } afterEach(() => { vi.restoreAllMocks() }) describe('graph adjacency index — watermark stamp + three-way load verdict', () => { it("save-with-stamp then reopen at the same committed generation → 'adopt'", 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() await index.close() }) it("stamp BEHIND the committed generation → 'catchup' with the exact gap reported", async () => { const storage = await makeStorage(5) await writeArtifact(storage, 5) setCommitted(storage, 11) const index = await reopen(storage) expect(index.watermarkVerdict()).toBe('catchup') expect(index.watermark()).toBe(5) expect(index.watermarkGap()).toEqual({ from: 5, to: 11 }) await index.close() }) it("stamp ABOVE the committed generation → 'rescan', said out loud", async () => { const storage = await makeStorage(9) await writeArtifact(storage, 9) 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') await index.close() }) 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 adjacency: SSTables, no stamp expect(await storage.getMetadata(GRAPH_ADJACENCY_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') await index.close() }) it("a store with no committed-generation capability keeps pre-stamp behavior → 'adopt'", async () => { const storage = await makeStorage(null) await writeArtifact(storage, null) const index = await reopen(storage) expect(index.watermarkVerdict()).toBe('adopt') expect(index.watermark()).toBeNull() await index.close() }) it('STAMP-AFTER-DATA: the stamp is the last saveMetadata of the flush, after both trees’ SSTable + manifest writes', async () => { const storage = await makeStorage(2) const idMapper = new EntityIdMapper({ storage, storageKey: 'test:graph:idMapper' }) await idMapper.init() const index = new GraphAdjacencyIndex(storage, {}, idMapper) const a = uuidv4() const b = uuidv4() await index.addVerb( makeVerb(uuidv4(), a, b), BigInt(idMapper.getOrAssign(a)), BigInt(idMapper.getOrAssign(b)), 1n ) 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(GRAPH_ADJACENCY_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) // Both trees flushed durable bytes before the stamp landed. expect( keys.slice(0, stampAt).some(k => k.startsWith('graph-lsm-verbs-source')), 'verbs-by-source tree wrote before the stamp' ).toBe(true) expect( keys.slice(0, stampAt).some(k => k.startsWith('graph-lsm-verbs-target')), 'verbs-by-target tree wrote before the stamp' ).toBe(true) const record = (await storage.getMetadata(GRAPH_ADJACENCY_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') await index.close() }) it('a pending stamp also lands on the close() shutdown path, after the final tree flushes', async () => { const storage = await makeStorage(6) const idMapper = new EntityIdMapper({ storage, storageKey: 'test:graph:idMapper' }) await idMapper.init() const index = new GraphAdjacencyIndex(storage, {}, idMapper) const a = uuidv4() const b = uuidv4() await index.addVerb( makeVerb(uuidv4(), a, b), BigInt(idMapper.getOrAssign(a)), BigInt(idMapper.getOrAssign(b)), 1n ) index.stampWatermark(6) await index.close() // no explicit flush — close() flushes, then stamps const record = (await storage.getMetadata(GRAPH_ADJACENCY_STAMP_KEY)) as { watermark: number } expect(record?.watermark).toBe(6) }) })