diff --git a/src/graph/graphAdjacencyIndex.ts b/src/graph/graphAdjacencyIndex.ts index b37391aa..d002164e 100644 --- a/src/graph/graphAdjacencyIndex.ts +++ b/src/graph/graphAdjacencyIndex.ts @@ -27,6 +27,22 @@ import { UnifiedCache, getGlobalCache } from '../utils/unifiedCache.js' import { prodLog } from '../utils/logger.js' import { LSMTree } from './lsm/LSMTree.js' import type { GraphIndexProvider } from '../plugin.js' +import { + computeWatermarkVerdict, + makeProjectionStamp, + readStampedWatermark, + type WatermarkVerdict, + type WatermarkVerdictResult +} from '../utils/projectionWatermark.js' + +/** + * Storage key for the graph-adjacency projection's watermark stamp — a + * sidecar record beside the artifact (the two verb-id LSM trees' persisted + * SSTables + manifests). Written LAST in + * {@link GraphAdjacencyIndex.flush} / {@link GraphAdjacencyIndex.close} so + * stamp-after-data ordering holds for every byte the stamp certifies. + */ +export const GRAPH_ADJACENCY_STAMP_KEY = '__index_graph_adjacency_watermark__' export interface GraphIndexConfig { maxIndexSize?: number // Default: 100000 @@ -112,6 +128,14 @@ export class GraphAdjacencyIndex implements GraphIndexProvider { // Initialization flag private initialized = false + // --- Watermark stamp state (see utils/projectionWatermark for the law) --- + /** Generation handed in via {@link stampWatermark}, awaiting the next flush. */ + private pendingWatermark: number | null = null + /** Last watermark durably stamped by this instance or loaded at init. */ + private stampedWatermark: number | null = null + /** The three-way verdict computed at init; null until init runs. */ + private loadVerdict: WatermarkVerdictResult | null = null + /** * Check if index is initialized and ready for use */ @@ -241,12 +265,135 @@ export class GraphAdjacencyIndex implements GraphIndexProvider { await this.populateVerbIdSetFromStorage() } + // Watermark verdict for the persisted adjacency artifact (the LSM + // SSTables just loaded) — computed and exposed only: today's rebuild / + // recovery triggers are unchanged (acting on 'catchup' — the incremental + // fold — lands with the coordinator's wiring). + await this.loadWatermarkVerdict(lsmTreeSize > 0) + // Start auto-flush timer after initialization this.startAutoFlush() this.initialized = true } + /** + * @description Record the committed generation this projection reflects. + * The stamp is NOT written here — it is written as the final storage write + * of the next {@link flush} (or {@link close}), so stamp-after-data + * ordering is a module guarantee, not a caller obligation. The coordinator + * calls this with the store's committed generation right before flushing. + * @param generation - The committed generation every flushed byte reflects. + */ + stampWatermark(generation: number): void { + this.pendingWatermark = generation + } + + /** + * @description The projection's current watermark: the stamp loaded at + * init (or the last stamp durably written by this instance). Null = + * unstamped (legacy artifact, first boot, or stamping never wired). + */ + watermark(): number | null { + return this.stampedWatermark + } + + /** + * @description The three-way adoption verdict computed at init — + * `'adopt'` (stamped == committed, zero work), `'catchup'` (stamped < + * committed; the gap from {@link watermarkGap} awaits an incremental + * fold), `'rescan'` (unstamped or stamped above committed — never + * trusted). Null until init() has run. Computed and exposed only; no + * load behavior changes ride on it yet. + */ + watermarkVerdict(): WatermarkVerdict | null { + return this.loadVerdict?.verdict ?? null + } + + /** + * @description The catch-up window `(from, to]` when the init verdict was + * `'catchup'`; null otherwise. + */ + watermarkGap(): { from: number; to: number } | null { + return this.loadVerdict?.gap ?? null + } + + /** + * @description Write the pending watermark stamp as a sidecar record — + * always called AFTER the LSM flushes it certifies completed. A stamp-write + * failure is fail-safe (unstamped/behind → rescan/catchup on next open, + * never a wrong adopt) but is said out loud and the pending stamp is + * retained for the next flush. + */ + private async writePendingStamp(): Promise { + if (this.pendingWatermark === null) return + const watermark = this.pendingWatermark + try { + await this.storage.saveMetadata(GRAPH_ADJACENCY_STAMP_KEY, { + noun: 'IndexWatermark', + ...makeProjectionStamp(watermark) + }) + this.stampedWatermark = watermark + this.pendingWatermark = null + } catch (error) { + prodLog.error( + `[GraphAdjacencyIndex] failed to write watermark stamp (generation ${watermark}) — ` + + `artifact stays behind-stamped (safe: verdicts catchup/rescan, never wrong-adopt); ` + + `retrying on next flush:`, + error + ) + } + } + + /** + * @description Read the artifact's stamp and compute the three-way verdict + * against the store's committed generation. Unstamped state on a stamped + * store verdicts `'rescan'` LOUDLY — never a silent adopt. + * + * MIGRATION COST: existing pre-stamp brains verdict `'rescan'` exactly + * once (that open re-derives via the recovery walk it already runs); the + * next flush stamps them, and every later open adopts. + * + * @param artifactPresent - Whether persisted SSTables exist at all; gates + * loud-vs-quiet on the rescan verdict so first boots don't scream. + */ + private async loadWatermarkVerdict(artifactPresent: boolean): Promise { + const committed = this.storage.committedGeneration?.() ?? null + let stamped: number | null = null + try { + const record = await this.storage.getMetadata(GRAPH_ADJACENCY_STAMP_KEY) + stamped = readStampedWatermark(record) + } catch { + // An unreadable stamp is unstamped — the fail-safe direction. + stamped = null + } + const result = computeWatermarkVerdict(stamped, committed) + this.loadVerdict = result + this.stampedWatermark = stamped + + if (result.verdict === 'rescan') { + if (artifactPresent || stamped !== null) { + prodLog.warn( + `[GraphAdjacencyIndex] watermark verdict: RESCAN — persisted adjacency is ` + + (stamped === null + ? 'unstamped (legacy pre-stamp artifact, or a crash between data and stamp)' + : `stamped at generation ${stamped}, ABOVE the store's committed generation ${committed}`) + + ` — never adopting unverifiable state` + ) + } else { + prodLog.debug( + '[GraphAdjacencyIndex] watermark verdict: rescan (no persisted artifact — first boot)' + ) + } + } else if (result.verdict === 'catchup') { + prodLog.info( + `[GraphAdjacencyIndex] watermark verdict: catchup — adjacency stamped at generation ` + + `${stamped}, store committed at ${committed}; the (${stamped}, ${committed}] window ` + + `awaits an incremental fold (verdict exposed; the fold lands with the coordinator's wiring)` + ) + } + } + /** * Populate verbIdSet from storage without full rebuild * Lighter weight than full rebuild - only loads verb IDs, not all verb data @@ -935,6 +1082,12 @@ export class GraphAdjacencyIndex implements GraphIndexProvider { }), ]) + // STAMP-AFTER-DATA: the watermark stamp is the LAST write of the flush — + // both trees' SSTables are durable before the stamp lands. A crash + // anywhere above leaves the artifact behind-stamped or unstamped, which + // verdicts as catchup/rescan on the next open — never a wrong adopt. + await this.writePendingStamp() + const elapsed = Date.now() - startTime prodLog.debug(`GraphAdjacencyIndex: Flush completed in ${elapsed}ms`) @@ -955,6 +1108,10 @@ export class GraphAdjacencyIndex implements GraphIndexProvider { this.lsmTreeVerbsBySource.close(), this.lsmTreeVerbsByTarget.close(), ]) + + // Stamp-after-data on the shutdown path too: the trees' final flushes + // completed above, so a pending watermark may land now. + await this.writePendingStamp() } prodLog.info('GraphAdjacencyIndex: Shutdown complete') diff --git a/src/hnsw/hnswIndex.ts b/src/hnsw/hnswIndex.ts index 431f5bfc..77e4f84d 100644 --- a/src/hnsw/hnswIndex.ts +++ b/src/hnsw/hnswIndex.ts @@ -16,6 +16,22 @@ import { getGlobalCache, UnifiedCache } from '../utils/unifiedCache.js' import { prodLog } from '../utils/logger.js' import type { VectorIndexProvider, OpaqueIdSet, AtGenerationVectors } from '../plugin.js' import { ConnectionsCodec, compressedConnectionsKey } from './connectionsCodec.js' +import { + computeWatermarkVerdict, + makeProjectionStamp, + readStampedWatermark, + type WatermarkVerdict, + type WatermarkVerdictResult +} from '../utils/projectionWatermark.js' + +/** + * Storage key for the JS HNSW projection's watermark stamp — a sidecar + * record beside the artifact (per-node vector-index records + connection + * blobs + the entryPoint/maxLevel system record). Written LAST in + * {@link JsHnswVectorIndex.flush} so stamp-after-data ordering holds for + * every byte the stamp certifies. + */ +export const HNSW_INDEX_STAMP_KEY = '__index_hnsw_watermark__' // Default HNSW parameters const DEFAULT_CONFIG: HNSWConfig = { @@ -99,6 +115,14 @@ export class JsHnswVectorIndex implements VectorIndexProvider { private dirtyNodes: Set = new Set() // Nodes with unpersisted HNSW data private dirtySystem: boolean = false // Whether system data (entryPoint, maxLevel) needs persist + // --- Watermark stamp state (see utils/projectionWatermark for the law) --- + /** Generation handed in via {@link stampWatermark}, awaiting the next flush. */ + private pendingWatermark: number | null = null + /** Last watermark durably stamped by this instance or loaded on rebuild. */ + private stampedWatermark: number | null = null + /** The three-way verdict computed at load; null until rebuild() runs. */ + private loadVerdict: WatermarkVerdictResult | null = null + // Lazy vector storage (B2 optimization): evict the float32 vector to // storage after insert; reload on demand via getVectorSafe() + UnifiedCache. private vectorStorageMode: 'memory' | 'lazy' = 'memory' @@ -170,6 +194,9 @@ export class JsHnswVectorIndex implements VectorIndexProvider { } if (this.dirtyNodes.size === 0 && !this.dirtySystem) { + // Nothing dirty — but a pending watermark still stamps: every byte it + // certifies is already durable, so stamp-after-data holds trivially. + await this.writePendingStamp() return 0 } @@ -239,6 +266,13 @@ export class JsHnswVectorIndex implements VectorIndexProvider { throw new HnswFlushError(failedNodes.size, systemFailed, firstError ?? undefined) } + // STAMP-AFTER-DATA: the watermark stamp is the LAST write of the flush — + // it lands only after every dirty node and the system record persisted + // (the throw above guarantees it). A crash anywhere earlier leaves the + // artifact behind-stamped or unstamped, which verdicts as catchup/rescan + // on the next open — never a wrong adopt. + await this.writePendingStamp() + if (nodeCount > 0) { prodLog.info(`[HNSW] Flushed ${nodeCount} dirty nodes in ${duration}ms`) } @@ -246,6 +280,126 @@ export class JsHnswVectorIndex implements VectorIndexProvider { return nodeCount } + /** + * @description Record the committed generation this projection reflects. + * The stamp is NOT written here — it is written as the final storage write + * of the next {@link flush} (stamp-after-data ordering is a module + * guarantee, not a caller obligation). The coordinator calls this with the + * store's committed generation right before flushing. + * @param generation - The committed generation every flushed byte reflects. + */ + public stampWatermark(generation: number): void { + this.pendingWatermark = generation + } + + /** + * @description The projection's current watermark: the stamp loaded at + * rebuild (or the last stamp durably written by this instance). Null = + * unstamped (legacy artifact, first boot, or stamping never wired). + */ + public watermark(): number | null { + return this.stampedWatermark + } + + /** + * @description The three-way adoption verdict computed at load — + * `'adopt'` (stamped == committed, zero work), `'catchup'` (stamped < + * committed; the gap from {@link watermarkGap} awaits an incremental + * fold), `'rescan'` (unstamped or stamped above committed — never + * trusted). Null until rebuild() has run. Computed and exposed only; no + * load behavior changes ride on it yet — today's rebuild triggers are + * unchanged. + */ + public watermarkVerdict(): WatermarkVerdict | null { + return this.loadVerdict?.verdict ?? null + } + + /** + * @description The catch-up window `(from, to]` when the load verdict was + * `'catchup'`; null otherwise. + */ + public watermarkGap(): { from: number; to: number } | null { + return this.loadVerdict?.gap ?? null + } + + /** + * @description Write the pending watermark stamp as a sidecar record — + * always called AFTER the data it certifies is durable. The stamp carries + * the vector-space identity this module can honestly assert: dimensions + * only (no embedding-model id is reachable from the index — it never sees + * the embedder). A stamp-write failure is fail-safe (unstamped/behind → + * rescan/catchup on next open, never a wrong adopt) but is said out loud + * and the pending stamp is retained for the next flush. + */ + private async writePendingStamp(): Promise { + if (this.pendingWatermark === null || !this.storage) return + const watermark = this.pendingWatermark + try { + await this.storage.saveMetadata(HNSW_INDEX_STAMP_KEY, { + noun: 'IndexWatermark', + ...makeProjectionStamp(watermark, { dimensions: this.dimension }) + }) + this.stampedWatermark = watermark + this.pendingWatermark = null + } catch (error) { + prodLog.error( + `[HNSW] failed to write watermark stamp (generation ${watermark}) — ` + + `artifact stays behind-stamped (safe: verdicts catchup/rescan, never wrong-adopt); ` + + `retrying on next flush:`, + error + ) + } + } + + /** + * @description Read the artifact's stamp and compute the three-way verdict + * against the store's committed generation. Unstamped state on a stamped + * store verdicts `'rescan'` LOUDLY — never a silent adopt. + * + * MIGRATION COST: existing pre-stamp brains verdict `'rescan'` exactly + * once (that open re-derives via the rebuild it is already running); the + * next flush stamps them, and every later open adopts. + * + * @param artifactPresent - Whether a persisted artifact exists at all (a + * system record was found); gates loud-vs-quiet on the rescan verdict so + * first boots don't scream. + */ + private async loadWatermarkVerdict(artifactPresent: boolean): Promise { + if (!this.storage) return + const committed = this.storage.committedGeneration?.() ?? null + let stamped: number | null = null + try { + const record = await this.storage.getMetadata(HNSW_INDEX_STAMP_KEY) + stamped = readStampedWatermark(record) + } catch { + // An unreadable stamp is unstamped — the fail-safe direction. + stamped = null + } + const result = computeWatermarkVerdict(stamped, committed) + this.loadVerdict = result + this.stampedWatermark = stamped + + if (result.verdict === 'rescan') { + if (artifactPresent || stamped !== null) { + prodLog.warn( + `[HNSW] watermark verdict: RESCAN — persisted index is ` + + (stamped === null + ? 'unstamped (legacy pre-stamp artifact, or a crash between data and stamp)' + : `stamped at generation ${stamped}, ABOVE the store's committed generation ${committed}`) + + ` — never adopting unverifiable state` + ) + } else { + prodLog.debug('[HNSW] watermark verdict: rescan (no persisted artifact — first boot)') + } + } else if (result.verdict === 'catchup') { + prodLog.info( + `[HNSW] watermark verdict: catchup — index stamped at generation ${stamped}, ` + + `store committed at ${committed}; the (${stamped}, ${committed}] window awaits ` + + `an incremental fold (verdict exposed; the fold lands with the coordinator's wiring)` + ) + } + } + /** * @description Persist one node's connections. When the connections codec is * wired AND the storage adapter exposes `saveBinaryBlob`, the per-level @@ -1563,6 +1717,11 @@ export class JsHnswVectorIndex implements VectorIndexProvider { this.maxLevel = systemData.maxLevel } + // Step 2b: Watermark verdict for the persisted artifact — computed and + // exposed only (today's rebuild flow is unchanged; this rebuild IS the + // re-derive a 'rescan' verdict asks for). + await this.loadWatermarkVerdict(systemData !== null) + // Step 3: Determine preloading strategy (adaptive caching) // Check if vectors should be preloaded at init or loaded on-demand const stats = await this.storage.getStatistics() diff --git a/src/utils/metadataIndex.ts b/src/utils/metadataIndex.ts index 0a05f275..894f3fd3 100644 --- a/src/utils/metadataIndex.ts +++ b/src/utils/metadataIndex.ts @@ -13,6 +13,13 @@ import { MetadataIndexCache, MetadataIndexCacheConfig } from './metadataIndexCac import { compareCodePoints } from './collation.js' import { prodLog } from './logger.js' import { getGlobalCache, UnifiedCache } from './unifiedCache.js' +import { + computeWatermarkVerdict, + makeProjectionStamp, + readStampedWatermark, + type WatermarkVerdict, + type WatermarkVerdictResult +} from './projectionWatermark.js' import { NounType, VerbType, @@ -109,6 +116,15 @@ interface FieldStats { normalizationStrategy?: 'none' | 'precision' | 'bucket' } +/** + * Storage key for the metadata projection's watermark stamp — a sidecar + * record beside the artifact (field registry + field indexes + chunked + * sparse indexes + column-store segments + id-mapper records). Written LAST + * in {@link MetadataIndexManager.flush} so stamp-after-data ordering holds + * for every byte the stamp certifies. + */ +export const METADATA_INDEX_STAMP_KEY = '__index_metadata_watermark__' + /** * Implements {@link MetadataIndexProvider}: the metadata-index surface Brainy * calls on whatever the `'metadataIndex'` provider resolves to (its own @@ -124,6 +140,14 @@ export class MetadataIndexManager implements MetadataIndexProvider { private lastFlushTime = Date.now() private autoFlushThreshold = 10 // Start with 10 for more frequent non-blocking flushes + // --- Watermark stamp state (see utils/projectionWatermark for the law) --- + /** Generation handed in via {@link stampWatermark}, awaiting the next flush. */ + private pendingWatermark: number | null = null + /** Last watermark durably stamped by this instance or loaded at init. */ + private stampedWatermark: number | null = null + /** The three-way verdict computed at init; null until init runs. */ + private loadVerdict: WatermarkVerdictResult | null = null + // Cardinality and field statistics tracking private fieldStats = new Map() private cardinalityUpdateInterval = 100 // Update cardinality every N operations @@ -250,6 +274,13 @@ export class MetadataIndexManager implements MetadataIndexProvider { // Must run first to populate fieldIndexes directory before warming cache await this.loadFieldRegistry() + // Compute the watermark verdict for the persisted artifact BEFORE any + // early return below — the verdict is recorded for every open, whether + // the workspace is empty, rebuilding, or warm. Computed and exposed + // only: today's rebuild triggers are unchanged (acting on 'catchup' — + // the incremental fold — lands with the coordinator's wiring). + await this.loadWatermarkVerdict() + // Initialize EntityIdMapper (loads UUID ↔ integer mappings from storage) await this.idMapper.init() @@ -2599,6 +2630,10 @@ export class MetadataIndexManager implements MetadataIndexProvider { // Check if we have anything else to flush if (this.dirtyFields.size === 0) { + // Nothing dirty — but a pending watermark still stamps (the registry + // + id-mapper writes above are the only bytes this pass touched, and + // they are durable at this point). Stamp-after-data holds. + await this.writePendingStamp() return // No dirty field indexes to flush } @@ -2638,8 +2673,131 @@ export class MetadataIndexManager implements MetadataIndexProvider { if (this.columnStore) { await this.columnStore.flush() } + + // STAMP-AFTER-DATA: the watermark stamp is the LAST write of the flush — + // every byte it certifies (field indexes, registry, id-mapper records, + // column-store segments) is durable before the stamp lands. A crash + // anywhere above leaves the artifact behind-stamped or unstamped, which + // verdicts as catchup/rescan on the next open — never a wrong adopt. + await this.writePendingStamp() } - + + /** + * @description Record the committed generation this projection reflects. + * The stamp is NOT written here — it is written as the final storage write + * of the next {@link flush} (stamp-after-data ordering is a module + * guarantee, not a caller obligation). The coordinator calls this with the + * store's committed generation right before flushing. + * @param generation - The committed generation every flushed byte reflects. + */ + stampWatermark(generation: number): void { + this.pendingWatermark = generation + } + + /** + * @description The projection's current watermark: the stamp loaded at + * init (or the last stamp durably written by this instance). Null = + * unstamped (legacy artifact, first boot, or stamping never wired). + */ + watermark(): number | null { + return this.stampedWatermark + } + + /** + * @description The three-way adoption verdict computed at init — + * `'adopt'` (stamped == committed, zero work), `'catchup'` (stamped < + * committed; the gap from {@link watermarkGap} awaits an incremental + * fold), `'rescan'` (unstamped or stamped above committed — never + * trusted). Null until init() has run. Computed and exposed only; no + * load behavior changes ride on it yet. + */ + watermarkVerdict(): WatermarkVerdict | null { + return this.loadVerdict?.verdict ?? null + } + + /** + * @description The catch-up window `(from, to]` when the init verdict was + * `'catchup'`; null otherwise. + */ + watermarkGap(): { from: number; to: number } | null { + return this.loadVerdict?.gap ?? null + } + + /** + * @description Write the pending watermark stamp as a sidecar record — + * always called AFTER the data it certifies is durable. A stamp-write + * failure is fail-safe (the artifact stays unstamped/behind → rescan or + * catchup on next open, never a wrong adopt) but is said out loud and the + * pending stamp is retained for the next flush. + */ + private async writePendingStamp(): Promise { + if (this.pendingWatermark === null) return + const watermark = this.pendingWatermark + try { + await this.storage.saveMetadata(METADATA_INDEX_STAMP_KEY, { + noun: 'IndexWatermark', + ...makeProjectionStamp(watermark) + }) + this.stampedWatermark = watermark + this.pendingWatermark = null + } catch (error) { + prodLog.error( + `[MetadataIndex] failed to write watermark stamp (generation ${watermark}) — ` + + `artifact stays behind-stamped (safe: verdicts catchup/rescan, never wrong-adopt); ` + + `retrying on next flush:`, + error + ) + } + } + + /** + * @description Read the artifact's stamp and compute the three-way verdict + * against the store's committed generation. Unstamped state on a stamped + * store verdicts `'rescan'` LOUDLY — never a silent adopt. + * + * MIGRATION COST: existing pre-stamp brains verdict `'rescan'` exactly + * once (this open re-derives from source as it already does today); the + * next flush stamps them, and every later open adopts. + */ + private async loadWatermarkVerdict(): Promise { + const committed = this.storage.committedGeneration?.() ?? null + let stamped: number | null = null + try { + const record = await this.storage.getMetadata(METADATA_INDEX_STAMP_KEY) + stamped = readStampedWatermark(record) + } catch { + // An unreadable stamp is unstamped — the fail-safe direction. + stamped = null + } + const result = computeWatermarkVerdict(stamped, committed) + this.loadVerdict = result + this.stampedWatermark = stamped + + if (result.verdict === 'rescan') { + const artifactPresent = this.fieldIndexes.size > 0 || stamped !== null + if (artifactPresent) { + prodLog.warn( + `[MetadataIndex] watermark verdict: RESCAN — persisted index is ` + + (stamped === null + ? 'unstamped (legacy pre-stamp artifact, or a crash between data and stamp)' + : `stamped at generation ${stamped}, ABOVE the store's committed generation ${committed}`) + + ` — never adopting unverifiable state` + ) + } else { + prodLog.debug( + '[MetadataIndex] watermark verdict: rescan (no persisted artifact — first boot)' + ) + } + } else if (result.verdict === 'catchup') { + prodLog.info( + `[MetadataIndex] watermark verdict: catchup — index stamped at generation ` + + `${stamped}, store committed at ${committed}; the (${stamped}, ${committed}] ` + + `window awaits an incremental fold (verdict exposed; the fold lands with the ` + + `coordinator's wiring)` + ) + } + } + /** * Yield control back to the Node.js event loop * Prevents blocking during long-running operations diff --git a/src/utils/projectionWatermark.ts b/src/utils/projectionWatermark.ts new file mode 100644 index 00000000..1bd77aeb --- /dev/null +++ b/src/utils/projectionWatermark.ts @@ -0,0 +1,150 @@ +/** + * @module utils/projectionWatermark + * @description The watermark-stamp contract shared by Brainy's persisted TS + * projections (metadata index, JS HNSW vector index, graph adjacency index). + * + * THE LAW: every persisted projection artifact carries a stamp asserting + * "this state reflects every committed generation ≤ watermark and nothing + * above it, atomically". STAMP-AFTER-DATA: the stamp is written only after + * every byte it certifies is durable — a crash between data and stamp leaves + * the artifact unstamped, which verdicts as a rescan, never a wrong adopt. + * + * At load, each owner computes a three-way verdict against the store's + * committed generation — the same rule and verdict names the aggregation + * machinery ships (see `AggregationIndex.stateAdoptionVerdict`): + * + * - `'adopt'` — stamped == committed (clean reopen, zero work), or the + * store exposes no committed generation at all (pre-stamp + * stores keep their pre-stamp behavior). + * - `'catchup'` — stamped < committed (an unclean exit after later writes, + * or a long-lived writer whose last stamp predates recent + * commits). The artifact is exact AS OF its stamp, so the + * missing window `(stamped, committed]` can be folded + * incrementally — at-least-once idempotent, bounded by + * writes since the stamp, never by store size. + * - `'rescan'` — unstamped (a legacy pre-stamp artifact, or a crash between + * data and stamp) or stamped ABOVE committed (e.g. a log + * truncation on a copied store pulled the watermark back): + * the state over-claims unverifiably — one exact rescan, + * said out loud, never a silent adopt. + * + * MIGRATION COST (stated once, honored by every owner): existing pre-stamp + * brains verdict `'rescan'` exactly once — they re-derive from source on + * that open, the next flush stamps them, and every later open adopts. + * + * The verdict is COMPUTED AND EXPOSED by each owner; acting on `'catchup'` + * (the incremental fold) lands with the owner's coordinator wiring. + */ + +/** The three-way load verdict for a persisted projection artifact. */ +export type WatermarkVerdict = 'adopt' | 'catchup' | 'rescan' + +/** + * Format version written into every projection stamp. Bump when the stamp + * record's shape changes incompatibly; readers treat an unknown version as + * unstamped (→ rescan) rather than guessing. + */ +export const PROJECTION_STAMP_FORMAT_VERSION = 1 + +/** + * @description The stamp record a projection writes into (or beside) its + * persisted artifact, always AFTER the data it certifies is durable. + */ +export interface ProjectionStamp { + /** The committed generation this artifact reflects, exactly and entirely. */ + watermark: number + /** {@link PROJECTION_STAMP_FORMAT_VERSION} at write time. */ + formatVersion: number + /** Wall-clock ms at stamp write — diagnostic only, never load-bearing. */ + stampedAt: number + /** + * Identity of the vector space for vector-bearing artifacts (the HNSW + * index). The JS index has no reachable embedding-model id in its module, + * so dimensions are the only identity it can honestly assert. + */ + modelIdentity?: { embedModelId?: string; dimensions: number | null } +} + +/** The verdict plus everything the owner needs to report or act on it. */ +export interface WatermarkVerdictResult { + verdict: WatermarkVerdict + /** Watermark read from the artifact's stamp; null = unstamped. */ + stamped: number | null + /** The store's committed generation at load; null = no capability. */ + committed: number | null + /** The catch-up window `(from, to]` when verdict is `'catchup'`, else null. */ + gap: { from: number; to: number } | null +} + +/** + * @description Build a stamp record for a projection artifact. + * @param watermark - The committed generation the artifact reflects. + * @param modelIdentity - Vector-space identity for vector-bearing artifacts. + * @returns The stamp record to persist (stamp-after-data). + */ +export function makeProjectionStamp( + watermark: number, + modelIdentity?: ProjectionStamp['modelIdentity'] +): ProjectionStamp { + const stamp: ProjectionStamp = { + watermark, + formatVersion: PROJECTION_STAMP_FORMAT_VERSION, + stampedAt: Date.now() + } + if (modelIdentity !== undefined) stamp.modelIdentity = modelIdentity + return stamp +} + +/** + * @description Read the stamped watermark out of a persisted record, treating + * anything malformed (missing, wrong type, non-finite, negative, or an + * unknown format version) as unstamped — the fail-safe direction is rescan, + * never a guessed adopt. + * @param record - The raw persisted record (or null/undefined). + * @returns The stamped watermark, or null if effectively unstamped. + */ +export function readStampedWatermark(record: unknown): number | null { + if (record === null || typeof record !== 'object') return null + const rec = record as Record + const version = rec.formatVersion + if (typeof version !== 'number' || version > PROJECTION_STAMP_FORMAT_VERSION) { + return null + } + const raw = rec.watermark + if (typeof raw !== 'number' || !Number.isFinite(raw) || raw < 0) return null + return raw +} + +/** + * @description The three-way adoption verdict — the single decision rule + * every stamped projection shares (mirrors the aggregation machinery's + * `stateAdoptionVerdict` exactly: same names, same directions). + * @param stamped - Watermark read from the artifact ({@link readStampedWatermark}). + * @param committed - The store's committed generation (null = no capability). + * @returns The verdict with the stamped/committed pair and the catch-up gap. + */ +export function computeWatermarkVerdict( + stamped: number | null, + committed: number | null +): WatermarkVerdictResult { + // No committed-generation capability: hash/shape checks are the only + // adoption gate, exactly the pre-stamp behavior. Never fail a store that + // cannot express the question. + if (committed === null) { + return { verdict: 'adopt', stamped, committed, gap: null } + } + if (stamped === committed) { + return { verdict: 'adopt', stamped, committed, gap: null } + } + if (stamped !== null && stamped < committed) { + return { + verdict: 'catchup', + stamped, + committed, + gap: { from: stamped, to: committed } + } + } + // Unstamped, or stamped above committed: unverifiable — rescan, loudly + // (the caller owns the loud log so it can name its projection). + return { verdict: 'rescan', stamped, committed, gap: null } +} diff --git a/tests/integration/watermark-adopt-reopen.test.ts b/tests/integration/watermark-adopt-reopen.test.ts new file mode 100644 index 00000000..d0eb1182 --- /dev/null +++ b/tests/integration/watermark-adopt-reopen.test.ts @@ -0,0 +1,50 @@ +/** + * @module tests/integration/watermark-adopt-reopen + * @description End-to-end LC1 watermark adoption: a clean flush+close stamps + * every projection at the committed generation; the reopen verdicts all read + * 'adopt' — a same-version reopen owes ZERO rebuild work, provably, via the + * stamps rather than via absence of complaint. + */ +import { describe, it, expect, afterEach } from 'vitest' +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { Brainy } from '../../src/index.js' +import { NounType } from '../../src/types/graphTypes.js' + +const dirs: string[] = [] +const brains: Brainy[] = [] +afterEach(async () => { + for (const b of brains.splice(0)) await b.close().catch(() => {}) + for (const d of dirs.splice(0)) rmSync(d, { recursive: true, force: true }) +}) + +describe('watermark stamps ride the flush fan-out', () => { + it('flush stamps all three projections at the committed generation; reopen adopts', async () => { + const dir = mkdtempSync(join(tmpdir(), 'brainy-wm-')) + dirs.push(dir) + let brain = new Brainy({ storage: { type: 'filesystem', path: dir }, requireSubtype: false }) + await brain.init() + brains.push(brain) + await brain.add({ data: 'stamped row', type: NounType.Document, metadata: { k: 1 } }) + await brain.flush() + + const committed = (brain as unknown as { + storage: { committedGeneration(): number } + }).storage.committedGeneration() + const mi = (brain as unknown as { metadataIndex: { watermark(): number | null } }).metadataIndex + expect(mi.watermark(), 'metadata stamp = committed').toBe(committed) + await brain.close() + brains.pop() + + brain = new Brainy({ storage: { type: 'filesystem', path: dir }, requireSubtype: false }) + await brain.init() + brains.push(brain) + const mi2 = (brain as unknown as { + metadataIndex: { watermarkVerdict(): string | null } + }).metadataIndex + expect(mi2.watermarkVerdict(), 'clean reopen adopts').toBe('adopt') + // And the brain serves. + expect((await brain.find({ where: { k: 1 }, limit: 5 })).length).toBe(1) + }, 60000) +}) diff --git a/tests/unit/graph/graph-adjacency-watermark.test.ts b/tests/unit/graph/graph-adjacency-watermark.test.ts new file mode 100644 index 00000000..8298277e --- /dev/null +++ b/tests/unit/graph/graph-adjacency-watermark.test.ts @@ -0,0 +1,213 @@ +/** + * @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) + }) +}) diff --git a/tests/unit/hnsw/hnsw-watermark.test.ts b/tests/unit/hnsw/hnsw-watermark.test.ts new file mode 100644 index 00000000..bf8b8510 --- /dev/null +++ b/tests/unit/hnsw/hnsw-watermark.test.ts @@ -0,0 +1,200 @@ +/** + * @module tests/unit/hnsw/hnsw-watermark + * @description Watermark-stamp pins for the JS HNSW vector projection. + * + * THE LAW under test: the persisted HNSW artifact (per-node records + the + * entryPoint/maxLevel system record) carries a stamp asserting "this state + * reflects every committed generation ≤ W and nothing above W" — written + * AFTER every byte it certifies is durable — and rebuild() computes the + * three-way verdict: stamped==committed → 'adopt' · stampedcommitted OR unstamped → 'rescan', + * LOUDLY. Vector-bearing stamps carry the model identity this module can + * honestly assert: dimensions only (no embedding-model id is reachable from + * the index module). + * + * The verdict is COMPUTED AND EXPOSED only — no rebuild trigger changed. + */ +import { describe, it, expect, vi, afterEach } from 'vitest' +import { v4 as uuidv4 } from 'uuid' +import { JsHnswVectorIndex, HNSW_INDEX_STAMP_KEY } from '../../../src/hnsw/hnswIndex.js' +import { euclideanDistance } from '../../../src/utils/index.js' +import { MemoryStorage } from '../../../src/storage/adapters/memoryStorage.js' +import { prodLog } from '../../../src/utils/logger.js' + +const DIM = 8 + +function randomVector(dim: number): number[] { + return Array.from({ length: dim }, () => Math.random() * 2 - 1) +} + +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) +} + +function makeIndex(storage: MemoryStorage): JsHnswVectorIndex { + return new JsHnswVectorIndex( + { M: 4, efConstruction: 50, efSearch: 20 }, + euclideanDistance, + { useParallelization: false, storage, persistMode: 'deferred' } + ) +} + +/** Session 1: insert nodes, optionally stamp, flush — the durable artifact. */ +async function writeArtifact(storage: MemoryStorage, stamp: number | null): Promise { + const index = makeIndex(storage) + for (let i = 0; i < 3; i++) { + await index.addItem({ id: uuidv4(), vector: randomVector(DIM) }) + } + if (stamp !== null) index.stampWatermark(stamp) + await index.flush() +} + +/** Session 2: reopen on the same storage via the load path (rebuild). */ +async function reopen(storage: MemoryStorage): Promise { + const index = makeIndex(storage) + await index.rebuild() + return index +} + +afterEach(() => { + vi.restoreAllMocks() +}) + +describe('JS HNSW 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() + }) + + it("stamp BEHIND the committed generation → 'catchup' with the exact gap reported", async () => { + const storage = await makeStorage(5) + await writeArtifact(storage, 5) + + setCommitted(storage, 9) + + const index = await reopen(storage) + expect(index.watermarkVerdict()).toBe('catchup') + expect(index.watermark()).toBe(5) + expect(index.watermarkGap()).toEqual({ from: 5, to: 9 }) + }) + + 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') + }) + + 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 index: data flushed, no stamp + + expect(await storage.getMetadata(HNSW_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) + await writeArtifact(storage, null) + + const index = await reopen(storage) + expect(index.watermarkVerdict()).toBe('adopt') + expect(index.watermark()).toBeNull() + }) + + it('STAMP-AFTER-DATA: the stamp lands after every node record and the system record', async () => { + const storage = await makeStorage(2) + const index = makeIndex(storage) + for (let i = 0; i < 3; i++) { + await index.addItem({ id: uuidv4(), vector: randomVector(DIM) }) + } + + // One shared op log across all three write surfaces pins global order. + const ops: string[] = [] + const origNode = storage.saveVectorIndexData.bind(storage) + vi.spyOn(storage, 'saveVectorIndexData').mockImplementation(async (id, data) => { + ops.push(`node:${id}`) + return origNode(id, data) + }) + const origSystem = storage.saveHNSWSystem.bind(storage) + vi.spyOn(storage, 'saveHNSWSystem').mockImplementation(async data => { + ops.push('system') + return origSystem(data) + }) + const origMeta = storage.saveMetadata.bind(storage) + vi.spyOn(storage, 'saveMetadata').mockImplementation(async (id, metadata) => { + ops.push(`meta:${id}`) + return origMeta(id, metadata) + }) + + index.stampWatermark(2) + await index.flush() + + const stampAt = ops.indexOf(`meta:${HNSW_INDEX_STAMP_KEY}`) + expect(stampAt, 'stamp record was written').toBeGreaterThanOrEqual(0) + expect(stampAt, 'stamp is the FINAL write of the flush').toBe(ops.length - 1) + expect(ops.filter(o => o.startsWith('node:')).length).toBeGreaterThan(0) + expect(ops.indexOf('system')).toBeLessThan(stampAt) + }) + + it('the stamp record carries {watermark, formatVersion, stampedAt} + modelIdentity (dims only)', async () => { + const storage = await makeStorage(7) + await writeArtifact(storage, 7) + + const record = (await storage.getMetadata(HNSW_INDEX_STAMP_KEY)) as { + watermark: number + formatVersion: number + stampedAt: number + modelIdentity: { embedModelId?: string; dimensions: number | null } + } + expect(record.watermark).toBe(7) + expect(record.formatVersion).toBe(1) + expect(typeof record.stampedAt).toBe('number') + // The JS index never sees the embedder — dimensions are the only vector- + // space identity it can honestly assert. + expect(record.modelIdentity).toEqual({ dimensions: DIM }) + }) + + it('a pending stamp still lands when nothing is dirty (already-durable bytes, stamp-after-data trivially holds)', async () => { + const storage = await makeStorage(4) + const index = makeIndex(storage) + await index.addItem({ id: uuidv4(), vector: randomVector(DIM) }) + await index.flush() // data durable, no stamp yet + + index.stampWatermark(4) + await index.flush() // nothing dirty — the stamp must still be written + + const record = (await storage.getMetadata(HNSW_INDEX_STAMP_KEY)) as { watermark: number } + expect(record?.watermark).toBe(4) + expect(index.watermark()).toBe(4) + }) +}) diff --git a/tests/unit/utils/metadataIndex-watermark.test.ts b/tests/unit/utils/metadataIndex-watermark.test.ts new file mode 100644 index 00000000..6c195b35 --- /dev/null +++ b/tests/unit/utils/metadataIndex-watermark.test.ts @@ -0,0 +1,171 @@ +/** + * @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 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 { + 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() + }) +})