diff --git a/src/graph/lsm/LSMTree.ts b/src/graph/lsm/LSMTree.ts index bfd48df4..e19ec145 100644 --- a/src/graph/lsm/LSMTree.ts +++ b/src/graph/lsm/LSMTree.ts @@ -574,6 +574,7 @@ export class LSMTree { * Load SSTables from storage based on manifest */ private async loadSSTables(): Promise { + const failures: string[] = [] const loadPromises: Promise[] = [] this.manifest.sstables.forEach((level, sstableId) => { @@ -598,7 +599,12 @@ export class LSMTree { } } } catch (error) { + // A per-SSTable load failure means the persisted adjacency is INCOMPLETE. + // Record it and fail the whole load closed (below): a partially-loaded + // tree that still publishes its full manifest count via size() would + // serve silent-empty traversals as truth (the cold-load swallow class). prodLog.warn(`LSMTree: Failed to load SSTable ${sstableId}`, error) + failures.push(sstableId) } })() @@ -606,6 +612,20 @@ export class LSMTree { }) await Promise.all(loadPromises) + + if (failures.length > 0) { + // Fail closed. loadManifest()'s catch resets sstables/totalRelationships/ + // sstablesByLevel to honest-empty, so size() reports 0 and the graph + // self-heal (_initializeGraphIndex size()===0 → rebuild) restores the index + // from the canonical records. Honest-partial is never published. + throw new Error( + `LSMTree(${this.config.storagePrefix}): ${failures.length} of ` + + `${this.manifest.sstables.size} SSTable(s) failed to load ` + + `(${failures.join(', ')}) — failing the load closed so size() reports 0 ` + + `and the graph self-heal rebuilds from canonical.` + ) + } + prodLog.info(`LSMTree: Loaded ${this.manifest.sstables.size} SSTables`) } diff --git a/src/hnsw/hnswIndex.ts b/src/hnsw/hnswIndex.ts index b4948bf8..605d2a0b 100644 --- a/src/hnsw/hnswIndex.ts +++ b/src/hnsw/hnswIndex.ts @@ -25,6 +25,29 @@ const DEFAULT_CONFIG: HNSWConfig = { ml: 16 // Max level } +/** + * @description Thrown by {@link JsHnswVectorIndex.flush} when one or more dirty + * nodes (or the system record) could not be persisted. The failed nodes remain + * in the dirty set for the next flush; this error tells the caller the flush did + * NOT achieve durability instead of a node-count that lies. Mandate: loud + * errors, never quiet losses. + */ +export class HnswFlushError extends Error { + constructor( + public readonly failedNodeCount: number, + public readonly systemFailed: boolean, + public override readonly cause?: Error + ) { + super( + `HNSW flush did not achieve durability: ${failedNodeCount} node(s) failed to ` + + `persist${systemFailed ? ' and the system record (entryPoint/maxLevel) failed' : ''}. ` + + `Failed nodes remain dirty for retry.` + + (cause ? ` First error: ${cause.message}` : '') + ) + this.name = 'HnswFlushError' + } +} + /** * Implements {@link VectorIndexProvider}: the vector-index surface Brainy calls * on whatever the `'vector'` factory returns (its own `JsHnswVectorIndex`, or a native @@ -150,41 +173,69 @@ export class JsHnswVectorIndex implements VectorIndexProvider { const startTime = Date.now() const nodeCount = this.dirtyNodes.size - // Batch persist all dirty nodes concurrently + // Batch persist all dirty nodes concurrently. A node whose connections FAIL + // to persist must stay dirty (retried on the next flush) — clearing it would + // silently drop the write forever. Track failures; only successfully- + // persisted (or deleted) nodes leave the dirty set, so nodes added to it + // during this flush are preserved. + const failedNodes = new Set() + let firstError: Error | null = null + if (this.dirtyNodes.size > 0) { const batchSize = 50 // Reasonable batch size for cloud storage const nodeIds = Array.from(this.dirtyNodes) for (let i = 0; i < nodeIds.length; i += batchSize) { const batch = nodeIds.slice(i, i + batchSize) - const promises = batch.map(nodeId => { + const promises = batch.map(async nodeId => { const noun = this.nouns.get(nodeId) - if (!noun) return Promise.resolve() // Node was deleted - - return this.persistNodeConnections(nodeId, noun).catch(error => { - console.error(`[HNSW flush] Failed to persist node ${nodeId}:`, error) - }) + if (!noun) return // Node was deleted — drop it from the dirty set. + try { + await this.persistNodeConnections(nodeId, noun) + } catch (error) { + failedNodes.add(nodeId) + if (firstError === null) firstError = error as Error + prodLog.error(`[HNSW flush] Failed to persist node ${nodeId}: ${(error as Error).message}`) + } }) - await Promise.allSettled(promises) + await Promise.all(promises) } - this.dirtyNodes.clear() + // Remove only nodes that were persisted (or deleted mid-flush); keep the + // failed ones dirty for the next attempt. + for (const nodeId of nodeIds) { + if (!failedNodes.has(nodeId)) this.dirtyNodes.delete(nodeId) + } } - // Persist system data if dirty + // Persist system data if dirty — keep it dirty on failure so the next flush + // retries rather than losing the entry-point/maxLevel update. + let systemFailed = false if (this.dirtySystem) { - await this.storage.saveHNSWSystem({ - entryPointId: this.entryPointId, - maxLevel: this.maxLevel - }).catch(error => { - console.error('[HNSW flush] Failed to persist system data:', error) - }) - - this.dirtySystem = false + try { + await this.storage.saveHNSWSystem({ + entryPointId: this.entryPointId, + maxLevel: this.maxLevel + }) + this.dirtySystem = false + } catch (error) { + systemFailed = true + if (firstError === null) firstError = error as Error + prodLog.error(`[HNSW flush] Failed to persist system data: ${(error as Error).message}`) + } } const duration = Date.now() - startTime + + // Loud failure: if ANY node or the system record could not be persisted the + // flush did not achieve durability. Throw so callers (close(), explicit + // flush(), the flush-request watcher) see the failure instead of a success + // count that lies. The failed nodes/system stay dirty for retry. + if (failedNodes.size > 0 || systemFailed) { + throw new HnswFlushError(failedNodes.size, systemFailed, firstError ?? undefined) + } + if (nodeCount > 0) { prodLog.info(`[HNSW] Flushed ${nodeCount} dirty nodes in ${duration}ms`) } @@ -396,13 +447,14 @@ export class JsHnswVectorIndex implements VectorIndexProvider { this.maxLevel = nounLevel this.nouns.set(id, noun) - // Persist system data for first noun (previously skipped) + // Persist system data for first noun (previously skipped). Surface a + // persist failure loudly — the entry point is the root of the whole index; + // silently dropping it while addItem() returns the id would strand every + // future search on a rootless index. Mandate: never a quiet loss. if (this.storage && this.persistMode === 'immediate') { await this.storage.saveHNSWSystem({ entryPointId: this.entryPointId, maxLevel: this.maxLevel - }).catch(error => { - console.error('Failed to persist initial HNSW system data:', error) }) } else if (this.persistMode === 'deferred') { this.dirtySystem = true diff --git a/src/storage/adapters/fileSystemStorage.ts b/src/storage/adapters/fileSystemStorage.ts index e728cc41..27f1af89 100644 --- a/src/storage/adapters/fileSystemStorage.ts +++ b/src/storage/adapters/fileSystemStorage.ts @@ -17,6 +17,7 @@ import { WriterLockInfo } from '../baseStorage.js' import { getBrainyVersion } from '../../utils/index.js' +import { isAbsentError } from '../../utils/errorClassification.js' // Node.js modules - dynamically imported to avoid issues in browser environments let fs: any @@ -446,8 +447,12 @@ export class FileSystemStorage extends BaseStorage { return null } - console.error(`Error reading object from ${pathStr}:`, error) - return null + // A real storage fault (EIO/EACCES/EMFILE/…) is NOT "object absent". The + // ENOENT branch (above) already returns null, and the corrupted-JSON + // branch (above) is a deliberate concurrent-write tolerance; a genuine + // fault reaching here must propagate loudly rather than masquerade as a + // missing object — which would corrupt reads and drive needless rebuilds. + throw error } } @@ -1207,8 +1212,14 @@ export class FileSystemStorage extends BaseStorage { await this.ensureInitialized() try { return await fs.promises.readFile(this.blobPath(key)) - } catch { - return null + } catch (err) { + // Absent blob → null (the documented contract). A real fault + // (EIO/EACCES/EMFILE/…) must NOT be masked as "absent": doing so makes a + // present-but-unreadable blob look missing and drives a needless rebuild + // or an empty read (the native provider consumes this). Mandate: loud + // errors, never quiet losses. + if (isAbsentError(err)) return null + throw err } } diff --git a/src/storage/baseStorage.ts b/src/storage/baseStorage.ts index ed4755c7..dd1d3fee 100644 --- a/src/storage/baseStorage.ts +++ b/src/storage/baseStorage.ts @@ -3234,10 +3234,22 @@ export abstract class BaseStorage extends BaseStorageAdapter { const priorCounted = isCountedVisibility(record?.visibility) if (priorType) { if (priorCounted) { + // Symmetric with the counted-add increment in saveNounMetadata_internal(): + // decrement BOTH the user-facing scalar total (getNounCount / counts.json) AND + // the per-type bucket, then persist. The scalar decrement was previously + // omitted here, so deletes permanently inflated getNounCount() — the stale + // scalar wins pagination via Math.max(totalNounCount, collected.length) and is + // persisted by scheduleCountPersist(). With this, the invariant + // `totalNounCount === Σ nounCountsByType` holds across add / update-flip / delete. + this.decrementEntityCount(priorType) const idx = TypeUtils.getNounIndex(priorType) if (this.nounCountsByType[idx] > 0) { this.nounCountsByType[idx]-- } + this.scheduleCountPersist().catch(() => { + // Ignore persist errors — the in-memory count is authoritative; a later + // operation retries the persist. + }) } // Symmetric subtype decrement — same non-empty-string guard as the write path. @@ -3392,16 +3404,30 @@ export abstract class BaseStorage extends BaseStorageAdapter { await this.ensureInitialized() // Direct O(1) delete with ID-first path. Read the canonical record BEFORE - // removing it so the verb-subtype decrement is sourced from the edge's own - // metadata (`verb` type + `subtype`) rather than an id-keyed cache — symmetric - // with the increment in `saveVerbMetadata_internal()`. Verb deletes do not - // touch `verbCountsByType` in this path (matching prior behavior), so no - // visibility read is needed here. + // removing it so every decrement is sourced from the edge's own metadata + // (`verb` type, `subtype`, `visibility`) rather than an id-keyed cache — + // symmetric with the increments in `saveVerbMetadata_internal()`. const path = getVerbMetadataPath(id) const record = await this.readCanonicalObject(path) await this.deleteCanonicalObject(path) const priorVerb = record?.verb as VerbType | undefined + // Symmetric count decrement (previously OMITTED — verb deletes touched neither the + // scalar total nor the per-type bucket, so both inflated permanently). A COUNTED + // edge bumped BOTH the scalar (incrementVerbCount in saveVerbMetadata_internal) and + // the per-type bucket (the unconditional bump in saveVerb_internal that the metadata + // path keeps for counted edges). Delete must undo both, gated on the SAME visibility + // as the add, so `totalVerbCount === Σ verbCountsByType` holds across delete. + const priorCounted = isCountedVisibility(record?.visibility) + if (priorVerb && priorCounted) { + this.decrementVerbCount(priorVerb) + const idx = TypeUtils.getVerbIndex(priorVerb) + if (this.verbCountsByType[idx] > 0) this.verbCountsByType[idx]-- + this.scheduleCountPersist().catch(() => { + // Ignore persist errors — in-memory count is authoritative; a later op retries. + }) + } + const priorSubtype = typeof record?.subtype === 'string' && (record.subtype as string).length > 0 ? (record.subtype as string) : undefined diff --git a/src/utils/errorClassification.ts b/src/utils/errorClassification.ts new file mode 100644 index 00000000..a3e7dad8 --- /dev/null +++ b/src/utils/errorClassification.ts @@ -0,0 +1,37 @@ +/** + * @module utils/errorClassification + * @description Shared classification of caught errors into "genuine absence" vs + * "real fault" — the antidote to blind `catch { return null }` handlers that + * cannot tell ENOENT (the object is legitimately not on disk) from EIO / EACCES + * / EMFILE / … (a transient or permission fault on data that IS on disk). + * Masking a fault as absence yields wrong results (a present record read as + * "not found") or a needless rebuild. Mandate: loud errors, never quiet losses. + */ + +/** + * The error `code`s that denote GENUINE absence of a file/object. Only `ENOENT` + * ("no such file or directory") qualifies — on every platform Node maps a + * missing file/directory to ENOENT, and no other errno means "simply not + * there". Every other errno (EIO, EACCES, EPERM, EMFILE, ENFILE, EBUSY, + * ENOTDIR, EISDIR, ELOOP) is a real fault and must propagate, as must any error + * without an errno `code` (parse/decompress failures, generic Errors). + * `ENOTDIR`/`EISDIR` are deliberately faults: a path component of the wrong + * type is corruption, not benign absence. Named constant so a future + * genuine-absence code can be added in one reviewed place. + */ +const ABSENCE_CODES: ReadonlySet = new Set(['ENOENT']) + +/** + * @description True IFF `e` represents genuine absence (an ENOENT-class errno), + * for which returning `null`/`[]`/`undefined` is the correct answer. Returns + * `false` for every real fault, so the canonical call site is: + * `catch (e) { if (isAbsentError(e)) return null; throw e }`. + * + * @param e - The caught value (typed `unknown`; non-objects are never absence). + * @returns Whether the error means "the thing is simply not there". + */ +export function isAbsentError(e: unknown): boolean { + if (e === null || typeof e !== 'object') return false + const code = (e as { code?: unknown }).code + return typeof code === 'string' && ABSENCE_CODES.has(code) +} diff --git a/tests/integration/count-invariant.test.ts b/tests/integration/count-invariant.test.ts new file mode 100644 index 00000000..3896b625 --- /dev/null +++ b/tests/integration/count-invariant.test.ts @@ -0,0 +1,109 @@ +/** + * @module tests/integration/count-invariant + * @description Pattern-C acceptance: the user-facing scalar total and the + * per-type array are ONE consistent projection — `getNounCount() === Σ + * nounCountsByType` (and the verb mirror) after any interleaving of add / + * update-visibility-flip / delete, AND across a reopen. + * + * The bug (finding 5): delete decremented the per-type array but never the + * scalar for nouns, and neither for verbs — so `getNounCount()`/`getVerbCount()` + * inflated permanently (the stale scalar wins pagination via + * `Math.max(total, collected.length)` and is persisted). The fix restores the + * symmetric decrement the code always intended (its own comments say "symmetric + * with the increments"). This test would report an inflated count before the + * fix and the exact count after. + */ +import { describe, it, expect, beforeEach, afterEach } from 'vitest' +import * as fs from 'node:fs' +import * as os from 'node:os' +import * as path from 'node:path' +import { Brainy } from '../../src/brainy.js' +import { NounType, VerbType } from '../../src/types/graphTypes.js' + +const sum = (a: Uint32Array): number => a.reduce((s, c) => s + c, 0) + +describe('count invariant — scalar total === Σ per-type, across delete + reopen (finding 5)', () => { + let dir: string + let brain: any + + const open = async (d: string) => { + const b = new Brainy({ + requireSubtype: false, + storage: { type: 'filesystem', path: d }, + dimensions: 384, + silent: true + }) + await b.init() + return b + } + + beforeEach(async () => { + process.env.BRAINY_DETERMINISTIC_EMBEDDINGS = 'true' + dir = fs.mkdtempSync(path.join(os.tmpdir(), 'brainy-count-inv-')) + brain = await open(dir) + }) + + afterEach(async () => { + try { await brain.close() } catch { /* already closed */ } + fs.rmSync(dir, { recursive: true, force: true }) + }) + + it('noun delete decrements the scalar total, not just the per-type bucket', async () => { + const things: string[] = [] + for (let i = 0; i < 5; i++) things.push(await brain.add({ data: `thing ${i}`, type: NounType.Thing })) + for (let i = 0; i < 3; i++) await brain.add({ data: `concept ${i}`, type: NounType.Concept }) + + const storage = brain['storage'] + expect(await storage.getNounCount()).toBe(8) + expect(sum(storage.getNounCountsByType())).toBe(8) + + // Delete 2 Things — the scalar must drop with the bucket (pre-fix it stayed 8). + await brain.remove(things[0]) + await brain.remove(things[1]) + + expect(await storage.getNounCount()).toBe(6) + expect(sum(storage.getNounCountsByType())).toBe(6) + // The invariant: scalar === Σ per-type. + expect(await storage.getNounCount()).toBe(sum(storage.getNounCountsByType())) + }) + + it('verb delete decrements BOTH the scalar total AND the per-type bucket', async () => { + const ids: string[] = [] + for (let i = 0; i < 4; i++) ids.push(await brain.add({ data: `node ${i}`, type: NounType.Thing })) + const edges: string[] = [] + edges.push(await brain.relate({ from: ids[0], to: ids[1], type: VerbType.RelatedTo })) + edges.push(await brain.relate({ from: ids[1], to: ids[2], type: VerbType.RelatedTo })) + edges.push(await brain.relate({ from: ids[2], to: ids[3], type: VerbType.Contains })) + edges.push(await brain.relate({ from: ids[0], to: ids[3], type: VerbType.Contains })) + + const storage = brain['storage'] + expect(await storage.getVerbCount()).toBe(4) + expect(sum(storage.getVerbCountsByType())).toBe(4) + + // Unrelate one edge — pre-fix, verb delete touched NEITHER counter (both stayed 4). + await brain.unrelate(edges[0]) + + expect(await storage.getVerbCount()).toBe(3) + expect(sum(storage.getVerbCountsByType())).toBe(3) + expect(await storage.getVerbCount()).toBe(sum(storage.getVerbCountsByType())) + }) + + it('the corrected counts persist and survive a reopen', async () => { + const things: string[] = [] + for (let i = 0; i < 6; i++) things.push(await brain.add({ data: `t${i}`, type: NounType.Thing })) + const a = await brain.relate({ from: things[0], to: things[1], type: VerbType.RelatedTo }) + await brain.relate({ from: things[1], to: things[2], type: VerbType.RelatedTo }) + await brain.remove(things[5]) + await brain.unrelate(a) + await brain.flush() + await brain.close() + + // Reopen from disk — counts.json must carry the corrected totals. + brain = await open(dir) + const storage = brain['storage'] + expect(await storage.getNounCount()).toBe(5) + expect(await storage.getVerbCount()).toBe(1) + expect(await storage.getNounCount()).toBe(sum(storage.getNounCountsByType())) + expect(await storage.getVerbCount()).toBe(sum(storage.getVerbCountsByType())) + }) +}) diff --git a/tests/unit/graph/lsm-partial-load-failclosed.test.ts b/tests/unit/graph/lsm-partial-load-failclosed.test.ts new file mode 100644 index 00000000..7082c0c6 --- /dev/null +++ b/tests/unit/graph/lsm-partial-load-failclosed.test.ts @@ -0,0 +1,76 @@ +/** + * @module tests/unit/graph/lsm-partial-load-failclosed + * @description LSMTree partial-load fail-closed guard (Finding 3 of the honest + * index-readiness spine plan). `loadSSTables()` used to swallow a per-SSTable + * load failure with only a `prodLog.warn`, then `loadManifest()` still published + * the FULL persisted `totalRelationships` count — so on a partial cold load, + * `size()`/`isHealthy()` reported the full count and "healthy" while a subset of + * the tree's edges were silently unqueryable. Fixed by failing the whole load + * closed on ANY per-SSTable failure: `loadManifest()`'s existing catch then + * resets `sstables`/`totalRelationships`/`sstablesByLevel` to honest-empty, so + * `size()` reports 0 and the graph layer's existing `size()===0` self-heal + * rebuilds from canonical records — the dishonest partial state can no longer be + * observed. + */ +import { describe, it, expect } from 'vitest' +import { LSMTree } from '../../../src/graph/lsm/LSMTree.js' +import { MemoryStorage } from '../../../src/storage/adapters/memoryStorage.js' + +describe('LSMTree partial SSTable load fails closed (honest size())', () => { + it('one SSTable fails to load → size() reports 0, not the manifest count', async () => { + const storage = new MemoryStorage() + await storage.init() + const prefix = 'test-lsm-partial' + + const t1 = new LSMTree(storage, { storagePrefix: prefix, enableCompaction: false }) + await t1.init() + await t1.add('a', 'b'); await t1.flush() // SSTable #1 + await t1.add('c', 'd'); await t1.flush() // SSTable #2 + expect(t1.size()).toBeGreaterThanOrEqual(2) + await t1.close() + + // Find an SSTable id from the persisted manifest. + const manifest = await storage.getMetadata(`${prefix}-manifest`) + const sstableIds = Object.keys((manifest!.data as any).sstables) + expect(sstableIds.length).toBeGreaterThanOrEqual(2) + + // Reopen with ONE SSTable load forced to fail (deterministic). + const t2 = new LSMTree(storage, { storagePrefix: prefix, enableCompaction: false }) + const realGet = storage.getMetadata.bind(storage) + ;(storage as any).getMetadata = async (key: string) => { + if (key === `${prefix}-${sstableIds[1]}`) throw new Error('simulated SSTable load failure') + return realGet(key) + } + await t2.init() + + // BEFORE THE FIX: t2.size() === (full manifest count) ← LIE (partial load, full count) + // AFTER THE FIX: t2.size() === 0 ← fail-closed → self-heal rebuilds + expect(t2.size()).toBe(0) + expect(t2.isHealthy()).toBe(true) // honestly-empty-but-initialized, not "healthy with a lie" + + await t2.close() + }) + + it('a clean load with no failures still reports the full honest count', async () => { + const storage = new MemoryStorage() + await storage.init() + const prefix = 'test-lsm-clean' + + const t1 = new LSMTree(storage, { storagePrefix: prefix, enableCompaction: false }) + await t1.init() + await t1.add('a', 'b'); await t1.flush() + await t1.add('c', 'd'); await t1.flush() + const expectedSize = t1.size() + expect(expectedSize).toBeGreaterThanOrEqual(2) + await t1.close() + + // Reopen with no injected failures — a normal, fully-successful cold load. + const t2 = new LSMTree(storage, { storagePrefix: prefix, enableCompaction: false }) + await t2.init() + + expect(t2.size()).toBe(expectedSize) + expect(t2.isHealthy()).toBe(true) + + await t2.close() + }) +}) diff --git a/tests/unit/hnsw/flush-failure.test.ts b/tests/unit/hnsw/flush-failure.test.ts new file mode 100644 index 00000000..14bdff6e --- /dev/null +++ b/tests/unit/hnsw/flush-failure.test.ts @@ -0,0 +1,157 @@ +/** + * HNSW deferred-flush durability tests (Finding 6 — spine-plan Part B, + * "blind catch" audit). + * + * `JsHnswVectorIndex.flush()` used to swallow a per-node + * `persistNodeConnections` failure (and a system-record `saveHNSWSystem` + * failure) behind `console.error`, then unconditionally clear the dirty set + * and report the pre-flush node count as "success". A transient write fault + * therefore silently dropped that node's connections from durable storage + * forever, with `flush()` having lied about it. These tests pin the fix: + * a node (or the system record) that fails to persist stays in the + * dirty/retry set, and `flush()` throws {@link HnswFlushError} instead of + * returning a count. The immediate-mode first-noun `saveHNSWSystem` swallow + * (while `addItem()` still returned the id) is covered too. + */ + +import { describe, it, expect, vi } from 'vitest' +import { v4 as uuidv4 } from 'uuid' +import { JsHnswVectorIndex, HnswFlushError } from '../../../src/hnsw/hnswIndex.js' +import { euclideanDistance } from '../../../src/utils/index.js' +import { MemoryStorage } from '../../../src/storage/adapters/memoryStorage.js' + +// Helper: generate a random vector of given dimension (mirrors lazy-vectors.test.ts). +function randomVector(dim: number): number[] { + return Array.from({ length: dim }, () => Math.random() * 2 - 1) +} + +describe('HNSW deferred-flush durability (Finding 6)', () => { + const dim = 8 + + it('a clean flush persists every dirty node and clears the dirty set', async () => { + const storage = new MemoryStorage() + const index = new JsHnswVectorIndex( + { M: 4, efConstruction: 50, efSearch: 20 }, + euclideanDistance, + { useParallelization: false, storage, persistMode: 'deferred' } + ) + + for (let i = 0; i < 5; i++) { + await index.addItem({ id: uuidv4(), vector: randomVector(dim) }) + } + + // Sanity: inserting several items in deferred mode does dirty something. + expect((index as any).dirtyNodes.size).toBeGreaterThan(0) + + const flushed = await index.flush() + expect(typeof flushed).toBe('number') + expect((index as any).dirtyNodes.size).toBe(0) + expect((index as any).dirtySystem).toBe(false) + }) + + it('retains a node whose connections failed to persist and surfaces HnswFlushError instead of reporting success', async () => { + const storage = new MemoryStorage() + const index = new JsHnswVectorIndex( + { M: 4, efConstruction: 50, efSearch: 20 }, + euclideanDistance, + { useParallelization: false, storage, persistMode: 'deferred' } + ) + + const ids: string[] = [] + for (let i = 0; i < 5; i++) { + const id = uuidv4() + ids.push(id) + await index.addItem({ id, vector: randomVector(dim) }) + } + + // The very first node is guaranteed to become a neighbor of the second + // insert (it is the only existing node in the graph at that point), so it + // is deterministically present in the dirty set before any fault. + const failId = ids[0] + const dirtyBefore = (index as any).dirtyNodes as Set + expect(dirtyBefore.has(failId)).toBe(true) + + const originalSave = storage.saveVectorIndexData.bind(storage) + const spy = vi + .spyOn(storage, 'saveVectorIndexData') + .mockImplementation(async (nounId, hnswData) => { + if (nounId === failId) { + throw Object.assign(new Error('simulated write fault'), { code: 'EIO' }) + } + return originalSave(nounId, hnswData) + }) + + await expect(index.flush()).rejects.toBeInstanceOf(HnswFlushError) + + const dirtyAfter = (index as any).dirtyNodes as Set + // The failed node stays dirty for the next retry ... + expect(dirtyAfter.has(failId)).toBe(true) + // ... and every node that DID persist successfully leaves the dirty set — + // the failure of one node must not re-dirty (or fail to clear) the rest. + expect(dirtyAfter.size).toBe(1) + + // Recovery: once the fault clears, the retained node persists and the + // flush reports success again (the intended retry path). + spy.mockRestore() + const flushed = await index.flush() + expect(typeof flushed).toBe('number') + expect((index as any).dirtyNodes.size).toBe(0) + }) + + it('surfaces a system-record persist failure as HnswFlushError and keeps dirtySystem set for retry', async () => { + const storage = new MemoryStorage() + const index = new JsHnswVectorIndex( + { M: 4, efConstruction: 50, efSearch: 20 }, + euclideanDistance, + { useParallelization: false, storage, persistMode: 'deferred' } + ) + + // First noun in deferred mode marks dirtySystem (entryPoint/maxLevel). + await index.addItem({ id: uuidv4(), vector: randomVector(dim) }) + expect((index as any).dirtySystem).toBe(true) + + const spy = vi + .spyOn(storage, 'saveHNSWSystem') + .mockRejectedValue( + Object.assign(new Error('simulated system write fault'), { code: 'EIO' }) + ) + + let caught: unknown + try { + await index.flush() + } catch (error) { + caught = error + } + + expect(caught).toBeInstanceOf(HnswFlushError) + expect((caught as HnswFlushError).systemFailed).toBe(true) + // The system record must stay dirty — a lost entry point/maxLevel update + // must be retried, not silently dropped. + expect((index as any).dirtySystem).toBe(true) + + spy.mockRestore() + const flushed = await index.flush() + expect(typeof flushed).toBe('number') + expect((index as any).dirtySystem).toBe(false) + }) + + it('surfaces the immediate-mode first-noun system persist failure via a rejecting addItem()', async () => { + const storage = new MemoryStorage() + const index = new JsHnswVectorIndex( + { M: 4, efConstruction: 50, efSearch: 20 }, + euclideanDistance, + { useParallelization: false, storage } // default persistMode: 'immediate' + ) + + vi.spyOn(storage, 'saveHNSWSystem').mockRejectedValue( + Object.assign(new Error('simulated system write fault'), { code: 'EIO' }) + ) + + // Previously this swallowed the error (console.error) and addItem() + // still resolved with the id — stranding a brand-new index whose root + // (entry point) was never actually persisted. Now it must reject. + await expect( + index.addItem({ id: uuidv4(), vector: randomVector(dim) }) + ).rejects.toThrow('simulated system write fault') + }) +})