feat: canonical enumeration mode for export — storage-walked, canon-complete, with an index-drift report
This commit is contained in:
parent
999d0ebbcf
commit
4d196af41b
6 changed files with 543 additions and 20 deletions
31
RELEASES.md
31
RELEASES.md
|
|
@ -31,6 +31,37 @@ is sometimes cited as a 7.x removal — those methods never existed on 7.x; the
|
|||
|
||||
---
|
||||
|
||||
## Unreleased (canonical enumeration mode for export — storage-walked, canon-complete)
|
||||
|
||||
From a fleet data-migration program's requirement for whole-brain exports that are
|
||||
provably canon-complete: `export()`'s default enumeration for a whole-brain/predicate
|
||||
selector is a generation-correct paginated `find()` walk — a projection query riding
|
||||
the metadata index as an acceleration structure. Production has documented both of the
|
||||
index's failure classes: a lost/stale posting can silently OMIT a canonical record from
|
||||
an export, and a stale posting can silently INCLUDE a phantom row. Neither is visible
|
||||
to the caller today.
|
||||
|
||||
- **New: `export(selector, { enumeration: 'canonical' })`** (default remains `'index'` —
|
||||
unchanged behavior on this release). Canonical mode walks every live noun/verb
|
||||
directly off the storage adapter's canonical shard layout (`storage.getNouns()` /
|
||||
`getVerbs()` — the same primitive `repairIndex()`'s recount and every index-heal
|
||||
walk use) instead of the metadata/graph indexes, then applies the selector as a
|
||||
plain predicate over the walked records. This guarantees canon-completeness — index
|
||||
corruption cannot hide a live record from the export — at the cost of an O(N) walk
|
||||
regardless of selector selectivity. Relations are also walked canonically in this
|
||||
mode, for every selector, not just the whole-brain case. Requires the LIVE current
|
||||
generation: called on a historical `asOf()` view or a speculative `with()` overlay it
|
||||
throws `CanonicalEnumerationUnavailableError` rather than silently mixing generations
|
||||
or missing an overlay's own entities — `enumeration: 'index'` (the default) is
|
||||
unaffected and still composes with `asOf()`/`with()` as before.
|
||||
- **New: `export(selector, { enumeration: 'canonical', reportIndexDrift: true })`** —
|
||||
also runs the index-based enumeration and diffs it against canonical ground truth,
|
||||
attaching `PortableGraph.drift: { canonicalOnly: string[], indexOnly: string[] }`
|
||||
(canon-present ids the index missed; index-visible ids canon-absent — phantoms).
|
||||
Migration-audit evidence, not a repair: nonzero drift is reported loudly
|
||||
(`console.warn` with the counts) and nothing is auto-healed — run `brain.repairIndex()`
|
||||
to reconcile the metadata index once drift is confirmed.
|
||||
|
||||
## v8.10.1 — 2026-07-24 (the no-hot-retry contract + warm()'s metadata surface under native providers)
|
||||
|
||||
From a production incident: a native-provider op ground 38-40s inside a transaction,
|
||||
|
|
|
|||
22
src/db/db.ts
22
src/db/db.ts
|
|
@ -66,7 +66,7 @@ import {
|
|||
import { v4 as uuidv4 } from '../universal/uuid.js'
|
||||
import { coerceNewEntityId, resolveEntityId, ORIGINAL_ID_KEY } from '../utils/idNormalization.js'
|
||||
import { EntityNotFoundError } from '../errors/notFound.js'
|
||||
import { SpeculativeOverlayError } from './errors.js'
|
||||
import { SpeculativeOverlayError, CanonicalEnumerationUnavailableError } from './errors.js'
|
||||
import type { GenerationStore } from './generationStore.js'
|
||||
import type { ChangedIds, TransactReceipt, TxOperation } from './types.js'
|
||||
import { entityMatchesFind, resolveEntityField, UnsupportedWhereOperatorError } from './whereMatcher.js'
|
||||
|
|
@ -520,14 +520,32 @@ export class Db<T = any> {
|
|||
* (no generation history) — distinct from `persist()` (native whole-brain snapshot
|
||||
* that preserves history). Restore with `brain.import(backup)`.
|
||||
*
|
||||
* `options.enumeration: 'canonical'` (default: `'index'`) walks the storage
|
||||
* adapter's canonical noun/verb layout directly instead of the metadata/graph
|
||||
* indexes, guaranteeing canon-completeness against index corruption — see
|
||||
* {@link ExportOptions.enumeration}. It requires the LIVE, current-generation
|
||||
* view: called on a historical `asOf()` pin or a speculative `with()` overlay it
|
||||
* throws {@link CanonicalEnumerationUnavailableError} rather than silently mixing
|
||||
* generations or missing the overlay's own entities.
|
||||
*
|
||||
* @param selector - WHAT to export (omit for the whole brain). See {@link ExportSelector}.
|
||||
* @param options - HOW to export (vectors / VFS bytes / edge policy). See {@link ExportOptions}.
|
||||
* @param options - HOW to export (vectors / VFS bytes / edge policy / enumeration mode). See {@link ExportOptions}.
|
||||
* @returns A versioned, portable `PortableGraph` document.
|
||||
* @throws {@link CanonicalEnumerationUnavailableError} if `enumeration:'canonical'` is
|
||||
* requested on a historical or speculative-overlay view.
|
||||
* @example
|
||||
* const backup = await brain.now().export({ collection: id }, { includeVectors: true })
|
||||
* @example
|
||||
* // Canon-complete audit export, with an index-drift report attached.
|
||||
* const audit = await brain.now().export({}, { enumeration: 'canonical', reportIndexDrift: true })
|
||||
* if (audit.drift) console.log(audit.drift.canonicalOnly, audit.drift.indexOnly)
|
||||
*/
|
||||
async export(selector: ExportSelector = {}, options: ExportOptions = {}): Promise<PortableGraph> {
|
||||
this.assertUsable('export')
|
||||
if (options.enumeration === 'canonical') {
|
||||
if (this.overlay) throw new CanonicalEnumerationUnavailableError(this.gen, 'overlay')
|
||||
if (this.isHistorical()) throw new CanonicalEnumerationUnavailableError(this.gen, 'historical')
|
||||
}
|
||||
return exportGraph(this, this.host.storage, selector, options)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -23,8 +23,12 @@
|
|||
* serve the full query surface via at-generation index materialization.
|
||||
* - {@link GenerationCompactedError} — `asOf()` asked for a generation whose
|
||||
* immutable records were reclaimed by `compactHistory()`.
|
||||
* - {@link CanonicalEnumerationUnavailableError} — `export()`'s
|
||||
* `enumeration:'canonical'` mode was called on a historical `asOf()` view or a
|
||||
* speculative `with()` overlay; the canonical storage walk only ever answers
|
||||
* "what is live right now."
|
||||
*
|
||||
* All three are exported from the package root (`@soulcraft/brainy`).
|
||||
* All are exported from the package root (`@soulcraft/brainy`).
|
||||
*/
|
||||
|
||||
/**
|
||||
|
|
@ -160,6 +164,64 @@ export class GenerationCompactedError extends Error {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @description Thrown by `db.export(selector, { enumeration: 'canonical' })` when
|
||||
* the `Db` it is called on is not the live, current-generation view: a historical
|
||||
* `brain.asOf(g)` pin, or a speculative `db.with()` overlay.
|
||||
*
|
||||
* Canonical enumeration mode walks the storage adapter's canonical shard layout
|
||||
* directly (`storage.getNouns()`/`getVerbs()`) instead of the metadata/graph
|
||||
* indexes — but that walk has no generation parameter, it can only ever answer
|
||||
* "what is live right now." Serving it against a historical pin would silently
|
||||
* mix generations (today's canonical records under yesterday's selector), and
|
||||
* against a speculative overlay it would silently miss the overlay's own
|
||||
* in-memory entities (which never touched storage). Both are exactly the kind of
|
||||
* silently-wrong result canonical mode exists to prevent elsewhere — so this
|
||||
* boundary throws instead.
|
||||
*
|
||||
* `enumeration: 'index'` (the default) is unaffected: it composes with
|
||||
* `asOf()`/`with()` exactly as before, via the generation-correct `find()` walk.
|
||||
*
|
||||
* @example
|
||||
* const past = await brain.asOf(g1)
|
||||
* try {
|
||||
* await past.export({}, { enumeration: 'canonical' })
|
||||
* } catch (err) {
|
||||
* if (err instanceof CanonicalEnumerationUnavailableError) {
|
||||
* // Time-travel export: use the default index-based enumeration instead.
|
||||
* await past.export({}, { enumeration: 'index' })
|
||||
* }
|
||||
* }
|
||||
*/
|
||||
export class CanonicalEnumerationUnavailableError extends Error {
|
||||
/** The view's pinned generation. */
|
||||
public readonly generation: number
|
||||
/** Why canonical mode cannot serve this view. */
|
||||
public readonly reason: 'historical' | 'overlay'
|
||||
|
||||
/**
|
||||
* @param generation - The view's pinned generation.
|
||||
* @param reason - `'historical'` (a past `asOf()` pin) or `'overlay'` (a speculative `with()`).
|
||||
*/
|
||||
constructor(generation: number, reason: 'historical' | 'overlay') {
|
||||
const what =
|
||||
reason === 'historical'
|
||||
? `a historical view pinned at generation ${generation}`
|
||||
: `a speculative with() overlay (base generation ${generation})`
|
||||
super(
|
||||
`export()'s enumeration:'canonical' requires the live, current-generation view — ` +
|
||||
`it was called on ${what}. The canonical storage walk has no generation parameter, ` +
|
||||
`so it can only answer "what is live right now"; serving it here would silently ` +
|
||||
`mix generations (historical) or miss the overlay's own in-memory entities ` +
|
||||
`(overlay). Use enumeration:'index' (the default) for a time-travel or what-if ` +
|
||||
`export, or pin brain.now() for a live canonical export.`
|
||||
)
|
||||
this.name = 'CanonicalEnumerationUnavailableError'
|
||||
this.generation = generation
|
||||
this.reason = reason
|
||||
}
|
||||
}
|
||||
|
||||
/** One entity/relationship left in an unreconciled state by a failed rollback. */
|
||||
export interface UnreconciledRecord {
|
||||
/** The entity or relationship id. */
|
||||
|
|
|
|||
|
|
@ -27,7 +27,7 @@
|
|||
|
||||
import { Entity, Relation, Result } from '../types/brainy.types.js'
|
||||
import { NounType, VerbType } from '../types/graphTypes.js'
|
||||
import { StorageAdapter } from '../coreTypes.js'
|
||||
import { StorageAdapter, HNSWVerbWithMetadata } from '../coreTypes.js'
|
||||
import { getBrainyVersion } from '../utils/version.js'
|
||||
import { TxOperation } from './types.js'
|
||||
|
||||
|
|
@ -96,6 +96,67 @@ export interface ExportOptions {
|
|||
includeSystem?: boolean
|
||||
/** Which edges to include (default: `'induced'`). */
|
||||
edges?: 'induced' | 'incident' | 'none'
|
||||
/**
|
||||
* How the whole-brain / predicate selector (no `ids`/`collection`/`connected`/
|
||||
* `vfsPath`) resolves its candidate id set:
|
||||
*
|
||||
* - `'index'` (DEFAULT — unchanged behavior) — the generation-correct
|
||||
* paginated `find()` walk. Fast (O(matches), not O(N)) but rides the
|
||||
* metadata index as an acceleration structure: a lost/stale index posting
|
||||
* can silently OMIT a canonical record, and a stale posting pointing at a
|
||||
* record that no longer matches can silently produce a phantom (dropped
|
||||
* later by the same predicate re-check `'canonical'` mode also runs, so
|
||||
* phantoms never reach `entities` — but they ARE lost silently unless
|
||||
* {@link reportIndexDrift} is set).
|
||||
* - `'canonical'` — walks every live noun/verb directly off the storage
|
||||
* adapter's canonical shard layout (`storage.getNouns()`/`getVerbs()` —
|
||||
* the same primitive `repairIndex()`'s recount and every index-heal walk
|
||||
* use), then applies the selector as a plain predicate over the walked
|
||||
* records. This GUARANTEES canon-completeness — the metadata/graph
|
||||
* indexes are never consulted, so their corruption cannot hide a live
|
||||
* record — at the cost of an O(N) walk regardless of selector
|
||||
* selectivity (unlike the index path's O(matches)). Structural selectors
|
||||
* (`ids`/`collection`/`connected`/`vfsPath`) resolve their node set
|
||||
* exactly as in `'index'` mode either way (they never rode the metadata
|
||||
* index); `'canonical'` additionally walks relations canonically for
|
||||
* EVERY selector, since a lost adjacency-index posting can hide a
|
||||
* relation regardless of how the node set was produced. Requires a
|
||||
* storage adapter and the CURRENT generation — throws on a historical
|
||||
* `asOf()` view or a speculative `with()` overlay (see
|
||||
* {@link CanonicalEnumerationUnavailableError}), because the canonical
|
||||
* walk has no notion of "as of a past generation."
|
||||
*/
|
||||
enumeration?: 'index' | 'canonical'
|
||||
/**
|
||||
* Only meaningful with `enumeration:'canonical'` (ignored otherwise): ALSO
|
||||
* run the `'index'` enumeration in parallel and diff it against the
|
||||
* canonical ground truth, attaching the result as {@link PortableGraph.drift}.
|
||||
* Never auto-heals anything — this is migration-audit evidence, reported
|
||||
* loudly (`console.warn` with the counts) whenever either list is non-empty,
|
||||
* never silently. Default: false.
|
||||
*/
|
||||
reportIndexDrift?: boolean
|
||||
}
|
||||
|
||||
/**
|
||||
* @description Index-vs-canonical drift for one `export({ enumeration: 'canonical',
|
||||
* reportIndexDrift: true })` call. Only populated for the whole-brain / predicate
|
||||
* selector (structural selectors never consulted the metadata index for their node
|
||||
* set, so there is nothing to diff — both lists are empty for those).
|
||||
*/
|
||||
export interface ExportIndexDrift {
|
||||
/**
|
||||
* Ids the canonical storage walk confirmed (live, selector-matching) that the
|
||||
* index-based `find()` enumeration did NOT return — canonical records the
|
||||
* metadata index has lost track of.
|
||||
*/
|
||||
canonicalOnly: string[]
|
||||
/**
|
||||
* Ids the index-based `find()` enumeration returned for this selector that
|
||||
* canonical ground truth (the storage walk + the same predicate check) does
|
||||
* NOT support — phantom index rows (stale or cross-bucket postings).
|
||||
*/
|
||||
indexOnly: string[]
|
||||
}
|
||||
|
||||
/** Controls how a `PortableGraph` is applied on `import()`. */
|
||||
|
|
@ -151,6 +212,8 @@ export interface PortableGraph {
|
|||
relations: PortableGraphRelation[]
|
||||
blobs?: Record<string, string>
|
||||
danglingIds?: string[]
|
||||
/** Present only when `export()` was called with `reportIndexDrift: true`. */
|
||||
drift?: ExportIndexDrift
|
||||
stats: { entityCount: number; relationCount: number; blobCount: number; vectorDimensions?: number }
|
||||
}
|
||||
|
||||
|
|
@ -271,11 +334,19 @@ export function validatePortableGraph(data: unknown): PortableGraphValidation {
|
|||
/**
|
||||
* @description Serialize part or all of a graph (read through `reader` at its pinned
|
||||
* generation) into a portable `PortableGraph` document.
|
||||
*
|
||||
* `enumeration:'canonical'` (see {@link ExportOptions.enumeration}) requires a
|
||||
* storage adapter and the current generation: it throws
|
||||
* {@link CanonicalEnumerationUnavailableError} if `storage` is absent, and the
|
||||
* caller (`Db.export()`) throws the same error before this runs if the view is
|
||||
* historical or a speculative overlay — the canonical storage walk has no
|
||||
* generation parameter, so it can only ever answer "as of right now."
|
||||
*
|
||||
* @param reader - Generation-correct read surface (`Db` or `Brainy`).
|
||||
* @param storage - Storage adapter (used only for VFS blob bytes when `includeContent`).
|
||||
* @param storage - Storage adapter (VFS blob bytes when `includeContent`; the
|
||||
* canonical noun/verb walk when `enumeration:'canonical'`).
|
||||
* @param selector - WHAT to export (omit for the whole brain).
|
||||
* @param options - HOW to export (vectors / file bytes / edge policy).
|
||||
* @param dimensions - Embedding dimensionality for the manifest.
|
||||
* @param options - HOW to export (vectors / file bytes / edge policy / enumeration mode).
|
||||
*/
|
||||
export async function exportGraph<T = any>(
|
||||
reader: PortableGraphReader<T>,
|
||||
|
|
@ -287,13 +358,34 @@ export async function exportGraph<T = any>(
|
|||
includeVectors = false,
|
||||
includeContent = false,
|
||||
includeSystem = false,
|
||||
edges = 'induced'
|
||||
edges = 'induced',
|
||||
enumeration = 'index',
|
||||
reportIndexDrift = false
|
||||
} = options
|
||||
|
||||
// 1. Resolve the node-id set.
|
||||
const idSet = await resolveSelector(reader, selector, includeSystem)
|
||||
if (enumeration === 'canonical' && !storage) {
|
||||
throw new Error(
|
||||
`export(): enumeration:'canonical' requires a storage adapter, but none was supplied ` +
|
||||
`to this reader. Use enumeration:'index' (the default), or export through a Db/Brainy ` +
|
||||
`that carries its storage adapter.`
|
||||
)
|
||||
}
|
||||
const wantDrift = enumeration === 'canonical' && reportIndexDrift
|
||||
|
||||
// 1. Resolve the node-id set (+ the index's raw candidate set, only when diffing it).
|
||||
const { idSet, indexCandidateIds } = await resolveSelector(
|
||||
reader,
|
||||
storage,
|
||||
selector,
|
||||
includeSystem,
|
||||
enumeration,
|
||||
wantDrift
|
||||
)
|
||||
|
||||
// 2. Read canonical entities (reserved fields top-level), applying any predicate filter.
|
||||
// Identical for both enumeration modes: 'canonical' only changes WHICH ids reach this
|
||||
// loop, never how a candidate is verified — so the two modes can disagree on candidacy,
|
||||
// never on what counts as a match.
|
||||
const usePredicate = hasPredicate(selector)
|
||||
const entityMap = new Map<string, Entity<T>>()
|
||||
const entities: PortableGraphEntity[] = []
|
||||
|
|
@ -307,8 +399,31 @@ export async function exportGraph<T = any>(
|
|||
}
|
||||
const keptIds = new Set(entityMap.keys())
|
||||
|
||||
// 3. Edges per policy.
|
||||
const { relations, danglingIds } = await collectEdges(reader, keptIds, edges)
|
||||
// 2b. Finalize the drift report now that ground truth (keptIds) is known.
|
||||
let drift: ExportIndexDrift | undefined
|
||||
if (wantDrift) {
|
||||
const indexIds = indexCandidateIds ?? new Set<string>()
|
||||
const canonicalOnly = [...keptIds].filter((id) => !indexIds.has(id))
|
||||
const indexOnly = [...indexIds].filter((id) => !keptIds.has(id))
|
||||
drift = { canonicalOnly, indexOnly }
|
||||
if (canonicalOnly.length > 0 || indexOnly.length > 0) {
|
||||
console.warn(
|
||||
`[Brainy] export() index drift: ${canonicalOnly.length} canonical-only id(s) ` +
|
||||
`(canon-present, the index-based enumeration missed them) and ${indexOnly.length} ` +
|
||||
`index-only id(s) (index-visible, canon-absent — phantom rows). ` +
|
||||
`See the returned PortableGraph's 'drift' field for the exact ids. Nothing was ` +
|
||||
`auto-healed — run brain.repairIndex() to reconcile the metadata index.`
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// 3. Edges per policy. Canonical mode ALSO walks verbs canonically for every
|
||||
// selector (not just whole-brain) — a lost adjacency-index posting can hide a
|
||||
// relation regardless of how the node set was produced.
|
||||
const { relations, danglingIds } =
|
||||
enumeration === 'canonical'
|
||||
? await collectEdgesCanonical(storage!, keptIds, edges)
|
||||
: await collectEdges(reader, keptIds, edges)
|
||||
|
||||
// 4. VFS blob bytes (only when requested).
|
||||
let blobs: Record<string, string> | undefined
|
||||
|
|
@ -330,6 +445,7 @@ export async function exportGraph<T = any>(
|
|||
relations,
|
||||
...(blobs && blobCount > 0 ? { blobs } : {}),
|
||||
...(danglingIds && danglingIds.length > 0 ? { danglingIds } : {}),
|
||||
...(drift ? { drift } : {}),
|
||||
stats: {
|
||||
entityCount: entities.length,
|
||||
relationCount: relations.length,
|
||||
|
|
@ -477,12 +593,30 @@ function hasPredicate(s: ExportSelector): boolean {
|
|||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* @param reader - Generation-correct read surface.
|
||||
* @param storage - Storage adapter (only touched when `enumeration:'canonical'`
|
||||
* resolves the whole-brain/predicate branch).
|
||||
* @param s - The export selector.
|
||||
* @param includeSystem - Whether `visibility:'system'` entities are wanted.
|
||||
* @param enumeration - `'index'` (default) or `'canonical'` — see {@link ExportOptions.enumeration}.
|
||||
* Only affects the whole-brain/predicate branch (the `else` below): structural
|
||||
* selectors (`ids`/`collection`/`connected`/`vfsPath`) never rode the metadata
|
||||
* index for their node set, so they resolve identically either way.
|
||||
* @param wantIndexCandidates - When true (only meaningful with `enumeration:'canonical'`
|
||||
* on the whole-brain/predicate branch), ALSO run the index-based walk and return
|
||||
* its raw candidate set as `indexCandidateIds`, for {@link ExportIndexDrift}.
|
||||
*/
|
||||
async function resolveSelector<T>(
|
||||
reader: PortableGraphReader<T>,
|
||||
storage: StorageAdapter | undefined,
|
||||
s: ExportSelector,
|
||||
includeSystem: boolean
|
||||
): Promise<Set<string>> {
|
||||
includeSystem: boolean,
|
||||
enumeration: 'index' | 'canonical',
|
||||
wantIndexCandidates: boolean
|
||||
): Promise<{ idSet: Set<string>; indexCandidateIds?: Set<string> }> {
|
||||
let idSet: Set<string>
|
||||
let indexCandidateIds: Set<string> | undefined
|
||||
if (s.ids && s.ids.length) {
|
||||
idSet = new Set(s.ids)
|
||||
} else if (s.collection ?? s.memberOf) {
|
||||
|
|
@ -491,15 +625,23 @@ async function resolveSelector<T>(
|
|||
idSet = await resolveConnected(reader, s.connected)
|
||||
} else if (s.vfsPath) {
|
||||
idSet = await resolveVfsPath(reader, s.vfsPath, s.recursive ?? true, s.depth)
|
||||
} else if (enumeration === 'canonical') {
|
||||
idSet = await enumerateAllCanonical(storage!)
|
||||
if (wantIndexCandidates) indexCandidateIds = await enumerateAllIndexed(reader, s)
|
||||
} else {
|
||||
idSet = await enumerateAll(reader, s)
|
||||
idSet = await enumerateAllIndexed(reader, s)
|
||||
}
|
||||
if (!includeSystem) idSet.delete(VFS_ROOT_ID)
|
||||
return idSet
|
||||
return { idSet, indexCandidateIds }
|
||||
}
|
||||
|
||||
/** Whole-brain / predicate enumeration via generation-correct paginated `find()`. */
|
||||
async function enumerateAll<T>(reader: PortableGraphReader<T>, s: ExportSelector): Promise<Set<string>> {
|
||||
/**
|
||||
* @description Whole-brain / predicate enumeration via generation-correct
|
||||
* paginated `find()`. The metadata index is an acceleration structure over
|
||||
* this candidate set — see {@link enumerateAllCanonical} for the storage-level
|
||||
* counterpart that never consults it.
|
||||
*/
|
||||
async function enumerateAllIndexed<T>(reader: PortableGraphReader<T>, s: ExportSelector): Promise<Set<string>> {
|
||||
const params: any = {}
|
||||
if (s.type !== undefined) params.type = s.type
|
||||
if (s.subtype !== undefined) params.subtype = s.subtype
|
||||
|
|
@ -517,6 +659,46 @@ async function enumerateAll<T>(reader: PortableGraphReader<T>, s: ExportSelector
|
|||
return ids
|
||||
}
|
||||
|
||||
/**
|
||||
* @description Canonical (storage-level) counterpart of {@link enumerateAllIndexed}:
|
||||
* walks every live noun directly off the storage adapter's canonical shard layout
|
||||
* (`storage.getNouns()` — the same primitive `repairIndex()`'s recount and every
|
||||
* index-heal walk use) instead of going through the metadata index. Guarantees
|
||||
* canon-completeness — a lost or stale metadata-index posting cannot cause a
|
||||
* canonical record to be silently missing from the returned set — at the cost of
|
||||
* an O(N) walk regardless of selector selectivity (unlike the index path's
|
||||
* O(matches)). Returns the RAW candidate id set; `exportGraph`'s caller applies
|
||||
* `matchesPredicate` per-entity via `reader.get()` afterward, exactly as the index
|
||||
* path does, so both paths share one predicate-evaluation code path and can only
|
||||
* disagree on candidacy, never on what a match means.
|
||||
*
|
||||
* Mirrors `find()`'s default hidden-tier policy (always hides `'internal'` and
|
||||
* `'system'` here — `enumerateAllIndexed` never opts either back in via `find()`
|
||||
* either, since `ExportOptions.includeSystem` is applied later, per-entity, and
|
||||
* only reachable for ids a selector already named directly) so the two
|
||||
* enumeration modes produce identical id sets when the index is healthy.
|
||||
*/
|
||||
async function enumerateAllCanonical(storage: StorageAdapter): Promise<Set<string>> {
|
||||
const ids = new Set<string>()
|
||||
let offset = 0
|
||||
let cursor: string | undefined
|
||||
// eslint-disable-next-line no-constant-condition
|
||||
while (true) {
|
||||
const page = await storage.getNouns({ pagination: { limit: ENUM_PAGE, offset, cursor } })
|
||||
for (const item of page.items) {
|
||||
if (item.visibility === 'internal' || item.visibility === 'system') continue
|
||||
ids.add(item.id)
|
||||
}
|
||||
if (!page.hasMore || page.items.length === 0) break
|
||||
if (page.nextCursor !== undefined) {
|
||||
cursor = page.nextCursor
|
||||
} else {
|
||||
offset += ENUM_PAGE
|
||||
}
|
||||
}
|
||||
return ids
|
||||
}
|
||||
|
||||
async function resolveCollectionSubtree<T>(
|
||||
reader: PortableGraphReader<T>,
|
||||
rootId: string,
|
||||
|
|
@ -718,6 +900,74 @@ async function collectEdges<T>(
|
|||
return dangling.size > 0 ? { relations, danglingIds: Array.from(dangling) } : { relations }
|
||||
}
|
||||
|
||||
/** Converts a canonical verb record (as returned by `storage.getVerbs()`) into the wire shape. */
|
||||
function hnswVerbToPortableGraphRelation(v: HNSWVerbWithMetadata): PortableGraphRelation {
|
||||
const br: PortableGraphRelation = { id: v.id, from: v.sourceId, to: v.targetId, type: v.verb as string }
|
||||
if (v.subtype !== undefined) br.subtype = v.subtype
|
||||
if (v.visibility !== undefined && v.visibility !== 'public') br.visibility = v.visibility
|
||||
if (v.weight !== undefined) br.weight = v.weight
|
||||
if (v.confidence !== undefined) br.confidence = v.confidence
|
||||
if (v.metadata && Object.keys(v.metadata as any).length) br.metadata = v.metadata
|
||||
return br
|
||||
}
|
||||
|
||||
/**
|
||||
* @description Canonical (storage-level) counterpart of {@link collectEdges}:
|
||||
* walks every live verb directly off `storage.getVerbs()` — the same primitive
|
||||
* `repairIndex()`'s recount and every index-heal walk use — instead of the graph
|
||||
* adjacency index (`reader.related()`), so a lost/stale adjacency posting cannot
|
||||
* cause a canonical relationship to be silently dropped from the export. Used for
|
||||
* EVERY selector in `enumeration:'canonical'` mode, not just the whole-brain
|
||||
* branch: relations can be blinded by adjacency-index corruption regardless of
|
||||
* how `idSet` (the kept node ids) was produced.
|
||||
*
|
||||
* Mirrors `related()`'s default hidden-tier policy (always hides `'internal'`
|
||||
* and `'system'` — `collectEdges` never opts either back in via `related()`
|
||||
* either) so the two enumeration modes produce identical relation sets when the
|
||||
* index is healthy.
|
||||
*/
|
||||
async function collectEdgesCanonical(
|
||||
storage: StorageAdapter,
|
||||
idSet: Set<string>,
|
||||
edges: 'induced' | 'incident' | 'none'
|
||||
): Promise<{ relations: PortableGraphRelation[]; danglingIds?: string[] }> {
|
||||
if (edges === 'none') return { relations: [] }
|
||||
|
||||
const relations: PortableGraphRelation[] = []
|
||||
const dangling = new Set<string>()
|
||||
const seen = new Set<string>()
|
||||
let offset = 0
|
||||
let cursor: string | undefined
|
||||
|
||||
// eslint-disable-next-line no-constant-condition
|
||||
while (true) {
|
||||
const page = await storage.getVerbs({ pagination: { limit: ENUM_PAGE, offset, cursor } })
|
||||
for (const v of page.items) {
|
||||
if (seen.has(v.id)) continue
|
||||
if (v.visibility === 'internal' || v.visibility === 'system') continue
|
||||
const fromIn = idSet.has(v.sourceId)
|
||||
const toIn = idSet.has(v.targetId)
|
||||
if (edges === 'induced') {
|
||||
if (!fromIn || !toIn) continue
|
||||
} else if (!fromIn && !toIn) {
|
||||
continue // 'incident': neither endpoint kept — irrelevant to this export
|
||||
}
|
||||
if (fromIn && !toIn) dangling.add(v.targetId)
|
||||
if (toIn && !fromIn) dangling.add(v.sourceId)
|
||||
seen.add(v.id)
|
||||
relations.push(hnswVerbToPortableGraphRelation(v))
|
||||
}
|
||||
if (!page.hasMore || page.items.length === 0) break
|
||||
if (page.nextCursor !== undefined) {
|
||||
cursor = page.nextCursor
|
||||
} else {
|
||||
offset += ENUM_PAGE
|
||||
}
|
||||
}
|
||||
|
||||
return dangling.size > 0 ? { relations, danglingIds: Array.from(dangling) } : { relations }
|
||||
}
|
||||
|
||||
async function collectBlobs<T>(
|
||||
storage: StorageAdapter | undefined,
|
||||
entityMap: Map<string, Entity<T>>
|
||||
|
|
|
|||
|
|
@ -185,6 +185,7 @@ export type {
|
|||
PortableGraphRelation,
|
||||
ExportSelector,
|
||||
ExportOptions,
|
||||
ExportIndexDrift,
|
||||
ImportOptions,
|
||||
ImportResult,
|
||||
PortableGraphValidation
|
||||
|
|
@ -194,7 +195,8 @@ export {
|
|||
SpeculativeOverlayError,
|
||||
GenerationCompactedError,
|
||||
StoreInconsistentError,
|
||||
PendingFlushDurabilityError
|
||||
PendingFlushDurabilityError,
|
||||
CanonicalEnumerationUnavailableError
|
||||
} from './db/errors.js'
|
||||
export type { UnreconciledRecord } from './db/errors.js'
|
||||
export type {
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@
|
|||
* subtype-required default.
|
||||
*/
|
||||
|
||||
import { describe, it, expect, beforeEach, afterEach } from 'vitest'
|
||||
import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import * as fs from 'node:fs/promises'
|
||||
import * as os from 'node:os'
|
||||
|
|
@ -19,6 +19,7 @@ import { createTestConfig } from '../../helpers/test-factory'
|
|||
import { NounType, VerbType } from '../../../src/types/graphTypes'
|
||||
import { validatePortableGraph } from '../../../src/db/portableGraph'
|
||||
import type { PortableGraph } from '../../../src/db/portableGraph'
|
||||
import { CanonicalEnumerationUnavailableError } from '../../../src/db/errors'
|
||||
|
||||
describe('8.0 portable graph export/import (PortableGraph v1)', () => {
|
||||
let brain: Brainy
|
||||
|
|
@ -293,3 +294,162 @@ describe('8.0 export includeContent (VFS blobs, filesystem)', () => {
|
|||
}
|
||||
})
|
||||
})
|
||||
|
||||
describe('8.0 export enumeration:"canonical" — canon-complete against index blindness', () => {
|
||||
let brain: Brainy
|
||||
|
||||
beforeEach(async () => {
|
||||
brain = new Brainy(createTestConfig())
|
||||
await brain.init()
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
await brain.close()
|
||||
})
|
||||
|
||||
it('(i) equals the index-based export when the index is healthy — same entity ids, relations, vectors', async () => {
|
||||
const a = await brain.add({ data: 'Alice', type: NounType.Person, subtype: 'employee' })
|
||||
const b = await brain.add({ data: 'Bob', type: NounType.Person, subtype: 'employee' })
|
||||
const c = await brain.add({ data: 'Acme', type: NounType.Organization, subtype: 'vendor' })
|
||||
await brain.relate({ from: a, to: b, type: VerbType.FriendOf, subtype: 'close' })
|
||||
await brain.relate({ from: a, to: c, type: VerbType.WorksWith, subtype: 'full-time' })
|
||||
|
||||
const indexExport = await brain.export({}, { includeVectors: true, enumeration: 'index' })
|
||||
const canonicalExport = await brain.export({}, { includeVectors: true, enumeration: 'canonical' })
|
||||
|
||||
expect(canonicalExport.entities.map((e) => e.id).sort()).toEqual(
|
||||
indexExport.entities.map((e) => e.id).sort()
|
||||
)
|
||||
expect(canonicalExport.relations.map((r) => r.id).sort()).toEqual(
|
||||
indexExport.relations.map((r) => r.id).sort()
|
||||
)
|
||||
expect(canonicalExport.entities.map((e) => e.id).sort()).toEqual([a, b, c].sort())
|
||||
for (const e of canonicalExport.entities) {
|
||||
expect(e.vector?.length).toBeGreaterThan(0)
|
||||
}
|
||||
expect(canonicalExport.drift).toBeUndefined() // reportIndexDrift not requested
|
||||
})
|
||||
|
||||
it('(ii) survives simulated metadata-index blindness; the index export misses the record; drift names it canonicalOnly', async () => {
|
||||
const staff = await brain.add({
|
||||
data: 'Staff',
|
||||
type: NounType.Person,
|
||||
subtype: 'employee',
|
||||
metadata: { role: 'staff' }
|
||||
})
|
||||
const other = await brain.add({
|
||||
data: 'Other',
|
||||
type: NounType.Person,
|
||||
subtype: 'employee',
|
||||
metadata: { role: 'staff' }
|
||||
})
|
||||
|
||||
// Surgically poison the metadata index (the lowest-level seam the existing
|
||||
// find() phantom-row guard tests use — see find-index-integrity-guard.test.ts,
|
||||
// which does the mirror-image ADD case) so the predicate query
|
||||
// enumeration:'index' issues (find({ type: Person })) never returns `staff` —
|
||||
// a real canonical record the index has lost track of, the exact
|
||||
// canon-present/index-missing state canonical mode exists to survive.
|
||||
const mi = (brain as any).metadataIndex
|
||||
const original = mi.getIdsForFilter.bind(mi)
|
||||
mi.getIdsForFilter = async (filter: any, opts?: any): Promise<string[]> => {
|
||||
const ids: string[] = await original(filter, opts)
|
||||
return ids.filter((id: string) => id !== staff)
|
||||
}
|
||||
|
||||
try {
|
||||
const indexExport = await brain.export({ type: NounType.Person }, { enumeration: 'index' })
|
||||
expect(indexExport.entities.map((e) => e.id)).not.toContain(staff)
|
||||
expect(indexExport.entities.map((e) => e.id)).toContain(other)
|
||||
|
||||
const canonicalExport = await brain.export(
|
||||
{ type: NounType.Person },
|
||||
{ enumeration: 'canonical', reportIndexDrift: true }
|
||||
)
|
||||
expect(canonicalExport.entities.map((e) => e.id)).toContain(staff)
|
||||
expect(canonicalExport.entities.map((e) => e.id)).toContain(other)
|
||||
expect(canonicalExport.drift?.canonicalOnly).toEqual([staff])
|
||||
expect(canonicalExport.drift?.indexOnly).toEqual([])
|
||||
} finally {
|
||||
mi.getIdsForFilter = original
|
||||
}
|
||||
})
|
||||
|
||||
it('(iii) drift report shape + loud console.warn only when nonzero', async () => {
|
||||
const staff = await brain.add({ data: 'Staff', type: NounType.Person, subtype: 'employee' })
|
||||
await brain.add({ data: 'Other', type: NounType.Person, subtype: 'employee' })
|
||||
|
||||
const mi = (brain as any).metadataIndex
|
||||
const original = mi.getIdsForFilter.bind(mi)
|
||||
mi.getIdsForFilter = async (filter: any, opts?: any): Promise<string[]> => {
|
||||
const ids: string[] = await original(filter, opts)
|
||||
return ids.filter((id: string) => id !== staff)
|
||||
}
|
||||
|
||||
const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {})
|
||||
try {
|
||||
const drifted = await brain.export(
|
||||
{ type: NounType.Person },
|
||||
{ enumeration: 'canonical', reportIndexDrift: true }
|
||||
)
|
||||
expect(drifted.drift).toEqual({ canonicalOnly: [staff], indexOnly: [] })
|
||||
expect(warnSpy).toHaveBeenCalledTimes(1)
|
||||
expect(warnSpy.mock.calls[0].join(' ')).toMatch(/drift/i)
|
||||
} finally {
|
||||
mi.getIdsForFilter = original
|
||||
warnSpy.mockClear()
|
||||
}
|
||||
|
||||
// Healthy index: drift is reported (both lists present) but never warned about.
|
||||
try {
|
||||
const healthy = await brain.export(
|
||||
{ type: NounType.Person },
|
||||
{ enumeration: 'canonical', reportIndexDrift: true }
|
||||
)
|
||||
expect(healthy.drift).toEqual({ canonicalOnly: [], indexOnly: [] })
|
||||
expect(warnSpy).not.toHaveBeenCalled()
|
||||
} finally {
|
||||
warnSpy.mockRestore()
|
||||
}
|
||||
})
|
||||
|
||||
it('(iv) throws CanonicalEnumerationUnavailableError on a historical asOf() view and a speculative with() overlay', async () => {
|
||||
const a = '22222222-2222-4222-8222-222222222222'
|
||||
const b = '33333333-3333-4333-8333-333333333333'
|
||||
await brain.transact([{ op: 'add', id: a, data: 'First', type: NounType.Thing, subtype: 'x' }])
|
||||
const g1 = brain.generation()
|
||||
await brain.transact([{ op: 'add', id: b, data: 'Second', type: NounType.Thing, subtype: 'x' }])
|
||||
|
||||
const past = await brain.asOf(g1)
|
||||
try {
|
||||
await expect(past.export({}, { enumeration: 'canonical' })).rejects.toThrow(
|
||||
CanonicalEnumerationUnavailableError
|
||||
)
|
||||
// The default (index) mode is unaffected — still a valid time-travel export.
|
||||
const backup = await past.export()
|
||||
expect(backup.entities.map((e) => e.id)).toContain(a)
|
||||
} finally {
|
||||
await past.release()
|
||||
}
|
||||
|
||||
const speculativeId = '11111111-1111-4111-8111-111111111111'
|
||||
const view = await brain.now().with([
|
||||
{ op: 'add', id: speculativeId, data: 'Speculative', type: NounType.Thing, subtype: 'x' }
|
||||
])
|
||||
try {
|
||||
await expect(view.export({}, { enumeration: 'canonical' })).rejects.toThrow(
|
||||
CanonicalEnumerationUnavailableError
|
||||
)
|
||||
} finally {
|
||||
await view.release()
|
||||
}
|
||||
})
|
||||
|
||||
it('throws a plain Error when enumeration:"canonical" has no storage adapter to walk', async () => {
|
||||
const { exportGraph } = await import('../../../src/db/portableGraph')
|
||||
const readerOnly = { get: async () => null, find: async () => [], related: async () => [] }
|
||||
await expect(
|
||||
exportGraph(readerOnly as any, undefined, {}, { enumeration: 'canonical' })
|
||||
).rejects.toThrow(/enumeration:'canonical' requires a storage adapter/)
|
||||
})
|
||||
})
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue