.json; this script is the one door
- * that composes an entry and lands it in the shared repo, so it is never
- * hand-written and never forked across repos again.
- *
- * Two modes:
- *
- * 1. Generate + publish (default):
- * node wall-entry.mjs --product --version --date \
- * --from-changelog
- * Derives an entry from the CHANGELOG.md entry for (headline = the
- * entry's first bullet, items = every bullet, trimmed of its trailing
- * commit hash), then:
- * - clones (or, if a cached clone already exists, fetches and resets)
- * the releases repo into a local cache directory,
- * - prepends the entry to /.json, newest first — replacing
- * any existing entry for the same version so a re-run is idempotent,
- * - validates the file's shape before and after,
- * - commits the change as "chore(wall):
" and pushes main.
- * A failure at any step (clone, validation, commit, push, a
- * non-fast-forward remote) exits non-zero naming the cure. Nothing is
- * ever skipped — the wall either lands correctly or the release fails.
- *
- * 2. Dry run:
- * node wall-entry.mjs --dry-run --product --version \
- * --date --from-changelog
- * Derives the entry exactly as above and prints it, along with the file
- * it would be written to, but touches no clone and no remote — usable
- * from a fresh checkout with no cache and no network.
- *
- * 3. Validate only (--check):
- * node wall-entry.mjs --check --file
- * Validates an arbitrary wall file's exact key set (top-level and
- * per-entry), field types, and strict-descending semver ordering with
- * no duplicates. Read-only; never writes. Exit 0 = clean, exit 1 =
- * named violations printed to stderr.
- *
- * The remote and the local cache directory are each overridable
- * (--remote / --cache-dir, or WALL_ENTRY_RELEASES_REMOTE /
- * WALL_ENTRY_RELEASES_CACHE_DIR) so tests can point at a throwaway local
- * bare repo and a throwaway cache directory — never the real remote or the
- * real developer cache.
- *
- * No dependencies beyond the system `git` binary — CHANGELOG parsing,
- * semver comparison, and JSON shape checking are all hand-rolled below.
- */
-
-import { readFileSync, writeFileSync, existsSync, mkdirSync } from 'node:fs'
-import { execFileSync } from 'node:child_process'
-import { homedir } from 'node:os'
-import { dirname, join } from 'node:path'
-
-const DEFAULT_REMOTE = 'git@source.soulcraft.com:soulcraftlabs/releases.git'
-
-/** @returns {string} */
-function defaultCacheDir() {
- const base = process.env.XDG_CACHE_HOME || join(homedir(), '.cache')
- return join(base, 'soulcraft-releases')
-}
-
-// Required on every entry; "thumb" is optional (may be absent, or present as
-// string | null) — matching the HQ contract's {..., thumb?}.
-const ENTRY_REQUIRED_KEYS = ['version', 'date', 'headline', 'items', 'url']
-const ENTRY_OPTIONAL_KEYS = ['thumb']
-const ENTRY_ALLOWED_KEYS = [...ENTRY_REQUIRED_KEYS, ...ENTRY_OPTIONAL_KEYS]
-const FILE_KEYS = ['product', 'entries']
-
-// The public permalink pattern, by product. Every entry MUST carry an https
-// permalink: HQ's parser rejects a wall whose entries carry url: null (the
-// whole feed became unreadable on 2026-09-02). A product whose forge repo is
-// private links its PUBLIC package page on The Source instead of a release
-// page that would 404 for HQ's readers.
-const RELEASE_URL_PATTERNS = {
- 'open-brainy': (version) => `https://source.soulcraft.com/soulcraftlabs/open-brainy/releases/tag/v${version}`,
- 'brainy': (version) => `https://source.soulcraft.com/soulcraft/-/packages/npm/@soulcraft%2Fbrainy/${version}`,
-}
-
-/**
- * Parse argv into a flag map. `--flag value` sets a string; `--flag` alone
- * (end of argv, or followed by another `--flag`) sets boolean true.
- * @param {string[]} argv
- * @returns {Record}
- */
-function parseArgs(argv) {
- /** @type {Record} */
- const args = {}
- for (let i = 0; i < argv.length; i++) {
- const a = argv[i]
- if (!a.startsWith('--')) continue
- const key = a.slice(2)
- const next = argv[i + 1]
- if (next === undefined || next.startsWith('--')) {
- args[key] = true
- } else {
- args[key] = next
- i++
- }
- }
- return args
-}
-
-/**
- * Print a loud, named error and exit 1. Every refusal in this script goes
- * through here so the failure mode is always the same shape: "wall-entry: ".
- * @param {string} message
- * @returns {never}
- */
-function fail(message) {
- console.error(`wall-entry: ${message}`)
- process.exit(1)
-}
-
-/**
- * @param {string} version
- * @returns {{major: number, minor: number, patch: number, pre: string | null} | null}
- */
-function parseSemver(version) {
- const m = /^(\d+)\.(\d+)\.(\d+)(?:-([0-9A-Za-z.-]+))?$/.exec(version)
- if (!m) return null
- return { major: Number(m[1]), minor: Number(m[2]), patch: Number(m[3]), pre: m[4] ?? null }
-}
-
-/**
- * @param {string} a
- * @param {string} b
- * @returns {number} positive if a > b, negative if a < b, 0 if equal.
- */
-function compareSemver(a, b) {
- const pa = parseSemver(a)
- const pb = parseSemver(b)
- if (!pa || !pb) throw new Error(`cannot compare non-semver versions "${a}" vs "${b}"`)
- if (pa.major !== pb.major) return pa.major - pb.major
- if (pa.minor !== pb.minor) return pa.minor - pb.minor
- if (pa.patch !== pb.patch) return pa.patch - pb.patch
- if (pa.pre === pb.pre) return 0
- if (pa.pre === null) return 1 // a release outranks any prerelease of the same core version
- if (pb.pre === null) return -1
- return pa.pre < pb.pre ? -1 : pa.pre > pb.pre ? 1 : 0
-}
-
-/**
- * Validate a wall file's full shape: top-level keys ("product", "entries" —
- * no more, no less), per-entry keys and field types ("thumb" optional), and
- * strict-descending semver ordering with no duplicates. Collects every
- * violation instead of failing on the first, so a caller reports the whole
- * picture in one pass.
- * @param {unknown} data
- * @returns {string[]} Violation messages; empty means the file is clean.
- */
-function validateShape(data) {
- /** @type {string[]} */
- const errors = []
-
- if (typeof data !== 'object' || data === null || Array.isArray(data)) {
- return ['top level: expected a JSON object']
- }
- const obj = /** @type {Record} */ (data)
-
- const topKeys = Object.keys(obj)
- const missingTop = FILE_KEYS.filter((k) => !(k in obj))
- const extraTop = topKeys.filter((k) => !FILE_KEYS.includes(k))
- if (missingTop.length) errors.push(`top level: missing key(s) ${missingTop.join(', ')}`)
- if (extraTop.length) errors.push(`top level: unexpected key(s) ${extraTop.join(', ')}`)
-
- if (typeof obj.product !== 'string' || obj.product.trim() === '') {
- errors.push('top level: "product" must be a non-empty string')
- }
- if (!Array.isArray(obj.entries)) {
- errors.push('top level: "entries" must be an array')
- return errors // nothing further to check without an array
- }
-
- const entries = /** @type {unknown[]} */ (obj.entries)
- entries.forEach((rawEntry, i) => {
- const label = `entries[${i}]`
- if (typeof rawEntry !== 'object' || rawEntry === null || Array.isArray(rawEntry)) {
- errors.push(`${label}: expected an object`)
- return
- }
- const entry = /** @type {Record} */ (rawEntry)
- const keys = Object.keys(entry)
- const missing = ENTRY_REQUIRED_KEYS.filter((k) => !(k in entry))
- const extra = keys.filter((k) => !ENTRY_ALLOWED_KEYS.includes(k))
- if (missing.length) errors.push(`${label}: missing key(s) ${missing.join(', ')}`)
- if (extra.length) errors.push(`${label}: unexpected key(s) ${extra.join(', ')}`)
-
- if (typeof entry.version !== 'string' || !parseSemver(entry.version)) {
- errors.push(`${label}: "version" must be a semver string (got ${JSON.stringify(entry.version)})`)
- }
- if (typeof entry.date !== 'string' || !/^\d{4}-\d{2}-\d{2}$/.test(entry.date) || Number.isNaN(Date.parse(entry.date))) {
- errors.push(`${label}: "date" must be a YYYY-MM-DD string (got ${JSON.stringify(entry.date)})`)
- }
- if (typeof entry.headline !== 'string' || entry.headline.trim() === '') {
- errors.push(`${label}: "headline" must be a non-empty string`)
- }
- if (!Array.isArray(entry.items) || entry.items.length === 0 || entry.items.some((it) => typeof it !== 'string' || it.trim() === '')) {
- errors.push(`${label}: "items" must be a non-empty array of non-empty strings`)
- }
- if (typeof entry.url !== 'string' || !/^https:\/\/\S+$/.test(entry.url)) {
- errors.push(`${label}: "url" must be an https permalink — never null; HQ's parser rejects the whole feed`)
- }
- if ('thumb' in entry && !(entry.thumb === null || typeof entry.thumb === 'string')) {
- errors.push(`${label}: "thumb" must be a string or null when present`)
- }
- })
-
- // Ordering: newest first, strictly descending, no duplicate versions —
- // checked only over entries whose version parsed (a bad version is
- // already reported above; comparing it too would just be noise).
- const versioned = entries
- .map((e, i) => ({ i, version: /** @type {any} */ (e)?.version }))
- .filter((e) => typeof e.version === 'string' && parseSemver(e.version))
- for (let i = 0; i < versioned.length - 1; i++) {
- const a = versioned[i]
- const b = versioned[i + 1]
- const cmp = compareSemver(a.version, b.version)
- if (cmp === 0) {
- errors.push(`entries[${a.i}] and entries[${b.i}]: duplicate version ${a.version}`)
- } else if (cmp < 0) {
- errors.push(`entries[${a.i}] (${a.version}) sits above entries[${b.i}] (${b.version}) — not newest-first`)
- }
- }
-
- return errors
-}
-
-/**
- * Extract one version's entry body from a standard-version-style CHANGELOG.md
- * (headings `### [version](url) (date)`, followed by `- bullet (hash)` lines
- * until the next heading or EOF).
- * @param {string} changelog
- * @param {string} version
- * @returns {string[]} Bullet lines, trimmed of their leading "- " and
- * trailing " (hash)".
- */
-function extractChangelogBullets(changelog, version) {
- const lines = changelog.split('\n')
- const headingRe = /^### \[([^\]]+)\]\(.*\)\s*\(\d{4}-\d{2}-\d{2}\)\s*$/
- let start = -1
- for (let i = 0; i < lines.length; i++) {
- const m = headingRe.exec(lines[i])
- if (m && m[1] === version) {
- start = i + 1
- break
- }
- }
- if (start === -1) {
- fail(
- `version ${version} has no CHANGELOG entry yet — run this after the CHANGELOG step composes "### [${version}]", not before`,
- )
- }
- /** @type {string[]} */
- const bullets = []
- for (let i = start; i < lines.length; i++) {
- if (headingRe.test(lines[i])) break // next entry starts
- const bulletMatch = /^- (.+?)(?:\s\(([0-9a-f]{6,40})\))?$/.exec(lines[i].trim())
- if (lines[i].trim().startsWith('- ') && bulletMatch) {
- const text = bulletMatch[1].trim()
- if (text) bullets.push(text)
- }
- }
- if (bullets.length === 0) {
- fail(`version ${version}'s CHANGELOG entry has no bullets to derive a headline/items from`)
- }
- return bullets
-}
-
-/**
- * Derive a wall entry from a CHANGELOG.md.
- * @param {{product: string, version: string, date: string, changelogPath: string, url?: string, thumb?: string | null}} opts
- * @returns {{version: string, date: string, headline: string, items: string[], url: string, thumb: string | null}}
- */
-function deriveEntry({ product, version, date, changelogPath, url, thumb }) {
- if (!parseSemver(version)) fail(`--version "${version}" is not a semver string`)
- if (!/^\d{4}-\d{2}-\d{2}$/.test(date) || Number.isNaN(Date.parse(date))) {
- fail(`--date "${date}" is not a YYYY-MM-DD date`)
- }
- if (!existsSync(changelogPath)) fail(`--from-changelog "${changelogPath}" does not exist`)
-
- const changelog = readFileSync(changelogPath, 'utf8')
- const items = extractChangelogBullets(changelog, version)
- const headline = items[0]
-
- const pattern = RELEASE_URL_PATTERNS[product]
- if (url === undefined && pattern === undefined) {
- throw new Error(`wall-entry: no permalink pattern for product "${product}" — add one to RELEASE_URL_PATTERNS or pass --url; entries never carry url: null`)
- }
- const resolvedUrl = url !== undefined ? url : pattern(version)
- const resolvedThumb = thumb !== undefined ? thumb : null
-
- return { version, date, headline, items, url: resolvedUrl, thumb: resolvedThumb }
-}
-
-/**
- * Load and shape-validate a wall file.
- * @param {string} filePath
- * @returns {Record}
- */
-function loadWallFile(filePath) {
- if (!existsSync(filePath)) fail(`"${filePath}" does not exist`)
- /** @type {unknown} */
- let data
- try {
- data = JSON.parse(readFileSync(filePath, 'utf8'))
- } catch (err) {
- fail(`"${filePath}" is not valid JSON: ${/** @type {Error} */ (err).message}`)
- }
- const errors = validateShape(data)
- if (errors.length) {
- fail(`"${filePath}" fails shape validation —\n ${errors.join('\n ')}`)
- }
- return /** @type {Record} */ (data)
-}
-
-/**
- * Run a git command, throwing an Error whose message is git's own stderr
- * (trimmed) on failure — every caller wraps this to name the cure.
- * @param {string[]} args
- * @param {string} cwd
- * @returns {string} stdout, trimmed.
- */
-function git(args, cwd) {
- try {
- return execFileSync('git', args, { cwd, encoding: 'utf8', stdio: ['ignore', 'pipe', 'pipe'] }).trim()
- } catch (err) {
- const stderr = /** @type {any} */ (err).stderr
- const message = (typeof stderr === 'string' && stderr.trim()) || /** @type {Error} */ (err).message
- throw new Error(message)
- }
-}
-
-/**
- * Ensure a clean, up-to-date local clone of the releases repo at
- * `cacheDir`, checked out on `main` — cloning fresh if `cacheDir` has no
- * `.git`, otherwise fetching and hard-resetting onto `origin/main` (so a
- * stray local commit or edit left by a previous failed run can never leak
- * into the next one).
- * @param {string} remote
- * @param {string} cacheDir
- */
-function ensureReleasesClone(remote, cacheDir) {
- if (existsSync(join(cacheDir, '.git'))) {
- try {
- git(['remote', 'set-url', 'origin', remote], cacheDir)
- git(['fetch', '--prune', 'origin'], cacheDir)
- git(['checkout', 'main'], cacheDir)
- git(['reset', '--hard', 'origin/main'], cacheDir)
- git(['clean', '-fd'], cacheDir)
- } catch (err) {
- fail(
- `cannot refresh the cached releases checkout at "${cacheDir}" from "${remote}" — ${/** @type {Error} */ (err).message}\n` +
- ` cure: delete "${cacheDir}" and re-run so it re-clones from scratch, or confirm SSH access with "ssh -T git@source.soulcraft.com"`,
- )
- }
- return
- }
-
- mkdirSync(dirname(cacheDir), { recursive: true })
- try {
- git(['clone', remote, cacheDir], dirname(cacheDir))
- } catch (err) {
- fail(
- `cannot clone "${remote}" — ${/** @type {Error} */ (err).message}\n` +
- ` cure: confirm SSH access with "ssh -T git@source.soulcraft.com" and that the soulcraftlabs/releases repo exists yet`,
- )
- }
- try {
- git(['checkout', 'main'], cacheDir)
- } catch (err) {
- fail(
- `cloned "${remote}" into "${cacheDir}" but could not check out "main" — ${/** @type {Error} */ (err).message}\n` +
- ` cure: confirm the releases repo's default branch is named "main"`,
- )
- }
-}
-
-/**
- * Prepend `entry` to the wall at `/.json`, replacing any
- * existing entry for the same version (idempotent re-runs), validating
- * before and after, committing, and pushing — or refusing loudly, naming
- * the cure, at whichever step fails.
- * @param {{version: string, date: string, headline: string, items: string[], url: string, thumb: string | null}} entry
- * @param {string} product
- * @param {string} remote
- * @param {string} cacheDir
- */
-function publishEntry(entry, product, remote, cacheDir) {
- ensureReleasesClone(remote, cacheDir)
-
- const filePath = join(cacheDir, `${product}.json`)
- if (!existsSync(filePath)) {
- fail(
- `"${filePath}" does not exist in the releases repo — cure: seed "${product}.json" at the repo root first (it must exist before any release rail can prepend to it)`,
- )
- }
- const wall = loadWallFile(filePath)
-
- if (wall.product !== product) {
- fail(`"${filePath}" has product "${wall.product}", but --product "${product}" was given — refusing a cross-product write`)
- }
-
- const replacing = wall.entries.some((e) => e.version === entry.version)
- wall.entries = [entry, ...wall.entries.filter((e) => e.version !== entry.version)]
-
- const postErrors = validateShape(wall)
- if (postErrors.length) {
- fail(`the entry for ${entry.version} would leave "${filePath}" invalid —\n ${postErrors.join('\n ')}`)
- }
-
- writeFileSync(filePath, JSON.stringify(wall, null, 2) + '\n', 'utf8')
-
- const status = git(['status', '--porcelain', '--', `${product}.json`], cacheDir)
- if (status === '') {
- console.log(`wall-entry: "${product}.json" already carries an identical entry for ${entry.version} — nothing to commit or push`)
- return
- }
-
- try {
- git(['add', `${product}.json`], cacheDir)
- git(['commit', '-m', `chore(wall): ${product} ${entry.version}`], cacheDir)
- } catch (err) {
- fail(`cannot commit the wall entry in "${cacheDir}" — ${/** @type {Error} */ (err).message}\n cure: inspect "${cacheDir}" by hand and re-run once its git state is clean`)
- }
-
- try {
- git(['push', 'origin', 'main'], cacheDir)
- } catch (err) {
- fail(
- `push to "${remote}" failed (likely a non-fast-forward — another release landed on main first) — ${/** @type {Error} */ (err).message}\n` +
- ` cure: re-run this release step; it re-fetches and resets onto the latest origin/main before retrying`,
- )
- }
-
- const sha = git(['rev-parse', 'HEAD'], cacheDir)
- console.log(
- `wall-entry: ${replacing ? 'replaced' : 'wrote'} v${entry.version} in "${product}.json" (${wall.entries.length} entries, newest first) — pushed ${sha} to ${remote} main`,
- )
-}
-
-function main() {
- const args = parseArgs(process.argv.slice(2))
-
- if (args.check) {
- const filePath = /** @type {string | undefined} */ (args.file)
- if (!filePath) fail('--check needs --file ')
- const wall = loadWallFile(/** @type {string} */ (filePath))
- console.log(`wall-entry --check: "${filePath}" OK — product "${wall.product}", ${wall.entries.length} entries, newest-first, no duplicates`)
- process.exit(0)
- }
-
- // Generate mode (default, also covers --dry-run): --product, --version,
- // --date, --from-changelog required.
- const product = /** @type {string | undefined} */ (args.product)
- const version = /** @type {string | undefined} */ (args.version)
- const date = /** @type {string | undefined} */ (args.date)
- const fromChangelog = /** @type {string | undefined} */ (args['from-changelog'])
-
- const missing = []
- if (!product) missing.push('--product')
- if (!version) missing.push('--version')
- if (!date) missing.push('--date')
- if (!fromChangelog) missing.push('--from-changelog')
- if (missing.length) {
- fail(
- `missing required flag(s): ${missing.join(', ')}\n` +
- 'Usage:\n' +
- ' wall-entry.mjs --product --version --date --from-changelog [--dry-run]\n' +
- ' wall-entry.mjs --check --file ',
- )
- }
-
- const urlArg = args.url === true ? undefined : /** @type {string | undefined} */ (args.url)
- const thumbArg = args.thumb === true ? undefined : /** @type {string | undefined} */ (args.thumb)
-
- const entry = deriveEntry({
- product: /** @type {string} */ (product),
- version: /** @type {string} */ (version),
- date: /** @type {string} */ (date),
- changelogPath: /** @type {string} */ (fromChangelog),
- url: urlArg,
- thumb: thumbArg,
- })
-
- const remote = /** @type {string} */ (args.remote ?? process.env.WALL_ENTRY_RELEASES_REMOTE ?? DEFAULT_REMOTE)
- const cacheDir = /** @type {string} */ (args['cache-dir'] ?? process.env.WALL_ENTRY_RELEASES_CACHE_DIR ?? defaultCacheDir())
-
- if (args['dry-run']) {
- console.log(`wall-entry --dry-run: would write to "${join(cacheDir, `${product}.json`)}" in ${remote} (main), pushed as "chore(wall): ${product} ${version}"`)
- console.log(JSON.stringify(entry, null, 2))
- process.exit(0)
- }
-
- publishEntry(entry, /** @type {string} */ (product), remote, cacheDir)
-}
-
-main()
diff --git a/src/aggregation/AggregationIndex.ts b/src/aggregation/AggregationIndex.ts
index d3a1fd74..ecd16228 100644
--- a/src/aggregation/AggregationIndex.ts
+++ b/src/aggregation/AggregationIndex.ts
@@ -14,22 +14,7 @@
*/
import type { StorageAdapter, HNSWNounWithMetadata } from '../coreTypes.js'
-import { parseFieldAddress, readEntityFieldAddress } from '../db/fieldAddressing.js'
-import type { HNSWNounWithMetadata as AddressedEntity } from '../coreTypes.js'
-
-/**
- * Read a user-supplied field name under the one addressing law (sealed
- * 2026-08-03): bare / `metadata.` = the user's metadata field, `system.` =
- * the ruled engine scalar, malformed = typed refusal. The aggregation engine
- * NEVER resolves names any other way — the pre-law resolver made bare
- * `subtype`/`confidence` read engine scalars, silently shadowing user fields.
- */
-function readAddressed(e: unknown, name: string): unknown {
- return readEntityFieldAddress(
- e as AddressedEntity,
- parseFieldAddress(name, 'entity')
- )
-}
+import { resolveEntityField } from '../coreTypes.js'
import type {
AggregateDefinition,
AggregateGroupState,
@@ -44,7 +29,6 @@ import { matchesMetadataFilter } from '../utils/metadataFilter.js'
import { compareCodePoints } from '../utils/collation.js'
import { bucketTimestamp } from './timeWindows.js'
import { NounType } from '../types/graphTypes.js'
-import { prodLog } from '../utils/logger.js'
/** Persistence key for aggregate definitions */
const DEFINITIONS_KEY = '__aggregation_definitions__'
@@ -103,22 +87,10 @@ function matchesSource(entity: Record, source: AggregateDefinit
if (entity.service !== source.service) return false
}
- // Where filter — resolve each filtered field through resolveEntityField,
- // the SAME single source of truth groupBy uses (top-level standard fields
- // + custom metadata). Matching only the metadata sub-object made
- // where:{subtype}/{visibility}/… a silent no-op: reserved fields never
- // live in the custom bag, so those filters could never match anything.
+ // Metadata where filter — match against the entity's metadata sub-object
if (source.where && Object.keys(source.where).length > 0) {
- const e = entity as unknown as HNSWNounWithMetadata
- for (const [key, condition] of Object.entries(source.where)) {
- // Evaluate ONE field at a time under a neutral key: the address may be
- // dotted ('system.subtype'), and the filter evaluator would otherwise
- // walk dots as a nested path instead of treating the key as an address.
- const value = readAddressed(e, key)
- if (!matchesMetadataFilter({ v: value }, { v: condition } as Record)) {
- return false
- }
- }
+ const metadata = (entity.metadata ?? entity) as Record
+ if (!matchesMetadataFilter(metadata, source.where)) return false
}
return true
@@ -148,11 +120,11 @@ function computeGroupKeys(
for (const dim of groupBy) {
if (typeof dim === 'string') {
- const val = readAddressed(e, dim)
+ const val = resolveEntityField(e, dim)
const v = val !== undefined && val !== null ? String(val) : '__null__'
for (const k of keys) k[dim] = v
} else if ('unnest' in dim) {
- const val = readAddressed(e, dim.field)
+ const val = resolveEntityField(e, dim.field)
const raw = Array.isArray(val) ? val : val !== undefined && val !== null ? [val] : []
// Distinct elements: an entity with duplicate tags counts once per distinct tag.
const elems = Array.from(new Set(raw.map(x => String(x))))
@@ -164,7 +136,7 @@ function computeGroupKeys(
keys = next
} else {
// Time-windowed field
- const val = readAddressed(e, dim.field)
+ const val = resolveEntityField(e, dim.field)
const v = typeof val === 'number' ? bucketTimestamp(val, dim.window) : '__null__'
for (const k of keys) k[dim.field] = v
}
@@ -193,7 +165,7 @@ function computeGroupKey(
* in metadata are both handled in one place.
*/
function getNumericField(entity: Record, field: string): number | undefined {
- const val = readAddressed(entity as unknown as HNSWNounWithMetadata, field)
+ const val = resolveEntityField(entity as unknown as HNSWNounWithMetadata, field)
if (typeof val === 'number' && !isNaN(val)) return val
if (typeof val === 'string') {
const num = parseFloat(val)
@@ -355,39 +327,6 @@ export class AggregationIndex {
/** Track aggregates with stale MIN/MAX (need lazy recompute) */
private staleMinMax = new Map>()
- /** Resolves when init() has finished loading persisted definitions/state. */
- private initPromise: Promise | null = null
-
- /** True once init() has settled (success or failure). */
- private initDone = false
-
- /**
- * Aggregates registered by the app before init() finished loading persisted
- * state, awaiting reconciliation: init() adopts the persisted state when the
- * definition hash matches; anything left unadopted when init settles resolves
- * to a backfill. Deciding backfill eagerly at define time was the boot-order
- * bug that wiped valid persisted state on every restart — the synchronous
- * defineAggregate() always beats the async init().
- */
- private pendingAdopt = new Set()
-
- /**
- * Aggregates adopted with a BEHIND stamp: name → the exact generation
- * window `(from, to]` whose writes the adopted state has not seen. The
- * owner (Brainy) drains this via {@link getPendingCatchUps} +
- * {@link reconcileEntity} + {@link finishCatchUp} BEFORE serving queries —
- * cost bounded by the window's affected entities, never store size.
- */
- private pendingCatchUp = new Map()
-
- /**
- * In-flight rescan targets. While a name has a staging map, ALL
- * contributions (the walk's and concurrent write hooks') land there instead
- * of the live map; the live map keeps serving until {@link finishBackfill}
- * swaps the staging map in atomically.
- */
- private backfillStaging = new Map>()
-
constructor(storage: StorageAdapter, nativeProvider?: AggregationProvider) {
this.storage = storage
this.nativeProvider = nativeProvider
@@ -397,163 +336,28 @@ export class AggregationIndex {
/**
* Initialize: load persisted definitions and state, detect changes, rebuild stale.
- *
- * Idempotent — repeated calls return the same promise. Definitions registered
- * *before* this completes (the normal boot order: `defineAggregate()` is
- * synchronous and always beats this async load) are reconciled rather than
- * clobbered: the app's definition wins, and its persisted state is adopted
- * when the definition hash matches — backfill happens only on a real change.
*/
- init(): Promise {
- if (!this.initPromise) {
- this.initPromise = this.loadPersisted().finally(() => {
- this.resolvePendingAdoptToBackfill()
- this.initDone = true
- })
- }
- return this.initPromise
- }
-
- /**
- * Await the persisted-state load (if one was started) and settle every
- * pending adoption decision. After this resolves, `getPendingBackfills()`
- * is authoritative: a name is listed iff it genuinely needs a rescan.
- * Query paths must await this before consulting backfill state.
- */
- async ready(): Promise {
- if (this.initPromise) {
- try {
- await this.initPromise
- } catch {
- // The owner already surfaced the load failure loudly; backfill covers.
- }
- }
- this.resolvePendingAdoptToBackfill()
- }
-
- /**
- * Any definition still awaiting state adoption has no persisted state to
- * adopt (or init never ran / failed) — it must backfill.
- */
- private resolvePendingAdoptToBackfill(): void {
- if (this.pendingAdopt.size > 0) {
- prodLog.info(
- `[Aggregation] no adoptable persisted state for: ${Array.from(this.pendingAdopt).join(', ')} — flagged for backfill`
- )
- }
- for (const name of this.pendingAdopt) this.needsBackfill.add(name)
- this.pendingAdopt.clear()
- }
-
- /**
- * The adoption verdict for persisted state, against the store's committed
- * watermark (SELF-ENGINE-LIFECYCLE-SPRINT ask (b) — behind-stamp is no
- * longer a whole-store rescan):
- *
- * - `'adopt'` — stamp equals the watermark (clean), or the store has no
- * watermark capability (hash-only adoption, the pre-stamp behavior).
- * - `'catchup'` — stamp is BEHIND the watermark (an unclean exit after
- * later writes, or a long-lived writer whose last flush predates recent
- * writes). The state is exact AS OF its stamp, so it is adopted and the
- * missing window `(stamp, committed]` is reconciled INCREMENTALLY per
- * affected entity via time-travel reads — bounded by writes since the
- * last flush, never by store size. The owner drains
- * {@link getPendingCatchUps} before serving queries.
- * - `'rescan'` — no stamp (pre-stamp state on a stamped store) or stamp
- * AHEAD of the watermark (e.g. a fact-log truncation on a copied store
- * pulled the watermark back): the state over-counts unverifiably; one
- * exact rescan, said out loud.
- */
- private stateAdoptionVerdict(
- name: string,
- stateData: unknown
- ): 'adopt' | 'catchup' | 'rescan' {
- const committed = this.storage.committedGeneration?.() ?? null
- if (committed === null) return 'adopt'
- const raw = (stateData as Record).sourceGeneration
- const stamped = typeof raw === 'number' ? raw : null
- if (stamped === committed) return 'adopt'
- if (stamped !== null && stamped < committed) {
- this.pendingCatchUp.set(name, { from: stamped, to: committed })
- prodLog.info(
- `[Aggregation] '${name}': persisted state is at generation ${stamped}, store is at ` +
- `${committed} — adopting and reconciling the ${committed - stamped}-generation window ` +
- `incrementally (no store rescan)`
- )
- return 'catchup'
- }
- prodLog.warn(
- `[Aggregation] '${name}': persisted state is at generation ${stamped ?? 'unstamped'} ` +
- `but the store's committed generation is ${committed} — rescanning instead of adopting`
- )
- return 'rescan'
- }
-
- private async loadPersisted(): Promise {
+ async init(): Promise {
// Load persisted definitions
const savedDefs = await this.storage.getMetadata(DEFINITIONS_KEY)
if (savedDefs && typeof savedDefs === 'object' && savedDefs.definitions) {
const defs = savedDefs.definitions as Array
for (const def of defs) {
- const savedHash = def._hash || ''
-
- if (this.definitions.has(def.name)) {
- // The app re-registered this aggregate before the load finished.
- // The app's definition wins — never clobber it with the persisted
- // copy. Adopt the persisted state when the definition is unchanged
- // AND no write has landed for it yet (a landed write would be lost
- // by adoption; the hook flips such names to backfill).
- const appHash = this.definitionHashes.get(def.name) || ''
- if (appHash === savedHash && this.pendingAdopt.has(def.name)) {
- const stateData = await this.storage.getMetadata(`${STATE_KEY_PREFIX}${def.name}__`)
- const verdict =
- stateData && stateData.groups
- ? this.stateAdoptionVerdict(def.name, stateData)
- : 'rescan'
- if (verdict !== 'rescan') {
- const groupMap = new Map()
- for (const group of stateData!.groups as AggregateGroupState[]) {
- groupMap.set(serializeGroupKey(group.groupKey), group)
- }
- this.states.set(def.name, groupMap)
- this.pendingAdopt.delete(def.name)
- this.needsBackfill.delete(def.name)
- prodLog.info(
- `[Aggregation] '${def.name}': adopted persisted state (${groupMap.size} groups) — ` +
- (verdict === 'catchup' ? 'incremental catch-up pending' : 'no rescan')
- )
- }
- // No/invalid persisted state: stays in pendingAdopt and resolves
- // to backfill when init settles.
- }
- continue
- }
-
- // Not registered this session — restore definition + state from
- // persistence.
this.definitions.set(def.name, def)
const currentHash = hashDefinition(def)
+ const savedHash = def._hash || ''
+ // Load persisted state
const stateData = await this.storage.getMetadata(`${STATE_KEY_PREFIX}${def.name}__`)
- const restoreVerdict =
- stateData && stateData.groups && savedHash === currentHash
- ? this.stateAdoptionVerdict(def.name, stateData)
- : 'rescan'
- if (restoreVerdict !== 'rescan') {
- // Definition unchanged — load state (exact as of its stamp; a
- // 'catchup' verdict reconciles the missing window incrementally).
+ if (stateData && stateData.groups && savedHash === currentHash) {
+ // Definition unchanged — load state
const groupMap = new Map()
- for (const group of stateData!.groups as AggregateGroupState[]) {
+ for (const group of stateData.groups as AggregateGroupState[]) {
const serialized = serializeGroupKey(group.groupKey)
groupMap.set(serialized, group)
}
this.states.set(def.name, groupMap)
- this.needsBackfill.delete(def.name)
- prodLog.info(
- `[Aggregation] '${def.name}': restored definition + adopted persisted state (${groupMap.size} groups)` +
- (restoreVerdict === 'catchup' ? ' — incremental catch-up pending' : '')
- )
} else {
// Definition changed or no saved state — start fresh and backfill from
// existing entities (the owner drains needsBackfill on first query).
@@ -570,35 +374,15 @@ export class AggregationIndex {
}
}
- // Restore native provider state from persistence — GATED by the same
- // adoption verdict as caller-side state (the unconditional adopt was an
- // asymmetry: a stale native blob restored over a moved store silently
- // over/under-counted). 'adopt' restores; 'catchup' restores too (the
- // incremental reconciliation drives the provider through
- // incrementalUpdate over the exact missing window); 'rescan' SKIPS the
- // blob — the flagged rebuild repopulates the provider from source.
- // Legacy unstamped envelopes verdict as rescan, loudly, never silently.
+ // Restore native provider state from persistence
if (this.nativeProvider?.restoreState) {
const nativeState = await this.storage.getMetadata('__aggregation_native_state__')
- const blob =
- nativeState && typeof nativeState === 'string'
- ? nativeState
- : nativeState && typeof nativeState === 'object' && nativeState.data
- ? (nativeState.data as string)
- : null
- if (blob !== null) {
- const verdict = this.stateAdoptionVerdict(
- '__native__',
- nativeState && typeof nativeState === 'object' ? (nativeState as Record) : {}
- )
- if (verdict === 'adopt' || verdict === 'catchup') {
- this.nativeProvider.restoreState(blob)
- } else {
- prodLog.warn(
- `[Aggregation] native provider state not adopted (verdict: ${verdict}) — ` +
- `the flagged rescan repopulates the provider from source`
- )
- }
+ if (nativeState && typeof nativeState === 'string') {
+ this.nativeProvider.restoreState(nativeState)
+ } else if (nativeState && typeof nativeState === 'object' && nativeState.data) {
+ // flush() persists `{ data: serializeState() }`, so `data` is the
+ // provider's serialized state string.
+ this.nativeProvider.restoreState(nativeState.data as string)
}
}
}
@@ -614,37 +398,24 @@ export class AggregationIndex {
}))
await this.storage.saveMetadata(DEFINITIONS_KEY, { definitions: defsToSave })
- // Persist dirty states, stamped with the committed generation they
- // reflect. The stamp is what makes reopen-adoption verifiable: state at a
- // different generation than the store's committed watermark is stale (an
- // unclean shutdown after later writes) or over-counts (a fact-log
- // truncation on a copied store pulled the watermark BACK below the
- // stamp) — either way the answer is one exact rescan, never a silent
- // adopt. Read the generation after collecting groups so any racing
- // commit resolves toward rescan, not wrong-adopt.
+ // Persist dirty states
for (const name of this.dirty) {
const stateMap = this.states.get(name)
if (stateMap) {
const groups = Array.from(stateMap.values())
- const sourceGeneration = this.storage.committedGeneration?.() ?? null
await this.storage.saveMetadata(
`${STATE_KEY_PREFIX}${name}__`,
- sourceGeneration === null ? { groups } : { groups, sourceGeneration }
+ { groups }
)
}
}
- // Persist native provider state — stamped. noteSourceGeneration lets the
- // provider bake the committed watermark into its OWN envelope before
- // serializing (so a native-side reopen can verify honesty without our
- // wrapper); the wrapper carries the same stamp for OUR adoption verdict.
+ // Persist native provider state
if (this.nativeProvider?.serializeState) {
- const nativeGen = this.storage.committedGeneration?.() ?? null
- if (nativeGen !== null) this.nativeProvider.noteSourceGeneration?.(nativeGen)
const nativeState = this.nativeProvider.serializeState()
await this.storage.saveMetadata(
'__aggregation_native_state__',
- nativeGen === null ? { data: nativeState } : { data: nativeState, sourceGeneration: nativeGen }
+ { data: nativeState }
)
}
@@ -681,19 +452,10 @@ export class AggregationIndex {
this.definitions.set(def.name, def)
this.definitionHashes.set(def.name, newHash)
- // First sight this session, before init() settled: defer the backfill
- // decision — init() adopts the persisted state on hash match, and anything
- // left unadopted resolves to backfill. Deciding eagerly here wiped valid
- // persisted state on every restart.
- if (!this.states.has(def.name) && !this.initDone) {
- this.states.set(def.name, new Map())
- this.pendingAdopt.add(def.name)
- }
// Reset state if definition changed or doesn't exist yet, and flag it for
// backfill so already-stored entities are counted (write-time hooks only see
// future writes). The owner drains this on the next query via getPendingBackfills().
- else if (!this.states.has(def.name) || (oldHash && oldHash !== newHash)) {
- this.pendingAdopt.delete(def.name)
+ if (!this.states.has(def.name) || (oldHash && oldHash !== newHash)) {
this.states.set(def.name, new Map())
this.needsBackfill.add(def.name)
}
@@ -714,8 +476,6 @@ export class AggregationIndex {
this.definitionHashes.delete(name)
this.states.delete(name)
this.staleMinMax.delete(name)
- this.pendingAdopt.delete(name)
- this.needsBackfill.delete(name)
// Notify native provider
if (this.nativeProvider?.removeAggregate) {
@@ -753,17 +513,9 @@ export class AggregationIndex {
return Array.from(this.needsBackfill)
}
- /**
- * Begin a rescan into a STAGING map. The live state is not touched — it
- * keeps serving (possibly stale, but flagged pending) until the rescan
- * completes and swaps in atomically. A mid-walk failure drops the staging
- * map via {@link abortBackfill} and loses nothing: wiping live state before
- * a scan that could throw was the destructive-before-durable defect.
- * Contributions (walk + concurrent write hooks) land in staging while it
- * exists, so the swapped-in result reflects writes that raced the walk.
- */
+ /** Clear an aggregate's state so a full rescan cannot double-count. */
beginBackfill(name: string): void {
- this.backfillStaging.set(name, new Map())
+ this.states.set(name, new Map())
// Reset native provider state for this aggregate too, if present.
const def = this.definitions.get(name)
if (def && this.nativeProvider?.removeAggregate && this.nativeProvider?.defineAggregate) {
@@ -772,15 +524,6 @@ export class AggregationIndex {
}
}
- /**
- * Abandon an in-flight rescan after a failure: drop the staging map, keep
- * the live state serving, leave the aggregate flagged as pending so a later
- * attempt rescans. The failure itself must be surfaced loudly by the owner.
- */
- abortBackfill(name: string): void {
- this.backfillStaging.delete(name)
- }
-
/** Feed one already-stored entity into a single aggregate during backfill. */
backfillEntity(name: string, entity: Record): void {
if (isAggregateEntity(entity)) return
@@ -794,146 +537,14 @@ export class AggregationIndex {
}
}
- /** Swap the rebuilt staging state in atomically; persists on next flush(). */
+ /** Mark an aggregate's backfill complete; rebuilt state persists on next flush(). */
finishBackfill(name: string): void {
- const staged = this.backfillStaging.get(name)
- if (staged) {
- this.states.set(name, staged)
- this.backfillStaging.delete(name)
- }
this.needsBackfill.delete(name)
this.dirty.add(name)
}
- // ============= Incremental Catch-Up (behind-stamp adoption) =============
-
- /** The aggregates adopted behind the watermark, with their exact missing windows. */
- getPendingCatchUps(): Array<{ name: string; from: number; to: number }> {
- return Array.from(this.pendingCatchUp, ([name, w]) => ({ name, ...w }))
- }
-
- /**
- * Reconcile ONE entity's contribution across a catch-up window using the
- * same exact delta algebra the write-time hooks use: remove the
- * contribution the adopted state counted (the entity AS OF the stamp),
- * add the contribution it should count (AS OF the window's end). `null`
- * on either side means the entity did not exist then. Composes exactly
- * with live hooks because every application is a precise old/new pair —
- * order between catch-up and post-window writes cannot drift the totals.
- */
- reconcileEntity(
- name: string,
- id: string,
- before: Record | null,
- after: Record | null
- ): void {
- const def = this.definitions.get(name)
- if (!def) return
- if (before && after) {
- if (isAggregateEntity(after)) return
- const oldMatches = matchesSource(before, def.source)
- const newMatches = matchesSource(after, def.source)
- if (this.nativeProvider && (oldMatches || newMatches)) {
- this.applyNativeResults(
- name,
- this.nativeProvider.incrementalUpdate(name, def, after, 'update', before)
- )
- return
- }
- if (oldMatches) this.removeContribution(name, def, before)
- if (newMatches) this.addContribution(name, def, after)
- return
- }
- if (after) {
- if (isAggregateEntity(after) || !matchesSource(after, def.source)) return
- if (this.nativeProvider) {
- this.applyNativeResults(name, this.nativeProvider.incrementalUpdate(name, def, after, 'add'))
- } else {
- this.addContribution(name, def, after)
- }
- return
- }
- if (before) {
- if (isAggregateEntity(before) || !matchesSource(before, def.source)) return
- if (this.nativeProvider) {
- this.applyNativeResults(name, this.nativeProvider.incrementalUpdate(name, def, before, 'delete'))
- } else {
- this.removeContribution(name, def, before)
- }
- }
- }
-
- /** Whether the native provider offers the parallel whole-rebuild path. */
- hasProviderRebuild(): boolean {
- return typeof this.nativeProvider?.rebuildAggregate === 'function'
- }
-
- /** The catch-up window for `name` is fully reconciled; state is current. */
- finishCatchUp(name: string): void {
- this.pendingCatchUp.delete(name)
- this.dirty.add(name)
- }
-
- /**
- * A catch-up could not complete (window unreadable, affected set over the
- * bound, …): demote to an exact rescan, loudly — never serve un-reconciled.
- */
- demoteCatchUpToBackfill(name: string, reason: string): void {
- this.pendingCatchUp.delete(name)
- this.needsBackfill.add(name)
- prodLog.warn(`[Aggregation] '${name}': catch-up demoted to full rescan — ${reason}`)
- }
-
- /**
- * Rebuild an aggregate through the native provider's parallel path
- * (SELF-ENGINE-LIFECYCLE-SPRINT ask (c) — `rebuildAggregate` existed on
- * the provider contract but was never invoked; the JS walk fed
- * per-entity FFI calls instead). Returns false when no provider rebuild
- * exists — the caller streams the JS walk as before.
- */
- rebuildWithProvider(name: string, entities: Array>): boolean {
- const def = this.definitions.get(name)
- if (!def || !this.nativeProvider?.rebuildAggregate) return false
- const rebuilt = this.nativeProvider.rebuildAggregate(
- def,
- entities.filter(e => !isAggregateEntity(e) && matchesSource(e, def.source))
- )
- this.states.set(name, rebuilt)
- this.backfillStaging.delete(name)
- this.needsBackfill.delete(name)
- this.dirty.add(name)
- return true
- }
-
- /**
- * A write-path hook could not see the entity it needed (e.g. a delete
- * whose before-image was unavailable): flag EVERY defined aggregate for
- * an exact rescan, loudly — the counts must never silently drift
- * (SELF-ENGINE-LIFECYCLE-SPRINT ask (d): the gated hook used to SKIP).
- */
- flagAllForRescan(reason: string): void {
- for (const name of this.definitions.keys()) this.needsBackfill.add(name)
- prodLog.warn(
- `[Aggregation] all ${this.definitions.size} aggregate(s) flagged for rescan — ${reason}`
- )
- }
-
// ============= Write-Time Hooks =============
- /**
- * A write is landing for an aggregate whose persisted-state adoption is still
- * pending — adopting after this write would lose its contribution. Settle the
- * decision now: an exact rescan instead of adoption. The window is the few
- * milliseconds between a boot-time defineAggregate() and init() completing,
- * so this rarely fires; when it does, correctness wins over the walk.
- */
- private resolveAdoptOnWrite(name: string): void {
- if (this.pendingAdopt.has(name)) {
- this.pendingAdopt.delete(name)
- this.needsBackfill.add(name)
- }
- }
-
/**
* Called when an entity is added. Updates all matching aggregates.
*/
@@ -942,7 +553,6 @@ export class AggregationIndex {
for (const [name, def] of this.definitions) {
if (!matchesSource(entity, def.source)) continue
- this.resolveAdoptOnWrite(name)
if (this.nativeProvider) {
const results = this.nativeProvider.incrementalUpdate(name, def, entity, 'add')
@@ -969,10 +579,6 @@ export class AggregationIndex {
const oldMatches = matchesSource(oldEntity, def.source)
const newMatches = matchesSource(newEntity, def.source)
- if (oldMatches || newMatches) {
- this.resolveAdoptOnWrite(name)
- }
-
if (this.nativeProvider && (oldMatches || newMatches)) {
const results = this.nativeProvider.incrementalUpdate(name, def, newEntity, 'update', oldEntity)
this.applyNativeResults(name, results)
@@ -999,7 +605,6 @@ export class AggregationIndex {
for (const [name, def] of this.definitions) {
if (!matchesSource(entity, def.source)) continue
- this.resolveAdoptOnWrite(name)
if (this.nativeProvider) {
const results = this.nativeProvider.incrementalUpdate(name, def, entity, 'delete')
@@ -1152,7 +757,7 @@ export class AggregationIndex {
def: AggregateDefinition,
entity: Record
): void {
- const stateMap = (this.backfillStaging.get(aggName) ?? this.states.get(aggName))!
+ const stateMap = this.states.get(aggName)!
// Fan out: an unnest dimension makes one entity contribute to several groups.
for (const groupKey of computeGroupKeys(entity, def.groupBy)) {
@@ -1180,7 +785,7 @@ export class AggregationIndex {
// distinctCount tracks distinct values of ANY type (strings, numbers, booleans),
// keyed by their string form — NOT numeric-coerced, since its primary use is
// categorical (distinct categories / users / tags), not numeric columns.
- const raw = readAddressed(entity as unknown as HNSWNounWithMetadata, metricDef.field!)
+ const raw = resolveEntityField(entity as unknown as HNSWNounWithMetadata, metricDef.field!)
if (raw !== undefined && raw !== null) {
if (!state.valueCounts) state.valueCounts = {}
const key = String(raw)
@@ -1210,7 +815,7 @@ export class AggregationIndex {
def: AggregateDefinition,
entity: Record
): void {
- const stateMap = (this.backfillStaging.get(aggName) ?? this.states.get(aggName))!
+ const stateMap = this.states.get(aggName)!
// Fan out: reverse the entity's contribution from every group it joined.
for (const groupKey of computeGroupKeys(entity, def.groupBy)) {
@@ -1224,7 +829,7 @@ export class AggregationIndex {
state.count = Math.max(0, state.count - 1)
state.sum = Math.max(0, state.sum - 1)
} else if (metricDef.op === 'distinctCount') {
- const raw = readAddressed(entity as unknown as HNSWNounWithMetadata, metricDef.field!)
+ const raw = resolveEntityField(entity as unknown as HNSWNounWithMetadata, metricDef.field!)
if (raw !== undefined && raw !== null && state.valueCounts) {
const key = String(raw)
const c = state.valueCounts[key]
@@ -1266,7 +871,7 @@ export class AggregationIndex {
* Apply results from native provider back into the state maps.
*/
private applyNativeResults(aggName: string, results: AggregateGroupState[]): void {
- const stateMap = (this.backfillStaging.get(aggName) ?? this.states.get(aggName))!
+ const stateMap = this.states.get(aggName)!
for (const group of results) {
const serialized = serializeGroupKey(group.groupKey)
stateMap.set(serialized, group)
diff --git a/src/brainy.ts b/src/brainy.ts
index da04577e..32b8757b 100644
--- a/src/brainy.ts
+++ b/src/brainy.ts
@@ -25,8 +25,7 @@ import {
} from './storage/brainFormat.js'
import type { BrainFormat } from './storage/brainFormat.js'
import { StorageAdapter, Vector, DistanceFunction, EmbeddingFunction, GraphVerb, STANDARD_ENTITY_FIELDS } from './coreTypes.js'
-import { isZeroNormVector } from './utils/distance.js'
-import type { HNSWNoun, HNSWNounWithMetadata, HNSWVerbWithMetadata, EntityVisibility } from './coreTypes.js'
+import type { HNSWNounWithMetadata, HNSWVerbWithMetadata, EntityVisibility } from './coreTypes.js'
import {
defaultEmbeddingFunction,
cosineDistance,
@@ -47,10 +46,8 @@ import {
pageRank,
MinHeap
} from './graph/analyticsFallback.js'
-import { runGraphAudit, type GraphAuditReport } from './graph/graphAudit.js'
import { createPipeline } from './streaming/pipeline.js'
import { configureLogger, LogLevel, prodLog } from './utils/logger.js'
-import { warnOnLowOsLimits } from './utils/osLimits.js'
import { setGlobalCache } from './utils/unifiedCache.js'
import type { UnifiedCache } from './utils/unifiedCache.js'
import { rankIndicesByScore, reorderByIndices } from './utils/resultRanking.js'
@@ -64,10 +61,7 @@ import type {
PathOptions,
MetadataIndexProvider,
OpaqueIdSet,
- AtGenerationVectors,
- VectorIndexProvider,
- GraphIndexProvider,
- ProviderMaintenanceDebt
+ AtGenerationVectors
} from './plugin.js'
import type {
BrainyPlugin,
@@ -76,7 +70,6 @@ import type {
} from './plugin.js'
import { ConnectionsCodec } from './hnsw/connectionsCodec.js'
import { TransactionManager } from './transaction/TransactionManager.js'
-import { transactTimeoutBudget } from './transaction/Transaction.js'
import { RevisionConflictError } from './transaction/RevisionConflictError.js'
import { EntityNotFoundError, RelationNotFoundError } from './errors/notFound.js'
import {
@@ -92,13 +85,12 @@ import { findCallerLocation } from './utils/callerLocation.js'
import {
SaveNounMetadataOperation,
SaveNounOperation,
- AddToVectorIndexOperation,
+ AddToHNSWOperation,
AddToMetadataIndexOperation,
SaveVerbMetadataOperation,
SaveVerbOperation,
AddToGraphIndexOperation,
- RemoveFromVectorIndexOperation,
- ReplaceInVectorIndexOperation,
+ RemoveFromHNSWOperation,
RemoveFromMetadataIndexOperation,
RemoveFromGraphIndexOperation,
UpdateNounMetadataOperation,
@@ -145,16 +137,12 @@ import {
ScoreExplanation,
FillSubtypeRule,
FillSubtypeRules,
- FillSubtypesResult,
- RepairReport,
- RepairFamilyReport
+ FillSubtypesResult
} from './types/brainy.types.js'
import { NounType, VerbType, TypeUtils } from './types/graphTypes.js'
import {
splitNounMetadataRecord,
- splitVerbMetadataRecord,
- buildNounMetadataRecord,
- buildVerbMetadataRecord
+ splitVerbMetadataRecord
} from './types/reservedFields.js'
import { BrainyInterface } from './types/brainyInterface.js'
import type { IntegrationHub, IntegrationHubConfig } from './integrations/core/IntegrationHub.js'
@@ -164,8 +152,6 @@ import { AggregationIndex } from './aggregation/AggregationIndex.js'
import { AggregateMaterializer } from './aggregation/materializer.js'
import type { AggregateDefinition, AggregateQueryParams, AggregateResult } from './types/brainy.types.js'
import type { MigrationProgress } from './types/brainy.types.js'
-import type { IndexedProjectionPath, WaitForIndexedOptions } from './types/brainy.types.js'
-import { WaitForIndexedTimeoutError } from './types/brainy.types.js'
import { resolveJsHnswConfig, DEFAULT_RECALL } from './utils/recallPreset.js'
import * as fs from 'node:fs'
import * as os from 'node:os'
@@ -181,14 +167,6 @@ import {
type ImportResult
} from './db/portableGraph.js'
import { GenerationStore, type CommitBeforeImages } from './db/generationStore.js'
-import type { FactScanHandle, FactMarkerRecord } from './db/factLog.js'
-import {
- ENTITY_TREE_STAMP_PATH,
- readFamilyStamp,
- verifyFamilyStamp,
- writeFamilyStamp,
- type FamilyStamp
-} from './db/familyStamp.js'
import {
ChangeFeed,
type BrainyChangeEvent,
@@ -197,31 +175,11 @@ import {
} from './events/changeFeed.js'
import { isDeterministicEmbedMode } from './embeddings/deterministicEmbedMode.js'
import { GenerationConflictError, StoreInconsistentError } from './db/errors.js'
-import { BrainyError, GraphIndexNotReadyError, MetadataIndexNotReadyError, MigrationInProgressError, VectorIndexNotReadyError } from './errors/brainyError.js'
-import {
- assessIndexReadiness,
- assessProviderHealth,
- assessProviderRebuild,
- describeRebuildProgress
-} from './utils/indexReadiness.js'
-import { reconstructNounWrapper } from './db/factLog.js'
-import { asBrainyFieldRefusal } from './db/fieldAddressing.js'
-import {
- readLogAuthority,
- runLogCompletenessOracle,
- flipToLogAuthority,
- recordDigest,
- nounEntityTruth,
- LOG_AUTHORITY_PATH,
- type LogAuthorityRecord,
- type LogAuthorityStorage,
- type OracleReport
-} from './db/logAuthority.js'
+import { BrainyError, GraphIndexNotReadyError, MetadataIndexNotReadyError, MigrationInProgressError } from './errors/brainyError.js'
import { MemoryStorage } from './storage/adapters/memoryStorage.js'
import type {
CompactHistoryOptions,
CompactHistoryResult,
- HistoryStats,
TransactOptions,
TransactReceipt,
TxLogEntry,
@@ -231,7 +189,7 @@ import type {
HistoryVersion
} from './db/types.js'
import { stableDeepEqual } from './db/stableEqual.js'
-import type { VersionedIndexProvider, ProviderInvariantReport } from './plugin.js'
+import type { VersionedIndexProvider } from './plugin.js'
import type { Operation, TransactionFunction } from './transaction/types.js'
/**
@@ -295,8 +253,6 @@ type ResolvedBrainyConfig = Required<
| 'retention'
| 'eagerEmbeddings'
| 'migrationWaitTimeoutMs'
- | 'transactionBudgetFloorMs'
- | 'persistence'
>
> &
Pick<
@@ -309,8 +265,6 @@ type ResolvedBrainyConfig = Required<
| 'retention'
| 'eagerEmbeddings'
| 'migrationWaitTimeoutMs'
- | 'transactionBudgetFloorMs'
- | 'persistence'
>
/**
@@ -394,22 +348,6 @@ interface PlannedTransact {
* rejected batch (CAS conflict, failed apply) emits nothing.
*/
changeEvents: PendingChangeEvent[]
- /**
- * V2 marker records riding the batch's ONE commit fact (e.g. the
- * deferred-embedding pending markers) — same generation, same atomic
- * append as the batch itself. A rejected batch appends no fact, so no
- * marker outlives its write.
- */
- markerRecords: FactMarkerRecord[]
- /**
- * Ids the batch's `{ op: 'update' }` unvector door (`vector: []`) needs to
- * decrement on the vectored-noun ledger — consumed by `transact()` with a
- * proper `await this.storage.noteVectorUnlanded?.(id)` per id, AFTER
- * `commitTransaction` resolves (never for a rejected batch). Kept separate
- * from `postCommit` (`Array<() => void>`, called synchronously, fire-and-
- * forget) because the ledger hook is async and must be awaited.
- */
- vectorUnlands: string[]
}
/**
@@ -428,88 +366,6 @@ class InsertPreconditionExistsSignal extends Error {
}
}
-/**
- * @description The derived-index families a read may depend on. A read that
- * consults none of them (a canonical-storage read: `get`, an entity
- * enumeration, a VFS content/dir read) is index-independent and must never
- * block on another family's one-time migration. Used by the family-scoped
- * migration gate ({@link Brainy.awaitMigrationLock}).
- */
-export type IndexFamily = 'vector' | 'metadata' | 'graph'
-
-/**
- * @description Honest per-surface outcome for {@link Brainy.warm}. Literal
- * meanings — never conflate the first two:
- * - `'warmed'` — the provider's own `warm?()` hook ran (vector/graph), or the
- * surface's full-hydration seam loaded EVERY shard/field/segment from
- * storage (metadata; graph's fallback path). The surface is genuinely at
- * steady-state cost for the next operation.
- * - `'probed'` — no `warm?()` hook was available, so a best-effort read
- * (e.g. one `search()` call) faulted in *some* backing storage as a side
- * effect. Real work happened, but it is NOT the same guarantee as
- * `'warmed'` — never reported as `'warmed'`.
- * - `'unavailable'` — nothing ran: no hook, no hydration seam, and (for the
- * vector probe fallback) nothing to probe (an empty index or unknown
- * vector dimension). The surface is unchanged by this `warm()` call.
- */
-export type WarmOutcome = 'warmed' | 'probed' | 'unavailable'
-
-/**
- * @description Result of {@link Brainy.warm}: one {@link WarmOutcome} +
- * elapsed time per index surface, plus the total wall-clock time for the
- * whole call. `durationMs` is measured around exactly the work described by
- * that surface's `outcome` (e.g. the vector entry's `durationMs` times the
- * provider `warm()` call OR the probe `search()` call — whichever ran).
- */
-export interface WarmReport {
- vector: { outcome: WarmOutcome; durationMs: number }
- metadata: { outcome: WarmOutcome; durationMs: number }
- graph: { outcome: WarmOutcome; durationMs: number }
- /** Total wall-clock time for the whole `warm()` call (all three surfaces). */
- totalDurationMs: number
-}
-
-/**
- * @description Result of {@link Brainy.maintenanceDebt}: one outcome per
- * index surface, mirroring {@link WarmReport}'s shape.
- * - `'reported'` — the active provider for this surface implements
- * `maintenanceDebt?()` and its {@link ProviderMaintenanceDebt} payload is
- * attached verbatim under `debt`.
- * - `'unavailable'` — the active provider does not implement the hook, so
- * nothing is known; brainy never estimates or infers a payload on its
- * behalf.
- */
-export type MaintenanceDebtOutcome = 'reported' | 'unavailable'
-
-/**
- * @description Per-surface result of {@link Brainy.maintenanceDebt}. Brainy
- * performs no thresholding, polling, or estimation over this data — it is a
- * pure passthrough of each active provider's own self-report (the provider
- * owns the numbers; the operator owns the policy).
- */
-export interface MaintenanceDebtReport {
- vector: { outcome: MaintenanceDebtOutcome; debt?: ProviderMaintenanceDebt }
- metadata: { outcome: MaintenanceDebtOutcome; debt?: ProviderMaintenanceDebt }
- graph: { outcome: MaintenanceDebtOutcome; debt?: ProviderMaintenanceDebt }
-}
-
-/**
- * How long a failed aggregation-backfill walk suppresses fresh walk attempts.
- * Within the window, queries rethrow the recorded failure instantly (loud,
- * cheap); after it, one new attempt is allowed. Bounds the damage of a
- * caller-side tight retry loop against a deterministically-failing store.
- */
-const AGGREGATION_BACKFILL_RETRY_COOLDOWN_MS = 30_000
-
-/**
- * Time budget for the auto-compaction pass at close() (8.9.0). Bounds how long
- * a clean shutdown spends reclaiming history backlog — an early stop is a
- * consistent prefix and the next close/explicit pass resumes. Explicit
- * `compactHistory()` calls are unbounded unless the caller passes their own
- * `timeBudgetMs` (maintenance windows choose their own budgets).
- */
-const CLOSE_COMPACTION_BUDGET_MS = 5_000
-
/**
* The main Brainy class - Clean, Beautiful, Powerful
* REAL IMPLEMENTATION - No stubs, no mocks
@@ -561,24 +417,9 @@ export class Brainy implements BrainyInterface {
* store has assigned the batch generation by then; for single-op writes it
* reads the post-write watermark. The arrow body reads `generationStore`
* lazily, so it is safe to define before `init()` assigns the store.
- * Metadata/vector index writes use the bootstrap-honest twin
- * {@link indexWriteGeneration} below.
*/
private readonly graphWriteGeneration = (): bigint =>
BigInt(this.generationStore.generation())
- /**
- * The metadata/vector twin of {@link graphWriteGeneration}, honest about
- * bootstrap: while generation stamping is inactive (init-time
- * infrastructure writes, e.g. the VFS root, applied via
- * `runWithoutGeneration`) there IS no commit generation — this resolves to
- * `undefined` so a provider records "unstamped", never a fabricated 0.
- * The graph thunk keeps its non-optional `bigint` contract (no graph
- * writes occur during bootstrap).
- */
- private readonly indexWriteGeneration = (): bigint | undefined =>
- this._generationStampingActive
- ? BigInt(this.generationStore.generation())
- : undefined
/** Lazily built host surface shared by every `Db` value of this brain. */
private _dbHost?: DbHost
/**
@@ -649,6 +490,8 @@ export class Brainy implements BrainyInterface {
/** One-shot guard so the degraded-reads warning fires once per degraded window
* (reset when the degraded state clears). See {@link warnIfReadsDegraded}. */
private _degradedReadWarned = false
+ /** One-shot guard so the metadata cold-open consistency probe runs once per brain. */
+ private _metadataConsistencyProbed = false
/** Graph-adjacency cold-load consistency: verified-live this session (one-shot). */
private _graphAdjacencyVerified = false
/** Re-entrancy guard: a verify (rebuild → reads) is in flight. */
@@ -657,10 +500,6 @@ export class Brainy implements BrainyInterface {
private _metadataVerified = false
/** Re-entrancy guard for {@link verifyMetadataLive}. */
private _metadataVerifying = false
- /** Vector-index cold-read guard: verified-serving this session (one-shot). */
- private _vectorVerified = false
- /** Re-entrancy guard for {@link verifyVectorLive}. */
- private _vectorVerifying = false
/**
* Coordinated migration LOCK (#18): dedup guards so the "upgrading, blocking"
* and "upgrade complete, resumed" lines each log once per migration window,
@@ -737,59 +576,6 @@ export class Brainy implements BrainyInterface {
private _hub?: IntegrationHub // Integration Hub for external tools
private _pendingMigrationRunner?: MigrationRunner // Deferred migration runner for large datasets
private _aggregationIndex?: AggregationIndex // Incremental aggregation engine
- private _aggregationBackfillFlight: Promise | null = null // Single-flight backfill walk
- private _aggregationCatchUpFlight: Promise | null = null // Single-flight behind-stamp catch-up
-
- // ENGINE-OWNED PERSISTENCE CADENCE (SELF-ENGINE-LIFECYCLE-SPRINT):
- // write-count / interval / idle triggers → ONE background flush at a time.
- // Write acks NEVER await it; a failed background flush is LOUD and re-armed.
- private _persistDirtyWrites = 0
- private _persistLastFlushAt = Date.now()
- /**
- * Whether a write has been committed since the last flush that ran. THE
- * ENGINE DOES NO PERIODIC WORK WITHOUT A CAUSE: a brain nobody has written
- * to has nothing to make durable, and a flush over it must cost nothing and
- * say nothing. Before this, a flush called every provider, stamped the
- * watermarks, persisted the generation counter and re-stamped the entity
- * tree whether or not anything had changed — roughly 28 writes for a store
- * that had not moved.
- *
- * WHAT THIS DOES NOT EXPLAIN, stated so nobody reads it as solved: a
- * production process holding 21 brains printed "All indexes flushed to disk
- * in 216-601ms" per brain every ~35s and idled at 1.26 cores with no writes
- * for ten minutes. This engine's cadence is WRITE-DRIVEN — every trigger
- * runs through noteWriteForPersistence, which only a committed write calls —
- * so something was calling flush() on those brains, and this gate makes such
- * a call free rather than accounting for it. The caller is still unidentified.
- */
- private _dirtySinceLastFlush = false
- private _persistIdleTimer: ReturnType | null = null
- private _persistBackgroundFlight: Promise | null = null
-
- // DEFERRED EMBEDDING (MT5): pending markers are LOG RECORDS — an
- // embed.pending record rides the deferred write's own commit fact and
- // embed.landed rides the landing commit; this set is the in-memory
- // fast-path index, rebuilt at open by folding the log's marker records.
- // ONE background worker drains it. A crash can delay a vector, never
- // lose one.
- private _pendingEmbedIds = new Set()
- private _embedWorkerFlight: Promise | null = null
-
- // OPEN-PATH FIX: the background embedding-engine warm kicked off (never
- // awaited) by `performInit()` when `eagerEmbeddings` resolves true. Stored
- // for observability only — `embed()`/`embeddingManager.embed()` already
- // await the engine's OWN singleton init promise internally, so nothing
- // needs to explicitly await this field for correctness. Never rejects on
- // its own: a `.catch` narrates the failure and swallows it so a failed
- // warm never surfaces as an unhandled rejection.
- private _embeddingWarmPromise: Promise | null = null
-
- /** The stored log-authority switch, read once at open (default: tree). */
- private _logAuthority: LogAuthorityRecord = { authority: 'tree' }
- // A failed walk latches its error: retries within the cooldown rethrow it
- // instantly instead of re-walking, so a tight caller-side retry loop costs
- // one loud error per query, never a full store walk per query.
- private _aggregationBackfillFailure: { at: number; error: Error } | null = null
private _materializer?: AggregateMaterializer // Debounced materialization of aggregate results
/**
* Fields registered via `brain.trackField()` — drives optional value validation on
@@ -844,48 +630,13 @@ export class Brainy implements BrainyInterface {
// applies only to instances that were never closed.
private closed = false
- // Index-build-at-open state. `lazyRebuildCompleted` predates the health-gate
- // law (it named a first-QUERY lazy rebuild) and stays for `getIndexStatus()`
- // API compatibility, but its truth changed: a needed rebuild now runs
- // unconditionally at open() (see `rebuildIndexesIfNeeded`), never deferred to
- // a read, so this simply flips true once that open-time step has run.
- // `lazyRebuildInProgress` / `lazyRebuildPromise` (the first-query rebuild's
- // concurrency guard) are retired with the lazy-build path they served —
- // `ensureIndexesLoaded()` is a read-time CHECK now, never a build.
+ // Lazy rebuild state (Production-scale lazy loading)
+ // Prevents race conditions when multiple queries trigger rebuild simultaneously
+ private lazyRebuildInProgress = false
private lazyRebuildCompleted = false
-
- // Read-gate narration dedup: a degraded-but-serving or not-ready health
- // report narrates via prodLog.warn ONCE per (provider, report.generation) —
- // never once per read. Keyed on the provider instance itself.
- /**
- * The last health narration emitted per provider, keyed by its CONTENT.
- *
- * This used to dedupe on the provider's `generation` counter, which bumps on
- * every ledger mutation and every rebuild boundary — so a provider that
- * bumps its generation on routine work re-emitted the same unchanged health
- * line on every read that consulted it, and a provider that never bumped
- * could suppress a line whose reasons had genuinely changed. The dedupe key
- * is now what the line SAYS: an unchanged verdict is silent however the
- * generation moves, and a changed verdict is always heard.
- */
- private _lastNarratedHealth = new Map()
+ private lazyRebuildPromise: Promise | null = null
constructor(config?: BrainyConfig) {
- // The reserved-field write policy died with the field-addressing law:
- // every metadata name is the user's now (engine scalars write via their
- // dedicated params and read at `system.*`), so there is nothing left for
- // the policy to govern. A config still passing it refuses loudly rather
- // than being silently ignored.
- if (config && 'reservedFieldPolicy' in (config as Record)) {
- throw new Error(
- `reservedFieldPolicy was removed by the field-addressing law: metadata field ` +
- `names are never reserved anymore — every name in the metadata bag is the ` +
- `user's and works like any other field. Set engine scalars via their ` +
- `dedicated params (confidence, weight, subtype, …) and query them as ` +
- `system.. Remove the reservedFieldPolicy option.`
- )
- }
-
// Normalize configuration with defaults
this.config = this.normalizeConfig(config)
@@ -986,14 +737,14 @@ export class Brainy implements BrainyInterface {
* extends FileSystemStorage`) inherit new methods Brainy adds to
* `FileSystemStorage` / `BaseStorage` automatically — `typeof` walks the
* prototype chain, so there's no in-package version skew to worry about as
- * long as the plugin's own dist resolves `@soulcraftlabs/brainy` dynamically
+ * long as the plugin's own dist resolves `@soulcraft/brainy` dynamically
* (which Cortex 2.2.x onward does — see
* `node_modules/@soulcraft/cor/dist/storage/mmapFileSystemStorage.js`).
*
* This helper exists for the **build/install** failure modes the import
* resolution can't catch:
* - Stale `node_modules` left over from a prior `bun install` against
- * `@soulcraftlabs/brainy ≤7.20.x`.
+ * `@soulcraft/brainy ≤7.20.x`.
* - Lockfile drift pinning brainy below the version that introduced the
* method.
* - Docker layer caches that reuse a `node_modules` from an earlier image.
@@ -1132,86 +883,6 @@ export class Brainy implements BrainyInterface {
configureLogger({ level: LogLevel.DEBUG }) // Enable verbose logging
}
- // OPEN-PATH NARRATION: phase timing across the five named stretches of
- // init — storage init / generation-store open+fold / index init+gate /
- // VFS bootstrap / embedding-warm-started. Each `markPhase()` call records
- // elapsed ms SINCE THE PREVIOUS checkpoint, so the buckets always sum to
- // the pre-integration/warmOnOpen total.
- //
- // THE LAW THIS ENFORCES: an open is never silent for more than
- // OPEN_HEARTBEAT_MS. A production service opening a 16 GB store logged
- // NOTHING for three minutes and then began work — the operator could not
- // tell a slow open from a hung one, and restarted into the same wall.
- // Two mechanisms, both on the always-visible narration channel (the old
- // breakdown used `prodLog.warn`, which production clamps away — that is
- // why the three minutes were silent):
- // - a heartbeat that names the phase currently running and its elapsed
- // wall, every OPEN_HEARTBEAT_MS, for as long as the open lasts;
- // - one line per phase AS IT ENDS, naming its wall and its cause, for
- // any phase over OPEN_PHASE_NARRATE_MS.
- // The heartbeat is unref'd and cleared in the `finally` below, so it can
- // neither hold the process open nor outlive a failed init. It cannot fire
- // inside a phase that blocks the event loop synchronously; such a phase
- // must narrate its own progress (the generation-log fold does).
- const OPEN_HEARTBEAT_MS = 5_000
- const OPEN_PHASE_NARRATE_MS = 2_000
- /** Phase order + what each one is paying for, quoted in its narration. */
- const OPEN_PHASES: ReadonlyArray<{ name: string; cause: string }> = [
- { name: 'storage-init', cause: 'opening the store and loading its count ledger' },
- {
- name: 'generation-store-open-fold',
- cause: 'opening the generation store: crash-recovery replay/fold, derived-family registration, format handshake'
- },
- { name: 'index-init-gate', cause: 'constructing the derived indexes and gating them for serving' },
- { name: 'vfs-bootstrap', cause: 'bootstrapping the virtual filesystem' },
- { name: 'embedding-warm-started', cause: 'starting the background embedding warm' }
- ]
- const initStart = Date.now()
- let lastPhaseCheckpoint = initStart
- let currentPhaseIndex = 0
- const phaseTimingsMs: Record = {}
- const openHeartbeat: ReturnType = setInterval(() => {
- const phase = OPEN_PHASES[currentPhaseIndex]
- if (!phase) return
- prodLog.narrate(
- `[Brainy] open: still in phase ${currentPhaseIndex + 1}/${OPEN_PHASES.length} ` +
- `"${phase.name}" after ${Math.round((Date.now() - lastPhaseCheckpoint) / 1000)}s ` +
- `(${Math.round((Date.now() - initStart) / 1000)}s into the open) — ${phase.cause}`
- )
- }, OPEN_HEARTBEAT_MS)
- if (typeof openHeartbeat.unref === 'function') openHeartbeat.unref()
- /**
- * Narrate one STEP inside a phase when it turns out to be expensive.
- * A phase that costs a minute and names only itself tells an operator
- * where to look but not what to look at; this names the step. Silent
- * under OPEN_PHASE_NARRATE_MS, so a fast open says nothing extra.
- */
- const step = async (name: string, cause: string, run: () => Promise): Promise => {
- const startedAt = Date.now()
- try {
- return await run()
- } finally {
- const elapsed = Date.now() - startedAt
- if (elapsed >= OPEN_PHASE_NARRATE_MS) {
- prodLog.narrate(`[Brainy] open: step "${name}" took ${elapsed}ms — ${cause}`)
- }
- }
- }
- const markPhase = (name: string): void => {
- const now = Date.now()
- const elapsed = now - lastPhaseCheckpoint
- phaseTimingsMs[name] = elapsed
- lastPhaseCheckpoint = now
- const finished = OPEN_PHASES[currentPhaseIndex]
- if (elapsed >= OPEN_PHASE_NARRATE_MS && finished && finished.name === name) {
- prodLog.narrate(
- `[Brainy] open: phase ${currentPhaseIndex + 1}/${OPEN_PHASES.length} ` +
- `"${name}" finished in ${elapsed}ms — ${finished.cause}`
- )
- }
- currentPhaseIndex++
- }
-
try {
// Auto-detect and activate plugins BEFORE storage setup
// so plugin-provided storage factories (e.g., filesystem override from cor) are available
@@ -1227,13 +898,6 @@ export class Brainy implements BrainyInterface {
this.storage = await this.setupStorage()
await this.storage.init()
- // OS-limit detection (once per process, Linux-only, measurement-only):
- // warn NOW about RLIMIT_NOFILE / vm.max_map_count values that will bite
- // at pool scale, instead of letting the operator meet them as EMFILE or
- // a failed mmap deep inside an index open. Fire-and-forget — the check
- // never affects open.
- void warnOnLowOsLimits()
-
// Acquire the writer lock for filesystem (and other locking-capable) backends.
// Skipped in reader mode and on backends that don't support multi-process locking.
// Throws if another live writer holds the directory (unless force: true).
@@ -1272,7 +936,7 @@ export class Brainy implements BrainyInterface {
`and the flush-request RPC are disabled for this directory. ` +
`Likely fix: clean install (\`rm -rf node_modules bun.lockb && ` +
`bun install\`) or rebuild your container image to refresh ` +
- `\`@soulcraftlabs/brainy\` to ≥7.21. See docs/concepts/storage-adapters.md.`
+ `\`@soulcraft/brainy\` to ≥7.21. See docs/concepts/storage-adapters.md.`
)
} else {
console.warn(
@@ -1283,12 +947,6 @@ export class Brainy implements BrainyInterface {
}
}
- // PHASE 1 of 5 — "storage init": plugin/legacy-layout bootstrap,
- // storage adapter construction+init, the OS-limit check, and the
- // writer-lock claim, all folded into one bucket (everything above this
- // line since performInit started).
- markPhase('storage-init')
-
// 8.0 generational MVCC: open the record layer BEFORE any index is
// created or loaded. Crash recovery may rewrite canonical entity files
// (restoring before-images of an uncommitted transaction), and every
@@ -1297,54 +955,9 @@ export class Brainy implements BrainyInterface {
// instances skip recovery (readers never write; the next writer
// repairs).
this.generationStore = new GenerationStore(this.storage)
- const generationOpenResult = await step(
- 'generation-store.open',
- 'reading the generation manifest and committed ranges, opening the fact log and the ' +
- 'packed segment tier, and folding any crash-recovery replay',
- () => this.generationStore.open({ readOnly: this.config.mode === 'reader' })
- )
-
- // The generation fact log is CANONICAL state, not a derived index — no
- // sweeper, GC, or blob-lifecycle path may ever delete under it. Declare
- // its namespace as a protected family (rebuildable: false — a lost fact
- // segment is NOT reconstructable) so the storage layer REFUSES such
- // deletes; refusal beats trust. Feature-detected + idempotent per name.
- if (
- this.config.mode !== 'reader' &&
- this.generationStore.getFactLog() &&
- typeof this.storage.registerDerivedFamily === 'function'
- ) {
- await this.storage.registerDerivedFamily({
- name: 'generation-facts',
- members: ['_generations/facts/'],
- namespace: true,
- rebuildable: false
- })
- }
-
- // Fact-scan capability: wire the storage seam through which index
- // providers (which hold only `storage`) reach the fact log. A closure
- // over the LIVE log — restore/reopen swaps the instance transparently —
- // so a provider's heal can switch from the enumeration walk to one
- // sequential fact scan whenever the log exists.
- if (typeof (this.storage as BaseStorage).setFactScanSource === 'function') {
- ;(this.storage as BaseStorage).setFactScanSource({
- factLog: () => this.generationStore?.getFactLog() ?? null,
- // The committed watermark, exposed as a capability so providers
- // never parse the store's private manifest format.
- committedGeneration: () => this.generationStore?.committedGeneration() ?? 0
- })
- }
-
- // Entity-tree stamp coherence: compare the stamped sourceGeneration +
- // rollup invariants against the log head + live counters. Loud on
- // genuine incoherence (repairIndex heals), silent on absent/coherent,
- // benign-behind refreshes at the next flush. Never blocks open.
- await step(
- 'verify-entity-tree-stamp',
- 'comparing the entity tree\'s stamped generation and rollups against the store',
- () => this.verifyEntityTreeStamp()
- )
+ const generationOpenResult = await this.generationStore.open({
+ readOnly: this.config.mode === 'reader'
+ })
// 8.0 ⇄ native-provider version handshake: load the on-disk brain-format
// marker (`_system/brain-format.json`) into an in-memory field NOW —
@@ -1356,11 +969,7 @@ export class Brainy implements BrainyInterface {
// them from the canonical records and then re-stamps the marker AFTER the
// rebuild verifies (non-destructive: a crash mid-rebuild leaves the old /
// absent marker, so the next open idempotently re-rebuilds).
- this._brainFormat = await step(
- 'read-brain-format',
- 'reading the on-disk format marker that decides whether the derived indexes are stale',
- () => readBrainFormat(this.storage)
- )
+ this._brainFormat = await readBrainFormat(this.storage)
this._indexEpochStale =
this._brainFormat === null || this._brainFormat.indexEpoch !== EXPECTED_INDEX_EPOCH
@@ -1371,19 +980,9 @@ export class Brainy implements BrainyInterface {
// upgrade verifies + stamps; retained on failure. No-op for a reader, for
// non-filesystem storage, or for a brain with no persisted data.
if (this._indexEpochStale && this.config.migrationBackup && !this.isReadOnly) {
- await step(
- 'pre-upgrade-backup',
- 'snapshotting the brain directory before a one-time format rebuild (migrationBackup)',
- () => this.createMigrationBackupIfNeeded()
- )
+ await this.createMigrationBackupIfNeeded()
}
- // PHASE 2 of 5 — "generation-store open+fold": GenerationStore
- // construction+open (crash-recovery replay/rollback fold), the
- // derived-family registration, the fact-scan seam, the entity-tree
- // stamp check, the brain-format handshake, and the pre-upgrade backup.
- markPhase('generation-store-open-fold')
-
// Provider: embeddings (reassign embedder if plugin provides one)
const embeddingProvider = this.pluginRegistry.getProvider('embeddings')
if (embeddingProvider) {
@@ -1458,42 +1057,6 @@ export class Brainy implements BrainyInterface {
this.graphIndex = graphIndex
}
- // Fact-log v2 mint seam: after-image records carry minted dense ints,
- // and the ONE authority for those assignments is the metadata index's
- // id mapper (append-only getOrAssign — a rebuilt mapper reproduces
- // them exactly). The generation store cannot know the mapper, so the
- // mint thunk is injected here, immediately after the index is ready;
- // installing it is what flips the fact log's LIVE writes to the v2
- // segment format. A configuration whose mapper is unavailable throws
- // at mint time — an int of 0 is never written.
- this.generationStore.setIntMinter((kind, id) => {
- const mapper = this.metadataIndex?.getIdMapper?.()
- if (!mapper || typeof mapper.getOrAssign !== 'function') {
- throw new Error(
- `fact log v2: cannot mint the ${kind} int for ${id} — the metadata index's ` +
- `id mapper is unavailable on this configuration; refusing to write an ` +
- `after-image without a reproducible int`
- )
- }
- const minted = mapper.getOrAssign(id, undefined)
- const asBigint = typeof minted === 'bigint' ? minted : BigInt(minted)
- // THE RESERVED-ROOT EXEMPTION: the VFS root (the all-zeros UUID) is
- // minted int 0 BY CONSTRUCTION at genesis on existing brains — the
- // one legitimate zero in the id space. Zero for ANY other id is a
- // corrupt mint and refuses. (Without this, every existing brain's
- // adoption oracle false-flagged its own root and refused the flip.)
- const isReservedRoot =
- asBigint === 0n && id === '00000000-0000-0000-0000-000000000000'
- if (asBigint < 0n || (asBigint === 0n && !isReservedRoot)) {
- throw new Error(
- `fact log v2: the id mapper minted ${asBigint} for ${kind} ${id} — ` +
- `minted ints are positive (int 0 is reserved for the VFS root alone); ` +
- `refusing to write`
- )
- }
- return asBigint
- })
-
// Eager cold-load (readiness contract). A provider that persists its
// derived state exposes init?(): trigger the load NOW — AFTER
// metadataIndex.init() above (the id-mapper is hydrated first, so a
@@ -1538,86 +1101,26 @@ export class Brainy implements BrainyInterface {
`[Brainy] Rebuilding indexes after crash recovery rolled back ` +
`${generationOpenResult.rolledBackGenerations} uncommitted transaction(s)`
)
- // SELF-REBUILD DEFERENCE, same law as the open gate: a provider that
- // is already rebuilding itself from canonical is doing exactly this
- // work. Kicking a second rebuild on top of it is redundant at best.
- // Safe by ordering: the crash-recovery fold ran in the generation
- // store's open, BEFORE any provider was constructed, so a provider
- // rebuilding now is reading the repaired canonical records.
- const kick = async (leg: string, provider: { rebuild: () => Promise }) => {
- const rebuilding = assessProviderRebuild(provider)
- if (rebuilding) {
- prodLog.narrate(
- `[Brainy] crash-recovery rebuild: the ${leg} provider is already ` +
- `${describeRebuildProgress(rebuilding)} from canonical — not kicking a second one.`
- )
- return
- }
- await provider.rebuild()
- }
await Promise.all([
- kick('metadata', this.metadataIndex),
- kick('vector', this.index as unknown as { rebuild: () => Promise }),
- kick('graph', this.graphIndex)
+ this.metadataIndex.rebuild(),
+ this.index.rebuild(),
+ this.graphIndex.rebuild()
])
}
- // METADATA WATERMARK CATCHUP: the JS metadata index computed its
- // three-way watermark verdict inside metadataIndex.init() above,
- // against the generation store's now-FINAL committed generation (the
- // crash-recovery fold above — the durable-at-ack replay of acked
- // writes whose canonical bytes hadn't reached disk — has already run,
- // and any rolled-back-transaction rebuild just above already brought
- // every index current, so the verdict is consumed here whether or not
- // that rebuild ran). Consumed BEFORE the rebuild gate below and BEFORE
- // this open serves any read — the cure for the class of bug where
- // canonical get()/counts recover a crash-window write but find()
- // keeps serving the metadata index's pre-crash state (the index
- // flushes only periodically, not per-commit).
- await this.consumeMetadataWatermarkVerdict(generationOpenResult.rolledBackGenerations > 0)
-
// 8.0 versioned-provider replay-gap check: a provider whose persisted
// index generation is behind the storage layer's committed generation
// replays the gap itself (post-commit applier contract) — surface the
// gap for observability.
for (const provider of this.versionedIndexProviders()) {
const providerGen = provider.generation()
- // Defensive finite-integer guard: committedGeneration() is validated
- // at the store's open (torn artifacts discard, narrated) — but a
- // RangeError here would kill the whole open, so the consumer guards
- // too. A non-finite value narrates and skips the gap check (the
- // provider's own replay contract still governs).
- const committedRaw = this.generationStore.committedGeneration()
- if (!Number.isSafeInteger(committedRaw) || committedRaw < 0) {
- prodLog.warn(
- `[Brainy] committed generation is non-integer (${String(committedRaw)}) at ` +
- `init — torn-artifact survivor; skipping the provider replay-gap check`
- )
- continue
- }
- const committed = BigInt(committedRaw)
+ const committed = BigInt(this.generationStore.committedGeneration())
if (providerGen < committed) {
prodLog.info(
`[Brainy] Versioned index provider is at generation ${providerGen} ` +
`(storage committed: ${committed}) — provider replays the gap per ` +
`the post-commit applier contract`
)
- } else if (providerGen > committed) {
- // The AHEAD direction is incoherence, not a replay gap: the provider's
- // persisted index claims writes the store no longer has — the signature
- // of a torn copy or a log truncation that pulled the committed
- // watermark back (crash recovery, byte-copy of a live store). A replay
- // can never converge on it and index answers may reference vanished
- // writes. Name it loudly at open so it is never diagnosed from a
- // silent journal; the provider's own coherence check / heal walk (or
- // brain.repairIndex()) is the cure.
- prodLog.warn(
- `[Brainy] Versioned index provider is AHEAD of the store: provider ` +
- `generation ${providerGen} vs committed ${committed}. This store was ` +
- `likely copied from a live service or truncated during crash recovery. ` +
- `Derived-index answers may reference rolled-back writes until the ` +
- `provider heals from canonical (brain.repairIndex() forces it).`
- )
}
}
@@ -1642,38 +1145,12 @@ export class Brainy implements BrainyInterface {
}).backfillBlobHistoryRefCountsIfNeeded()
}
- // LEG C (zero-norm/unvector-door law): migrate a legacy zero-norm VFS
- // root BEFORE the vector-leg open gate below ever compares the
- // canonical vectored-noun count against the vector index's size — see
- // migrateLegacyZeroNormVfsRootIfNeeded's JSDoc for why this is a safe
- // O(1) exception to "nothing at open may scale with brain size", and
- // why it must run here rather than waiting on VirtualFileSystem's own
- // (VFS-instance-gated) lazy migration.
- await this.migrateLegacyZeroNormVfsRootIfNeeded()
-
- // Rebuild indexes if needed for existing data. Runs to completion before
- // init() returns — there is no more first-query lazy path, so the flag
- // below (kept for getIndexStatus() API compatibility) simply flips true
- // once this open-time step has run.
- await step(
- 'rebuild-indexes-if-needed',
- 'the derived-index gate: each family\'s readiness verdict, and any build it asks for',
- () => this.rebuildIndexesIfNeeded()
- )
- this.lazyRebuildCompleted = true
+ // Rebuild indexes if needed for existing data
+ await this.rebuildIndexesIfNeeded()
// Check for pending data migrations
await this.checkMigrations()
- // PHASE 3 of 5 — "index init+gate": provider wiring (embeddings,
- // cache, roaring, msgpack, sort:topK, distance), HNSW/metadata/graph
- // index construction, the eager cold-load, id-resolver + connections-
- // codec wiring, crash-recovery index rebuild, the replay-gap check,
- // legacy VFS blob adoption, blob-history backfill, the legacy
- // zero-norm VFS root migration, and the rebuildIndexesIfNeeded() gate
- // + migration check.
- markPhase('index-init-gate')
-
// Register shutdown hooks for graceful count flushing (once globally)
if (!Brainy.shutdownHooksRegisteredGlobally) {
this.registerShutdownHooks()
@@ -1732,11 +1209,7 @@ export class Brainy implements BrainyInterface {
// Initialize VFS: Ensure VFS is ready when accessed as property
// This eliminates need for separate vfs.init() calls - zero additional complexity
this._vfs = new VirtualFileSystem(this)
- await step(
- 'vfs.init',
- 'creating or adopting the VFS root and wiring the path resolver',
- () => this._vfs!.init()
- )
+ await this._vfs.init()
this._vfsInitialized = true // Mark VFS as fully initialized
// 8.0 MVCC: infrastructure bootstrap (VFS root, etc.) is now the
@@ -1746,140 +1219,15 @@ export class Brainy implements BrainyInterface {
this._generationStampingActive = true
}
- // LOG-AUTHORITY SWITCH (checked at open only). A STORED artifact
- // always wins: an already-flipped brain runs durable-at-ack; an
- // explicitly-recorded tree posture is honored. With NO artifact, the
- // 10.0.0 FLEET DEFAULT is ADOPT-AT-OPEN (config logAuthority:
- // 'adopt'): the verification oracle gates the flip — curable
- // divergences are baseline-backfilled, the brain flips ONLY on green,
- // and a brain that cannot go green STAYS tree-authoritative LOUDLY
- // with the refusal recorded (cheap subsequent opens; an operator
- // re-runs adoptLogAuthority() after fixing the divergence).
- // 'defer' is the documented opt-out: no automatic adoption.
- if (!this.isReadOnly) {
- const storedArtifact = await this.storage
- .readRawObject(LOG_AUTHORITY_PATH)
- .catch(() => null)
- const authority = await step(
- 'read-log-authority',
- 'reading the stored storage-authority artifact',
- () => readLogAuthority(this.storage)
- )
- this._logAuthority = authority
- if (authority.authority === 'log') {
- this.generationStore.setLogDurability('at-ack')
- prodLog.info('[Brainy] storage authority: generation log (durable-at-ack enabled)')
- } else if (
- storedArtifact === null &&
- this.config.logAuthority === 'adopt' &&
- this.generationStore.getFactLog() !== null
- ) {
- try {
- await step(
- 'adopt-log-authority',
- 'the adoption oracle: verifying the log against canonical before flipping this ' +
- 'brain to durable-at-ack, and backfilling any curable divergence',
- () => this.adoptLogAuthority()
- )
- prodLog.info(
- '[Brainy] storage authority adopted at open: generation log ' +
- '(fleet default; oracle green; durable-at-ack enabled)'
- )
- } catch (err) {
- // The guarded ruling: a brain that cannot verify STAYS tree,
- // loudly, with the refusal recorded so subsequent opens are
- // cheap. Never a silent half-state; never a failed open.
- const reason = (err as Error).message
- prodLog.warn(
- `[Brainy] log-authority adoption REFUSED at open — this brain stays ` +
- `tree-authoritative until an operator resolves the divergence and ` +
- `re-runs adoptLogAuthority(). Reason: ${reason}`
- )
- try {
- const refusal: LogAuthorityRecord = {
- authority: 'tree',
- adoptRefusal: { at: Date.now(), reason: reason.slice(0, 500) }
- }
- await this.storage.writeRawObject(LOG_AUTHORITY_PATH, refusal)
- this._logAuthority = refusal
- } catch {
- // Unrecordable refusal = the next open retries the oracle —
- // the conservative outcome.
- }
- }
- }
- }
-
- // MT5 crash recovery — REPLAY, NOT LISTING: the pending-embed markers
- // live IN the generation log (embed.pending rides the deferred write's
- // own fact; embed.landed rides the landing commit), so recovery folds
- // the log's marker records back into the in-memory set — after the
- // one-time bridge migrates any sidecar files a pre-log build left
- // behind — and resumes the worker in the background. A crash between
- // a deferred write's ack and its background embed DELAYED a vector;
- // this is where it lands.
- if (!this.isReadOnly) {
- try {
- await step(
- 'bridge-pending-embed-sidecars',
- 'migrating any pre-log deferred-embed marker files into the generation log',
- () => this.bridgeLegacyPendingEmbedSidecars()
- )
- await step(
- 'recover-pending-embeds',
- 'folding the generation log\'s deferred-embed markers back into the pending set',
- () => this.recoverPendingEmbedsFromLog()
- )
- if (this._pendingEmbedIds.size > 0) {
- prodLog.info(
- `[Brainy] ${this._pendingEmbedIds.size} deferred embed(s) pending from a previous ` +
- `session — resuming in the background`
- )
- const t = setTimeout(() => this.kickEmbedWorker(), 0)
- ;(t as { unref?: () => void }).unref?.()
- }
- } catch (err) {
- prodLog.warn(
- `[Brainy] pending-embed recovery failed: ${(err as Error).message} — ` +
- `the log's markers remain durable; recovery retries next open`
- )
- }
- }
-
- // PHASE 4 of 5 — "VFS bootstrap": shutdown-hook registration, blob
- // storage init, the provider-summary log, flipping `initialized`,
- // the migration-lock wait, VFS construction+init, flipping generation
- // stamping active, the log-authority adopt/oracle check, and
- // pending-embed crash recovery.
- markPhase('vfs-bootstrap')
-
- // Eager embedding initialization — BACKGROUND WARM (open-path fix).
+ // Eager embedding initialization.
//
- // Adaptive default (8.0): the WASM embedding engine eagerly WARMS
+ // Adaptive default (8.0): the WASM embedding engine eagerly initializes
// during init() WHENEVER it is the active embedder — i.e. no native
// 'embeddings' provider has taken over — and the instance is a writer
// (not reader-mode) outside of unit tests. The WASM module (≈93MB with
- // the embedded model) takes 90-140s to compile on throttled CPUs.
- //
- // Historically this AWAITED `embeddingManager.init()` INLINE, so every
- // writer's open() blocked on the compile — N concurrent opens all
- // queued on the ONE process-global singleton (an ~80x contention
- // multiplier measured in a production restart storm: 90,017ms busy vs
- // 1,117ms quiet). The engine only needs to be ready before the FIRST
- // REAL embed() call, not before init() returns, so this now only
- // STARTS the warm and moves on — init() never waits for it.
- //
- // No double-await needed for correctness: `this.embed()` (~line 15420)
- // delegates to `this.embedder`, which for the default engine is
- // `embeddingManager.getEmbeddingFunction()` → `embeddingManager.embed()`
- // (src/embeddings/EmbeddingManager.ts). That method calls `await
- // this.init()` FIRST, and `init()` itself serializes every concurrent
- // caller onto ONE shared `globalInitPromise` — so the first real
- // embed() automatically waits for whichever finishes first: this
- // background warm (if still running) or a fresh init() (if the warm
- // hasn't reached this code yet, e.g. `eagerEmbeddings: false`).
- // Verified by reading both call sites; `_embeddingWarmPromise` below
- // is stored for observability only, never re-awaited by embed().
+ // the embedded model) takes 90-140s to compile on throttled CPUs; paying
+ // that during boot rather than on the first embed()-driven call is the
+ // right default for the overwhelmingly common single-process server.
//
// Skipped automatically when:
// - a native 'embeddings' provider is registered (it owns embeddings;
@@ -1887,8 +1235,8 @@ export class Brainy implements BrainyInterface {
// - reader-mode (readers don't embed — they query existing vectors),
// - unit-test mode (tests must stay fast and use the mock embedder).
//
- // `eagerEmbeddings: false` keeps meaning "no warm at all" — fully lazy,
- // the first embed() call pays the full cost inline, same as before.
+ // `eagerEmbeddings: false` is the explicit override to force lazy init
+ // (first-embed) even when this instance is the active embedder.
const isUnitTestMode = isDeterministicEmbedMode()
const eager = this.config.eagerEmbeddings ?? true
if (
@@ -1897,45 +1245,9 @@ export class Brainy implements BrainyInterface {
this.config.mode !== 'reader' &&
!isUnitTestMode
) {
- const warmStart = Date.now()
- console.log('Background embedding-engine warm started (init() does not wait for it)...')
- this._embeddingWarmPromise = embeddingManager
- .init()
- .then(() => {
- prodLog.info(
- `[Brainy] background embedding-engine warm complete in ${Date.now() - warmStart}ms`
- )
- })
- .catch((err) => {
- // Loud, never silent: a warm that fails to compile must be
- // heard NOW, not discovered as a mystery latency spike on
- // whichever request happens to trigger the first real embed().
- // That first embed() call still retries init() itself (the
- // singleton promise contract above) and surfaces its own typed
- // error to its caller — this is the immediate, background echo.
- prodLog.warn(
- `[Brainy] background embedding-engine warm FAILED: ` +
- `${(err as Error).message} — the first embed() call will retry ` +
- `initialization and surface the error there`
- )
- })
- }
-
- // PHASE 5 of 5 — "embedding-warm-started": just the synchronous cost
- // of kicking off the background warm above (the warm's own compile
- // time is NOT included — that's the whole point of backgrounding it).
- markPhase('embedding-warm-started')
- {
- const totalOpenMs = Date.now() - initStart
- if (totalOpenMs > 2000) {
- const phaseList = Object.entries(phaseTimingsMs)
- .map(([name, ms]) => `${name}=${ms}ms`)
- .join(', ')
- prodLog.narrate(
- `[Brainy] slow open: ${totalOpenMs}ms total (${phaseList}) — see the ` +
- `phase breakdown above to find which one to investigate first`
- )
- }
+ console.log('Eager embedding initialization enabled...')
+ await embeddingManager.init()
+ console.log('Embedding engine ready')
}
// Integration Hub initialization
@@ -1961,17 +1273,6 @@ export class Brainy implements BrainyInterface {
})
}
- // Eager index warm (operator opt-in, `warmOnOpen: true`). Runs AFTER
- // every step above — index construction, crash recovery, migrations,
- // VFS bootstrap — never in place of any of it, so `warm()` always
- // operates on a fully-initialized brain. Blocking BY DESIGN: the
- // operator traded a longer startup for a first-request that runs at
- // steady-state cost instead of paying demand-load latency on the
- // critical path. See the `warmOnOpen` JSDoc in brainy.types.ts.
- if (this.config.warmOnOpen) {
- await this.warm()
- }
-
// Resolve ready Promise - consumers awaiting brain.ready will now proceed
if (this._readyResolve) {
this._readyResolve()
@@ -1981,22 +1282,7 @@ export class Brainy implements BrainyInterface {
if (this._readyReject) {
this._readyReject(error instanceof Error ? error : new Error(String(error)))
}
- // Machine-readable init failures pass through UNWRAPPED — the writer-lock
- // conflict documents an err.code/err.lockInfo contract ("callers detect
- // this case via err.code"), and wrapping in a fresh Error silently
- // stripped both, leaving consumers only a message to regex against.
- if (error instanceof Error && (error as Error & { code?: string }).code === 'BRAINY_WRITER_LOCKED') {
- throw error
- }
- // Wrap with the original as `cause` so the originating frame (a plugin's
- // own file:line, e.g. a provider boot failure) survives to the caller's
- // log — a plain string interpolation discards both stack and cause.
- const message = error instanceof Error ? error.message : String(error)
- throw new Error(`Failed to initialize Brainy: ${message}`, { cause: error })
- } finally {
- // The open is over — succeeded or failed. Stop the heartbeat here so a
- // failed init never leaves a timer narrating a phase nobody is running.
- clearInterval(openHeartbeat)
+ throw new Error(`Failed to initialize Brainy: ${error}`)
}
}
@@ -2014,112 +1300,76 @@ export class Brainy implements BrainyInterface {
* NOTE: Registers globally (once for all instances) to avoid MaxListenersExceededWarning
*/
private registerShutdownHooks(): void {
- /**
- * The signal-path shutdown. THREE LAWS, each written by a production
- * shutdown that looked clean and wasn't:
- *
- * 1. PER-INSTANCE ISOLATION. This used to be one `try` around a loop over
- * every open brain: the first instance whose flush rejected aborted the
- * loop, so every remaining brain kept its writer lock and its unwritten
- * markers — and the process still exited 0. A pool of brains failed in
- * a batch, not one at a time.
- * 2. THE MARKER IS PART OF SHUTDOWN. Flushing the indexes without closing
- * the generation store leaves the clean-shutdown marker unwritten, so
- * the NEXT open reads the store as crashed and folds the whole
- * generation log — measured in tens of seconds on a real store, paid on
- * every restart, after a shutdown the operator saw exit 0.
- * 3. THE LOCK IS ALWAYS GIVEN UP. In a `finally`, per instance: a process
- * on its way out holds nothing.
- */
const flushOnShutdown = async () => {
console.log('Shutdown signal received - flushing pending data...')
- let flushedCount = 0
- let failedCount = 0
- // Snapshot: close() splices Brainy.instances while we iterate.
- for (const instance of [...Brainy.instances]) {
- if (!instance.initialized) continue
- try {
- // Flush all buffered data (parallel across components, this brain only).
- await Promise.all([
- (async () => {
- if (instance.storage && typeof instance.storage.flushCounts === 'function') {
- await instance.storage.flushCounts()
- }
- })(),
- (async () => {
- if (instance.metadataIndex && typeof instance.metadataIndex.flush === 'function') {
- await instance.metadataIndex.flush()
- }
- })(),
- (async () => {
- if (instance.graphIndex && typeof instance.graphIndex.flush === 'function') {
- await instance.graphIndex.flush()
- }
- })(),
- (async () => {
- if (instance.index && typeof instance.index.flush === 'function') {
- await instance.index.flush()
- }
- })()
- ])
-
- // Close the generation store: persists the counter, advances the
- // fold checkpoint, and stamps the clean-shutdown marker LAST — the
- // one step that decides whether the next open adopts or folds. Law 2.
- if (instance.generationStore && !instance.isReadOnly) {
- await instance.generationStore.close()
- }
-
- // Close components to stop timers that would prevent clean process exit
- await Promise.all([
- (async () => {
- if (instance.graphIndex && typeof instance.graphIndex.close === 'function') {
- await instance.graphIndex.close()
- }
- })(),
- (async () => {
- const index = instance.index as JsHnswVectorIndex & VectorIndexOptionalHooks
- if (index && typeof index.close === 'function') {
- await index.close()
- }
- })(),
- (async () => {
- const metadataIndex = instance.metadataIndex as MetadataIndexManager & MetadataIndexOptionalHooks
- if (metadataIndex && typeof metadataIndex.close === 'function') {
- await metadataIndex.close()
- }
- })()
- ])
- flushedCount++
- } catch (error) {
- failedCount++
- console.error('Failed to flush one Brainy instance on shutdown:', error)
- } finally {
- // Law 3 — the lock and the watcher go regardless.
- try {
- if (instance.storage && typeof instance.storage.stopFlushRequestWatcher === 'function') {
- instance.storage.stopFlushRequestWatcher()
- }
- } catch (error) {
- console.error('Failed to stop the flush-request watcher on shutdown:', error)
- }
- try {
- if (instance.storage && typeof instance.storage.releaseWriterLock === 'function') {
- await instance.storage.releaseWriterLock()
- }
- } catch (error) {
- console.error('Failed to release the writer lock on shutdown:', error)
+ try {
+ let flushedCount = 0
+ for (const instance of Brainy.instances) {
+ if (instance.initialized) {
+ // Flush all buffered data, then close to release resources (timers, handles)
+ await Promise.all([
+ (async () => {
+ if (instance.storage && typeof instance.storage.flushCounts === 'function') {
+ await instance.storage.flushCounts()
+ }
+ })(),
+ (async () => {
+ if (instance.metadataIndex && typeof instance.metadataIndex.flush === 'function') {
+ await instance.metadataIndex.flush()
+ }
+ })(),
+ (async () => {
+ if (instance.graphIndex && typeof instance.graphIndex.flush === 'function') {
+ await instance.graphIndex.flush()
+ }
+ })(),
+ (async () => {
+ if (instance.index && typeof instance.index.flush === 'function') {
+ await instance.index.flush()
+ }
+ })()
+ ])
+ // Close components to stop timers that would prevent clean process exit
+ await Promise.all([
+ (async () => {
+ if (instance.graphIndex && typeof instance.graphIndex.close === 'function') {
+ await instance.graphIndex.close()
+ }
+ })(),
+ (async () => {
+ const index = instance.index as JsHnswVectorIndex & VectorIndexOptionalHooks
+ if (index && typeof index.close === 'function') {
+ await index.close()
+ }
+ })(),
+ (async () => {
+ const metadataIndex = instance.metadataIndex as MetadataIndexManager & MetadataIndexOptionalHooks
+ if (metadataIndex && typeof metadataIndex.close === 'function') {
+ await metadataIndex.close()
+ }
+ })(),
+ // Release the writer lock so a successor process can take over.
+ // No-op for readers and for backends without locking.
+ (async () => {
+ if (instance.storage && typeof instance.storage.releaseWriterLock === 'function') {
+ await instance.storage.releaseWriterLock()
+ }
+ })(),
+ // Stop the flush-request watcher to release its interval timer.
+ (async () => {
+ if (instance.storage && typeof instance.storage.stopFlushRequestWatcher === 'function') {
+ instance.storage.stopFlushRequestWatcher()
+ }
+ })(),
+ ])
+ flushedCount++
}
}
- }
- if (flushedCount > 0) {
- console.log(`Flushed successfully (${flushedCount} instance${flushedCount > 1 ? 's' : ''})`)
- }
- if (failedCount > 0) {
- console.error(
- `${failedCount} Brainy instance${failedCount > 1 ? 's' : ''} did not complete shutdown — ` +
- `their writer locks were released, but their next open will run crash recovery.`
- )
+ if (flushedCount > 0) {
+ console.log(`Flushed successfully (${flushedCount} instance${flushedCount > 1 ? 's' : ''})`)
+ }
+ } catch (error) {
+ console.error('Failed to flush on shutdown:', error)
}
}
@@ -2127,32 +1377,13 @@ export class Brainy implements BrainyInterface {
// kept as statics so the last live instance's close() can deregister them
// — the signal handles they hold are ref'd and would otherwise keep the
// process alive forever after every brain is closed.
- /**
- * Exit the process ONLY when Brainy is the sole handler for this signal.
- *
- * Registering a signal listener suppresses Node's default terminate
- * behaviour, so a library that attaches one must either exit or be sure
- * someone else will. Brainy attaching one AND exiting was the wrong half
- * of that choice for every host application with its own graceful
- * shutdown: both handlers run concurrently, and whichever finishes first
- * wins — a library flush finishing before an application's close()
- * terminated that close mid-flight, at exit code 0, with locks and
- * markers unwritten. When the host has its own handler (listener count
- * above our own), the host owns the exit; Brainy only makes its data
- * durable and steps aside.
- */
- const exitIfSoleShutdownOwner = (signal: 'SIGTERM' | 'SIGINT'): void => {
- if (process.listenerCount(signal) <= 1) {
- process.exit(0)
- }
- }
Brainy.sigtermListener = async () => {
await flushOnShutdown()
- exitIfSoleShutdownOwner('SIGTERM')
+ process.exit(0)
}
Brainy.sigintListener = async () => {
await flushOnShutdown()
- exitIfSoleShutdownOwner('SIGINT')
+ process.exit(0)
}
Brainy.beforeExitListener = async () => {
// Self-deregister FIRST: Node re-emits 'beforeExit' after every event-
@@ -2203,10 +1434,7 @@ export class Brainy implements BrainyInterface {
* re-initializing — using a closed Brainy is a consumer bug, not a lazy-init
* opportunity.
*/
- private async ensureInitialized(opts?: {
- bypassMigrationLock?: boolean
- needs?: IndexFamily[]
- }): Promise {
+ private async ensureInitialized(opts?: { bypassMigrationLock?: boolean }): Promise {
if (this.closed) {
throw new Error('Brainy instance is not initialized: it was closed via close(). Create a new instance.')
}
@@ -2216,17 +1444,12 @@ export class Brainy implements BrainyInterface {
// Coordinated migration LOCK (#18): every data-plane read and write funnels
// through here, so this is the single choke point that holds operations while
// a native provider runs its one-time 7.x → 8.0 rebuild-from-canonical — no
- // op touches a half-built index. The gate is FAMILY-SCOPED: `needs` names the
- // derived-index families this operation actually consults, so a read served
- // entirely from canonical storage (`needs: []`) or from a healthy family
- // never blocks on an UNRELATED family's migration. `needs` omitted = the
- // conservative whole-brain wait (writes, and any read not yet classified).
- // Observability (`health`/`checkHealth`) and the lock-clearing path
- // (`stampBrainFormat`, which does not route through here) opt out entirely so
- // an operator can always watch progress and a native provider can stamp.
+ // op touches a half-built index. Observability (`health`/`checkHealth`) and
+ // the lock-clearing path (`stampBrainFormat`, which does not route through
+ // here) opt out so an operator can always watch progress and cor can stamp.
// A brain that never migrates pays one boolean check (see awaitMigrationLock).
if (!opts?.bypassMigrationLock) {
- await this.awaitMigrationLock(opts?.needs)
+ await this.awaitMigrationLock()
}
}
@@ -2391,485 +1614,12 @@ export class Brainy implements BrainyInterface {
* deletes — the before-image + per-id-chain set.
* @param run - The single-op's existing operation batch builder (the
* `tx => {…}` body previously passed straight to `executeTransaction`).
- * @param precommit - Optional CAS precondition, run under the commit mutex.
- * @param pendingEvents - Change-feed events to stamp and emit post-commit.
- * @param records - Optional v2 marker records (e.g. the deferred-embedding
- * lifecycle markers) riding this write's commit fact — same generation,
- * one atomic append. Refused on generation-less bootstrap writes.
*/
- /**
- * Storage-root-relative prefix of the RETIRED sidecar pending-embed marker
- * files (pre-log builds persisted one raw object per pending embed here).
- * The markers live IN the generation log now (`embed.pending` /
- * `embed.landed` records); this prefix survives ONLY for the one-time
- * migration bridge ({@link bridgeLegacyPendingEmbedSidecars}) — no other
- * code path writes, lists, or deletes it.
- */
- private static readonly PENDING_EMBED_PREFIX = '_system/pending_embeds/'
-
- /**
- * @description Mark a deferred embed pending (MT5): the id joins the
- * in-memory fast-path set and the returned `embed.pending` record is
- * threaded onto the deferred write's OWN commit fact — same generation,
- * same atomic append, and (in at-ack log durability) the same covering
- * fsync as the write itself. The marker can never be orphaned from its
- * write nor the write from its marker: a failed commit appends no fact,
- * so no durable marker exists either (the in-memory entry is harmless
- * and reaped by the worker). Recovery folds the marker back out of the
- * log at open ({@link recoverPendingEmbedsFromLog}).
- */
- private enqueuePendingEmbed(id: string): FactMarkerRecord {
- this._pendingEmbedIds.add(id)
- return { type: 'embed.pending', id, enqueuedAt: Date.now() }
- }
-
- /**
- * @description Clear a pending embed from the in-memory set. The DURABLE
- * clear is the `embed.landed` record riding the landing commit's own fact
- * (or, for a row deleted before its embed landed, the row's tombstone
- * fact) — the recovery fold consumes those; nothing here touches storage.
- * One honest residue: a pending row whose entity still exists but carries
- * no data is reaped in memory only, so it re-folds at the next open and
- * is re-reaped there — a bounded no-op, never a lost vector.
- */
- private clearPendingEmbed(id: string): void {
- this._pendingEmbedIds.delete(id)
- }
-
- /**
- * @description Rebuild the pending-embed set by REPLAYING the generation
- * log's marker records (recovery = replay, not listing): `embed.pending`
- * arms an id, `embed.landed` disarms it, and a noun tombstone disarms it
- * too (a row deleted before its embed landed owes no vector). What
- * survives the fold is exactly the set of acknowledged deferred writes
- * whose vectors have not landed.
- *
- * BOUND (honest): no durable low-water mark exists for the earliest
- * unconsumed pending, so the fold scans the log's committed facts from
- * generation 1 — a sequential read of the log at open, O(log bytes).
- * It is SKIPPED WHOLESALE when the log has never had a v2 tail
- * ({@link FactLog.hasV2History} — v1 facts cannot carry marker records),
- * so pre-cutover brains pay nothing; on a mixed log the scan still reads
- * the v1 segments (a segment's format is only known from its bytes) but
- * they fold to nothing, so the DECODE cost is bounded by v2 history.
- * Storage without a fact log hosts no durable markers at all — the
- * pending set is session-local there, matching that storage's overall
- * durability posture.
- */
- private async recoverPendingEmbedsFromLog(): Promise {
- const log = this.generationStore.getFactLog()
- if (!log || !log.hasV2History()) return
- const scan = log.scanFacts({ fromGeneration: 1 })
- for await (const batch of scan.batches()) {
- for (const fact of batch.facts) {
- for (const record of fact.records ?? []) {
- if (record.type === 'embed.pending') {
- this._pendingEmbedIds.add(record.id)
- } else if (record.type === 'embed.landed') {
- this._pendingEmbedIds.delete(record.id)
- }
- }
- for (const op of fact.ops) {
- if (op.kind === 'noun' && op.record === null) {
- this._pendingEmbedIds.delete(op.id)
- }
- }
- }
- }
- }
-
- /**
- * @description ONE-TIME LEGACY BRIDGE: a brain that deferred embeds under
- * a pre-log build persisted one sidecar marker file per pending embed
- * under {@link PENDING_EMBED_PREFIX}. At open, fold those ids into the
- * pending set AND migrate them: commit ONE fact carrying their
- * `embed.pending` records (the log is the markers' durable home now),
- * then delete the sidecar files — in that order, so a crash between the
- * two re-runs the bridge instead of losing a marker (a re-migrated
- * duplicate folds idempotently; at worst an already-landed embed re-runs
- * once — idempotent, never lost). Narrated loudly. Storage without a
- * fact log keeps its sidecars in place (there is no log to migrate into)
- * and folds them into memory only, exactly as loud.
- */
- private async bridgeLegacyPendingEmbedSidecars(): Promise {
- const markerPaths = await this.storage.listRawObjects(Brainy.PENDING_EMBED_PREFIX)
- if (markerPaths.length === 0) return
- const ids: string[] = []
- for (const path of markerPaths) {
- const id = path.slice(path.lastIndexOf('/') + 1)
- if (id) ids.push(id)
- }
- if (ids.length === 0) return
- for (const id of ids) this._pendingEmbedIds.add(id)
- if (!this.generationStore.getFactLog()) {
- prodLog.warn(
- `[Brainy] ${ids.length} legacy pending-embed sidecar marker(s) found, but this ` +
- `storage hosts no fact log to migrate them into — folded into memory; the ` +
- `sidecar files remain the durable recovery source on this configuration`
- )
- return
- }
- const enqueuedAt = Date.now()
- const markers: FactMarkerRecord[] = ids.map((id) => ({
- type: 'embed.pending',
- id,
- enqueuedAt
- }))
- // One migration commit: a zero-op fact carrying every legacy marker
- // (empty-ops facts are legal; the records leg makes this one visible).
- await this.generationStore.commitSingleOp({
- touched: {},
- records: markers,
- execute: async () => {}
- })
- for (const id of ids) {
- await this.storage.deleteRawObject(`${Brainy.PENDING_EMBED_PREFIX}${id}`).catch(() => {})
- }
- prodLog.info(
- `[Brainy] migrated ${ids.length} legacy pending-embed sidecar marker(s) into the ` +
- `generation log and removed the sidecar files (one-time bridge)`
- )
- }
-
- /**
- * @description Start (or skip into) the ONE deferred-embedding worker.
- * Never awaited by write paths; failures are LOUD and markers survive for
- * the next kick (next deferred write, or the next open's recovery).
- */
- private kickEmbedWorker(): void {
- if (this._embedWorkerFlight || this._pendingEmbedIds.size === 0 || this.isReadOnly) return
- this._embedWorkerFlight = this.runEmbedWorker()
- .catch((err) => {
- prodLog.error(
- `[Brainy] deferred-embed worker failed: ${(err as Error).message} — ` +
- `markers retained; retries at the next deferred write or open`
- )
- })
- .finally(() => {
- this._embedWorkerFlight = null
- if (this._pendingEmbedIds.size > 0) {
- // New arrivals during the run: schedule (never recurse) the next pass.
- const t = setTimeout(() => this.kickEmbedWorker(), 0)
- ;(t as { unref?: () => void }).unref?.()
- }
- })
- }
-
- /**
- * @description Drain the pending-embed set: embed each row's CURRENT data
- * (a row updated again before its turn embeds the latest content — the
- * marker set is idempotent per id) and swap the vector in ATOMICALLY
- * (ReplaceInVectorIndex → the in-place update; the row is never absent
- * from search). Orphans (row deleted, or no data) reap their markers.
- */
- private async runEmbedWorker(): Promise {
- const batch = Array.from(this._pendingEmbedIds)
- for (const id of batch) {
- try {
- const entity = await this.get(id, { includeVectors: true })
- if (!entity || entity.data === undefined || entity.data === null) {
- // Orphan reap: a deleted row's tombstone fact durably disarms the
- // marker at the next recovery fold; a data-less-but-present row
- // (edge case) re-folds and re-reaps — bounded, never a lost vector.
- this.clearPendingEmbed(id)
- continue
- }
- // Hang guard: a wedged embedder must not block every later pending
- // embed forever — time out LOUDLY, keep the marker, move on. (A
- // failure is retryable; an unbounded silent wait is the outlawed
- // shape.)
- const newVector = await Promise.race([
- this.embed(entity.data),
- new Promise((_, reject) => {
- const t = setTimeout(
- () => reject(new Error('deferred embed timed out after 60s')),
- 60_000
- )
- ;(t as { unref?: () => void }).unref?.()
- })
- ])
- if (!this.dimensions) {
- this.dimensions = newVector.length
- } else if (newVector.length !== this.dimensions) {
- throw new Error(
- `deferred embed produced ${newVector.length} dimensions, store expects ${this.dimensions}`
- )
- }
- const oldVector = (entity.vector as number[] | undefined) ?? []
- // The landing commit's fact carries the embed.landed record (vector
- // inline, per the v2 format) alongside the row's after-image — the
- // durable "this pending is consumed" that recovery's fold reads.
- await this.persistSingleOp(
- { nouns: [id] },
- async (tx) => {
- tx.addOperation(
- new SaveNounOperation(this.storage, {
- id,
- vector: newVector,
- connections: new Map(),
- level: 0
- })
- )
- tx.addOperation(
- new ReplaceInVectorIndexOperation(this.index, id, oldVector, newVector, this.indexWriteGeneration)
- )
- },
- undefined,
- undefined,
- [{ type: 'embed.landed', id, vector: newVector }],
- 'system:embed-landing'
- )
- // Vectored-noun ledger: the landing commit above carries a vector
- // write with NO accompanying metadata operation, so the
- // saveNounMetadata(..., hasVector) seam never fires for it — the
- // narrow storage hook is the only seam left. `oldVector.length===0`
- // (already known for free from the pre-embed read above) proves this
- // is a GENUINE first landing, not a re-embed of an already-vectored
- // row (e.g. a deferred update() on a row that already had a real
- // vector) — the latter must never double-count.
- if (oldVector.length === 0) {
- await this.storage.noteVectorLanded?.(id)
- }
- this.clearPendingEmbed(id)
- } catch (err) {
- prodLog.warn(
- `[Brainy] deferred embed for ${id} failed: ${(err as Error).message} — marker retained for retry`
- )
- }
- }
- }
-
- /**
- * @description The deferred-embedding BARRIER: resolves when every pending
- * embed has landed (vector searchable) or been reaped. The eventual-
- * vector-index contract's awaitable edge — tests and "must be searchable
- * before I proceed" callers use this; nothing else ever needs to wait.
- */
- public async awaitPendingEmbeds(): Promise {
- while (this._pendingEmbedIds.size > 0 || this._embedWorkerFlight) {
- this.kickEmbedWorker()
- await (this._embedWorkerFlight ?? Promise.resolve())
- }
- }
-
- /** The deferred-embedding backlog size (also on getIndexStatus().pendingEmbeds). */
- public pendingEmbedCount(): number {
- return this._pendingEmbedIds.size
- }
-
- /**
- * THE READ BARRIER: wait until a projection — or every projection — has
- * caught up to the CURRENT committed head, so a write-then-recall caller
- * has ONE honest await instead of a sleep-and-hope.
- *
- * Legs:
- * - `'semantic'` — waits for the deferred-embedding backlog to drain
- * (delegates to {@link awaitPendingEmbeds}, which keeps working
- * unchanged as this leg's engine). After it resolves, every previously
- * acknowledged write is vector-searchable.
- * - `'metadata'` / `'graph'` / `'aggregation'` — resolve IMMEDIATELY by
- * design today: these projections are updated inside the write path, so
- * by the time a write's promise resolves they already reflect it. Their
- * asynchrony arrives with the log-authority read path; the door's shape
- * freezes now so callers written against it keep working unchanged when
- * those legs become real waits.
- * - no argument — every projection at the head; today that reduces to the
- * semantic drain (the only asynchronous projection in the current
- * architecture).
- *
- * `opts.generation`: resolve as soon as the projection's watermark has
- * reached that committed generation. The pending-embed set carries no
- * generation stamps today, so the refinement is conservative — an empty
- * backlog resolves immediately (the watermark is at the head, hence ≥ any
- * committed generation); a non-empty backlog waits for the full drain, a
- * SUPERSET of the requested wait, never a partial one.
- *
- * `opts.timeoutMs`: on expiry the promise REJECTS with
- * {@link WaitForIndexedTimeoutError} — typed, carrying the leg and the
- * still-pending embed count, and naming the gauge to check
- * (`getIndexStatus().projections.semantic.pendingEmbeds`). Never a silent
- * partial wait: a timeout means the projection has NOT caught up.
- *
- * @example Write, then semantically recall — no polling, no sleeps
- * ```typescript
- * const id = await brain.add({
- * data: 'quarterly revenue narrative',
- * type: NounType.Document,
- * deferEmbedding: true,
- * metadata: { kind: 'report' }
- * })
- * await brain.waitForIndexed('semantic') // the barrier: vector landed + indexed
- * const hits = await brain.find({ query: 'revenue report', searchMode: 'semantic' })
- * // `id` is eligible to appear in `hits` — the recall is honest, not lucky.
- * ```
- *
- * @param path - The projection to wait on; omit to wait on all of them.
- * @param opts - Optional `generation` watermark target and `timeoutMs` bound.
- * @throws {WaitForIndexedTimeoutError} When `timeoutMs` expires before the
- * projection catches up.
- */
- public async waitForIndexed(
- path?: IndexedProjectionPath,
- opts?: WaitForIndexedOptions
- ): Promise {
- await this.ensureInitialized()
-
- // Synchronous projections: updated inside the write path today, so an
- // acknowledged write is already reflected — resolve immediately BY
- // DESIGN (honest, not a stub). When the log-authority read path makes
- // these legs asynchronous, only this body changes; the door's shape is
- // frozen now.
- if (path === 'metadata' || path === 'graph' || path === 'aggregation') {
- return
- }
-
- // 'semantic' — or no-arg, which today reduces to it: the deferred-embed
- // backlog is the only asynchronous projection in the current
- // architecture.
-
- // Generation refinement (conservative — see JSDoc): an empty backlog
- // means the semantic watermark is at the head, hence ≥ any committed G.
- if (opts?.generation !== undefined && this._pendingEmbedIds.size === 0) {
- return
- }
-
- const timeoutMs = opts?.timeoutMs
- const drained = this.awaitPendingEmbeds()
- if (timeoutMs === undefined) {
- return drained
- }
-
- // Typed timeout: reject LOUDLY with the leg + the live backlog gauge.
- // (`drained` never rejects — the worker catches its own failures — so
- // abandoning it on timeout cannot leak an unhandled rejection; the
- // backlog keeps draining in the background.)
- let timer: ReturnType | undefined
- try {
- await Promise.race([
- drained,
- new Promise((_, reject) => {
- timer = setTimeout(
- () =>
- reject(
- new WaitForIndexedTimeoutError(
- path ?? 'all',
- timeoutMs,
- this._pendingEmbedIds.size
- )
- ),
- timeoutMs
- )
- ;(timer as { unref?: () => void }).unref?.()
- })
- ])
- } finally {
- if (timer !== undefined) clearTimeout(timer)
- }
- }
-
- /**
- * @description The write-side persistence trigger (policy `'auto'`): count
- * the committed write, kick a single-flight BACKGROUND flush when the
- * write-count or interval threshold is crossed, and (re)arm the idle
- * timer. Never awaited by the write path — the ack is already durable at
- * the canonical layer; this schedules DERIVED-state persistence on the
- * engine's own cadence (callers never call flush() in hot paths).
- */
- private noteWriteForPersistence(): void {
- // THE DIRTY WITNESS. Set on every committed write — both commit paths
- // (single-op and transaction) end here, and the deferred-embed worker
- // lands its vectors through the single-op path — BEFORE the policy check,
- // so a `'manual'` consumer's explicit flush() is never skipped either.
- // Cleared by a flush that actually runs; see flush().
- this._dirtySinceLastFlush = true
- const cfg = this.config.persistence
- if (this.isReadOnly || cfg?.policy === 'manual') return
- this._persistDirtyWrites++
- const every = cfg?.flushEveryWrites ?? 512
- const intervalMs = cfg?.flushIntervalMs ?? 30_000
- const idleMs = cfg?.flushOnIdleMs ?? 2_000
-
- if (
- this._persistDirtyWrites >= every ||
- Date.now() - this._persistLastFlushAt >= intervalMs
- ) {
- this.kickBackgroundFlush('threshold')
- }
-
- if (this._persistIdleTimer) clearTimeout(this._persistIdleTimer)
- this.armIdleFlushTimer(idleMs, intervalMs)
- }
-
- /**
- * @description Arm the idle-flush timer — DEBOUNCED UNDER LOAD. The idle
- * trigger exists to make a QUIET system durable fast; it must never add
- * flush pressure to a BUSY one. When individual writes are slower than
- * the idle window (a contended disk), every inter-write gap looks like
- * "idle" and would fire a full flush per write — a measured 15-flush
- * amplifier during 100 contended adds on a production-shaped box. The
- * law: an idle fire landing within `intervalMs` of the last flush DEFERS
- * (re-arms for the remaining interval) rather than flushing — deferred,
- * never dropped, so a lone write on a then-quiet system still persists at
- * the interval boundary without any further write arriving; a genuinely
- * quiet system (last flush long past) flushes on idle exactly as before.
- */
- private armIdleFlushTimer(idleMs: number, intervalMs: number, delayMs = idleMs): void {
- // The idle-fire spacing floor: 10× the CONFIGURED idle window, capped by
- // the interval — always derived from idleMs, never from a deferred
- // re-arm delay (recomputing from the delay compounds into runaway
- // deferral). Scales with intent — a caller configuring a tiny idle
- // window gets fast idle-driven durability (small floor); default config
- // (2s idle / 30s interval) gets a 20s floor, capping the contended-disk
- // shape at ~1 idle flush per 20s instead of one per inter-write gap.
- const floorMs = Math.min(intervalMs, idleMs * 10)
- const timer = setTimeout(() => {
- this._persistIdleTimer = null
- if (this._persistDirtyWrites === 0) return
- const sinceFlush = Date.now() - this._persistLastFlushAt
- if (sinceFlush >= floorMs) {
- this.kickBackgroundFlush('idle')
- } else {
- // Deferred, never dropped: land exactly at the floor boundary.
- this.armIdleFlushTimer(idleMs, intervalMs, Math.max(idleMs, floorMs - sinceFlush))
- }
- }, delayMs)
- // Never hold the process open for a cadence timer.
- ;(timer as { unref?: () => void }).unref?.()
- this._persistIdleTimer = timer
- }
-
- /**
- * @description Start (or join) the ONE background flush. The dirty counter
- * resets at kick time so writes landing during the flush re-accumulate
- * toward the next trigger. A failure is LOUD and leaves the writes counted
- * again — silence is not an option, and neither is a retry storm (the next
- * trigger re-attempts).
- */
- private kickBackgroundFlush(reason: 'threshold' | 'idle'): void {
- if (this._persistBackgroundFlight) return
- const counted = this._persistDirtyWrites
- this._persistDirtyWrites = 0
- this._persistLastFlushAt = Date.now()
- this._persistBackgroundFlight = this.flush()
- .catch((err) => {
- this._persistDirtyWrites += counted // re-arm the trigger honestly
- prodLog.error(
- `[Brainy] background flush (${reason}) FAILED: ${(err as Error).message} — ` +
- `derived-state persistence retries at the next trigger; canonical data is unaffected`
- )
- })
- .finally(() => {
- this._persistBackgroundFlight = null
- })
- }
-
private async persistSingleOp(
touched: { nouns?: string[]; verbs?: string[] },
run: TransactionFunction,
precommit?: (before: CommitBeforeImages) => void,
- pendingEvents?: PendingChangeEvent[],
- records?: FactMarkerRecord[],
- origin?: string
+ pendingEvents?: PendingChangeEvent[]
): Promise<{ generation?: number; timestamp: number; degraded?: string[] }> {
// Change-feed capture: when this write will emit, hold a reference to the
// commit's before-images so `remove` events can carry the record's last
@@ -2884,15 +1634,6 @@ export class Brainy implements BrainyInterface {
: precommit
if (!this._generationStampingActive) {
- // Marker records ride a commit FACT — a generation-less bootstrap
- // write has none to ride. No bootstrap path defers embeds today;
- // refuse loudly rather than silently dropping a durable marker.
- if (records && records.length > 0) {
- throw new Error(
- 'persistSingleOp: marker records require a generation-stamped commit — ' +
- 'a bootstrap (generation-0) write cannot carry them'
- )
- }
// Init-time / infrastructure baseline write (e.g. the VFS root): apply
// WITHOUT creating a generation. Generation 0 is the freshly-materialized
// brain (bootstrap included); the first USER write is generation 1.
@@ -2913,13 +1654,7 @@ export class Brainy implements BrainyInterface {
captureAndCheck({ nouns, verbs } as CommitBeforeImages)
}
await this.generationStore.runWithoutGeneration(() =>
- this.transactionManager.executeTransaction(run, {
- timeout: transactTimeoutBudget(
- (touched.nouns?.length ?? 0) + (touched.verbs?.length ?? 0),
- undefined,
- this.config.transactionBudgetFloorMs
- )
- })
+ this.transactionManager.executeTransaction(run)
)
const timestamp = Date.now()
// Bootstrap writes are not generation-stamped; emit without one.
@@ -2931,16 +1666,7 @@ export class Brainy implements BrainyInterface {
receipt = await this.generationStore.commitSingleOp({
touched,
precommit: captureAndCheck,
- ...(records && records.length > 0 ? { records } : {}),
- ...(origin ? { origin } : {}),
- execute: () =>
- this.transactionManager.executeTransaction(run, {
- timeout: transactTimeoutBudget(
- (touched.nouns?.length ?? 0) + (touched.verbs?.length ?? 0),
- undefined,
- this.config.transactionBudgetFloorMs
- )
- })
+ execute: () => this.transactionManager.executeTransaction(run)
})
} catch (err) {
// A failed rollback that left the store inconsistent (a remove/update
@@ -2968,7 +1694,6 @@ export class Brainy implements BrainyInterface {
)
}
}
- this.noteWriteForPersistence()
return receipt
}
@@ -3034,29 +1759,6 @@ export class Brainy implements BrainyInterface {
}
}
- /**
- * @description Build the AGGREGATION view of an entity from a stored flat
- * metadata record — EVERY reserved field mapped to its top-level entity
- * name (stored `noun` → `type`), custom metadata in `metadata`. This must
- * mirror the add-path `entityForIndexing` shape exactly: the aggregation
- * engine resolves groupBy/where fields via `resolveEntityField`
- * (top-level standard fields + custom metadata), so a view that drops a
- * reserved field makes every aggregate grouped by that field decrement a
- * group that does not exist — counts then drift upward forever after
- * deletes (SELF-AGGREGATE-DELETE-DRIFT). Do not hand-roll subsets of this.
- * @param record - The stored flat metadata record (before-image or pre-delete read).
- * @returns The full-fidelity entity view for aggregation hooks.
- */
- private entityForAggFromRawRecord(record: Record): Record {
- const { reserved, custom } = splitNounMetadataRecord(record)
- const { noun, ...rest } = reserved
- return {
- type: noun,
- ...rest,
- metadata: custom
- }
- }
-
/**
* @description Add an entity (noun) to the brain. Embeds `data` into a vector and
* indexes the entity across all three intelligences — vector similarity, graph
@@ -3084,6 +1786,12 @@ export class Brainy implements BrainyInterface {
// Zero-config validation (static import for performance)
validateAddParams(params)
+ // Reserved fields arriving via the metadata bag (untyped callers — the
+ // compile-time guard stops TypeScript callers) are normalized to their
+ // canonical top-level location BEFORE any enforcement runs, so a
+ // remapped subtype participates in subtype-pairing enforcement and the
+ // indexed metadata bag carries only custom fields.
+ params = this.remapReservedAddMetadata(params)
// Tracked-field vocabulary enforcement (Layer 2). Walks both bags so a
// tracked field declared at top level (e.g. 'subtype') and one declared in
@@ -3144,98 +1852,50 @@ export class Brainy implements BrainyInterface {
}
// Get or compute vector
- // MT5 deferred embedding: ack at durability with a stub vector and a
- // pending marker riding the insert's OWN commit fact (same generation,
- // one atomic append — a marker-less committed row, the silently-missing-
- // vector shape, is structurally impossible). The background worker
- // embeds + inserts.
- const deferringEmbed = params.deferEmbedding === true && !params.vector
- let vector = deferringEmbed
- ? []
- : params.vector || (await this.embed(params.data))
+ const vector = params.vector || (await this.embed(params.data))
- // THE ZERO-NORM LAW (canonical write side): a zero-norm vector is not a
- // vector — it never crosses an engine boundary (the engine pair's seam
- // law). This engine's own cosine distance treats an all-zero vector
- // safely (a zero-norm operand always scores MAXIMUM distance — see
- // isZeroNormVector's JSDoc), but a downstream engine serving squared-
- // euclidean distance cannot tell it apart from a legitimate origin
- // point — a false attractor that silently darkened 150+ rows in a
- // production deployment. The index belt (AddToVectorIndexOperation)
- // already refuses to INDEX a zero-norm vector, but until now the
- // CANONICAL write still persisted it and the vectored-noun ledger
- // counted it — so a near-empty store whose only vectored row was
- // zero-norm read "canonical vectored > 0, index size 0" and threw a
- // not-ready error at open. Normalize HERE, before the dimension pin,
- // the vectored-ledger flag (`SaveNounMetadataOperation`'s `hasVector`),
- // and the index ops below ever see it, so it persists as the sanctioned
- // "unvectored" `[]` shape instead — the canonical write still succeeds.
- if (!deferringEmbed && vector.length > 0 && isZeroNormVector(vector)) {
- prodLog.warn(
- `[Brainy] add(): entity ${id} was given an explicit all-zero vector — ` +
- `a zero-norm vector is not a vector; persisted unvectored ([]) instead.`
+ // Ensure dimensions are set
+ if (!this.dimensions) {
+ this.dimensions = vector.length
+ } else if (vector.length !== this.dimensions) {
+ throw new Error(
+ `Vector dimension mismatch: expected ${this.dimensions}, got ${vector.length}`
)
- vector = []
}
- // Ensure dimensions are set (a deferred-embed stub carries no dimension
- // information — the worker's real vector goes through the same guard).
- // Gated on `vector.length > 0`, not `!deferringEmbed`: ANY insert whose
- // vector is the "unvectored" empty-array shape carries no dimension
- // information, deferred or not — an explicit `vector: []` (e.g. the VFS
- // root's zero-norm fix, see VirtualFileSystem.doInitializeRoot()) must
- // never pin `this.dimensions` to 0, which would poison every subsequent
- // real embed's dimension check for the life of the store.
- if (!deferringEmbed && vector.length > 0) {
- if (!this.dimensions) {
- this.dimensions = vector.length
- } else if (vector.length !== this.dimensions) {
- throw new Error(
- `Vector dimension mismatch: expected ${this.dimensions}, got ${vector.length}`
- )
+ // Prepare metadata for storage
+ // data is stored opaquely in the 'data' field - NOT spread into top-level metadata.
+ // Only metadata fields are queryable via find({ where }).
+ const storageMetadata = {
+ ...params.metadata,
+ // Preserve the caller's original (non-UUID) id when normalized, so reads
+ // can surface it. A real UUID passes through with no _originalId.
+ ...(originalId !== undefined && { [ORIGINAL_ID_KEY]: originalId }),
+ data: params.data,
+ noun: params.type,
+ ...(params.subtype !== undefined && { subtype: params.subtype }),
+ // visibility: stored only when not 'public' (absent === public, keeps records lean)
+ ...(params.visibility !== undefined &&
+ params.visibility !== 'public' && { visibility: params.visibility }),
+ service: params.service,
+ createdAt: Date.now(),
+ updatedAt: Date.now(),
+ _rev: 1,
+ ...(params.confidence !== undefined && { confidence: params.confidence }),
+ ...(params.weight !== undefined && { weight: params.weight }),
+ ...(params.createdBy && { createdBy: params.createdBy })
}
- }
-
- // Prepare metadata for storage: a v2 nested-bag record — engine fields
- // top-level, the user's bag nested VERBATIM (any name, including engine
- // spellings like `confidence` or `type`, is the user's and survives
- // faithfully; the field-addressing law).
- const storageMetadata = buildNounMetadataRecord(
- {
- data: params.data,
- noun: params.type,
- ...(params.subtype !== undefined && { subtype: params.subtype }),
- // visibility: stored only when not 'public' (absent === public, keeps records lean)
- ...(params.visibility !== undefined &&
- params.visibility !== 'public' && { visibility: params.visibility }),
- service: params.service,
- createdAt: Date.now(),
- updatedAt: Date.now(),
- _rev: 1,
- ...(params.confidence !== undefined && { confidence: params.confidence }),
- ...(params.weight !== undefined && { weight: params.weight }),
- ...(params.createdBy && { createdBy: params.createdBy })
- },
- {
- ...params.metadata,
- // Preserve the caller's original (non-UUID) id when normalized, so reads
- // can surface it. A real UUID passes through with no _originalId.
- ...(originalId !== undefined && { [ORIGINAL_ID_KEY]: originalId })
- }
- )
// Build entity structure for indexing (NEW - with top-level fields)
// Optional fields must use conditional spreading to match storageMetadata exactly.
// If undefined values are included as explicit keys, extractIndexableFields indexes
// them as '__NULL__' entries that removeFromIndex can never clean up (storageMetadata
// omits those keys entirely via conditional spreading, so the fields don't match).
- // No `level` here: engine plumbing never enters the indexing view — a
- // hardcoded level:0 landed in the SAME flattened index column as user
- // metadata named `level`, poisoning it multi-valued ([0, real]).
const entityForIndexing = {
id,
vector,
connections: new Map(),
+ level: 0,
type: params.type,
...(params.subtype !== undefined && { subtype: params.subtype }),
...(params.visibility !== undefined &&
@@ -3273,22 +1933,11 @@ export class Brainy implements BrainyInterface {
}
: undefined
- // MT5: the pending marker RIDES the insert's own commit fact (same
- // generation, one atomic append) — threaded to persistSingleOp below.
- // A failed commit appends nothing, so no orphaned durable marker can
- // exist; the in-memory entry is harmless and reaped by the worker.
- const embedMarkers: FactMarkerRecord[] | undefined = deferringEmbed
- ? [this.enqueuePendingEmbed(id)]
- : undefined
-
const runInsert: TransactionFunction = async (tx) => {
// Operation 1: Save metadata FIRST (TypeAwareStorage caching)
// isNew=true: skip pre-read for rollback (entity doesn't exist yet)
- // hasVector: the vectored-noun ledger counts this insert iff its
- // vector is real/non-empty (never true for a deferred embed, whose
- // stub `vector` is `[]` — it counts later, at landing).
tx.addOperation(
- new SaveNounMetadataOperation(this.storage, id, storageMetadata, true, vector.length > 0)
+ new SaveNounMetadataOperation(this.storage, id, storageMetadata, true)
)
// Operation 2: Save vector data
@@ -3302,23 +1951,14 @@ export class Brainy implements BrainyInterface {
}, true)
)
- // Operation 3: Add to HNSW index (after entity saved). Gated on
- // `vector.length > 0`, not `!deferringEmbed`: a deferred embed has
- // nothing to index yet (the worker's atomic update inserts the real
- // vector later), and an explicit `vector: []` insert (the VFS root's
- // zero-norm fix — permanently unvectored plumbing, never embedded)
- // is exactly the same "nothing to index yet" shape. The zero-norm
- // BELT (a real all-zero vector, non-empty) is enforced inside
- // AddToVectorIndexOperation itself — see its JSDoc.
- if (vector.length > 0) {
- tx.addOperation(
- new AddToVectorIndexOperation(this.index, id, vector, this.indexWriteGeneration)
- )
- }
+ // Operation 3: Add to HNSW index (after entity saved)
+ tx.addOperation(
+ new AddToHNSWOperation(this.index, id, vector)
+ )
// Operation 4: Add to metadata index
tx.addOperation(
- new AddToMetadataIndexOperation(this.metadataIndex, id, entityForIndexing, this.indexWriteGeneration)
+ new AddToMetadataIndexOperation(this.metadataIndex, id, entityForIndexing)
)
}
@@ -3348,7 +1988,7 @@ export class Brainy implements BrainyInterface {
const MAX_UPSERT_ATTEMPTS = 10
for (let attempt = 0; ; attempt++) {
try {
- await this.persistSingleOp({ nouns: [id] }, runInsert, insertPrecommit, addEvents, embedMarkers)
+ await this.persistSingleOp({ nouns: [id] }, runInsert, insertPrecommit, addEvents)
break
} catch (err) {
if (!(err instanceof InsertPreconditionExistsSignal)) {
@@ -3382,7 +2022,6 @@ export class Brainy implements BrainyInterface {
this._aggregationIndex.onEntityAdded(id, entityForIndexing)
}
- if (deferringEmbed) this.kickEmbedWorker()
return id
}
@@ -3544,9 +2183,7 @@ export class Brainy implements BrainyInterface {
*
*/
async get(id: string, options?: GetOptions): Promise | null> {
- // Canonical read: a get resolves an entity by id straight from storage and
- // consults no derived index — it must not wait on any family's migration.
- await this.ensureInitialized({ needs: [] })
+ await this.ensureInitialized()
this.warnIfReadsDegraded('get')
// Id normalization (8.0): a caller may read by their natural key — resolve
@@ -3605,8 +2242,7 @@ export class Brainy implements BrainyInterface {
* ```
*/
async batchGet(ids: string[], options?: GetOptions): Promise