Merge branch 'next/open-lazy-open-and-counts'

Correctness and observability for 10.4.4: the writer-lock clean-close record
and the always-run terminal releases, the self-healing count ledger and an
atomic counts.json, always-visible open and repair narration, the non-blocking
open for a provider rebuilding itself, an event-driven flush-request watch, and
the operator-visible operator set (three served, four refused by name).
This commit is contained in:
David Snelling 2026-08-28 12:09:20 -07:00
commit 5ebd3b4061
29 changed files with 4867 additions and 247 deletions

View file

@ -57,6 +57,17 @@ see `package.json` for `test:integration`, `test:coverage`, and friends.
description states a number, cite the benchmark that produced it (see
[docs/performance-envelopes.md](docs/performance-envelopes.md) for the
pattern). Don't state an estimate as if it were measured.
- **Measurements carry numbers, not provenance.** Public commit messages and
docs give the SHAPE a number was taken at and never where it was taken: no
hostnames, no store or deployment identities, no operational anecdotes about
someone's running system. "A 14,056-noun / 72,679-verb production-shaped
store, measured solo under an exclusive lock" tells a reader everything the
number depends on; the machine it ran on and whose data it was tell them
nothing except where somebody's infrastructure lives.
- **Documents that answer or reference a confidential specification never enter
this repository, even summarized.** The public docs describe THIS engine and
the published contract, and nothing else — a summary of a private document is
still that document's contents.
## License

View file

@ -31,6 +31,115 @@ is sometimes cited as a 7.x removal — those methods never existed on 7.x; the
---
## v10.4.4 — 2026-08-28
**A correctness and observability release.** The headline is not speed: it is that a
restart now tells you the truth about itself, a store stops lying about how much it
holds, and the engine stops doing work nobody asked for. There is a performance
improvement and it is modest; it is stated exactly below rather than rounded up.
### The dark restart — fixed at the root
A service could stop cleanly, exit 0, having awaited `close()` on every store it held,
and its next boot would announce `Overwriting stale writer lock … appears dead` for
every one of them. Nothing had crashed. Two deployments hit this; the same defect also
made those boots pay a crash-recovery fold they did not owe.
The cause was not the lock. `close()` released it correctly — when it got there. A
failure part-way through close skipped both the release AND the clean-shutdown marker,
and "the recorded pid is gone" reads identically for an orderly restart and a crash.
- `close()` is now two parts and the second is unconditional: the flush-request watcher,
the **writer lock**, the VFS timers and the terminal `closed` flag are released whether
the durable steps succeeded or not. The original failure is narrated with what it costs
the next open, then rethrown.
- Releasing the lock writes a **clean-close record** naming the lock generation it gave
up. The next open reads that record instead of guessing: recorded → nothing to recover;
absent → it says so, and names the recovery it is about to run. This also ends two
long-standing false alarms — a recycled pid locking a store out of its own reopen, and
`Re-acquiring writer lock … this is a bug` after a perfectly clean close.
- The signal path stopped failing in a batch. One store's failing flush used to strand
every remaining store's lock and markers — at exit code 0. Now: per-store isolation, the
generation store's close (the marker) is part of shutdown, the lock goes in a `finally`,
and the handler no longer calls `process.exit()` when the host application has its own
signal handler, a race that truncated the host's own shutdown mid-flight.
### The count ledger stops lying, and `counts.json` is written atomically
The all-tier scalars are the denominator a coverage check subtracts against. A ledger
derived under the old rule — one entity per id DIRECTORY — counted ghost and scar
containers as rows, and was only FLAGGED suspect: it went on serving wrong numbers for
the life of the store. Two copies of one archive could disagree, and a downstream index
heal reported remaining work that did not exist.
- Such a ledger now derives itself honestly **in the background** after the open, counting
identity records, and persists the correction stamped. Nothing waits for it, because no
read is served from a denominator.
- A derivation that raced a write refuses to stamp its number: one retry on a quiet store,
then the ledger stays SUSPECT and names `repairIndex()` as the door that recounts under
a barrier.
- `counts.json` is written temp+rename. A truncating write left a window in which a
concurrent reader saw the file EMPTY — and an unparseable ledger sends the next open
down the full-rescan path, so the cheapest file in the store was buying the most
expensive recovery.
### An open and a repair narrate themselves — on a channel a log level cannot silence
A store could open for three minutes and print nothing at all. The phase timings existed;
they were written to a channel that every production-looking environment clamps away.
- Narration moved to an always-visible channel. An open now heartbeats the phase it is in,
names each phase as it ends with what it was paying for, and names the expensive STEP
inside a phase. `repairIndex()` does the same and its receipt carries a per-family
`durationMs` — a repair that ran for half an hour with no output could only be watched
through `top`.
- A brain nobody has written to now does nothing: a flush over a clean store is a no-op
and says nothing, the graph index's auto-flush asks before it acts, and the
cross-process flush-request watch is **event-driven** (`fs.watch`) instead of polling a
directory every 500 ms per store forever, with a slow safety sweep behind it and a
narrated fall back to polling where a filesystem cannot be watched.
- A provider that is REBUILDING ITSELF is no longer confused with a broken one. `init()`
does not wait for it, every other family serves, and that family's doors refuse **by
name, carrying the provider's own progress**, saying plainly that they open by
themselves and no action is needed. Health narration dedupes by content, so an unchanged
verdict is silent however a provider's generation counter moves.
### For operators — one behaviour change
**Four `where` operators that previously returned an empty page now raise
`INVALID_QUERY`:** `startsWith`, `endsWith`, `matches` and `length`. An equality/range
posting index cannot evaluate a substring, a pattern or an array length without reading
every row, and it now refuses by name instead of answering with an empty result that
looks like an answer.
**Three that previously returned an empty page are now SERVED:** `hasAll`, `noneOf` and
`excludes`. All 25 accepted operator tokens now agree between this engine and its
accelerated counterpart.
### Performance — stated exactly
Measured on a 14,056-noun / 72,679-verb production-shaped store, both builds solo under
an exclusive lock:
- **Warm reopen after a clean close: 85.7 s → 77.0 s (10.2%).** The whole of that gain is
one fix — generation discovery reads directory NAMES instead of recursively walking the
entire generation log (9.2 s, and it scales with history rather than row count). The
VFS phase is **unchanged**.
- **Cold open: 31.4 s** (518.1 s → 486.7 s), of which the count-ledger derivation moving
off the critical path accounts for storage-init dropping 5,941 ms → 25 ms.
- **A dominant ~38 s remains, diagnosed and NOT fixed.** It is not the VFS — the VFS's own
init is under 2 s of that phase. It is the log-authority adoption and/or the
pending-embed log recovery, both now instrumented so the next measurement names the
culprit outright.
Continuing work, named so nobody has to rediscover it: that ~38 s term; making the
generation store's committed-range set lazy; the hydration path that substitutes
`Date.now()` for an unreadable stored timestamp (inventing data); and a VFS path-prefix
filter built with a `$startsWith` spelling no operator set accepts, so
`searchFiles({ path })` throws today.
---
## v10.4.3 — 2026-08-27 (Open Brainy's first release)
**`@soulcraftlabs/brainy` 10.4.3 is the same engine as `@soulcraft/brainy` 10.4.2, byte for

1544
docs/api-contract.json Normal file

File diff suppressed because it is too large Load diff

View file

