soulcraftlabs/open-brainy
2
0
Fork 0
Code Issues 1 Pull requests Projects Releases 341 Packages Wiki Activity Actions 4

feat(index): watermark stamps on every TS projection — adopt/catchup/rescan verdicts at load, stamp-after-data

Browse source 
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.
This commit is contained in:
David Snelling 2026-08-10 10:55:11 -07:00
parent 26c6025158
commit b35d87a7ab
8 changed files with 1259 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

157
src/graph/graphAdjacencyIndex.ts
View file

@ -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<void> {
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<void> {
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')

159
src/hnsw/hnswIndex.ts
View file

@ -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<string> = 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<void> {
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<void> {
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()

160
src/utils/metadataIndex.ts
View file

@ -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<string, FieldStats>()
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<void> {
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<void> {
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

150
src/utils/projectionWatermark.ts Normal file
View file

@ -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<string, unknown>
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 }
}

50
tests/integration/watermark-adopt-reopen.test.ts Normal file
View file

@ -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)
})

213
tests/unit/graph/graph-adjacency-watermark.test.ts Normal file
View file

@ -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' · stamped<committed → 'catchup'
* (gap reported) · stamped>committed 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<MemoryStorage> {
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<void> {
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<GraphAdjacencyIndex> {
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)
})
})

200
tests/unit/hnsw/hnsw-watermark.test.ts Normal file
View file

@ -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' · stamped<committed →
* 'catchup' (gap reported) · stamped>committed 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<MemoryStorage> {
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<void> {
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<JsHnswVectorIndex> {
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)
})
})

171
tests/unit/utils/metadataIndex-watermark.test.ts Normal file
View file

@ -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' · 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()
})
})