fix: spine hardening pass 1 (part) — count symmetry, honest partial-load, flush durability, read-fault propagation
Write/index-spine hardening, first batch of Pass 1. Each fix restores an invariant the surrounding code already intended; every one has a fail-before/pass-after test. - Pattern C, finding 5 (baseStorage): delete now decrements the user-facing scalar total symmetrically — deleteNounMetadata was decrementing only the per-type bucket, deleteVerbMetadata neither the bucket nor the scalar, so getNounCount()/getVerbCount() inflated permanently (the stale scalar wins pagination via Math.max and is persisted). Invariant now holds: scalar total === Σ per-type across add/update/delete and reopen. - Pattern A, finding 3 (graph/lsm/LSMTree): a partial SSTable-load failure no longer publishes the manifest's full relationship count as healthy. Any per-SSTable load failure throws after the batch, which resets to honest-empty and lets the existing size()===0 self-heal rebuild run — size()/isHealthy() can no longer lie about a partial load. - Pattern B, finding 6 (hnsw/hnswIndex): deferred flush() no longer clears dirty nodes whose connections failed to persist — failed nodes stay in the retry set, and flush() throws HnswFlushError instead of returning a lying node count. The immediate-mode first-noun saveHNSWSystem is un-swallowed, so addItem() rejects rather than returning an id for a rootless index. - Pattern B, finding 11 (part — storage reads): new shared isAbsentError() helper (utils/errorClassification, ENOENT-only absence) applied to loadBinaryBlob and readObjectFromPath — a real IO fault (EIO/EACCES/EMFILE) now propagates loudly instead of masquerading as "absent", which had driven needless rebuilds / empty reads (loadBinaryBlob feeds the native provider). Regression: 78 green across the 3 new suites + db-mvcc, generationStore, temporal-vfs, rollback-trapdoor, restore-nondestructive. Full gate runs before the Pass-1 release (David-gated). Remaining Pass 1: finding 11 getNoun/getVerb legs, finding 4 (ColumnStore), finding 8 (pending-flush), finding 10 (degraded), finding 7 (clear). Pattern A guards (1,2,9) as a follow-up release.
This commit is contained in:
parent
eb9c4eb963
commit
119087a75c
8 changed files with 518 additions and 30 deletions
|
|
@ -574,6 +574,7 @@ export class LSMTree {
|
||||||
* Load SSTables from storage based on manifest
|
* Load SSTables from storage based on manifest
|
||||||
*/
|
*/
|
||||||
private async loadSSTables(): Promise<void> {
|
private async loadSSTables(): Promise<void> {
|
||||||
|
const failures: string[] = []
|
||||||
const loadPromises: Promise<void>[] = []
|
const loadPromises: Promise<void>[] = []
|
||||||
|
|
||||||
this.manifest.sstables.forEach((level, sstableId) => {
|
this.manifest.sstables.forEach((level, sstableId) => {
|
||||||
|
|
@ -598,7 +599,12 @@ export class LSMTree {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
} catch (error) {
|
} 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)
|
prodLog.warn(`LSMTree: Failed to load SSTable ${sstableId}`, error)
|
||||||
|
failures.push(sstableId)
|
||||||
}
|
}
|
||||||
})()
|
})()
|
||||||
|
|
||||||
|
|
@ -606,6 +612,20 @@ export class LSMTree {
|
||||||
})
|
})
|
||||||
|
|
||||||
await Promise.all(loadPromises)
|
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`)
|
prodLog.info(`LSMTree: Loaded ${this.manifest.sstables.size} SSTables`)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -25,6 +25,29 @@ const DEFAULT_CONFIG: HNSWConfig = {
|
||||||
ml: 16 // Max level
|
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
|
* Implements {@link VectorIndexProvider}: the vector-index surface Brainy calls
|
||||||
* on whatever the `'vector'` factory returns (its own `JsHnswVectorIndex`, or a native
|
* 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 startTime = Date.now()
|
||||||
const nodeCount = this.dirtyNodes.size
|
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<string>()
|
||||||
|
let firstError: Error | null = null
|
||||||
|
|
||||||
if (this.dirtyNodes.size > 0) {
|
if (this.dirtyNodes.size > 0) {
|
||||||
const batchSize = 50 // Reasonable batch size for cloud storage
|
const batchSize = 50 // Reasonable batch size for cloud storage
|
||||||
const nodeIds = Array.from(this.dirtyNodes)
|
const nodeIds = Array.from(this.dirtyNodes)
|
||||||
|
|
||||||
for (let i = 0; i < nodeIds.length; i += batchSize) {
|
for (let i = 0; i < nodeIds.length; i += batchSize) {
|
||||||
const batch = nodeIds.slice(i, 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)
|
const noun = this.nouns.get(nodeId)
|
||||||
if (!noun) return Promise.resolve() // Node was deleted
|
if (!noun) return // Node was deleted — drop it from the dirty set.
|
||||||
|
try {
|
||||||
return this.persistNodeConnections(nodeId, noun).catch(error => {
|
await this.persistNodeConnections(nodeId, noun)
|
||||||
console.error(`[HNSW flush] Failed to persist node ${nodeId}:`, error)
|
} 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) {
|
if (this.dirtySystem) {
|
||||||
|
try {
|
||||||
await this.storage.saveHNSWSystem({
|
await this.storage.saveHNSWSystem({
|
||||||
entryPointId: this.entryPointId,
|
entryPointId: this.entryPointId,
|
||||||
maxLevel: this.maxLevel
|
maxLevel: this.maxLevel
|
||||||
}).catch(error => {
|
|
||||||
console.error('[HNSW flush] Failed to persist system data:', error)
|
|
||||||
})
|
})
|
||||||
|
|
||||||
this.dirtySystem = false
|
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
|
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) {
|
if (nodeCount > 0) {
|
||||||
prodLog.info(`[HNSW] Flushed ${nodeCount} dirty nodes in ${duration}ms`)
|
prodLog.info(`[HNSW] Flushed ${nodeCount} dirty nodes in ${duration}ms`)
|
||||||
}
|
}
|
||||||
|
|
@ -396,13 +447,14 @@ export class JsHnswVectorIndex implements VectorIndexProvider {
|
||||||
this.maxLevel = nounLevel
|
this.maxLevel = nounLevel
|
||||||
this.nouns.set(id, noun)
|
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') {
|
if (this.storage && this.persistMode === 'immediate') {
|
||||||
await this.storage.saveHNSWSystem({
|
await this.storage.saveHNSWSystem({
|
||||||
entryPointId: this.entryPointId,
|
entryPointId: this.entryPointId,
|
||||||
maxLevel: this.maxLevel
|
maxLevel: this.maxLevel
|
||||||
}).catch(error => {
|
|
||||||
console.error('Failed to persist initial HNSW system data:', error)
|
|
||||||
})
|
})
|
||||||
} else if (this.persistMode === 'deferred') {
|
} else if (this.persistMode === 'deferred') {
|
||||||
this.dirtySystem = true
|
this.dirtySystem = true
|
||||||
|
|
|
||||||
|
|
@ -17,6 +17,7 @@ import {
|
||||||
WriterLockInfo
|
WriterLockInfo
|
||||||
} from '../baseStorage.js'
|
} from '../baseStorage.js'
|
||||||
import { getBrainyVersion } from '../../utils/index.js'
|
import { getBrainyVersion } from '../../utils/index.js'
|
||||||
|
import { isAbsentError } from '../../utils/errorClassification.js'
|
||||||
|
|
||||||
// Node.js modules - dynamically imported to avoid issues in browser environments
|
// Node.js modules - dynamically imported to avoid issues in browser environments
|
||||||
let fs: any
|
let fs: any
|
||||||
|
|
@ -446,8 +447,12 @@ export class FileSystemStorage extends BaseStorage {
|
||||||
return null
|
return null
|
||||||
}
|
}
|
||||||
|
|
||||||
console.error(`Error reading object from ${pathStr}:`, error)
|
// A real storage fault (EIO/EACCES/EMFILE/…) is NOT "object absent". The
|
||||||
return null
|
// 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()
|
await this.ensureInitialized()
|
||||||
try {
|
try {
|
||||||
return await fs.promises.readFile(this.blobPath(key))
|
return await fs.promises.readFile(this.blobPath(key))
|
||||||
} catch {
|
} catch (err) {
|
||||||
return null
|
// 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
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -3234,10 +3234,22 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
||||||
const priorCounted = isCountedVisibility(record?.visibility)
|
const priorCounted = isCountedVisibility(record?.visibility)
|
||||||
if (priorType) {
|
if (priorType) {
|
||||||
if (priorCounted) {
|
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)
|
const idx = TypeUtils.getNounIndex(priorType)
|
||||||
if (this.nounCountsByType[idx] > 0) {
|
if (this.nounCountsByType[idx] > 0) {
|
||||||
this.nounCountsByType[idx]--
|
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.
|
// 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()
|
await this.ensureInitialized()
|
||||||
|
|
||||||
// Direct O(1) delete with ID-first path. Read the canonical record BEFORE
|
// 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
|
// removing it so every decrement is sourced from the edge's own metadata
|
||||||
// metadata (`verb` type + `subtype`) rather than an id-keyed cache — symmetric
|
// (`verb` type, `subtype`, `visibility`) rather than an id-keyed cache —
|
||||||
// with the increment in `saveVerbMetadata_internal()`. Verb deletes do not
|
// symmetric with the increments in `saveVerbMetadata_internal()`.
|
||||||
// touch `verbCountsByType` in this path (matching prior behavior), so no
|
|
||||||
// visibility read is needed here.
|
|
||||||
const path = getVerbMetadataPath(id)
|
const path = getVerbMetadataPath(id)
|
||||||
const record = await this.readCanonicalObject(path)
|
const record = await this.readCanonicalObject(path)
|
||||||
await this.deleteCanonicalObject(path)
|
await this.deleteCanonicalObject(path)
|
||||||
|
|
||||||
const priorVerb = record?.verb as VerbType | undefined
|
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
|
const priorSubtype = typeof record?.subtype === 'string' && (record.subtype as string).length > 0
|
||||||
? (record.subtype as string)
|
? (record.subtype as string)
|
||||||
: undefined
|
: undefined
|
||||||
|
|
|
||||||
37
src/utils/errorClassification.ts
Normal file
37
src/utils/errorClassification.ts
Normal file
|
|
@ -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<string> = 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)
|
||||||
|
}
|
||||||
109
tests/integration/count-invariant.test.ts
Normal file
109
tests/integration/count-invariant.test.ts
Normal file
|
|
@ -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()))
|
||||||
|
})
|
||||||
|
})
|
||||||
76
tests/unit/graph/lsm-partial-load-failclosed.test.ts
Normal file
76
tests/unit/graph/lsm-partial-load-failclosed.test.ts
Normal file
|
|
@ -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()
|
||||||
|
})
|
||||||
|
})
|
||||||
157
tests/unit/hnsw/flush-failure.test.ts
Normal file
157
tests/unit/hnsw/flush-failure.test.ts
Normal file
|
|
@ -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<string>
|
||||||
|
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<string>
|
||||||
|
// 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')
|
||||||
|
})
|
||||||
|
})
|
||||||
Loading…
Add table
Add a link
Reference in a new issue