Every persisted projection artifact (metadata field indexes + column segments, HNSW node records, graph adjacency LSM trees) now carries a stamp asserting 'this state reflects every committed generation ≤ W, atomically' — written LAST in each owner's flush (stamp-after-data: a crash between data and stamp = unstamped = rescan, never trust). At load, each owner computes the three-way verdict: stamped==committed → adopt (zero work) · behind → catchup (gap reported) · above/unstamped → RESCAN, loudly. Legacy artifacts re-derive once, then are stamped forever. Shared law in projectionWatermark.ts (the aggregation verdict machinery, generalized); vector artifacts carry model dimensions. Verdicts are computed and exposed (watermark()/watermarkVerdict()/watermarkGap()); rebuild triggers unchanged — acting on 'catchup' is the fold train. Pins: 22 unit (7 metadata · 8 hnsw · 7 graph, incl. spy-order stamp-after-data) + the end-to-end reopen-adopts pin.
171 lines
6.6 KiB
TypeScript
171 lines
6.6 KiB
TypeScript
/**
|
|
* @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' · stamped<committed → 'catchup' (gap
|
|
* reported) · stamped>committed OR unstamped → 'rescan', LOUDLY.
|
|
* Same rule, same verdict names as the shipped aggregation machinery
|
|
* (AggregationIndex.stateAdoptionVerdict).
|
|
*
|
|
* The verdict is COMPUTED AND EXPOSED only — these pins assert no rebuild
|
|
* trigger changed; acting on 'catchup' lands with the coordinator's wiring.
|
|
*/
|
|
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'
|
|
|
|
/** Fresh storage with a controllable committed generation. */
|
|
async function makeStorage(committed: number | null): Promise<MemoryStorage> {
|
|
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<void> {
|
|
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<MetadataIndexManager> {
|
|
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()
|
|
})
|
|
})
|