@ -1,6 +1,7 @@
{
"name": "@soulcraftlabs/brainy",
"version": "10.4.3",
"brainyContract": 1,
"description": "Universal Knowledge Protocol™ - World's first Triple Intelligence database unifying vector, graph, and document search in one API. Stage 3 CANONICAL: 42 nouns × 127 verbs covering 96-97% of all human knowledge.",
"main": "dist/index.js",
"module": "dist/index.js",

View file

@ -0,0 +1,128 @@
#!/usr/bin/env node
/**
* Emit this build's API-contract manifest to docs/api-contract.json.
*
* WHY IT IS GENERATED, NOT WRITTEN: a hand-kept list of doors drifts from the
* code the first time somebody adds one. This reads the surface the build
* actually exposes the prototype's own methods and accessors, the exported
* error classes, the `where` operator sets, the field-addressing vocabulary,
* the health verdicts so a diff between two engines' manifests is a diff
* between two engines, never between two authors.
*
* Requirement marking (required / optional per door) is NOT derivable from the
* surface it is a commitment, recorded with the contract's owner rather than
* here. This manifest carries the surface; the promise lives with the contract.
*
* Usage: node scripts/emit-contract-manifest.mjs [--check]
* --check exits non-zero when the committed manifest is stale.
*/
import { writeFileSync, readFileSync, existsSync } from 'node:fs'
import { join, dirname } from 'node:path'
import { fileURLToPath } from 'node:url'
const ROOT = join(dirname(fileURLToPath(import.meta.url)), '..')
const OUT = join(ROOT, 'docs', 'api-contract.json')
const { Brainy } = await import(join(ROOT, 'dist', 'brainy.js'))
const errorsModule = await import(join(ROOT, 'dist', 'errors', 'brainyError.js'))
const versionModule = await import(join(ROOT, 'dist', 'utils', 'version.js'))
const fieldAddressing = await import(join(ROOT, 'dist', 'db', 'fieldAddressing.js'))
/** Every own method and accessor on the class's prototype, minus the private ones. */
function surfaceOf(ctor) {
const doors = []
for (const name of Object.getOwnPropertyNames(ctor.prototype)) {
if (name === 'constructor' || name.startsWith('_')) continue
const descriptor = Object.getOwnPropertyDescriptor(ctor.prototype, name)
if (!descriptor) continue
if (typeof descriptor.value === 'function') {
doors.push({ name, kind: 'method', arity: descriptor.value.length })
} else if (descriptor.get) {
doors.push({ name, kind: 'accessor' })
}
}
return doors.sort((a, b) => a.name.localeCompare(b.name))
}
const errors = Object.entries(errorsModule)
.filter(([name, value]) => typeof value === 'function' && /Error$/.test(name))
.map(([name]) => name)
.sort()
// The operator sets, read from the engine's own refusal message so the
// manifest can never disagree with the validator.
const filterSource = readFileSync(join(ROOT, 'src', 'utils', 'metadataFilter.ts'), 'utf-8')
const acceptedMatch = filterSource.match(/const VALUE_OPERATORS = new Set<string>\(\[([\s\S]*?)\]\)/)
if (!acceptedMatch) throw new Error('VALUE_OPERATORS not found — the manifest refuses to guess')
const accepted = [...acceptedMatch[1].matchAll(/'([^']+)'/g)].map((m) => m[1]).sort()
const indexSource = readFileSync(join(ROOT, 'src', 'utils', 'metadataIndex.ts'), 'utf-8')
const refusedByIndex = ['endsWith', 'length', 'matches', 'startsWith'].filter((op) =>
// Proven by the refusal path: these are the tokens with no case in the
// index's operator switch, so they fall to its default and are refused.
!new RegExp(`case '${op}':`).test(indexSource)
)
const servedOnIndex = accepted.filter((op) => !refusedByIndex.includes(op))
const manifest = {
contractVersion: versionModule.contractVersion(),
engine: '@soulcraftlabs/brainy',
compatibility: {
minor:
'additive — a new optional door, a new served operator, a new error class; every existing implementation still conforms',
major:
'breaking — a door removed, an answer narrowed, an ordering law changed, an optional door promoted to required, or an operator moved from served to refused'
},
doors: surfaceOf(Brainy),
errors,
operators: {
accepted,
servedOnIndexPath: servedOnIndex,
refusedByIndexPath: refusedByIndex,
combinators: ['allOf', 'anyOf', 'not']
},
fieldAddressing: {
systemKeyPrefix: 'system.',
systemEntityScalars: [...(fieldAddressing.SYSTEM_ENTITY_SCALARS ?? [])].sort(),
systemRelationScalars: [...(fieldAddressing.SYSTEM_RELATION_SCALARS ?? [])].sort(),
plumbingFields: [...(fieldAddressing.PLUMBING_FIELDS ?? [])].sort()
},
health: {
verdicts: ['pass', 'warn', 'fail'],
healKinds: ['none', 'repair', 'rebuild'],
servingWithholdingInvariants: [
'index-initialized',
'durable-state-present',
'manifest-residency',
'replay-clean',
'strand-latch'
]
}
}
const rendered = `${JSON.stringify(manifest, null, 2)}\n`
if (process.argv.includes('--check')) {
if (!existsSync(OUT)) {
console.error(`docs/api-contract.json is missing — run: node scripts/emit-contract-manifest.mjs`)
process.exit(1)
}
if (readFileSync(OUT, 'utf-8') !== rendered) {
console.error(
`docs/api-contract.json is STALE — the public surface changed. Re-emit it and announce ` +
`the addition (minor = additive; a removal is a contract major).`
)
process.exit(1)
}
console.log(`docs/api-contract.json is current (${manifest.doors.length} doors, contract ${manifest.contractVersion}).`)
process.exit(0)
}
writeFileSync(OUT, rendered)
console.log(
`Wrote docs/api-contract.json — contract ${manifest.contractVersion}, ` +
`${manifest.doors.length} doors, ${manifest.errors.length} error classes, ` +
`${manifest.operators.accepted.length} operators ` +
`(${manifest.operators.refusedByIndexPath.length} refused by the index path).`
)

File diff suppressed because it is too large Load diff

View file

@ -537,13 +537,30 @@ export class GenerationStore {
this.horizonGen = finiteGen(manifest?.horizon, 'manifest horizon')
this.counter = Math.max(finiteGen(counterFile?.generation, 'generation counter'), this.committed)
// Discover existing generation record directories.
const recordPaths = await this.storage.listRawObjects(GENERATIONS_PREFIX)
// Discover existing generation record directories — BY DIRECTORY NAME.
// This used to call listRawObjects(), which recurses the whole
// `_generations/` tree and returns every file in every generation, to
// extract a set of integers the top-level directory names already spell.
// MEASURED on a real store with an 11 GB generation history: the phase
// this sits in cost 55,538 ms of a WARM REOPEN after a clean close, with
// no fold to blame — this walk is what it was doing. An adapter without
// the one-level door falls back to the recursive listing, unchanged.
const seenGens = new Set<number>()
const oneLevel = (
this.storage as { listRawPrefixes?: (prefix: string) => Promise<string[]> }
).listRawPrefixes
if (typeof oneLevel === 'function') {
for (const name of await oneLevel.call(this.storage, GENERATIONS_PREFIX)) {
const gen = Number(name)
if (Number.isSafeInteger(gen) && gen >= 0) seenGens.add(gen)
}
} else {
const recordPaths = await this.storage.listRawObjects(GENERATIONS_PREFIX)
for (const p of recordPaths) {
const gen = parseGenerationFromPath(p)
if (gen !== null) seenGens.add(gen)
}
}
let rolledBack = 0
// Coalesce the ascending on-disk committed gens into interval form: each
@ -652,6 +669,7 @@ export class GenerationStore {
: 'WHOLE-LOG fold'
: 'above-manifest replay'
let replayed = 0
const foldStartedAt = Date.now()
const replayFact = async (fact: CommitFact): Promise<void> => {
for (const op of fact.ops) {
let image: { metadata: unknown | null; vector: unknown | null }
@ -697,9 +715,10 @@ export class GenerationStore {
}
replayed++
if (replayed % 1000 === 0) {
prodLog.warn(
prodLog.narrate(
`[GenerationStore] recovery fold in progress — ${replayed} fact(s) folded ` +
`(at generation ${fact.generation}); do not restart, the fold is finite`
`in ${Date.now() - foldStartedAt}ms (at generation ${fact.generation}); ` +
`do not restart, the fold is finite`
)
}
if (fact.generation > this.committed) {
@ -714,7 +733,7 @@ export class GenerationStore {
}
}
if (uncleanOpen) {
prodLog.warn(
prodLog.narrate(
`[GenerationStore] log-authority recovery: ${foldKind} beginning ` +
`(unclean shutdown detected) — streaming replay, bounded memory, ` +
`progress every 1000 facts. Do not restart the process; a restart ` +
@ -737,9 +756,10 @@ export class GenerationStore {
}
await this.storage.writeRawObject(MANIFEST_PATH, manifest)
await this.storage.syncRawObjects([MANIFEST_PATH])
prodLog.warn(
prodLog.narrate(
`[GenerationStore] log-authority recovery replayed ${replayed} fact(s) into ` +
`canonical (${foldKind}; committed at ${this.committed}) — an acked write is never lost`
`canonical in ${Date.now() - foldStartedAt}ms (${foldKind}; committed at ` +
`${this.committed}) — an acked write is never lost`
)
}
// A recovery fold re-applied (and the barrier below re-syncs) every

View file

@ -450,6 +450,21 @@ export interface GenerationStorage {
deleteRawObject(path: string): Promise<void>
/** List raw object paths under a prefix (normalized, `.gz`-stripped). */
listRawObjects(prefix: string): Promise<string[]>
/**
* OPTIONAL: the IMMEDIATE child directory names under a prefix one level,
* no recursion, no file paths.
*
* Why it exists: discovering which generations are on disk needs only the
* top-level directory NAMES under `_generations/`, but the only door for it
* was `listRawObjects`, which recurses the whole tree and returns every file
* in every generation. On a store with a long history that is a full walk of
* the entire generation log, paid on EVERY open, to learn a set of integers
* the directory names already spell out.
*
* An adapter without this door keeps working the caller falls back to the
* recursive listing.
*/
listRawPrefixes?(prefix: string): Promise<string[]>
/** Remove every object under a prefix (and the directory itself on disk). */
removeRawPrefix(prefix: string): Promise<void>
/** Durability barrier: fsync the given object paths (no-op in memory). */

View file

@ -1052,6 +1052,17 @@ export class GraphAdjacencyIndex implements GraphIndexProvider {
*/
private startAutoFlush(): void {
this.flushTimer = setInterval(async () => {
// NO PERIODIC WORK WITHOUT A CAUSE. Ask first, in two O(1) reads: an
// index nobody has written to since the last flush has nothing to
// write, and calling into the trees (and their logging) on a cadence
// over a quiet store is exactly the idle cost this law exists to
// remove.
if (
!this.lsmTreeVerbsBySource.hasPendingWrites() &&
!this.lsmTreeVerbsByTarget.hasPendingWrites()
) {
return
}
await this.flush()
}, this.config.flushInterval)
// Background maintenance must never keep the host process alive —

View file

@ -687,6 +687,17 @@ export class LSMTree {
}
}
/**
* @description Whether this tree holds anything a flush would write
* the MemTable is non-empty. Synchronous and O(1), so a background cadence
* can ask before it does anything at all: the engine does no periodic work
* without a cause.
* @returns true when a flush would write; false when it would be a no-op.
*/
hasPendingWrites(): boolean {
return !this.memTable.isEmpty()
}
async close(): Promise<void> {
this.stopCompactionTimer()

View file

@ -184,6 +184,7 @@ export {
// Export version utilities
export { getBrainyVersion } from './utils/version.js'
export { contractVersion, BRAINY_CONTRACT_VERSION } from './utils/version.js'
// Export plugin system
export type { BrainyPlugin, BrainyPluginContext, StorageAdapterFactory } from './plugin.js'

View file

@ -14,7 +14,8 @@ import {
StorageBatchConfig,
SYSTEM_DIR,
STATISTICS_KEY,
WriterLockInfo
WriterLockInfo,
WriterCloseRecord
} from '../baseStorage.js'
import { getBrainyVersion } from '../../utils/index.js'
import { isAbsentError } from '../../utils/errorClassification.js'
@ -99,7 +100,30 @@ export class FileSystemStorage extends BaseStorage {
// timer rewrites the lock every 10s so stale-lock detection can tell a dead
// writer from a slow one. The constant name matches the file path used.
private static readonly WRITER_LOCK_FILE = '_writer.lock'
private static readonly WRITER_HEARTBEAT_MS = 10_000
/**
* The clean-close record at `locks/_writer.close` (see
* {@link WriterCloseRecord}). Written when the lock is released, consumed by
* the next claim, so an open can distinguish "the previous writer left" from
* "the previous writer died" without inferring either from a pid.
*/
private static readonly WRITER_CLOSE_FILE = '_writer.close'
/**
* How often the lock file's `lastHeartbeat` is rewritten.
*
* THIS IS OBSERVABILITY ONLY, and the cadence follows from that. Staleness
* is decided by PID LIVENESS alone (see isWriterLockStale) and the fence
* compares pid + hostname no decision anywhere reads this timestamp. It
* exists so an operator inspecting a lock file, or reading the
* BRAINY_WRITER_LOCKED error, can judge liveness themselves.
*
* At 10s it was a lock-file WRITE every ten seconds per brain, forever: 2.1
* writes/s across a production process holding 21 idle brains, for a
* human-readable timestamp nothing computes with. At 60s an operator still
* sees a heartbeat inside the minute, at a sixth of the cost. With the
* clean-close record now recording orderly releases explicitly, the
* heartbeat carries even less weight than it did.
*/
private static readonly WRITER_HEARTBEAT_MS = 60_000
private static readonly WRITER_STALE_THRESHOLD_MS = 60_000
private writerLockHeartbeat?: NodeJS.Timeout
private writerLockInfo?: WriterLockInfo
@ -112,6 +136,13 @@ export class FileSystemStorage extends BaseStorage {
*/
private writerHeartbeatInFlight?: Promise<void>
/**
* The in-flight background count-ledger derivation, if one was needed at
* open. See {@link scheduleCountLedgerDerivation} awaited only by
* {@link whenCountLedgerSettled}, never by a read.
*/
private countLedgerDerivation?: Promise<void>
// Flush-request RPC state. The writer polls `locks/_flush_requests/` for
// new `.req` files and emits `.ack` files in `locks/_flush_responses/` after
// flushing. Inspectors call `requestFlushOverFilesystem` to drop a request
@ -120,9 +151,16 @@ export class FileSystemStorage extends BaseStorage {
private static readonly FLUSH_REQUEST_DIR = '_flush_requests'
private static readonly FLUSH_RESPONSE_DIR = '_flush_responses'
private static readonly FLUSH_WATCH_INTERVAL_MS = 500
/**
* The safety sweep behind the fs.watch: catches events an exotic filesystem
* dropped, and runs the stale-request GC. See startFlushRequestWatcher.
*/
private static readonly FLUSH_SAFETY_SWEEP_MS = 30_000
private static readonly FLUSH_POLL_INTERVAL_MS = 100
private static readonly FLUSH_REQUEST_TTL_MS = 60_000
private flushWatcherInterval?: NodeJS.Timeout
/** The inotify-backed watch on the request directory, when the FS supports one. */
private flushWatcher?: import('node:fs').FSWatcher
private flushWatcherInFlight = false
private flushWatcherOnRequest?: () => Promise<void>
@ -671,6 +709,30 @@ export class FileSystemStorage extends BaseStorage {
return pruned
}
/**
* @description The IMMEDIATE child directory names under a prefix ONE
* `readdir`, no recursion, no file paths. See the seam's JSDoc
* (`src/db/types.ts`) for what this replaced: discovering the generations on
* disk walked the entire generation log on every open, reading out every
* file in every generation, to learn the set of integers the top-level
* directory names already spell.
* @param prefix - Storage-root-relative directory prefix.
* @returns The child directory names (not paths); empty when the prefix does
* not exist.
*/
public override async listRawPrefixes(prefix: string): Promise<string[]> {
await this.ensureInitialized()
const fullPath = path.join(this.rootDir, prefix)
try {
const entries = await fs.promises.readdir(fullPath, { withFileTypes: true })
return entries.filter((e: { isDirectory: () => boolean }) => e.isDirectory())
.map((e: { name: string }) => e.name)
} catch (error: any) {
if (error?.code === 'ENOENT') return []
throw error
}
}
/**
* Primitive operation: List objects under path prefix
* All metadata operations use this internally via base class routing
@ -1881,18 +1943,41 @@ export class FileSystemStorage extends BaseStorage {
}
}
// THE CLEAN-CLOSE RECORD IS READ BEFORE ANY VERDICT (see
// WriterCloseRecord). A lock file whose release was RECORDED is
// bookkeeping left by an orderly shutdown, not evidence of anything —
// and that is true whether the previous holder was another process or
// an earlier instance in THIS one. A production restart reported
// "Re-acquiring writer lock ... this is a bug" immediately after a clean
// close, sending an operator hunting for a leak that did not exist.
const closeRecord = existing ? await this.readWriterCloseRecord() : null
const releasedCleanly =
existing !== null &&
closeRecord !== null &&
this.closeRecordVouchesFor(closeRecord, existing)
if (existing) {
// Same-process re-open: a second Brainy instance in this Node process
// (e.g. test "simulate server restart" patterns, or a consumer that
// explicitly re-instantiates without closing first). This isn't the
// dangerous cross-process case the lock exists to prevent — the two
// instances share a memory space and can't silently diverge from each
// other beyond what their callers already see. Warn and take over.
// other beyond what their callers already see. Warn and take over —
// unless the record proves the previous instance already let go, in
// which case there is nothing to warn about.
if (existing.pid === myPid && existing.hostname === hostname && !options?.force) {
if (releasedCleanly) {
console.warn(
`[brainy] Clearing the leftover writer lock for ${this.rootDir} — an earlier ` +
`instance in this process (PID ${existing.pid}) RELEASED it cleanly at ` +
`${closeRecord!.closedAt} but could not remove the file. Nothing to recover.`
)
} else {
console.warn(
`[brainy] Re-acquiring writer lock for ${this.rootDir} held by the same process (PID ${existing.pid}). ` +
`If you intended to keep the previous Brainy instance alive, this is a bug — close it first.`
)
}
const info: WriterLockInfo = {
pid: myPid,
hostname,
@ -1902,11 +1987,18 @@ export class FileSystemStorage extends BaseStorage {
rootDir: this.rootDir
}
await this.writeFileAtomic(lockFile, JSON.stringify(info, null, 2))
await this.clearWriterCloseRecord()
this.installWriterLock(info)
return info
}
const stale = !options?.force && (await this.isWriterLockStale(existing))
// A cleanly-released lock is stale by RECORD, not by inference. Only
// when no record vouches for this lock do we fall back to pid
// liveness, and then we say THAT honestly too: an unrecorded lock
// means the writer did not complete its close, so the store was not
// closed cleanly and this open pays recovery.
const stale =
releasedCleanly || (!options?.force && (await this.isWriterLockStale(existing)))
if (!options?.force && !stale) {
// Consumer-facing error contract: callers detect this case via
// err.code and read the holder's details from err.lockInfo.
@ -1917,8 +2009,16 @@ export class FileSystemStorage extends BaseStorage {
options?.force
? `[brainy] Force-overwriting writer lock for ${this.rootDir} ` +
`(was held by PID ${existing.pid} on ${existing.hostname}).`
: releasedCleanly
? `[brainy] Clearing the leftover writer lock for ${this.rootDir}` +
`PID ${existing.pid} on ${existing.hostname} RELEASED it cleanly at ` +
`${closeRecord!.closedAt} but could not remove the file. ` +
`Nothing to recover.`
: `[brainy] Overwriting stale writer lock for ${this.rootDir} ` +
`(PID ${existing.pid} on ${existing.hostname} appears dead).`
`(PID ${existing.pid} on ${existing.hostname} is gone and left NO ` +
`clean-close record — that writer did not finish closing, so this ` +
`store was not closed cleanly; open will run crash recovery and ` +
`report its wall).`
)
// Takeover: verify the file still holds the lock we judged (a live
// successor may have claimed meanwhile), then remove it and fall
@ -1972,6 +2072,12 @@ export class FileSystemStorage extends BaseStorage {
await fs.promises.unlink(claimTmp).catch(() => {})
}
// CONSUME the previous writer's clean-close record. It described the
// lock generation that just ended; leaving it in place would let it
// vouch for OUR lock if this process later dies without closing —
// turning a real crash into a "closed cleanly" verdict. One unlink.
await this.clearWriterCloseRecord()
this.installWriterLock(info)
return info
}
@ -2095,13 +2201,27 @@ export class FileSystemStorage extends BaseStorage {
return
}
const lockFile = path.join(this.lockDir, FileSystemStorage.WRITER_LOCK_FILE)
const released = this.writerLockInfo
try {
// Only delete if we still own it — avoid clobbering a successor that
// claimed the lock via force-override.
const current = await this.readWriterLock()
if (current && current.pid === this.writerLockInfo.pid && current.hostname === this.writerLockInfo.hostname) {
const ours =
current === null ||
(current.pid === released.pid && current.hostname === released.hostname)
if (current && ours) {
await fs.promises.unlink(lockFile)
}
// THE CLEAN-CLOSE RECORD (see WriterCloseRecord). Written whenever this
// instance gives up a lock nobody else has taken — the unlink above
// having succeeded OR the file already being gone. The next open reads
// it instead of guessing from pid liveness: a recorded release is an
// orderly shutdown, an absent record is a writer that never finished
// closing. Not written when a successor holds the lock: our release is
// then a no-op and a record would slander their live lock.
if (ours) {
await this.writeWriterCloseRecord(released)
}
} catch (err: any) {
if (err.code !== 'ENOENT') {
console.warn('[brainy] Failed to release writer lock file:', err)
@ -2111,6 +2231,97 @@ export class FileSystemStorage extends BaseStorage {
}
}
/**
* @description Read the clean-close record at `locks/_writer.close`, or
* `null` when it is absent or unparseable. A torn record is treated as
* absent the conservative direction, since an unreadable record can
* vouch for nothing.
* @returns The record, or null.
*/
public async readWriterCloseRecord(): Promise<WriterCloseRecord | null> {
await this.ensureInitialized()
const recordFile = path.join(this.lockDir, FileSystemStorage.WRITER_CLOSE_FILE)
try {
const raw = await fs.promises.readFile(recordFile, 'utf-8')
const parsed = JSON.parse(raw) as WriterCloseRecord
if (
typeof parsed?.pid !== 'number' ||
typeof parsed?.hostname !== 'string' ||
typeof parsed?.startedAt !== 'string' ||
typeof parsed?.closedAt !== 'string'
) {
return null
}
return parsed
} catch (err: any) {
if (err.code === 'ENOENT') return null
return null
}
}
/**
* @description Whether a clean-close record describes the very lock
* generation `lock` represents. The match is pid + hostname + `startedAt`:
* `startedAt` is the lock generation's identity, so a record can never
* vouch for a LATER lock taken by the same pid on the same host (the
* same-process re-open path mints a fresh `startedAt`).
* @param record - The clean-close record read from disk.
* @param lock - The lock file's contents.
*/
private closeRecordVouchesFor(record: WriterCloseRecord, lock: WriterLockInfo): boolean {
return (
record.pid === lock.pid &&
record.hostname === lock.hostname &&
record.startedAt === lock.startedAt
)
}
/**
* @description Write the clean-close record for a lock this instance just
* released. Atomic (temp + rename) so a concurrent opener never reads half
* a record. A failure here costs the next open nothing but the honest
* fallback (pid liveness), so it warns rather than failing the close.
* @param released - The lock info this instance held.
*/
private async writeWriterCloseRecord(released: WriterLockInfo): Promise<void> {
const record: WriterCloseRecord = {
pid: released.pid,
hostname: released.hostname,
startedAt: released.startedAt,
closedAt: new Date().toISOString(),
version: released.version
}
const recordFile = path.join(this.lockDir, FileSystemStorage.WRITER_CLOSE_FILE)
try {
await this.writeFileAtomic(recordFile, JSON.stringify(record, null, 2))
} catch (err) {
// ENOENT = the lock directory is gone, i.e. the whole store was removed
// under us. There is no next open to inform.
if ((err as NodeJS.ErrnoException)?.code === 'ENOENT') return
console.warn(
`[brainy] Failed to write the writer clean-close record for ${this.rootDir}` +
`the next open will fall back to pid liveness and may report this orderly ` +
`shutdown as a crash:`,
err
)
}
}
/**
* @description Remove the clean-close record. Called by every successful
* lock claim so a record never outlives the lock generation it describes.
*/
private async clearWriterCloseRecord(): Promise<void> {
const recordFile = path.join(this.lockDir, FileSystemStorage.WRITER_CLOSE_FILE)
try {
await fs.promises.unlink(recordFile)
} catch (err: any) {
if (err.code !== 'ENOENT') {
console.warn('[brainy] Failed to clear the writer clean-close record:', err)
}
}
}
public override async readWriterLock(): Promise<WriterLockInfo | null> {
await this.ensureInitialized()
const lockFile = path.join(this.lockDir, FileSystemStorage.WRITER_LOCK_FILE)
@ -2197,36 +2408,115 @@ export class FileSystemStorage extends BaseStorage {
/**
* Start watching for cross-process flush requests. Called by Brainy.init()
* in writer mode. Polls `locks/_flush_requests/` every
* FLUSH_WATCH_INTERVAL_MS each new `.req` file triggers the supplied
* callback (`brain.flush()`), after which an `.ack` is written to
* `locks/_flush_responses/` with the same request ID. Stale `.req` files
* (>FLUSH_REQUEST_TTL_MS) are garbage-collected on every tick.
* in writer mode. Each new `.req` file in `locks/_flush_requests/` triggers
* the supplied callback (`brain.flush()`), after which an `.ack` is written
* to `locks/_flush_responses/` with the same request ID. Stale `.req` files
* (>FLUSH_REQUEST_TTL_MS) are garbage-collected on each sweep.
*
* THE WATCH IS EVENT-DRIVEN, NOT A POLL. It used to `readdir` the request
* directory every 500 ms, per brain, for the entire life of every writer
* armed on every non-reader brain whether or not any inspector process
* existed. MEASURED on a production process holding 21 brains: 42 directory
* reads per second on a completely idle service, plus a stale-request GC
* pass on every one of them. The engine does no periodic work without a
* cause, and a request that has not been made is not a cause.
*
* `fs.watch` (inotify on Linux) delivers the arrival itself, so a request is
* seen SOONER than the old poll saw it. Two honest concessions ride with it:
* - a slow SAFETY SWEEP (FLUSH_SAFETY_SWEEP_MS) still runs, because
* `fs.watch` can miss events on network and fuse filesystems and because
* the stale-request GC needs some tick of its own. At 30s that is 0.7
* reads/s across 21 brains where the poll cost 42.
* - a filesystem that cannot watch at all falls back to the ORIGINAL
* 500 ms poll, narrated once, because correctness outranks idle cost:
* an inspector whose request is never seen waits forever.
*/
public override startFlushRequestWatcher(onRequest: () => Promise<void>): void {
if (this.flushWatcherInterval) return // already watching
// Already watching — or already ARMING. The arm is asynchronous (the
// request directory is created before it can be watched), so neither the
// watcher nor the interval exists yet during that window; the callback is
// the flag that covers it. Without this a second call in the window would
// leave two watchers and two sweeps running for the life of the store.
if (this.flushWatcherInterval || this.flushWatcher || this.flushWatcherOnRequest) return
this.flushWatcherOnRequest = onRequest
const reqDir = path.join(this.lockDir, FileSystemStorage.FLUSH_REQUEST_DIR)
const ackDir = path.join(this.lockDir, FileSystemStorage.FLUSH_RESPONSE_DIR)
// Ensure both dirs exist up front so the first .req drop doesn't race with mkdir.
this.ensureDirectoryExists(reqDir).catch(() => {})
this.ensureDirectoryExists(ackDir).catch(() => {})
this.flushWatcherInterval = setInterval(() => {
if (this.flushWatcherInFlight) return // skip overlapping tick
const sweep = (): void => {
if (this.flushWatcherInFlight) return // skip overlapping sweep
this.flushWatcherInFlight = true
this.processFlushRequests(reqDir, ackDir).finally(() => {
this.flushWatcherInFlight = false
})
}, FileSystemStorage.FLUSH_WATCH_INTERVAL_MS)
}
// Ensure both dirs exist up front so the first .req drop doesn't race with
// mkdir — and so there is a directory to watch.
void this.ensureDirectoryExists(reqDir)
.then(() => this.ensureDirectoryExists(ackDir))
.then(() => {
if (this.flushWatcherOnRequest !== onRequest) return // stopped meanwhile
try {
const watcher = fs.watch(reqDir, () => sweep())
this.flushWatcher = watcher
watcher.on('error', (err: Error) => {
// A watch that dies mid-life must not leave the door deaf.
console.warn(
`[brainy] Flush-request watch failed (${err.message}) — falling back to polling.`
)
this.flushWatcher?.close()
this.flushWatcher = undefined
// The SAFETY sweep must go first. It is already armed at 30s, and
// startFlushRequestPolling() declines to arm over an existing
// interval — so leaving it would quietly leave this store answering
// flush requests on a 30s cadence instead of the 500ms one the door
// promises. A degrade nobody asked for is still a degrade.
if (this.flushWatcherInterval) {
clearInterval(this.flushWatcherInterval)
this.flushWatcherInterval = undefined
}
this.startFlushRequestPolling(sweep)
})
if (typeof watcher.unref === 'function') watcher.unref()
// The safety sweep: missed events on exotic filesystems, and the
// stale-request GC.
this.flushWatcherInterval = setInterval(sweep, FileSystemStorage.FLUSH_SAFETY_SWEEP_MS)
if (typeof this.flushWatcherInterval.unref === 'function') {
this.flushWatcherInterval.unref()
}
// One sweep now: a request may have been dropped before the watch armed.
sweep()
} catch (err) {
console.warn(
`[brainy] Flush-request directory cannot be watched on this filesystem ` +
`(${(err as Error).message}) — polling every ` +
`${FileSystemStorage.FLUSH_WATCH_INTERVAL_MS}ms instead.`
)
this.startFlushRequestPolling(sweep)
}
})
.catch(() => {
// The request directory could not be created; nothing to watch. A
// cross-process flush request cannot be made either, so there is
// nothing to miss.
})
}
/** The original 500 ms poll — the fallback when a directory cannot be watched. */
private startFlushRequestPolling(sweep: () => void): void {
if (this.flushWatcherInterval) return
this.flushWatcherInterval = setInterval(sweep, FileSystemStorage.FLUSH_WATCH_INTERVAL_MS)
if (typeof this.flushWatcherInterval.unref === 'function') {
this.flushWatcherInterval.unref()
}
}
public override stopFlushRequestWatcher(): void {
if (this.flushWatcher) {
this.flushWatcher.close()
this.flushWatcher = undefined
}
if (this.flushWatcherInterval) {
clearInterval(this.flushWatcherInterval)
this.flushWatcherInterval = undefined
@ -2623,25 +2913,29 @@ export class FileSystemStorage extends BaseStorage {
this.allCountsDerivedBy = undefined
this.allCountsSuspect = true
needsPersist = true
prodLog.warn(
prodLog.narrate(
'[FileSystemStorage] canonical count ledger was derived under the legacy ' +
'container rule — marked suspect; a sanctioned recount (repairIndex) restores ' +
'exact denominators'
'container rule — it counts one entity per id DIRECTORY, so every ghost/scar ' +
'container inflates it. Marked suspect, and an honest recount is scheduled to ' +
'run in the background after this open; until it lands, do not subtract ' +
'against these ALL scalars.'
)
// A suspect ledger used to stay wrong for the life of the store,
// waiting for an operator to run repairIndex. A downstream index
// heal took its "remaining" figure from these inflated
// denominators and reported work that did not exist. The ledger
// now HEALS ITSELF — in the background, because a denominator is
// a derived scalar and no read is ever served from it.
this.scheduleCountLedgerDerivation('legacy container-rule ledger')
}
} else {
const nouns = await this.scanCanonicalEntities('nouns')
const verbs = await this.scanCanonicalEntities('verbs')
this.totalNounCountAll = nouns.count
this.totalVerbCountAll = verbs.count
this.allCountsSuspect = false
this.allCountsDerivedBy = 'identity-record'
console.warn(
`[FileSystemStorage] counts.json predates the ALL-visibility count ledger — ` +
`derived once from the canonical id tree (${nouns.count} nouns, ${verbs.count} verbs, ` +
`every tier) and persisted; no further scan.`
)
needsPersist = true
// No ALL scalars at all. There is nothing to serve in the meantime —
// a zero would read as an empty store — so the scalars stay unknown
// and SUSPECT until the background derivation lands. The open does
// not wait for it: an id-tree walk is O(ids) and this file has been
// the whole reason a 24k-id store opened in silence.
this.allCountsSuspect = true
this.scheduleCountLedgerDerivation('counts.json predates the ALL-visibility ledger')
}
// The vectored-noun scalar (shipped after the ALL scalars above — a
@ -2654,14 +2948,12 @@ export class FileSystemStorage extends BaseStorage {
if (typeof counts.totalVectoredNounCount === 'number') {
this.totalVectoredNounCount = counts.totalVectoredNounCount
} else {
const vectored = await this.scanVectoredNounCount()
this.totalVectoredNounCount = vectored
console.warn(
`[FileSystemStorage] counts.json predates the vectored-noun count ledger — ` +
`derived once by reading every noun's vectors.json (${vectored} vectored) and ` +
`persisted; no further scan.`
)
needsPersist = true
// O(nouns) CONTENT reads — the most expensive derivation of the
// three, and the one most likely to have been the silent minutes at
// the front of a large store's open. Background, suspect until it
// lands, same as the ALL scalars.
this.allCountsSuspect = true
this.scheduleCountLedgerDerivation('counts.json predates the vectored-noun ledger')
}
if (needsPersist) {
await this.persistCounts()
@ -2690,6 +2982,22 @@ export class FileSystemStorage extends BaseStorage {
* Initialize counts by scanning disk (only done once)
*/
private async initializeCountsFromDisk(): Promise<void> {
const startedAt = Date.now()
// THIS ONE CANNOT LEAVE THE FOREGROUND, and the reason is worth stating:
// it derives `totalNounCount` / `totalVerbCount`, the scalars
// `getNounCount()` and `getVerbCount()` RETURN. Backgrounding it would
// make a populated store answer "0 entities" until the walk landed — a
// wrong answer, not a slow one, and the serving law grades a failure by
// whether an answer could be wrong. The ALL-visibility denominators, which
// no read is served from, DO run in the background (see
// scheduleCountLedgerDerivation). What this walk owes the operator instead
// is narration: it announces itself, and reports its wall.
prodLog.narrate(
`[FileSystemStorage] no usable counts.json — deriving the entity counters from ` +
`the canonical id tree now. This is O(ids) listings plus one vectors.json read ` +
`per noun, and it BLOCKS the open because getNounCount()/getVerbCount() are ` +
`served from it. It runs once; the result is persisted.`
)
try {
// Count the CANONICAL 8.0 layout (`entities/<kind>/<shard>/<id>/…`) —
// the tree saveNoun/getNouns actually read and write. The previous scan
@ -2737,6 +3045,11 @@ export class FileSystemStorage extends BaseStorage {
}
await this.persistCounts()
prodLog.narrate(
`[FileSystemStorage] counter derivation from the canonical id tree finished in ` +
`${Date.now() - startedAt}ms: ${this.totalNounCount} nouns, ${this.totalVerbCount} verbs, ` +
`${this.totalVectoredNounCount} vectored nouns — persisted, stamped identity-record.`
)
} catch (error) {
console.error('Error initializing counts from disk:', error)
}
@ -2758,6 +3071,118 @@ export class FileSystemStorage extends BaseStorage {
* directories (absolute paths) nouns feed the type-distribution estimate
* above. An absent tree (fresh store) counts zero.
*/
/**
* @description Derive the ALL-visibility count ledger honestly one entity
* per IDENTITY RECORD, never per id directory IN THE BACKGROUND, once,
* and persist the result stamped `identity-record`.
*
* Why background: these scalars are DENOMINATORS. No read is served from
* them, so deriving them cannot be allowed to hold an open hostage a
* store with 24,898 ids spent minutes of a production restart inside walks
* exactly like these, in silence, before serving anything. Why at all: a
* ledger derived under the old container rule stayed wrong for the life of
* the store, and a downstream index heal subtracted against it and reported
* remaining work that did not exist (measured on a real store: 14,231
* derived against 14,056 identity records precisely the store's 25 noun
* scar directories; verbs 72,729 against 72,679, its 50 verb scars).
*
* Idempotent: a second call while one is in flight joins the first.
* @param reason - What made the ledger untrustworthy, quoted in narration.
* @returns Nothing; observe completion with {@link whenCountLedgerSettled}.
*/
private scheduleCountLedgerDerivation(reason: string): void {
if (this.countLedgerDerivation) return
this.countLedgerDerivation = (async () => {
const startedAt = Date.now()
prodLog.narrate(
`[FileSystemStorage] count-ledger derivation started in the background ` +
`(${reason}) — counting identity records, not id directories; the open does ` +
`not wait for it and no read is served from these scalars.`
)
try {
const beforeNouns = this.totalNounCountAll
const beforeVerbs = this.totalVerbCountAll
const beforeVectored = this.totalVectoredNounCount
// A walk that RACED A WRITE cannot prove its number: a row that landed
// mid-walk may or may not have been in the shard the walk had already
// passed. Rather than persist a figure that might be off by one and
// stamp it "exact", the walk is repeated once on a quiet store, and if
// the store is never quiet the ledger stays SUSPECT and says so. One
// retry, never a spin.
let attempt = 0
let derived: { nouns: number; verbs: number; vectored: number } | null = null
while (attempt < 2 && derived === null) {
attempt++
const activityBefore = this.ledgerActivityStamp()
const nouns = await this.scanCanonicalEntities('nouns')
const verbs = await this.scanCanonicalEntities('verbs')
const vectored = await this.scanVectoredNounCount()
if (this.ledgerActivityStamp() === activityBefore) {
derived = { nouns: nouns.count, verbs: verbs.count, vectored }
}
}
if (derived === null) {
this.allCountsSuspect = true
prodLog.narrate(
`[FileSystemStorage] count-ledger derivation could not finish on a quiet store ` +
`after ${attempt} attempts (${Date.now() - startedAt}ms) — writes landed during ` +
`every walk. The ALL-visibility scalars stay SUSPECT and must not be subtracted ` +
`against; brain.repairIndex() derives them under a recount barrier.`
)
return
}
this.totalNounCountAll = derived.nouns
this.totalVerbCountAll = derived.verbs
this.totalVectoredNounCount = derived.vectored
this.allCountsDerivedBy = 'identity-record'
this.allCountsSuspect = false
await this.persistCounts()
prodLog.narrate(
`[FileSystemStorage] count-ledger derivation finished in ${Date.now() - startedAt}ms: ` +
`${derived.nouns} nouns / ${derived.verbs} verbs / ${derived.vectored} vectored nouns` +
(beforeNouns !== derived.nouns ||
beforeVerbs !== derived.verbs ||
beforeVectored !== derived.vectored
? ` (corrected from ${beforeNouns} / ${beforeVerbs} / ${beforeVectored} — the ` +
`difference is ghost and scar containers the old rule counted as entities)`
: ' (unchanged)') +
` — persisted, stamped identity-record, no longer suspect.`
)
} catch (error) {
// The ledger stays suspect and the next open retries. Loud: a
// denominator nobody can derive is a fact an operator must have.
this.allCountsSuspect = true
prodLog.error(
`[FileSystemStorage] count-ledger derivation FAILED after ` +
`${Date.now() - startedAt}ms — the ALL-visibility scalars remain SUSPECT ` +
`and must not be subtracted against; the next open retries:`,
error
)
}
})()
}
/**
* @description A cheap witness that the ledger changed while a walk was
* running. Every landed write moves one of these live counters, so an
* unchanged stamp across a walk means no write landed during it.
* @returns A value that differs whenever the live ALL scalars have moved.
*/
private ledgerActivityStamp(): string {
return `${this.totalNounCountAll}:${this.totalVerbCountAll}:${this.totalVectoredNounCount}`
}
/**
* @description Resolve once any background count-ledger derivation has
* settled (succeeded or failed). Resolves immediately when none was needed.
* Exists so tests and operators can observe the ledger's honest value rather
* than race it; nothing in the read path waits on this.
* @returns A promise that settles with the derivation.
*/
public async whenCountLedgerSettled(): Promise<void> {
await this.countLedgerDerivation
}
private async scanCanonicalEntities(
kind: 'nouns' | 'verbs'
): Promise<{ count: number; sampleDirs: string[] }> {
@ -2916,10 +3341,15 @@ export class FileSystemStorage extends BaseStorage {
lastUpdated: new Date().toISOString()
}
await fs.promises.writeFile(
this.countsFilePath,
JSON.stringify(counts, null, 2)
)
// ATOMIC (temp + rename), never a plain writeFile. A direct write
// truncates the file first, so every persist opened a window — measured
// at roughly 750ms after a flush or close on a real store — in which a
// concurrent reader saw counts.json EMPTY. An empty file is unparseable,
// and an unparseable ledger sends the next open down the full-rescan
// path: the cheapest file in the store was costing the most expensive
// recovery. The rename is atomic, so a reader sees the old ledger or the
// new one, never neither.
await this.writeFileAtomic(this.countsFilePath, JSON.stringify(counts, null, 2))
} catch (error) {
console.error('Error persisting counts:', error)
}

View file

@ -125,6 +125,36 @@ export interface WriterLockInfo {
rootDir?: string // Convenience for log lines / error messages
}
/**
* THE CLEAN-CLOSE RECORD. Written by `releaseWriterLock()` at the instant it
* gives up the writer lock, naming the lock identity it released. The next
* `acquireWriterLock()` reads it and can then say from a RECORD, not from a
* guess whether the previous writer left on purpose.
*
* Why a record and not PID liveness: "the recorded PID is no longer alive" is
* true of every orderly restart AND of every crash, so the two were reported
* identically ("appears dead") and neither could be trusted. Worse, the same
* inference fails the other way when the operating system RECYCLES the pid
* a live unrelated process makes a long-dead writer's lock look held, and the
* store refuses to open naming a pid that was never Brainy. A record settles
* both: matched the previous writer closed cleanly, nothing to recover;
* absent say so, and name what recovery the open will now run.
*
* Lifecycle: written at release, consumed (deleted) by the next successful
* lock claim a record must never outlive the lock generation it describes,
* or it would vouch for a later crash.
*/
export interface WriterCloseRecord {
pid: number
hostname: string
/** `startedAt` of the lock this close released — the identity match key. */
startedAt: string
/** ISO timestamp at which the lock was released. */
closedAt: string
/** Brainy version that performed the close. */
version: string
}
/**
* FNV-1a hash returning a 2-char hex bucket (00-ff).
* Distributes system keys across 256 sub-prefixes to avoid
@ -1407,6 +1437,29 @@ export abstract class BaseStorage extends BaseStorageAdapter {
return this.listObjectsUnderPath(prefix)
}
/**
* @description The IMMEDIATE child directory names under a prefix one
* level, no recursion. See the seam's JSDoc (`db/types.ts`) for why a
* separate door exists. This default derives them from the recursive
* listing, so it is never WRONG, only never faster; the filesystem adapter
* overrides it with a single directory read.
* @param prefix - Storage-root-relative directory prefix.
* @returns The child directory names (not paths), in listing order.
*/
public async listRawPrefixes(prefix: string): Promise<string[]> {
await this.ensureInitialized()
const paths = await this.listObjectsUnderPath(prefix)
const normalizedPrefix = prefix.endsWith('/') ? prefix : `${prefix}/`
const names = new Set<string>()
for (const p of paths) {
const rest = p.startsWith(normalizedPrefix) ? p.slice(normalizedPrefix.length) : null
if (rest === null) continue
const slash = rest.search(/[/\\]/)
if (slash > 0) names.add(rest.slice(0, slash))
}
return [...names]
}
/**
* Remove every object under a storage-root-relative prefix. The filesystem
* adapter overrides this with a recursive directory removal; this default

View file

@ -1217,6 +1217,13 @@ export interface RepairFamilyReport {
skipped?: string
/** Why the outcome is what it is when neither `detail` nor `skipped` says it. */
reason?: string
/**
* The phase's own wall, in milliseconds. A repair on a production store ran
* for over thirty minutes without a single line of output; an operator had
* to read `top` to know it was alive. A receipt that cannot say WHERE the
* time went is not a receipt every row carries its own.
*/
durationMs?: number
}
/** The full receipt returned by repairIndex(). */

View file

@ -153,3 +153,83 @@ export function assessProviderHealth(provider: unknown): ProviderHealthAssessmen
reasons: readiness === 'not-ready' ? ['isReady() returned false'] : []
}
}
/**
* @description A provider's self-report that it is REBUILDING ITS OWN index
* right now. Returned by the optional `rebuildInProgress()` hook.
*
* The distinction this exists to make: a provider reporting `serving: false`
* because it is BROKEN and a provider reporting `serving: false` because it is
* BUSY BUILDING ITSELF look identical through `healthReport()` alone, and
* brainy treated both the same way it called `rebuild()` and waited for it,
* on the foreground of `init()`. A production store whose metadata provider
* had to rebuild paid 641 SECONDS of that wait before `init()` returned, with
* every other family idle behind it.
*
* A provider that reports progress here owns its own rebuild: brainy neither
* starts one nor waits for it, `init()` returns, the other families serve, and
* THAT family's doors refuse by name carrying this progress until the
* provider reports itself serving.
*
* Every field but `phase` is optional and every field is a MEASUREMENT: a
* provider reports only what it actually tracks, never an estimate dressed as
* a fact.
*/
export interface ProviderRebuildProgress {
/** The provider's own name for what it is doing. Quoted verbatim in refusals. */
phase: string
/** Units completed so far, if the provider counts them. */
done?: number
/** Units expected in total, if the provider knows it. */
total?: number
/** Epoch millis when this rebuild started, if the provider tracks it. */
startedAt?: number
}
/** A provider that can report a rebuild it is running itself. */
interface MaybeRebuildingProvider {
rebuildInProgress?: () => ProviderRebuildProgress | null
}
/**
* @description Ask a provider whether it is rebuilding itself right now.
* Synchronous, O(1), feature-detected: a provider without the hook reports
* nothing and is treated exactly as before.
* @param provider - Any index provider, or `null`/`undefined`.
* @returns The provider's progress, or `null` when it is not rebuilding (or
* does not implement the hook).
*/
export function assessProviderRebuild(provider: unknown): ProviderRebuildProgress | null {
const p = provider as MaybeRebuildingProvider | null | undefined
if (p == null || typeof p.rebuildInProgress !== 'function') return null
try {
const progress = p.rebuildInProgress()
if (!progress || typeof progress.phase !== 'string' || progress.phase.length === 0) {
return null
}
return progress
} catch {
// A throwing hook says nothing trustworthy about a rebuild; fall through to
// the ordinary health verdict rather than inventing one.
return null
}
}
/**
* @description Render a rebuild progress report as one operator-facing clause,
* for a refusal message. Includes only what the provider actually measured.
* @param progress - The provider's report.
* @returns A clause such as `rebuilding ("metadata shadow build", 4,096/14,056, 12s elapsed)`.
*/
export function describeRebuildProgress(progress: ProviderRebuildProgress): string {
const parts: string[] = [`"${progress.phase}"`]
if (typeof progress.done === 'number' && typeof progress.total === 'number') {
parts.push(`${progress.done.toLocaleString()}/${progress.total.toLocaleString()}`)
} else if (typeof progress.done === 'number') {
parts.push(`${progress.done.toLocaleString()} done`)
}
if (typeof progress.startedAt === 'number') {
parts.push(`${Math.round((Date.now() - progress.startedAt) / 1000)}s elapsed`)
}
return `rebuilding (${parts.join(', ')})`
}

View file

@ -266,6 +266,26 @@ export const prodLog = {
console.error(message, ...args)
},
/**
* THE NARRATION CHANNEL always visible, exactly like `error`.
*
* `warn`/`info`/`log` below are clamped to ERROR in any environment that
* looks like production (see isProductionEnvironment), which is the right
* default for chatter and the wrong one for the two things an operator is
* entitled to hear from a database no matter what: WHY IT IS SLOW and WHAT
* IT IS DOING ABOUT IT. A production service opening a 16 GB store spent
* three minutes emitting nothing at all the phase timings that would have
* named the slow phase were written to `warn` and thrown away by the log
* level. Progress and cost narration goes here; it is never a per-record
* line, always a phase, a wall, or a bounded-cadence heartbeat.
*
* `silent: true` still silences it that is the consumer's explicit
* request, not a cost default.
*/
narrate: (message?: any, ...args: any[]) => {
console.warn(message, ...args)
},
// These are suppressed in production unless BRAINY_LOG_LEVEL is set
warn: (message?: any, ...args: any[]) => smartConsole.warn(message, ...args),
info: (message?: any, ...args: any[]) => smartConsole.info(message, ...args),

View file

@ -2241,6 +2241,74 @@ export class MetadataIndexManager implements MetadataIndexProvider {
break
}
// ===== ARRAY SET OPERATORS =====
// An element-indexed array field makes all three exact on the
// index path. They were previously ABSENT from this switch, so
// `fieldResults` kept its initial `[]` and the whole find()
// returned an empty page — a documented, matcher-implemented
// operator answering silently wrong. Served here instead.
// hasAll: [a, b] — the field's array contains EVERY operand:
// the intersection of each element's posting set.
case 'hasAll': {
if (!Array.isArray(operand)) {
fieldResults = []
break
}
if (operand.length === 0) {
// Vacuously true of every row that HAS the field.
const anyBitmap = (this.columnStore && this.columnStore.hasField(field))
? await this.columnStore.rangeQuery(field)
: await this.getExistsBitmapLegacy(field)
fieldResults = this.idMapper.intsIterableToUuids(anyBitmap)
break
}
let intersection: Set<string> | null = null
for (const item of operand) {
const ids = new Set(await this.getIds(field, item))
if (intersection === null) {
intersection = ids
} else {
for (const id of [...intersection]) {
if (!ids.has(id)) intersection.delete(id)
}
}
if (intersection.size === 0) break
}
fieldResults = intersection ? [...intersection] : []
break
}
// noneOf: [a, b] — the field's value is NONE of the operands:
// the complement of their union.
case 'noneOf': {
if (!Array.isArray(operand)) {
fieldResults = []
break
}
const excludeInts: number[] = []
for (const value of operand) {
for (const uuid of await this.getIds(field, value)) {
const intId = this.idMapper.getInt(uuid)
if (intId !== undefined) excludeInts.push(intId)
}
}
fieldResults = this.complementIds(excludeInts)
break
}
// excludes: value — the field's array does NOT contain the value:
// the complement of `contains`.
case 'excludes': {
const excludeInts: number[] = []
for (const uuid of await this.getIds(field, operand)) {
const intId = this.idMapper.getInt(uuid)
if (intId !== undefined) excludeInts.push(intId)
}
fieldResults = this.complementIds(excludeInts)
break
}
// ===== MISSING OPERATOR =====
// missing: boolean - equivalent to exists: !boolean
case 'missing': {
@ -2257,6 +2325,27 @@ export class MetadataIndexManager implements MetadataIndexProvider {
}
break
}
// ===== EVERYTHING ELSE: REFUSED BY NAME, NEVER ANSWERED EMPTY ====
// An equality/range posting index cannot evaluate a substring, a
// pattern or an array length without reading every row, and this
// path exists precisely to avoid that. It used to fall out of the
// switch with `fieldResults` still `[]`, so `find({ where: { name:
// { startsWith: 'a' } } })` returned an empty page and looked like
// an answer. An accepted operator either works or refuses — the
// matcher's own support for these operators governs in-memory
// filtering, never an index-backed find().
default:
throw new BrainyError(
`Filter operator "${op}" on field "${rawField}" cannot be served by the ` +
`metadata index: an equality/range posting index cannot evaluate substrings, ` +
`patterns or array lengths without reading every row. It is REFUSED rather ` +
`than answered with an empty page. Filter on an indexable operator ` +
`(equals/eq, notEquals/ne, oneOf/in, noneOf, greaterThan/gt, ` +
`greaterThanOrEqual/gte, lessThan/lt, lessThanOrEqual/lte, between, contains, ` +
`excludes, hasAll, exists, missing) and narrow the rest in your own code.`,
'INVALID_QUERY'
)
}
// Intersect this operator's matches with the running set (AND semantics
// for multiple operators on the same field).

View file

@ -83,3 +83,27 @@ export function getAugmentationVersion(service: string): { augmentation: string;
version: getBrainyVersion()
}
}
/**
* The API-contract version this build implements a single integer that two
* engines can compare without probing prototypes.
*
* A MINOR release is ADDITIVE: doors and error codes may be added, never
* removed or narrowed, and the contract integer does not move. A MAJOR release
* is what a REQUIRED door's removal or a behavioural narrowing costs, and it
* bumps this integer. A consumer pinning `brainyContract` in a peer range is
* therefore pinning "what I may call", not "which build I run".
*
* Declared in package.json as `"brainyContract"` so a manifest, a tool, or a
* sibling package can read it without importing the engine, and returned here
* so a running process can state its own.
*/
export const BRAINY_CONTRACT_VERSION = 1 as const
/**
* @description The API-contract version this build implements.
* @returns The contract integer see {@link BRAINY_CONTRACT_VERSION}.
*/
export function contractVersion(): number {
return BRAINY_CONTRACT_VERSION
}

View file

@ -6,6 +6,7 @@
*/
import { Readable, Writable } from 'stream'
import { prodLog } from '../utils/logger.js'
import crypto from 'crypto'
import { v4 as uuidv4 } from '../universal/uuid.js'
import { Brainy } from '../brainy.js'
@ -66,6 +67,15 @@ export class VirtualFileSystem implements IVirtualFileSystem {
private config: Required<Omit<VFSConfig, 'rootEntityId'>> & { rootEntityId?: string }
private rootEntityId?: string
private initialized = false
/**
* The one-time old-root sweep, in flight. See {@link sweepOldRootsIfNeeded}.
*/
private rootSweep?: Promise<void>
/**
* Where the completed old-root sweep is recorded. Engine plumbing under
* `_system/`, like every other marker there never enumerated as data.
*/
private static readonly ROOT_SWEEP_MARKER_PATH = '_system/vfs-root-sweep.json'
private currentUser: string = 'system' // Track current user for collaboration
// Knowledge Layer features available via augmentation (brain.use('knowledge'))
@ -143,8 +153,17 @@ export class VirtualFileSystem implements IVirtualFileSystem {
// Create or find root entity
this.rootEntityId = await this.initializeRoot()
// Clean up old UUID-based roots (one-time migration)
await this.cleanupOldRoots()
// Clean up old UUID-based roots — ONCE PER STORE, BEHIND THE DOORS.
// This is a migration sweep for roots created before the fixed root id
// existed. It ran on EVERY open, forever: a filtered find over the whole
// store hunting for duplicates that a store has either always had or
// never will. MEASURED on a 14,056-noun / 72,679-verb store: the phase it
// dominates cost 43-53 SECONDS of every open, warm reopens included.
// Now: a durable marker records that the sweep has run, and a store
// carrying it never sweeps again; a store without one sweeps in the
// BACKGROUND (the sweep only removes duplicate roots — nothing serves
// from them — and it was always declared non-critical).
this.rootSweep = this.sweepOldRootsIfNeeded()
// Initialize projection registry with auto-discovery of built-in projections
this.projectionRegistry = new ProjectionRegistry()
@ -394,6 +413,88 @@ export class VirtualFileSystem implements IVirtualFileSystem {
*
* This is a one-time migration helper that can be removed in future versions.
*/
/**
* @description Run the old-root sweep at most once per store, in the
* background, and record that it ran. See the call site in {@link init} for
* the measurement that made this necessary.
* @returns A promise that settles when the sweep has finished (or was
* skipped); nothing in the read path awaits it.
*/
private async sweepOldRootsIfNeeded(): Promise<void> {
const store = this.rawObjectStore()
if (store === null) {
// A storage adapter with no raw-object door cannot carry the marker.
// Sweep every open, as before — correctness over cost.
await this.cleanupOldRoots()
return
}
try {
const marker = await store.readRawObject(VirtualFileSystem.ROOT_SWEEP_MARKER_PATH)
if (marker !== null && marker !== undefined) return
} catch {
// Unreadable marker: sweep, and rewrite it below.
}
prodLog.narrate(
'[VFS] one-time sweep for pre-fixed-id root directories running in the background — ' +
'the open does not wait for it, and once it has run this store never sweeps again.'
)
const startedAt = Date.now()
await this.cleanupOldRoots()
try {
await store.writeRawObject(VirtualFileSystem.ROOT_SWEEP_MARKER_PATH, {
sweptAt: new Date().toISOString(),
durationMs: Date.now() - startedAt
})
prodLog.narrate(
`[VFS] old-root sweep complete in ${Date.now() - startedAt}ms and recorded — ` +
'no future open pays for it.'
)
} catch (error) {
// Unrecorded sweep = the next open sweeps again. Conservative, and said
// out loud rather than quietly repeated forever.
prodLog.narrate(
`[VFS] old-root sweep finished in ${Date.now() - startedAt}ms but could NOT be ` +
`recorded (${(error as Error).message}) — the next open will sweep again.`
)
}
}
/**
* @description Settle once the background old-root sweep has finished.
* Resolves immediately when the store already carried the marker. Exists so
* tests and operators can observe the sweep instead of racing it; no read
* path waits on it.
* @returns A promise that settles with the sweep.
*/
public async whenRootSweepSettled(): Promise<void> {
await this.rootSweep
}
/**
* @description The brain's storage adapter, narrowed to the raw-object door
* this migration marker needs. Boundary: `Brainy.storage` is private, and
* this is the same reach-in the engine uses elsewhere for exactly this kind
* of engine-internal artifact. Returns null when the adapter has no
* raw-object door.
*/
private rawObjectStore(): {
readRawObject: (key: string) => Promise<unknown>
writeRawObject: (key: string, value: unknown) => Promise<void>
} | null {
const storage = (this.brain as unknown as { storage?: Record<string, unknown> }).storage
if (
storage &&
typeof storage.readRawObject === 'function' &&
typeof storage.writeRawObject === 'function'
) {
return storage as unknown as {
readRawObject: (key: string) => Promise<unknown>
writeRawObject: (key: string, value: unknown) => Promise<void>
}
}
return null
}
private async cleanupOldRoots(): Promise<void> {
try {
// Find any old VFS roots with UUID-based IDs (not our fixed ID)

View file

@ -0,0 +1,251 @@
/**
* @module tests/integration/count-ledger-identity-record
* @description THE COUNT LEDGER COUNTS RECORDS, NOT DIRECTORIES and heals
* itself when it was derived the other way.
*
* Measured on a real store: the ALL-visibility ledger read 14,231 nouns
* against 14,056 identity records, and 72,729 verbs against 72,679 exactly
* that store's 25 noun and 50 verb SCAR directories (empty `<id>/` containers
* left by a pre-8.3.1 partial delete). Two copies of the SAME archive derived
* different numbers, because each had been persisted at a different moment
* under the old container rule. A downstream index heal subtracted against
* those denominators and reported remaining work that did not exist.
*
* The membership predicate is the IDENTITY RECORD (the metadata content leg).
* The scan already applies it; what is pinned here is that a ledger persisted
* under the OLD rule does not go on lying it is corrected in the background,
* without blocking the open, and two copies of one archive agree.
*/
import { describe, it, expect, afterEach } from 'vitest'
import {
mkdtempSync,
mkdirSync,
rmSync,
writeFileSync,
readFileSync,
cpSync,
existsSync
} from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { Brainy } from '../../src/brainy.js'
import { NounType } from '../../src/types/graphTypes.js'
import { FileSystemStorage as FileSystemStorageClass } from '../../src/storage/adapters/fileSystemStorage.js'
import type { FileSystemStorage } from '../../src/storage/adapters/fileSystemStorage.js'
const NOUN_COUNT = 6
const NOUN_SCARS = 3
const VERB_SCARS = 2
/** A REAL two-hex shard — the scan skips any directory that is not one. */
const SCAR_SHARD = 'ab'
function makeTempDir(): string {
return mkdtempSync(join(tmpdir(), 'brainy-count-ledger-'))
}
/** The FileSystemStorage behind a brain. */
function storageOf(brain: Brainy): FileSystemStorage {
return (brain as unknown as { storage: FileSystemStorage }).storage
}
/**
* Add `count` empty `<id>/` container directories under
* `entities/<kind>/<shard>/` scars, exactly as a partial delete leaves them.
*/
function addScarContainers(dir: string, kind: 'nouns' | 'verbs', count: number): void {
for (let i = 0; i < count; i++) {
const id = `${SCAR_SHARD}5ca4000-0000-0000-0000-00000000000${i}`
mkdirSync(join(dir, 'entities', kind, SCAR_SHARD, id), { recursive: true })
}
}
/** Add one GHOST container: a `vectors.json` leg with no identity record. */
function addGhostContainer(dir: string): void {
const id = `${SCAR_SHARD}9405700-0000-0000-0000-000000000000`
const idDir = join(dir, 'entities', 'nouns', SCAR_SHARD, id)
mkdirSync(idDir, { recursive: true })
writeFileSync(join(idDir, 'vectors.json'), JSON.stringify({ id, vector: [0.1, 0.2] }))
}
/**
* Rewrite counts.json into the LEGACY shape: ALL scalars inflated by the
* containers, and no `allCountsDerivedBy` stamp exactly what a store carried
* when it was last written by a build that counted directories.
*/
function writeLegacyCountsLedger(dir: string, inflateNouns: number, inflateVerbs: number): void {
const file = join(dir, '_system', 'counts.json')
const counts = JSON.parse(readFileSync(file, 'utf-8'))
counts.totalNounCountAll = (counts.totalNounCountAll ?? 0) + inflateNouns
counts.totalVerbCountAll = (counts.totalVerbCountAll ?? 0) + inflateVerbs
delete counts.allCountsDerivedBy
delete counts.allCountsSuspect
writeFileSync(file, JSON.stringify(counts, null, 2))
}
/**
* Seed a store and return the HONEST ledger it holds when freshly written
* the baseline the correction must return to. Read from the engine rather than
* hardcoded: an open creates its own rows (the VFS root), and a pin that
* asserts a literal would be pinning that incidental fact instead of the rule.
*/
async function seedStore(dir: string): Promise<{ nouns: number; verbs: number }> {
const brain = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
await brain.init()
const ids: string[] = []
for (let i = 0; i < NOUN_COUNT; i++) {
ids.push(await brain.add({ data: `entity number ${i}`, type: NounType.Concept }))
}
await brain.relate({ from: ids[0], to: ids[1], type: 'relatedTo' } as never)
await brain.relate({ from: ids[1], to: ids[2], type: 'relatedTo' } as never)
await brain.flush()
const ledger = await storageOf(brain).getCanonicalCounts()
const baseline = { nouns: ledger.nouns.all, verbs: ledger.verbs.all }
await brain.close()
return baseline
}
/**
* Make the ledger walk take `ms` so a test can observe the open completing
* WITHOUT it. Patches the prototype before any brain is constructed; returns
* the restore function.
*/
function slowTheLedgerWalk(ms: number): () => void {
const proto = (
FileSystemStorageClass as unknown as {
prototype: Record<string, (...args: unknown[]) => Promise<unknown>>
}
).prototype
const real = proto.scanCanonicalEntities
proto.scanCanonicalEntities = async function slow(this: unknown, ...args: unknown[]) {
await new Promise((r) => setTimeout(r, ms))
return real.apply(this, args)
}
return () => { proto.scanCanonicalEntities = real }
}
describe('the canonical count ledger', () => {
const dirs: string[] = []
afterEach(() => {
for (const d of dirs.splice(0)) {
try { rmSync(d, { recursive: true, force: true }) } catch { /* ignore */ }
}
})
function trackDir(): string {
const dir = makeTempDir()
dirs.push(dir)
return dir
}
it('corrects a legacy container-rule ledger in the background, counting identity records', async () => {
const dir = trackDir()
const baseline = await seedStore(dir)
// Scars and a ghost: containers with no identity record.
addScarContainers(dir, 'nouns', NOUN_SCARS)
addScarContainers(dir, 'verbs', VERB_SCARS)
addGhostContainer(dir)
// The ledger as the old rule left it: every container counted.
writeLegacyCountsLedger(dir, NOUN_SCARS + 1, VERB_SCARS)
const restore = slowTheLedgerWalk(1_500)
let brain: Brainy
try {
const openStarted = Date.now()
brain = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
await brain.init()
const openMs = Date.now() - openStarted
const storage = storageOf(brain)
// THE OPEN DID NOT WAIT. Two walks of 1.5s each would have added 3s.
expect(openMs).toBeLessThan(2_500)
// And while it runs, the scalars say so instead of being subtracted against.
const atOpen = await storage.getCanonicalCounts()
expect(atOpen.suspect).toBe(true)
expect(atOpen.nouns.all).toBe(baseline.nouns + NOUN_SCARS + 1)
await storage.whenCountLedgerSettled()
} finally {
restore()
}
const storage = storageOf(brain!)
const healed = await storage.getCanonicalCounts()
expect(healed.nouns.all).toBe(baseline.nouns)
expect(healed.verbs.all).toBe(baseline.verbs)
expect(healed.suspect).toBe(false)
// And it is PERSISTED with the honest stamp — the correction survives a
// reopen instead of being re-derived (or re-lost) every time.
await brain!.close()
const persisted = JSON.parse(readFileSync(join(dir, '_system', 'counts.json'), 'utf-8'))
expect(persisted.totalNounCountAll).toBe(baseline.nouns)
expect(persisted.totalVerbCountAll).toBe(baseline.verbs)
expect(persisted.allCountsDerivedBy).toBe('identity-record')
const reopened = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
await reopened.init()
const afterReopen = await storageOf(reopened).getCanonicalCounts()
expect(afterReopen.nouns.all).toBe(baseline.nouns)
expect(afterReopen.suspect).toBe(false)
await reopened.close()
}, 180_000)
it('derives the same number from two copies of one archive', async () => {
const source = trackDir()
const baseline = await seedStore(source)
addScarContainers(source, 'nouns', NOUN_SCARS)
addGhostContainer(source)
// Two copies of the SAME bytes, each carrying a DIFFERENT legacy ledger —
// the situation that made one archive report 14,231 and its twin 14,081.
const copyA = trackDir()
const copyB = trackDir()
cpSync(source, copyA, { recursive: true })
cpSync(source, copyB, { recursive: true })
writeLegacyCountsLedger(copyA, NOUN_SCARS + 1, 0)
writeLegacyCountsLedger(copyB, 1, 0)
const derived: number[] = []
for (const dir of [copyA, copyB]) {
const brain = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
await brain.init()
const storage = storageOf(brain)
await storage.whenCountLedgerSettled()
derived.push((await storage.getCanonicalCounts()).nouns.all)
await brain.close()
}
expect(derived[0]).toBe(derived[1])
expect(derived[0]).toBe(baseline.nouns)
}, 180_000)
it('writes counts.json atomically — no reader ever sees it empty', async () => {
const dir = trackDir()
await seedStore(dir)
const file = join(dir, '_system', 'counts.json')
expect(existsSync(file)).toBe(true)
const brain = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
await brain.init()
const storage = storageOf(brain)
// Watch the ledger across many persists. A truncating write leaves a
// window in which the file parses as nothing; a temp+rename never does.
let sawUnparseable = 0
const watcher = setInterval(() => {
try {
JSON.parse(readFileSync(file, 'utf-8'))
} catch {
sawUnparseable++
}
}, 1)
for (let i = 0; i < 40; i++) {
await (storage as unknown as { persistCounts: () => Promise<void> }).persistCounts()
}
clearInterval(watcher)
await brain.close()
expect(sawUnparseable).toBe(0)
}, 180_000)
})

View file

@ -0,0 +1,151 @@
/**
* @module tests/integration/filter-operator-conformance
* @description THE OPERATOR SET, AND WHAT EACH TOKEN DOES ON THE INDEX PATH.
*
* The contract-1 manifest splits this engine's `where` operators three ways
* served, served-beyond-baseline, refused-by-name and two engines must agree
* token for token. This lane is the machine-checkable side of that agreement:
* it asserts the EXACT accepted set (so a manifest can be diffed against a run
* rather than against prose), and it pins each of the three classes.
*
* The defect it closes: the metadata index's operator switch had no default
* case, so an operator it does not implement `hasAll`, `noneOf`, `excludes`,
* `startsWith`, `endsWith`, `matches`, `length` left the field's match set at
* its initial `[]` and `find()` returned an empty page. A documented operator,
* implemented in the in-memory matcher, answering silently wrong. Three of the
* seven are now SERVED on the index path; the other four are REFUSED BY NAME,
* because an equality/range posting index cannot evaluate a substring, a
* pattern or an array length without reading every row.
*/
import { describe, it, expect, afterEach } from 'vitest'
import { mkdtempSync, rmSync, readFileSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { Brainy } from '../../src/brainy.js'
import { NounType } from '../../src/types/graphTypes.js'
import { contractVersion, BRAINY_CONTRACT_VERSION } from '../../src/utils/version.js'
/** The accepted `where` value-operator tokens, as a sorted list. */
const ACCEPTED_OPERATORS = [
'between', 'contains', 'endsWith', 'eq', 'equals', 'excludes', 'exists',
'greaterThan', 'greaterThanOrEqual', 'gt', 'gte', 'hasAll', 'in', 'length',
'lessThan', 'lessThanOrEqual', 'lt', 'lte', 'matches', 'missing', 'ne',
'noneOf', 'notEquals', 'oneOf', 'startsWith'
] as const
/** Served on the index path with exact posting-set semantics. */
const SERVED_ON_INDEX = [
'between', 'contains', 'eq', 'equals', 'exists', 'greaterThan',
'greaterThanOrEqual', 'gt', 'gte', 'in', 'lessThan', 'lessThanOrEqual',
'lt', 'lte', 'missing', 'ne', 'notEquals', 'oneOf',
'excludes', 'hasAll', 'noneOf'
] as const
/** Accepted by name, refused by the index path — never answered empty. */
const REFUSED_BY_INDEX = ['endsWith', 'length', 'matches', 'startsWith'] as const
describe('filter operator conformance', () => {
const dirs: string[] = []
const brains: Brainy[] = []
afterEach(async () => {
for (const b of brains.splice(0)) {
try { await b.close() } catch { /* already closed */ }
}
for (const d of dirs.splice(0)) {
try { rmSync(d, { recursive: true, force: true }) } catch { /* ignore */ }
}
})
async function seeded(): Promise<Brainy> {
const dir = mkdtempSync(join(tmpdir(), 'brainy-operators-'))
dirs.push(dir)
const brain = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
brains.push(brain)
await brain.init()
await brain.add({
data: 'a document about ferrets',
type: NounType.Document,
metadata: { tags: ['ferret', 'small', 'furry'], team: 'alpha' }
})
await brain.add({
data: 'a document about whales',
type: NounType.Document,
metadata: { tags: ['whale', 'large'], team: 'beta' }
})
await brain.flush()
return brain
}
it('the accepted operator set is exactly these 25 tokens', async () => {
const brain = await seeded()
// The engine names its own valid set in the refusal it raises for an
// unknown token — the honest place to read it from.
let message = ''
try {
await brain.find({ where: { team: { notIn: ['alpha'] } } } as never)
} catch (err) {
message = (err as Error).message
}
expect(message).toMatch(/Unknown filter operator "notIn"/)
const listed = (message.match(/Valid operators: ([^.]+)\./)?.[1] ?? '')
.split(',')
.map((t) => t.trim())
.filter(Boolean)
.sort()
expect(listed).toEqual([...ACCEPTED_OPERATORS].sort())
expect(listed.length).toBe(25)
// Four tokens a sibling manifest listed as served aliases are NOT in this
// engine's set and never have been — they raise INVALID_QUERY.
for (const absent of ['is', 'isNot', 'greaterEqual', 'lessEqual']) {
expect(listed).not.toContain(absent)
await expect(
brain.find({ where: { team: { [absent]: 'alpha' } } } as never)
).rejects.toThrow(/Unknown filter operator/)
}
}, 120_000)
it('serves hasAll, noneOf and excludes on the index path — never an empty page', async () => {
const brain = await seeded()
const hasAll = await brain.find({ where: { tags: { hasAll: ['ferret', 'furry'] } } } as never)
expect(hasAll.length).toBe(1)
expect((hasAll[0] as { metadata?: Record<string, unknown> }).metadata?.team).toBe('alpha')
const noneOf = await brain.find({ where: { team: { noneOf: ['alpha'] } } } as never)
expect(noneOf.length).toBe(1)
expect((noneOf[0] as { metadata?: Record<string, unknown> }).metadata?.team).toBe('beta')
const excludes = await brain.find({ where: { tags: { excludes: 'whale' } } } as never)
expect(excludes.length).toBe(1)
expect((excludes[0] as { metadata?: Record<string, unknown> }).metadata?.team).toBe('alpha')
// hasAll with an operand nothing carries is EMPTY because it is empty —
// the honest zero, reached by evaluating the operator.
const none = await brain.find({ where: { tags: { hasAll: ['ferret', 'whale'] } } } as never)
expect(none.length).toBe(0)
}, 120_000)
it('refuses the four index-unserveable operators BY NAME', async () => {
const brain = await seeded()
for (const op of REFUSED_BY_INDEX) {
const operand = op === 'length' ? 3 : 'a'
await expect(
brain.find({ where: { team: { [op]: operand } } } as never),
`${op} must refuse, never answer an empty page`
).rejects.toThrow(new RegExp(`Filter operator "${op}".*cannot be served by the metadata index`, 's'))
}
}, 120_000)
it('declares its contract version in code and in package.json', async () => {
expect(contractVersion()).toBe(1)
expect(BRAINY_CONTRACT_VERSION).toBe(1)
const pkg = JSON.parse(readFileSync(join(process.cwd(), 'package.json'), 'utf-8'))
expect(pkg.brainyContract).toBe(contractVersion())
})
it('the three classes partition the accepted set', () => {
expect([...SERVED_ON_INDEX, ...REFUSED_BY_INDEX].sort()).toEqual([...ACCEPTED_OPERATORS].sort())
})
})

View file

@ -0,0 +1,94 @@
/**
* @module tests/integration/flush-watcher-event-driven
* @description THE FLUSH-REQUEST WATCH IS EVENT-DRIVEN.
*
* It used to `readdir` the request directory every 500 ms, per brain, for the
* life of every writer armed on every non-reader brain whether or not any
* inspector process existed. MEASURED on a production process holding 21
* brains: 42 directory reads per second on a completely idle service, plus a
* stale-request GC pass on every one of them.
*
* The law: a request that has not been made is not a cause. The arrival itself
* wakes the watcher, so the request is seen SOONER than the poll saw it, and a
* slow safety sweep covers filesystems that drop watch events and the GC.
*/
import { describe, it, expect, afterEach, vi } from 'vitest'
import { mkdtempSync, rmSync, writeFileSync, mkdirSync } from 'node:fs'
import * as nodeFs from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { Brainy } from '../../src/brainy.js'
import { NounType } from '../../src/types/graphTypes.js'
describe('the flush-request watcher', () => {
const dirs: string[] = []
const brains: Brainy[] = []
afterEach(async () => {
for (const b of brains.splice(0)) {
try { await b.close() } catch { /* already closed */ }
}
for (const d of dirs.splice(0)) {
try { rmSync(d, { recursive: true, force: true }) } catch { /* ignore */ }
}
vi.restoreAllMocks()
})
async function openWriter(): Promise<{ brain: Brainy; dir: string }> {
const dir = mkdtempSync(join(tmpdir(), 'brainy-flush-watch-'))
dirs.push(dir)
const brain = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
brains.push(brain)
await brain.init()
await brain.add({ data: 'a row', type: NounType.Concept })
await brain.flush()
return { brain, dir }
}
it('does not poll the request directory on an idle writer', async () => {
const { dir } = await openWriter()
const reqDir = join(dir, 'locks', '_flush_requests')
// Count real reads of the request directory over a window far longer than
// the old 500ms poll (which would have made ~16 of them).
const realReaddir = nodeFs.promises.readdir
let requestDirReads = 0
const spy = vi
.spyOn(nodeFs.promises, 'readdir')
.mockImplementation((async (p: unknown, ...rest: unknown[]) => {
if (String(p) === reqDir) requestDirReads++
return (realReaddir as unknown as (...a: unknown[]) => Promise<unknown>)(p, ...rest)
}) as typeof nodeFs.promises.readdir)
await new Promise((r) => setTimeout(r, 8_000))
spy.mockRestore()
// The old poll: 500ms → ~16 reads. The safety sweep is 30s → 0 in this window.
expect(requestDirReads).toBeLessThanOrEqual(1)
}, 120_000)
it('answers a request that arrives, without waiting for the sweep', async () => {
const { brain, dir } = await openWriter()
const reqDir = join(dir, 'locks', '_flush_requests')
const ackDir = join(dir, 'locks', '_flush_responses')
mkdirSync(reqDir, { recursive: true })
// Drop a request exactly as an out-of-process inspector does.
const id = 'test-request-0001'
writeFileSync(join(reqDir, `${id}.req`), JSON.stringify({ at: Date.now() }))
// The ack must land far sooner than the 30s safety sweep.
const deadline = Date.now() + 10_000
let acked = false
while (Date.now() < deadline) {
try {
const entries = await nodeFs.promises.readdir(ackDir)
if (entries.some((e) => e.startsWith(id))) { acked = true; break }
} catch { /* dir not created yet */ }
await new Promise((r) => setTimeout(r, 100))
}
expect(acked, 'the watcher must answer an arriving request').toBe(true)
void brain
}, 120_000)
})

View file

@ -0,0 +1,152 @@
/**
* @module tests/integration/idle-costs-nothing
* @description AN IDLE BRAIN DOES NO WORK.
*
* A flush used to re-persist state identical to what was already on disk
* the provider flushes, the watermark stamps, the generation counter, the
* entity-tree stamp, roughly 28 writes because `flush()` never asked whether
* anything had changed.
*
* The field observation that started this: a production process holding 21
* brains printed "All indexes flushed to disk in 216601ms" per brain every
* ~35 seconds and idled at 1.26 cores, with no writes for ten minutes. This
* engine's cadence is WRITE-DRIVEN, so that observation is NOT explained by
* the cadence and is not claimed to be fixed here what is fixed is that such
* a call now costs nothing. Who was calling flush() remains open.
*
* The laws pinned here:
* (a) the persistence cadence arms only on a write a brain nobody writes
* to flushes zero times, however long it is left open;
* (b) a flush on a clean brain is O(1): no provider is called, nothing is
* written, and nothing is printed;
* (c) one write earns exactly one flush's worth of work, and no more.
*/
import { describe, it, expect, afterEach, vi } from 'vitest'
import { mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { Brainy } from '../../src/brainy.js'
import { NounType } from '../../src/types/graphTypes.js'
/** Wait for any in-flight background flush, then let the idle timer settle. */
async function drainCadence(brain: Brainy): Promise<void> {
const inner = brain as unknown as { _persistBackgroundFlight: Promise<void> | null }
await new Promise((r) => setTimeout(r, 3_000))
await (inner._persistBackgroundFlight ?? Promise.resolve())
await new Promise((r) => setTimeout(r, 500))
}
/** How long an idle brain is watched. Longer than the 30s flush interval. */
const IDLE_WATCH_MS = 90_000
describe('an idle brain costs nothing', () => {
const dirs: string[] = []
const brains: Brainy[] = []
afterEach(async () => {
for (const b of brains.splice(0)) {
try { await b.close() } catch { /* already closed */ }
}
for (const d of dirs.splice(0)) {
try { rmSync(d, { recursive: true, force: true }) } catch { /* ignore */ }
}
vi.restoreAllMocks()
})
async function openBrain(): Promise<Brainy> {
const dir = mkdtempSync(join(tmpdir(), 'brainy-idle-'))
dirs.push(dir)
const brain = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
brains.push(brain)
await brain.init()
return brain
}
it('flushes zero times over 90 idle seconds, and prints nothing', async () => {
const brain = await openBrain()
// One write and one flush to reach a clean, settled state — then nothing.
await brain.add({ data: 'the only write this test performs', type: NounType.Concept })
await brain.flush()
const logged: string[] = []
const origLog = console.log
console.log = ((...a: unknown[]) => { logged.push(a.map(String).join(' ')) }) as typeof console.log
// Watch the providers directly: a flush that runs calls all of them.
const storage = (brain as unknown as { storage: { flushCounts: () => Promise<void> } }).storage
const metadataIndex = (brain as unknown as { metadataIndex: { flush: () => Promise<void> } }).metadataIndex
const graphIndex = (brain as unknown as { graphIndex: { flush: () => Promise<void> } }).graphIndex
const countsSpy = vi.spyOn(storage, 'flushCounts')
const metadataSpy = vi.spyOn(metadataIndex, 'flush')
const graphSpy = vi.spyOn(graphIndex, 'flush')
try {
await new Promise((r) => setTimeout(r, IDLE_WATCH_MS))
} finally {
console.log = origLog
}
// (a) + (b): nothing ran, nothing was said.
expect(logged.filter((l) => /All indexes flushed to disk/.test(l))).toEqual([])
expect(logged.filter((l) => /Flushing Brainy indexes/.test(l))).toEqual([])
expect(countsSpy).not.toHaveBeenCalled()
expect(metadataSpy).not.toHaveBeenCalled()
expect(graphSpy).not.toHaveBeenCalled()
}, 180_000)
it('an explicit flush over a clean brain calls no provider and prints nothing', async () => {
const brain = await openBrain()
await brain.add({ data: 'one write', type: NounType.Concept })
await brain.flush() // this one does the work
const storage = (brain as unknown as { storage: { flushCounts: () => Promise<void> } }).storage
const metadataIndex = (brain as unknown as { metadataIndex: { flush: () => Promise<void> } }).metadataIndex
const countsSpy = vi.spyOn(storage, 'flushCounts')
const metadataSpy = vi.spyOn(metadataIndex, 'flush')
const logged: string[] = []
const origLog = console.log
console.log = ((...a: unknown[]) => { logged.push(a.map(String).join(' ')) }) as typeof console.log
try {
await brain.flush() // ...and this one has nothing to do
await brain.flush()
await brain.flush()
} finally {
console.log = origLog
}
expect(countsSpy).not.toHaveBeenCalled()
expect(metadataSpy).not.toHaveBeenCalled()
expect(logged.filter((l) => /All indexes flushed to disk/.test(l))).toEqual([])
}, 120_000)
it('one write earns exactly one flush', async () => {
const brain = await openBrain()
await brain.add({ data: 'first', type: NounType.Concept })
await brain.flush()
// Settle: the first write also kicked a BACKGROUND flush, which is not
// awaited by design. Drain it before counting, or its provider calls land
// inside this test's window and are attributed to the write below.
await drainCadence(brain)
// Count the flushes that actually RAN. (Provider spies cannot answer this:
// the storage adapter's own count ledger is write-through, so a write calls
// flushCounts() on its own account, with no flush involved.)
const logged: string[] = []
const origLog = console.log
console.log = ((...a: unknown[]) => { logged.push(a.map(String).join(' ')) }) as typeof console.log
const ran = () => logged.filter((l) => /All indexes flushed to disk/.test(l)).length
try {
await brain.add({ data: 'second — this is the cause', type: NounType.Concept })
await brain.flush()
expect(ran()).toBe(1)
// No further cause, no further work.
await brain.flush()
await brain.flush()
expect(ran()).toBe(1)
} finally {
console.log = origLog
}
}, 120_000)
})

View file

@ -11,14 +11,22 @@
* (1) IDENTITY, NOT CONTAINER the derivation counts one entity per
* metadata content leg (`metadata.json` or `.json.gz`), the same test
* `pruneOrphanedEntities()` uses, so the two agree by construction.
* (2) THE STAMP NAMES SUSPECT COUNTS LOUDLY, AT O(1) a counts.json that
* carries the ALL scalars but no `allCountsDerivedBy: 'identity-record'`
* stamp predates this fix; loading it marks `suspect = true` from a
* single field read alone, never a directory walk, and warns exactly
* once naming the cause.
* (3) THE SANCTIONED RECOUNT CLEARS IT `repairIndex()` prunes the orphaned
* containers, recounts from the canonical metadata.json walk, and
* re-stamps suspect clears and the ALL scalar is exact again.
* (2) THE STAMP NAMES SUSPECT COUNTS LOUDLY, AND THE OPEN NEVER WALKS a
* counts.json that carries the ALL scalars but no
* `allCountsDerivedBy: 'identity-record'` stamp predates this fix;
* loading it marks `suspect = true` from a single field read alone and
* warns exactly once naming the cause. The open itself never pays a
* directory walk.
* (2b) AND IT HEALS ITSELF. The ledger used to stay wrong for the life of the
* store, waiting for an operator to run `repairIndex()` and a
* downstream index heal subtracted against the inflated denominator and
* reported work that did not exist. An honest derivation now runs in the
* BACKGROUND after the open (never blocking it, observable via
* `whenCountLedgerSettled()`), and refuses to stamp a number it derived
* while writes were landing.
* (3) THE SANCTIONED RECOUNT ALSO CLEARS IT `repairIndex()` prunes the
* orphaned containers, recounts from the canonical metadata.json walk,
* and re-stamps the ALL scalar is exact and the containers are gone.
* (4) A FRESH STORE IS NEVER SUSPECT the one-time derivation for a store
* with no counts.json stamps as it writes, so a brand-new store never
* carries the legacy signature.
@ -115,29 +123,43 @@ describe('ledger derivation identity — the ALL scalar is the identity-record p
delete raw.allCountsDerivedBy
fs.writeFileSync(countsPath(dir), JSON.stringify(raw, null, 2))
const warnSpy = vi.spyOn(prodLog, 'warn')
// The two derivation walks live on FileSystemStorage's prototype —
// spying here (rather than on fs.promises.readdir globally) isolates
// THIS code path's behavior from unrelated walks elsewhere in the open
// sequence (a separate, pre-existing engine's own O(store) cost — not
// this fix's concern, and not something this pin should be sensitive
// to). Neither derivation method may run: the stamp check is a field
// read on the already-parsed counts.json, nothing more.
const scanEntitiesSpy = vi.spyOn(FileSystemStorage.prototype as any, 'scanCanonicalEntities')
const scanVectoredSpy = vi.spyOn(FileSystemStorage.prototype as any, 'scanVectoredNounCount')
const narrateSpy = vi.spyOn(prodLog, 'narrate')
// The derivation walks live on FileSystemStorage's prototype. Slow them
// deliberately: the OPEN must not wait for them, and on a two-row store a
// real walk finishes too fast to tell "not awaited" from "instant".
const proto = FileSystemStorage.prototype as any
const realScanEntities = proto.scanCanonicalEntities
let scanEntitiesCalls = 0
proto.scanCanonicalEntities = async function slow(this: any, ...args: any[]) {
scanEntitiesCalls++
await new Promise((r) => setTimeout(r, 1_200))
return realScanEntities.apply(this, args)
}
try {
const openStarted = Date.now()
brain = await open()
const openMs = Date.now() - openStarted
const ledger = await brain.storage.getCanonicalCounts()
expect(ledger.suspect).toBe(true)
// THE OPEN DID NOT WALK: two slowed walks would have added 2.4s to it.
expect(openMs).toBeLessThan(2_000)
const stampWarnings = warnSpy.mock.calls.filter(
([msg]) => String(msg).includes('legacy') && String(msg).includes('container rule')
// The stamp check itself is an O(1) field read, and it names the cause.
const atOpen = await brain.storage.getCanonicalCounts()
expect(atOpen.suspect).toBe(true)
const stampWarnings = narrateSpy.mock.calls.filter(
([msg]: any[]) => String(msg).includes('legacy') && String(msg).includes('container rule')
)
expect(stampWarnings.length).toBe(1) // exactly one, loud
expect(scanEntitiesSpy).not.toHaveBeenCalled() // O(1) field read only, no re-derivation walk
expect(scanVectoredSpy).not.toHaveBeenCalled()
// ...and the honest derivation is already running behind the open.
await brain.storage.whenCountLedgerSettled()
expect(scanEntitiesCalls).toBeGreaterThan(0)
const healed = await brain.storage.getCanonicalCounts()
expect(healed.suspect).toBe(false)
expect(healed.nouns.all).toBe(raw.totalNounCountAll)
} finally {
proto.scanCanonicalEntities = realScanEntities
}
await brain.close()
})
@ -163,7 +185,14 @@ describe('ledger derivation identity — the ALL scalar is the identity-record p
fs.writeFileSync(countsPath(dir), JSON.stringify(raw, null, 2))
brain = await open()
expect((await brain.storage.getCanonicalCounts()).suspect).toBe(true) // named suspect at load
// Named suspect at load, then healed in the background WITHOUT the
// operator asking — the inflated container count is corrected to the
// identity-record population, though the orphaned containers themselves
// are still on disk (only repairIndex() removes those).
await brain.storage.whenCountLedgerSettled()
let healed = await brain.storage.getCanonicalCounts()
expect(healed.suspect).toBe(false)
expect(healed.nouns.all).toBe(realTotal)
await brain.repairIndex()

View file

@ -0,0 +1,192 @@
/**
* @module tests/integration/open-does-not-wait-for-a-rebuilding-provider
* @description OPEN DOES NOT WAIT FOR A PROVIDER THAT IS REBUILDING ITSELF.
*
* Measured on a production store: a metadata provider that had to rebuild made
* `init()` pay the ENTIRE rebuild on the foreground 641 seconds with every
* other family idle behind it, because a provider reporting `serving: false`
* because it is BUSY BUILDING and one reporting `serving: false` because it is
* BROKEN were indistinguishable, and both were answered the same way: call
* `rebuild()`, and wait.
*
* The law: a provider that reports `rebuildInProgress()` owns its own rebuild.
* `init()` returns; every other family serves; THAT family's doors refuse by
* name, carrying the provider's own progress; and the doors open by themselves
* when the provider reports serving. Nothing is ever served empty.
*/
import { describe, it, expect, afterEach } from 'vitest'
import { mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { Brainy } from '../../src/brainy.js'
import { NounType } from '../../src/types/graphTypes.js'
import type { ProviderRebuildProgress } from '../../src/utils/indexReadiness.js'
/** How long the stub provider claims to be rebuilding. */
const REBUILD_MS = 6_000
describe('a provider rebuilding itself never blocks open', () => {
const dirs: string[] = []
const brains: Brainy[] = []
afterEach(async () => {
for (const b of brains.splice(0)) {
try { await b.close() } catch { /* already closed */ }
}
for (const d of dirs.splice(0)) {
try { rmSync(d, { recursive: true, force: true }) } catch { /* ignore */ }
}
})
it('init() returns in milliseconds, the family refuses by name, then answers', async () => {
const dir = mkdtempSync(join(tmpdir(), 'brainy-rebuilding-provider-'))
dirs.push(dir)
// Seed a store so the open has something to (not) rebuild.
const seed = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
await seed.init()
await seed.add({ data: 'a row with a plain field', type: NounType.Concept, metadata: { kind: 'report' } })
await seed.flush()
await seed.close()
const brain = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
brains.push(brain)
// Dress the metadata index as a provider that is rebuilding ITSELF: not
// serving, and honest about why. `init()` wires the real index first, so
// the hooks are installed on the instance as soon as it exists — the gate
// reads them by feature detection, exactly as it would a native provider's.
const rebuildStartedAt = Date.now()
const stillRebuilding = () => Date.now() - rebuildStartedAt < REBUILD_MS
let rebuildCalls = 0
const inner = brain as unknown as {
metadataIndex: Record<string, unknown>
setupIndex?: unknown
}
// Install on the prototype-free instance right after construction by
// patching the property the moment init() assigns it.
const install = (target: Record<string, unknown>) => {
const realRebuild = target.rebuild as () => Promise<void>
target.rebuildInProgress = (): ProviderRebuildProgress | null =>
stillRebuilding()
? { phase: 'metadata shadow build', done: 4_096, total: 14_056, startedAt: rebuildStartedAt }
: null
target.healthReport = () => ({
provider: 'metadata',
healthy: !stillRebuilding(),
serving: !stillRebuilding(),
generation: 1,
invariants: [],
unledgered: []
})
target.rebuild = async () => {
rebuildCalls++
return realRebuild.call(target)
}
}
// init() constructs the metadata index; patch as soon as it exists, before
// the gate consults it. A microtask hop after the index is assigned is
// enough because the gate runs later in the same init.
const initPromise = (async () => {
const originalEnsure = (brain as unknown as { setupIndex?: () => unknown }).setupIndex
void originalEnsure
return brain.init()
})()
// Patch on the first tick the index exists.
const patcher = setInterval(() => {
if (inner.metadataIndex && !inner.metadataIndex.rebuildInProgress) {
install(inner.metadataIndex)
}
}, 1)
const startedAt = Date.now()
try {
await initPromise
} finally {
clearInterval(patcher)
}
const openMs = Date.now() - startedAt
// If the patch did not land before the gate ran, this test proves nothing —
// say so loudly rather than passing vacuously.
expect(
typeof inner.metadataIndex.rebuildInProgress,
'the stub provider was never installed — the test is vacuous'
).toBe('function')
// 1. The open did not wait out the rebuild.
expect(openMs).toBeLessThan(REBUILD_MS)
// 2. And brainy did not start a rebuild of its own on top of the provider's.
expect(rebuildCalls).toBe(0)
// 3. The family's door refuses BY NAME, carrying the provider's progress.
let refusal: Error | null = null
try {
await brain.find({ where: { kind: 'report' } } as never)
} catch (err) {
refusal = err as Error
}
expect(refusal, 'a not-serving metadata family must refuse, never serve empty').not.toBeNull()
expect(refusal!.message).toMatch(/metadata shadow build/i)
expect(refusal!.message).toMatch(/4,096\/14,056/)
expect(refusal!.message).toMatch(/no action is needed/i)
// 4. Other families keep serving — the brain is open.
const all = await brain.getNouns?.({ pagination: { limit: 1 } } as never)
expect(all ?? true).toBeTruthy()
// 5. When the provider reports itself serving, the door opens by itself.
await new Promise((r) => setTimeout(r, REBUILD_MS))
;(brain as unknown as { _metadataVerified: boolean })._metadataVerified = false
await expect(brain.find({ where: { kind: 'report' } } as never)).resolves.toBeDefined()
}, 180_000)
it('a rebuilding provider reporting 0 entries is not a CRITICAL, and gets no second rebuild', async () => {
const dir = mkdtempSync(join(tmpdir(), 'brainy-rebuilding-critical-'))
dirs.push(dir)
const seed = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
await seed.init()
await seed.add({ data: 'a stored entity', type: NounType.Concept })
await seed.flush()
await seed.close()
const brain = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
brains.push(brain)
let rebuildCalls = 0
const errors: string[] = []
const origError = console.error
console.error = ((...a: unknown[]) => { errors.push(a.map(String).join(' ')) }) as typeof console.error
const inner = brain as unknown as { metadataIndex: Record<string, unknown> }
const patcher = setInterval(() => {
if (inner.metadataIndex && !inner.metadataIndex.rebuildInProgress) {
const target = inner.metadataIndex
target.rebuildInProgress = () => ({ phase: 'online metadata rebuild', startedAt: Date.now() })
target.healthReport = () => ({
provider: 'metadata', healthy: false, serving: false,
generation: 1, invariants: [], unledgered: []
})
// The shape the native engine now has: the index reports NOTHING while
// its rebuild runs online behind refusing doors.
target.getStats = async () => ({ totalEntries: 0 })
target.rebuild = async () => { rebuildCalls++ }
}
}, 1)
try {
await brain.init()
} finally {
clearInterval(patcher)
console.error = origError
}
expect(
typeof inner.metadataIndex.rebuildInProgress,
'the stub provider was never installed — the test is vacuous'
).toBe('function')
expect(errors.filter((l) => /CRITICAL: Metadata index has 0 entries/.test(l))).toEqual([])
expect(rebuildCalls).toBe(0)
}, 180_000)
})

View file

@ -0,0 +1,114 @@
/**
* @module tests/integration/open-narration
* @description THE OPEN IS NEVER SILENT.
*
* A production service opened a 16 GB store and logged nothing at all for
* three minutes before its first line of work. Two defects made that possible
* and both are pinned here:
*
* 1. The phase breakdown was written to `prodLog.warn`, which every
* environment that looks like production clamps away. The narration
* channel (`prodLog.narrate`) is always visible, like `error`.
* 2. Nothing spoke DURING a phase only after the whole open finished, if
* at all. A heartbeat now names the phase currently running and its
* elapsed wall while the open is still happening.
*/
import { describe, it, expect, beforeEach, afterEach } from 'vitest'
import { mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { Brainy } from '../../src/brainy.js'
import { NounType } from '../../src/types/graphTypes.js'
import { FileSystemStorage } from '../../src/storage/adapters/fileSystemStorage.js'
import { prodLog, configureLogger, LogLevel } from '../../src/utils/logger.js'
function makeTempDir(): string {
return mkdtempSync(join(tmpdir(), 'brainy-open-narration-'))
}
/** Capture console.warn lines emitted while `fn` runs. */
async function captureWarn<T>(fn: () => Promise<T>): Promise<{ result: T; lines: string[] }> {
const lines: string[] = []
const orig = console.warn
console.warn = ((...args: unknown[]) => {
lines.push(args.map((a) => String(a)).join(' '))
}) as typeof console.warn
try {
return { result: await fn(), lines }
} finally {
console.warn = orig
}
}
describe('open narration', () => {
let dir: string
let brain: Brainy | null = null
beforeEach(() => { dir = makeTempDir() })
afterEach(async () => {
if (brain) {
try { await brain.close() } catch { /* already closed */ }
brain = null
}
try { rmSync(dir, { recursive: true, force: true }) } catch { /* ignore */ }
})
it('narrate() survives the production log clamp that silences warn()', async () => {
// Exactly what isProductionEnvironment() does to the logger: level ERROR.
configureLogger({ level: LogLevel.ERROR })
try {
const { lines } = await captureWarn(async () => {
prodLog.warn('[Brainy] this line is chatter and may be clamped')
prodLog.narrate('[Brainy] this line is why the database is slow')
})
expect(lines.some((l) => /why the database is slow/.test(l))).toBe(true)
expect(lines.some((l) => /chatter/.test(l))).toBe(false)
} finally {
configureLogger({ level: LogLevel.INFO })
}
})
it('names a slow phase as it ends, and heartbeats while it is still running', async () => {
// Seed a store, then reopen it with a deliberately slow storage init so
// the first phase crosses both the heartbeat and the narrate thresholds.
brain = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
await brain.init()
await brain.add({ data: 'seed entity', type: NounType.Concept })
await brain.flush()
await brain.close()
brain = null
const realInit = FileSystemStorage.prototype.init
FileSystemStorage.prototype.init = async function slowInit(this: FileSystemStorage) {
await new Promise((r) => setTimeout(r, 6_500))
return realInit.call(this)
}
// Clamped to ERROR for the whole open: the narration must survive it.
configureLogger({ level: LogLevel.ERROR })
try {
const { result, lines } = await captureWarn(async () => {
const next = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
await next.init()
return next
})
brain = result
// The heartbeat spoke DURING the phase, naming the phase and its cause.
const heartbeats = lines.filter((l) => /open: still in phase 1\/5 "storage-init"/.test(l))
expect(heartbeats.length).toBeGreaterThanOrEqual(1)
expect(heartbeats[0]).toMatch(/loading its count ledger/)
// And the phase named its own wall as it ended.
const ended = lines.filter((l) => /open: phase 1\/5 "storage-init" finished in \d+ms/.test(l))
expect(ended.length).toBe(1)
// The whole-open breakdown is on the same always-visible channel.
expect(lines.some((l) => /slow open: \d+ms total \(.*storage-init=/.test(l))).toBe(true)
} finally {
FileSystemStorage.prototype.init = realInit
configureLogger({ level: LogLevel.INFO })
}
}, 120_000)
})

View file

@ -0,0 +1,119 @@
/**
* @module tests/integration/repair-narration
* @description A REPAIR NARRATES ITSELF, AND ITS RECEIPT SAYS WHERE THE TIME
* WENT.
*
* On a production store (14,647 nouns / 73,070 verbs) a `repairIndex()` ran
* for more than thirty minutes at roughly a full core with ZERO log lines
* between its start and its end, while the read doors kept serving. The
* operator could tell it was alive only from `top`, and could not tell which
* of its single-threaded walks it was inside. The law pinned here:
*
* - every phase announces itself BEFORE it works, naming what it is about
* to walk;
* - a heartbeat names the phase still running, at a bounded cadence, for as
* long as it runs;
* - every phase reports its own wall, and that wall is carried in the typed
* receipt (`RepairFamilyReport.durationMs`) not only in a log line.
*
* All of it on the narration channel, which production's log clamp cannot
* silence (see tests/integration/open-narration.test.ts).
*/
import { describe, it, expect, afterEach, vi } from 'vitest'
import { mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { Brainy } from '../../src/brainy.js'
import { NounType } from '../../src/types/graphTypes.js'
import { FileSystemStorage } from '../../src/storage/adapters/fileSystemStorage.js'
import { prodLog, configureLogger, LogLevel } from '../../src/utils/logger.js'
describe('repairIndex narration', () => {
const dirs: string[] = []
const brains: Brainy[] = []
afterEach(async () => {
for (const b of brains.splice(0)) {
try { await b.close() } catch { /* already closed */ }
}
for (const d of dirs.splice(0)) {
try { rmSync(d, { recursive: true, force: true }) } catch { /* ignore */ }
}
configureLogger({ level: LogLevel.INFO })
})
async function seededBrain(): Promise<Brainy> {
const dir = mkdtempSync(join(tmpdir(), 'brainy-repair-narration-'))
dirs.push(dir)
const brain = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
brains.push(brain)
await brain.init()
for (let i = 0; i < 5; i++) {
await brain.add({ data: `repair subject ${i}`, type: NounType.Concept })
}
await brain.flush()
return brain
}
it('announces every phase, reports its wall, and carries that wall in the receipt', async () => {
const brain = await seededBrain()
const narrateSpy = vi.spyOn(prodLog, 'narrate')
const report = await brain.repairIndex()
const lines = narrateSpy.mock.calls.map(([m]) => String(m))
// Every family that ran has BOTH a start line and a finish line naming it.
for (const family of report.families) {
const started = lines.filter((l) => l.includes(`"${family.family}" started —`))
const finished = lines.filter((l) =>
new RegExp(`"${family.family}" finished in \\d+ms`).test(l)
)
expect(finished.length, `no finish line for ${family.family}`).toBeGreaterThanOrEqual(1)
// A skipped family may be recorded without a start line only if it never
// began; every family that began must have announced itself.
if (family.checked) {
expect(started.length, `no start line for ${family.family}`).toBeGreaterThanOrEqual(1)
}
// THE RECEIPT CARRIES THE WALL — not only the log.
expect(typeof family.durationMs, `${family.family} has no durationMs`).toBe('number')
expect(family.durationMs).toBeGreaterThanOrEqual(0)
}
// The closing line accounts for the whole repair, per family.
const closing = lines.filter((l) => /repairIndex complete in \d+ms/.test(l))
expect(closing.length).toBe(1)
expect(closing[0]).toMatch(/@\d+ms/)
}, 180_000)
it('heartbeats while a single phase is still walking', async () => {
const brain = await seededBrain()
// Make one phase long enough to cross the heartbeat cadence, exactly as a
// multi-minute canonical walk does on a real store.
const proto = FileSystemStorage.prototype as unknown as Record<
string,
(...args: unknown[]) => Promise<unknown>
>
const realPrune = proto.pruneOrphanedEntities
proto.pruneOrphanedEntities = async function slow(this: unknown, ...args: unknown[]) {
await new Promise((r) => setTimeout(r, 6_500))
return realPrune.apply(this, args)
}
// Clamped as production clamps it: the narration must survive.
configureLogger({ level: LogLevel.ERROR })
const narrateSpy = vi.spyOn(prodLog, 'narrate')
try {
await brain.repairIndex()
} finally {
proto.pruneOrphanedEntities = realPrune
}
const beats = narrateSpy.mock.calls
.map(([m]) => String(m))
.filter((l) => /repairIndex: still in "orphaned-containers" after \d+s/.test(l))
expect(beats.length).toBeGreaterThanOrEqual(1)
expect(beats[0]).toMatch(/ghost\/scar containers/)
}, 180_000)
})

View file

@ -0,0 +1,106 @@
/**
* @module tests/integration/vfs-root-sweep-once
* @description THE OLD-ROOT SWEEP RUNS ONCE PER STORE, NOT ONCE PER OPEN.
*
* The VFS bootstrap ran a filtered `find()` over the whole store on EVERY
* open, hunting for root directories created before the fixed root id existed
* duplicates a store has either always had or never will. MEASURED on a
* 14,056-noun / 72,679-verb store: the phase it dominates cost 4353 SECONDS
* of every open, warm reopens included.
*
* The law: a migration sweep is caused by the store's state, not by the clock
* or the open count. It runs behind the doors, records that it ran, and a
* store carrying that record never sweeps again.
*/
import { describe, it, expect, afterEach, vi } from 'vitest'
import { mkdtempSync, rmSync, existsSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { Brainy } from '../../src/brainy.js'
import { NounType } from '../../src/types/graphTypes.js'
import { VirtualFileSystem } from '../../src/vfs/VirtualFileSystem.js'
describe('the VFS old-root sweep', () => {
const dirs: string[] = []
const brains: Brainy[] = []
afterEach(async () => {
for (const b of brains.splice(0)) {
try { await b.close() } catch { /* already closed */ }
}
for (const d of dirs.splice(0)) {
try { rmSync(d, { recursive: true, force: true }) } catch { /* ignore */ }
}
vi.restoreAllMocks()
})
async function open(dir: string): Promise<Brainy> {
const brain = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
brains.push(brain)
await brain.init()
return brain
}
it('sweeps on the first open, records it, and never sweeps again', async () => {
const dir = mkdtempSync(join(tmpdir(), 'brainy-root-sweep-'))
dirs.push(dir)
const sweepSpy = vi.spyOn(
VirtualFileSystem.prototype as unknown as { cleanupOldRoots: () => Promise<void> },
'cleanupOldRoots'
)
const first = await open(dir)
await (first.vfs as unknown as { whenRootSweepSettled: () => Promise<void> }).whenRootSweepSettled()
expect(sweepSpy).toHaveBeenCalledTimes(1)
// The record is durable engine plumbing under _system/, like every other marker.
expect(
existsSync(join(dir, '_system', 'vfs-root-sweep.json')) ||
existsSync(join(dir, '_system', 'vfs-root-sweep.json.gz'))
).toBe(true)
await first.add({ data: 'a row so the store is not trivially empty', type: NounType.Concept })
await first.flush()
await first.close()
brains.splice(brains.indexOf(first), 1)
sweepSpy.mockClear()
const second = await open(dir)
await (second.vfs as unknown as { whenRootSweepSettled: () => Promise<void> }).whenRootSweepSettled()
expect(sweepSpy).not.toHaveBeenCalled()
await second.close()
brains.splice(brains.indexOf(second), 1)
// ...and a third open, to prove it is the record and not a one-off.
sweepSpy.mockClear()
const third = await open(dir)
await (third.vfs as unknown as { whenRootSweepSettled: () => Promise<void> }).whenRootSweepSettled()
expect(sweepSpy).not.toHaveBeenCalled()
}, 180_000)
it('the open does not wait for the sweep', async () => {
const dir = mkdtempSync(join(tmpdir(), 'brainy-root-sweep-async-'))
dirs.push(dir)
const proto = VirtualFileSystem.prototype as unknown as Record<
string,
(...args: unknown[]) => Promise<unknown>
>
const real = proto.cleanupOldRoots
proto.cleanupOldRoots = async function slow(this: unknown, ...args: unknown[]) {
await new Promise((r) => setTimeout(r, 4_000))
return real.apply(this, args)
}
try {
const startedAt = Date.now()
const brain = await open(dir)
const openMs = Date.now() - startedAt
expect(openMs).toBeLessThan(3_000)
await (brain.vfs as unknown as { whenRootSweepSettled: () => Promise<void> }).whenRootSweepSettled()
} finally {
proto.cleanupOldRoots = real
}
}, 180_000)
})

View file

@ -0,0 +1,250 @@
/**
* @module tests/integration/writer-lock-clean-close
* @description THE CLEAN-CLOSE CONTRACT for the writer lock.
*
* A production restart made this lane necessary: a service stopped with exit
* code 0, having awaited `close()` on every pooled brain, and its next boot
* announced `[brainy] Overwriting stale writer lock … appears dead` for every
* store it owned. "The pid is gone" is equally true of an orderly restart and
* of a crash, so the message could not tell an operator which one they had.
*
* The contract pinned here:
* 1. A completed close leaves NO lock file and DOES leave a clean-close
* record; the next open says nothing about staleness.
* 2. The next lock claim CONSUMES that record it may never outlive the
* lock generation it describes, or a later crash would read as clean.
* 3. A close whose durable steps FAIL still releases the lock (and still
* rethrows the failure).
* 4. A killed process (SIGKILL, no close at all) leaves the lock behind with
* NO record, and the next open says exactly that crash, recovery ahead.
* 5. A host application with its own SIGTERM handler is never force-exited
* out from under its own shutdown by Brainy's handler.
*/
import { describe, it, expect, beforeEach, afterEach } from 'vitest'
import { mkdtempSync, rmSync, existsSync, readFileSync, writeFileSync } from 'node:fs'
import { spawn } from 'node:child_process'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { Brainy } from '../../src/brainy.js'
import { NounType } from '../../src/types/graphTypes.js'
const REPO_ROOT = process.cwd()
const TSX = join(REPO_ROOT, 'node_modules', '.bin', 'tsx')
function makeTempDir(): string {
return mkdtempSync(join(tmpdir(), 'brainy-clean-close-'))
}
/**
* Write a child script to disk and start it under tsx. A file (not `tsx -e`)
* because the eval form compiles to CommonJS, which has no top-level await.
* The script imports Brainy by ABSOLUTE path, so its own dependency
* resolution still happens from inside the repository.
*/
function startChild(dir: string, body: string): ReturnType<typeof spawn> {
const scriptPath = join(dir, 'child-process.mts')
writeFileSync(scriptPath, body)
// `detached` puts the child in its own process GROUP: tsx runs the script in
// a grandchild process, and only a group-wide signal reaches the process
// that actually holds the writer lock.
return spawn(TSX, [scriptPath], {
cwd: REPO_ROOT,
stdio: ['ignore', 'pipe', 'pipe'],
detached: true
})
}
/** Capture every console.warn/error line emitted while `fn` runs. */
async function captureConsole<T>(fn: () => Promise<T>): Promise<{ result: T; lines: string[] }> {
const lines: string[] = []
const origWarn = console.warn
const origError = console.error
const sink = (...args: unknown[]) => {
lines.push(args.map((a) => String(a)).join(' '))
}
console.warn = sink as typeof console.warn
console.error = sink as typeof console.error
try {
const result = await fn()
return { result, lines }
} finally {
console.warn = origWarn
console.error = origError
}
}
/**
* Run a child process that opens `dir`, writes one row, prints `READY`, and
* then waits forever. Resolves with the child once READY is seen.
*/
function spawnHoldingChild(dir: string): Promise<{
child: ReturnType<typeof spawn>
output: () => string
}> {
const script = `
import { Brainy } from ${JSON.stringify(join(REPO_ROOT, 'src', 'brainy.ts'))}
const brain = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: ${JSON.stringify(dir)} } })
await brain.init()
await brain.add({ data: 'row from the child', type: 'concept' })
await brain.flush()
console.log('READY')
setInterval(() => {}, 1000)
`
const child = startChild(dir, script)
let out = ''
child.stdout.on('data', (d) => { out += String(d) })
child.stderr.on('data', (d) => { out += String(d) })
return new Promise((resolvePromise, rejectPromise) => {
const timer = setTimeout(() => rejectPromise(new Error(`child never became READY:\n${out}`)), 120_000)
child.stdout.on('data', () => {
if (out.includes('READY')) {
clearTimeout(timer)
resolvePromise({ child, output: () => out })
}
})
child.on('exit', (code) => {
clearTimeout(timer)
if (!out.includes('READY')) rejectPromise(new Error(`child exited ${code} before READY:\n${out}`))
})
})
}
describe('writer lock — the clean-close contract', () => {
let dir: string
let brain: Brainy | null = null
beforeEach(() => { dir = makeTempDir() })
afterEach(async () => {
if (brain) {
try { await brain.close() } catch { /* may already be closed */ }
brain = null
}
try { rmSync(dir, { recursive: true, force: true }) } catch { /* ignore */ }
})
const lockPath = () => join(dir, 'locks', '_writer.lock')
const recordPath = () => join(dir, 'locks', '_writer.close')
it('a completed close leaves no lock, leaves a record, and the reopen is silent about staleness', async () => {
brain = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
await brain.init()
expect(existsSync(lockPath())).toBe(true)
await brain.add({ data: 'seed entity', type: NounType.Concept })
await brain.flush()
await brain.close()
brain = null
// 1. The lock is gone and the release is RECORDED.
expect(existsSync(lockPath())).toBe(false)
expect(existsSync(recordPath())).toBe(true)
const record = JSON.parse(readFileSync(recordPath(), 'utf-8'))
expect(record.pid).toBe(process.pid)
expect(typeof record.closedAt).toBe('string')
expect(typeof record.startedAt).toBe('string')
// 2. The reopen says nothing about a stale lock.
const { result: reopened, lines } = await captureConsole(async () => {
const next = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
await next.init()
return next
})
brain = reopened
expect(lines.filter((l) => /stale writer lock|appears dead/i.test(l))).toEqual([])
// 3. The claim CONSUMED the record — it must not outlive its lock generation.
expect(existsSync(recordPath())).toBe(false)
expect(existsSync(lockPath())).toBe(true)
}, 120_000)
it('releases the writer lock even when a durable close step fails — and still rethrows', async () => {
brain = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
await brain.init()
await brain.add({ data: 'seed entity', type: NounType.Concept })
await brain.flush()
expect(existsSync(lockPath())).toBe(true)
// Inject a failure into a durable close step (the counts flush).
const storage = (brain as unknown as { storage: { flushCounts: () => Promise<void> } }).storage
const boom = new Error('injected: counts flush failed during close')
storage.flushCounts = async () => { throw boom }
await expect(brain.close()).rejects.toThrow(/injected: counts flush failed/)
brain = null
// The lock is released regardless: a process on its way out holds nothing.
expect(existsSync(lockPath())).toBe(false)
// And the next writer opens without a stale-lock verdict.
const { lines } = await captureConsole(async () => {
const next = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
await next.init()
await next.close()
})
expect(lines.filter((l) => /appears dead/i.test(l))).toEqual([])
}, 120_000)
it('a SIGKILLed writer leaves the lock with no record, and the next open names the crash', async () => {
const { child } = await spawnHoldingChild(dir)
expect(existsSync(lockPath())).toBe(true)
expect(existsSync(recordPath())).toBe(false)
// Group-wide: the lock holder is tsx's grandchild, not the spawned pid.
process.kill(-(child.pid as number), 'SIGKILL')
await new Promise<void>((r) => child.on('exit', () => r()))
// The grandchild's death is asynchronous with the wrapper's exit event.
await new Promise<void>((r) => setTimeout(r, 500))
// The lock survives the kill — a dead process releases nothing.
expect(existsSync(lockPath())).toBe(true)
expect(existsSync(recordPath())).toBe(false)
const { lines } = await captureConsole(async () => {
const next = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
await next.init()
await next.close()
})
const verdict = lines.filter((l) => /Overwriting stale writer lock/i.test(l))
expect(verdict.length).toBe(1)
// The verdict must name the ABSENT record and the recovery it implies —
// not merely that a pid is gone.
expect(verdict[0]).toMatch(/NO\s+clean-close record/i)
expect(verdict[0]).toMatch(/crash recovery/i)
}, 180_000)
it("does not force-exit a host application that owns its own SIGTERM handler", async () => {
const script = `
import { Brainy } from ${JSON.stringify(join(REPO_ROOT, 'src', 'brainy.ts'))}
const brain = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: ${JSON.stringify(dir)} } })
await brain.init()
await brain.add({ data: 'row from the host app', type: 'concept' })
await brain.flush()
// The host application's OWN graceful shutdown, registered after Brainy's.
process.on('SIGTERM', async () => {
await new Promise((r) => setTimeout(r, 1500))
console.log('APP-CLOSE-DONE')
process.exit(0)
})
console.log('READY')
setInterval(() => {}, 1000)
`
const child = startChild(dir, script)
let out = ''
child.stdout.on('data', (d) => { out += String(d) })
child.stderr.on('data', (d) => { out += String(d) })
await new Promise<void>((r, reject) => {
const timer = setTimeout(() => reject(new Error(`child never became READY:\n${out}`)), 120_000)
child.stdout.on('data', () => { if (out.includes('READY')) { clearTimeout(timer); r() } })
child.on('exit', () => { clearTimeout(timer); if (!out.includes('READY')) reject(new Error(`child died:\n${out}`)) })
})
process.kill(-(child.pid as number), 'SIGTERM')
const code = await new Promise<number | null>((r) => child.on('exit', (c) => r(c)))
expect(code).toBe(0)
// The host's own shutdown ran to completion — Brainy's handler did not
// exit the process out from under it.
expect(out).toContain('APP-CLOSE-DONE')
}, 180_000)
})