diff --git a/README.md b/README.md index 6b1109e9..19b69f25 100644 --- a/README.md +++ b/README.md @@ -349,6 +349,41 @@ npm install @soulcraft/brainy # Node.js — fully supported > **Deprecation Notice:** Browser support (OPFS, Web Workers, WASM embeddings) is deprecated in v7.10.0 and will be removed in v8.0.0. Brainy v8+ will be server-only. +## Single-Writer Model + +Brainy is **single-writer, many-reader** on filesystem storage. One writer +holds an exclusive lock on the data directory; any number of readers can +inspect it concurrently. Opening a second writer throws with the PID of the +existing one. + +```typescript +// Live application — writer mode is the default +const brain = new Brainy({ storage: { type: 'filesystem', rootDirectory: '/data/brain' } }) +await brain.init() + +// Out-of-band diagnostics from a separate process — safe to run while the +// writer is live +const reader = await Brainy.openReadOnly({ + storage: { type: 'filesystem', rootDirectory: '/data/brain' } +}) +await reader.requestFlush({ timeoutMs: 5000 }) +const stats = await reader.stats() +``` + +For incident debugging, use the `brainy inspect` CLI: + +```bash +brainy inspect stats /data/brain +brainy inspect find /data/brain --where '{"entityType":"booking"}' +brainy inspect explain /data/brain --where '{"entityType":"booking"}' +brainy inspect health /data/brain +``` + +See [the multi-process model](docs/concepts/multi-process.md) and the +[inspection guide](docs/guides/inspection.md) for the full story, including +stale-lock detection, the cross-process flush RPC, and what's not yet +enforced on cloud storage backends. + ## Contributing We welcome contributions! See **[CONTRIBUTING.md](CONTRIBUTING.md)** for guidelines. diff --git a/RELEASES.md b/RELEASES.md index 845083c0..3086332a 100644 --- a/RELEASES.md +++ b/RELEASES.md @@ -11,6 +11,111 @@ Collective. The SDK wraps it — most products never call Brainy directly. Read --- +## v7.21.0 — 2026-05-15 + +**Affected products:** All consumers using filesystem storage (Venue, Workshop dev, +local development). + +### Multi-process safety + first-class read-only inspector + +Filesystem storage now enforces **single-writer, many-reader**. Previously, opening +a second writer process against a live data directory silently produced wrong query +results — no error, no warning, just empty or stale data. This release replaces +that footgun with a hard refusal at init time and a dedicated read-only inspection +mode. + +#### New: `Brainy.openReadOnly()` +```ts +const reader = await Brainy.openReadOnly({ + storage: { type: 'filesystem', rootDirectory: '/data/brain' } +}) +const bookings = await reader.find({ where: { entityType: 'booking' } }) +``` +- Does NOT acquire the writer lock — coexists with a live writer. +- Every mutation method throws `Cannot mutate a read-only Brainy instance`. +- `flush()` and `close()` are safe (no-op + clean shutdown). + +#### New: writer lock at init +- `new Brainy({ ... })` (default `mode: 'writer'`) acquires + `/locks/_writer.lock` containing `{ pid, hostname, startedAt, lastHeartbeat, version }`. +- A second writer on the same directory **throws** with the PID, hostname, + heartbeat, and a pointer to `openReadOnly()`. +- Stale-lock detection: same hostname + dead PID OR heartbeat > 60s ago → overwritten with a warning. +- Heartbeat: writer rewrites `lastHeartbeat` every 10s (unref'd timer). +- Override: pass `{ force: true }` to bypass when you've verified the existing lock is stale. + +#### New: `brain.requestFlush({ timeoutMs })` +Cross-process RPC for inspectors to force a fresh snapshot: +- In-process: just calls `flush()`. +- Out-of-process: writes a request file, polls for ack. Times out gracefully. +- Watcher runs in every writer instance (filesystem only). + +#### New: `brain.stats()` +Operator-facing summary: `entityCount`, `entitiesByType`, `relationCount`, +`relationsByType`, `fieldRegistry`, `indexHealth`, `storage.backend`, `writerLock`, `version`. +Designed for `/api/health` endpoints and incident triage. + +#### New: `brain.explain(findParams)` +Shows which index path serves each `where` clause: `column-store` | `sparse-chunked` | `none`. +**This is the answer to "why is `find()` returning empty?"** — `path: "none"` means +the field has no index entries and the query will silently return `[]`. Includes +notes on likely causes (writer hasn't flushed; typo; field genuinely absent). + +#### New: `brain.health()` +Invariant-check battery: HNSW vs metadata count parity, field registry sanity, +`_seeded` entity sweep, writer heartbeat freshness. Returns `{ overall: 'pass' | 'warn' | 'fail', checks: [...] }`. + +#### New: `brainy inspect` CLI (13 subcommands) +All read-only by default (internally uses `openReadOnly()`): +``` +brainy inspect stats +brainy inspect find --type Event --where '{"status":"paid"}' --limit 20 +brainy inspect get +brainy inspect relations --direction both +brainy inspect explain --where '{"entityType":"booking"}' +brainy inspect health +brainy inspect sample --type Event --n 20 +brainy inspect fields +brainy inspect dump --type Event > backup.jsonl +brainy inspect watch --type Event +brainy inspect backup /backups/brain.tar +brainy inspect repair # writer mode — stop live writer first +brainy inspect diff +``` +Default behaviour: ask the writer to flush via the RPC first (skip with `--no-fresh`). + +### Brainy + Cortex + +Cortex segments (`*.cidx`) are immutable mmap files; `MANIFEST.json` uses atomic +rename. The single writer lock at `/locks/_writer.lock` covers both Brainy +and Cortex. Read-only inspectors mmap Cortex segments with zero coordination. + +### What's NOT enforced yet + +- **Cloud storage backends** (S3, GCS, R2, Azure) — no multi-process locking. + A best-effort warning logs in writer mode against non-filesystem backends. +- **The `find({ where })`-returns-0 root-cause bug** — tracked separately. The + new `brain.explain()` and `brain.health()` surface it loudly + (`path: 'none'`, `index-parity warn`) instead of letting it be silent. + +### Upgrade notes + +- **Default behaviour change:** opening a Brainy directory in writer mode while + another writer is live now THROWS where it previously silently produced wrong + results. If you have scripts that intentionally co-opened a writer directory, + switch them to `Brainy.openReadOnly()` or pass `{ force: true }`. +- **Cloud backends unchanged** — no lock acquisition for S3/GCS/R2/Azure. +- All existing APIs unchanged. Pure addition. + +### Docs + +- README "Single-Writer Model" section. +- `docs/concepts/multi-process.md` — full model + Cortex compatibility. +- `docs/guides/inspection.md` — operator recipes. +- JSDoc on every new API. + +--- + ## v7.19.10 — 2026-02-24 **Affected products:** All Bun/ESM consumers (Workshop, Venue, Academy, SDK) diff --git a/docs/concepts/multi-process.md b/docs/concepts/multi-process.md new file mode 100644 index 00000000..c62cc575 --- /dev/null +++ b/docs/concepts/multi-process.md @@ -0,0 +1,152 @@ +--- +title: Multi-Process Model +slug: concepts/multi-process +public: true +category: concepts +template: concept +order: 5 +description: How Brainy coordinates a single writer with any number of readers on a filesystem data directory — and how to safely inspect a live store. +next: + - guides/inspection +--- + +# Multi-Process Model + +Brainy is a **single-writer, many-reader** database when backed by filesystem +storage. This page explains the model, the guarantees, and the safe ways to +inspect a live store from a second process. + +## The rule + +For one data directory: + +- **One writer** at a time. The writer acquires an exclusive lock on the + directory at `init()` and releases it on `close()`. +- **Any number of readers**, concurrent with each other and with the writer. + Readers open via `Brainy.openReadOnly()` — they never touch the writer + lock. + +Any attempt to open a second writer on the same directory throws: + +``` +BrainyError: Another writer holds this Brainy directory. + PID: 1774431 on host app-host-1 + Started: 2026-05-15T14:22:11Z + Heartbeat: 2026-05-15T14:22:34Z + Version: 7.21.0 + Directory: /data/brain +``` + +This is intentional. Two writers sharing a directory would silently corrupt +in-memory indexes and produce wrong query results — the worst possible default +for an operations tool. + +## Why a lock? + +Brainy keeps its primary indexes (HNSW, metadata, graph adjacency) in memory. +On disk, those indexes are persisted incrementally as writes flush. A second +process opening the same directory: + +- Loads the *persisted* state into a fresh in-memory copy. +- Has no awareness of writes the first process buffered but hasn't flushed. +- Will overwrite the persisted state on its own next flush, racing the first + process and corrupting whichever wins. + +The fix is the lock: refuse to open a second writer. SQLite has done the same +since the late 1990s (`SQLITE_BUSY`). + +## What about Cortex? + +Brainy + Cortex compose cleanly under this model: + +- Cortex stores its column-index segments inside the same `rootDir` (under + `indexes/_column_index/{field}/`). +- Segments (`*.cidx` files) are **immutable** once written. Cortex mmaps them + read-only. +- The `MANIFEST.json` per field is updated via atomic rename — readers see + either the old or new manifest, never a torn file. + +A reader process can safely mmap Cortex segments alongside a live writer +without coordination. The single Brainy writer lock at +`/locks/_writer.lock` covers Cortex too, because Cortex segment +writes happen on the writer's side. + +## Stale-lock detection + +If a writer crashes or is forcibly killed, its lock file is left behind. To +avoid a permanently-jammed directory, Brainy treats a lock as stale when: + +1. The recorded `hostname` equals the current host (cross-host PID checks + are unsafe), AND +2. The recorded `pid` is no longer alive (`process.kill(pid, 0)` returns + `ESRCH`), OR the `lastHeartbeat` field is older than 60 seconds. + +A live writer rewrites `lastHeartbeat` every 10 seconds, so a hung writer +that's missed several heartbeats is treated as dead. Stale locks are +overwritten with a warning. + +If stale detection cannot prove the existing lock is dead — for example, a +crashed writer on a different host writing to a shared filesystem — pass +`{ force: true }` to override. A warning is logged either way. + +## Heartbeat and shutdown + +The heartbeat interval rewrites the lock file every 10 seconds. The timer +is unref'd, so it does not keep the event loop alive on its own. + +On normal shutdown the writer releases the lock in `close()`. The shutdown +hooks Brainy registers for `SIGTERM`, `SIGINT`, and `beforeExit` also +release the lock so a container restart doesn't strand the directory. + +## How to inspect a live writer + +Use `Brainy.openReadOnly()`. It does not acquire the writer lock, so it +coexists with whatever the writer is doing: + +```typescript +const reader = await Brainy.openReadOnly({ + storage: { type: 'filesystem', rootDirectory: '/data/brain' } +}) + +const stats = await reader.stats() +const bookings = await reader.find({ where: { entityType: 'booking' } }) + +await reader.close() +``` + +What the reader sees reflects the writer's most recent **flush** to disk. If +you need fresher state, ask the writer to flush before opening: + +```typescript +const reader = await Brainy.openReadOnly({ + storage: { type: 'filesystem', rootDirectory: '/data/brain' } +}) + +const acked = await reader.requestFlush({ timeoutMs: 5000 }) +if (!acked) { + console.warn('Writer did not respond; results reflect last natural flush.') +} + +const fresh = await reader.find({ where: { entityType: 'booking' } }) +``` + +The CLI `brainy inspect` subcommands all do this for you by default +(`--no-fresh` to opt out). + +## What's not enforced (yet) + +- **Cloud storage backends** (S3, GCS, R2, Azure) do not currently enforce + multi-process locking. Two processes pointing at the same bucket can both + succeed at `init()` in writer mode and clobber each other's writes. A + best-effort warning is logged in writer mode against a non-filesystem + backend. Lock semantics for cloud backends will land in a future release. +- **Long-running readers** do not automatically pick up new Cortex segments + the writer publishes. One-shot inspector calls re-open the store and see + fresh segments; a reader that stays open for hours sees its column store + as-of the time it opened. + +## Reading material + +- `Brainy.openReadOnly()` — [API reference](../api/brainy.md) +- `brainy inspect` — [inspection guide](../guides/inspection.md) +- Cortex columnar storage — see `node_modules/@soulcraft/cortex/README.md` diff --git a/docs/guides/inspection.md b/docs/guides/inspection.md new file mode 100644 index 00000000..c43c7f4f --- /dev/null +++ b/docs/guides/inspection.md @@ -0,0 +1,187 @@ +--- +title: Inspecting a Live Brainy +slug: guides/inspection +public: true +category: guides +template: guide +order: 30 +description: Operator recipes for diagnosing a running Brainy data directory — counts, queries, query plans, health checks, and snapshots — without stopping the live writer. +next: + - concepts/multi-process +--- + +# Inspecting a Live Brainy + +When something is wrong in production, you need to see what's actually in the +store. This guide covers the safe ways to query a running Brainy directory. + +## The cardinal rule + +**Never open a second writer on the same directory.** It will throw on +filesystem storage; on cloud storage it'll silently overwrite the live +writer's state. Use `Brainy.openReadOnly()` or the `brainy inspect` CLI +instead. + +## The CLI is the fastest path + +```bash +# What's in this brain? +brainy inspect stats /data/brain + +# Find specific entities +brainy inspect find /data/brain --type Event --where '{"status":"paid"}' --limit 20 + +# Single entity by ID +brainy inspect get /data/brain 0b7a9... + +# Why is this query returning empty? +brainy inspect explain /data/brain --where '{"entityType":"booking"}' + +# Quick invariants +brainy inspect health /data/brain + +# Random sample (no query needed) +brainy inspect sample /data/brain --type Event --n 20 + +# Tail new writes as they happen +brainy inspect watch /data/brain --type Event + +# Save a snapshot +brainy inspect backup /data/brain /backups/brain-$(date +%Y%m%d).tar +``` + +Every subcommand internally: + +1. Asks the live writer to flush via the cross-process RPC (skip with `--no-fresh`). +2. Opens the data directory via `Brainy.openReadOnly()`. +3. Runs the query. +4. Closes cleanly. + +Results are JSON by default. Add `--pretty` for indented output. + +## When a query returns surprising results + +If `find()` returns `0` for a query you expect to match: run `inspect +explain` first. It shows which index path will serve each `where` clause: + +```bash +$ brainy inspect explain /data/brain --where '{"entityType":"booking","status":"paid"}' +{ + "query": { "where": { "entityType": "booking", "status": "paid" } }, + "fieldPlan": [ + { "field": "entityType", "path": "none", "notes": "No index entries for field..." }, + { "field": "status", "path": "column-store", "notes": "O(log n) binary search..." } + ], + "warnings": [ + "Field \"entityType\" has no index entries. find() will return [] silently." + ] +} +``` + +The `"path": "none"` is the smoking gun. It means the field has no column +store manifest and no sparse chunked index — so `find()` will return `[]` +regardless of what's actually on disk. Likely causes: + +- The writer registered the field in memory but hasn't flushed. Run + `brain.requestFlush()` from the writer side, or use `brainy inspect + --fresh` (default). +- The field name has a typo or wrong casing. +- The field is genuinely absent from every entity. + +## Health checks + +`inspect health` runs a fixed battery of cheap invariant checks: + +```bash +$ brainy inspect health /data/brain +{ + "overall": "warn", + "checks": [ + { "name": "index-parity", "status": "pass", "message": "HNSW (1851) and metadata (1851) agree." }, + { "name": "field-registry", "status": "pass", "message": "23 fields registered for 1851 entities." }, + { "name": "seeded-records", "status": "warn", "message": "15 entities tagged _seeded:true." }, + { "name": "writer-heartbeat", "status": "pass", "message": "Writer healthy (PID 1774431...)." } + ] +} +``` + +Each check returns `pass`, `warn`, or `fail`. The exit code is `2` when any +check fails — useful for piping into monitoring or CI. + +## Programmatic inspection + +```typescript +import { Brainy } from '@soulcraft/brainy' + +const reader = await Brainy.openReadOnly({ + storage: { type: 'filesystem', rootDirectory: '/data/brain' } +}) + +// Force the writer to flush before reading +await reader.requestFlush({ timeoutMs: 5000 }) + +// What's in there? +const stats = await reader.stats() +console.log(`${stats.entityCount} entities`, stats.entitiesByType) + +// Why is this query empty? +const plan = await reader.explain({ where: { entityType: 'booking' } }) +for (const f of plan.fieldPlan) { + console.log(`${f.field} -> ${f.path}`) +} + +// Run invariants +const health = await reader.health() +console.log(health.overall) + +await reader.close() +``` + +Every mutation method (`add`, `update`, `delete`, `relate`, `commit`, +`fork`, `branch`, ...) throws on a read-only instance with a clear message. + +## Backups + +`brainy inspect backup` asks the writer to flush first, then tars the +directory. The snapshot reflects the writer's state at the moment of the +flush: + +```bash +brainy inspect backup /data/brain /backups/brain-2026-05-15.tar +``` + +For periodic backups (hourly, daily), schedule this via cron or your +container scheduler. For point-in-time recovery, use Brainy's COW +`commit()` API — the snapshots there are content-addressed and never +overwritten. + +## Comparing two stores + +`brainy inspect diff` returns a JSON summary of counts and a sample of +entity IDs present in one but not the other. Useful when debugging +replication or migrations: + +```bash +brainy inspect diff /data/brain-prod /data/brain-staging +``` + +Sample-based — for a full diff, dump both with `inspect dump` and compare +the JSONL. + +## Repairing a corrupted store + +If invariants fail and you suspect index corruption, `inspect repair` +opens the store in writer mode and rebuilds all indexes from raw storage. +**Stop the live writer first** — `repair` will throw if another writer +holds the lock. Add `--force` only if you have personally verified the +existing lock is stale. + +```bash +brainy inspect repair /data/brain +``` + +## Multi-process safety summary + +See [concepts/multi-process](../concepts/multi-process.md) for the lock +semantics, heartbeat behavior, and what's not yet enforced on cloud +backends. diff --git a/src/brainy.ts b/src/brainy.ts index 87c897f4..f2d690db 100644 --- a/src/brainy.ts +++ b/src/brainy.ts @@ -65,7 +65,10 @@ import { DistributedCoordinator, ShardManager, CacheSync, - ReadWriteSeparation + ReadWriteSeparation, + BaseOperationalMode, + ReaderMode, + HybridMode } from './distributed/index.js' import { Entity, @@ -83,6 +86,7 @@ import { RelateManyParams, BatchResult, BrainyConfig, + BrainyStats, ScoreExplanation } from './types/brainy.types.js' import { NounType, VerbType } from './types/graphTypes.js' @@ -182,6 +186,12 @@ export class Brainy implements BrainyInterface { private initialized = false private dimensions?: number + // Multi-process mode (writer is default; reader is set via mode: 'reader' or + // Brainy.openReadOnly()). `operationalMode.validateOperation('write')` is + // called at the top of every mutation method. Snapshots from `asOf()` also + // set this to ReaderMode so historical instances are protected the same way. + private operationalMode: BaseOperationalMode + // Ready Promise state (Unified readiness API) // Allows consumers to await brain.ready for initialization completion private _readyPromise: Promise | null = null @@ -198,6 +208,14 @@ export class Brainy implements BrainyInterface { // Normalize configuration with defaults this.config = this.normalizeConfig(config) + // Multi-process mode — default 'writer' uses HybridMode (read + write), + // 'reader' uses ReaderMode (read-only, all mutations throw). + // `WriterMode` from operationalModes.ts blocks reads, which is too strict + // for a Brainy instance — a writer needs to read its own data. + this.operationalMode = this.config.mode === 'reader' + ? new ReaderMode() + : new HybridMode() + // Configure memory limits // This must happen early, before any validation occurs if (this.config.maxQueryLimit !== undefined || this.config.reservedQueryMemory !== undefined) { @@ -223,6 +241,11 @@ export class Brainy implements BrainyInterface { this._readyResolve = resolve this._readyReject = reject }) + // Attach a default no-op rejection handler so that init-failure cases + // (e.g. another writer holds the lock) do not surface as Node + // "unhandled promise rejection" warnings when callers don't `await brain.ready`. + // Callers who DO await it still see the original rejection. + this._readyPromise.catch(() => { /* observed by ready() consumers, if any */ }) // Track this instance for shutdown hooks Brainy.instances.push(this) @@ -230,6 +253,70 @@ export class Brainy implements BrainyInterface { // Index and storage are initialized in init() because they may need each other } + /** + * Open a Brainy store in read-only mode and initialize it. + * + * Convenience factory equivalent to + * `new Brainy({ ...config, mode: 'reader' })` followed by `init()`. The + * resulting instance: + * + * - Does NOT acquire the writer lock — coexists with a live writer process + * and with any number of other readers on the same data directory. + * - Throws `Cannot mutate a read-only Brainy instance` from every mutation + * method (`add`, `addMany`, `update`, `delete`, `deleteMany`, `relate`, + * `unrelate`, `commit`, `branch`, `merge`, `deleteBranch`). + * - Reflects the state of the writer's last successful `flush()`. To force + * the writer to flush before opening, call `requestFlush()` on a separate + * handle first, or pass `--fresh` to the `brainy inspect` CLI. + * + * Typical use cases: + * - Operator diagnostics during incidents (`brainy inspect` uses this). + * - Read-replica processes on the same machine. + * - Long-running analytics scripts that should not contend with the writer. + * + * @param config Same options as `new Brainy(config)`. The `mode` field is + * ignored and forced to `'reader'`. + * @returns An initialized, read-only Brainy instance. + * + * @example + * ```typescript + * const reader = await Brainy.openReadOnly({ + * storage: { type: 'filesystem', rootDirectory: '/data/brainy-data/tenant' } + * }) + * const bookings = await reader.find({ where: { entityType: 'booking' } }) + * await reader.close() + * ``` + */ + static async openReadOnly(config: BrainyConfig): Promise> { + const brain = new Brainy({ ...config, mode: 'reader' }) + await brain.init() + return brain + } + + /** + * Whether this instance is read-only (set via `mode: 'reader'`, + * `Brainy.openReadOnly()`, or `asOf()`). When true, all mutation methods + * throw. + */ + get isReadOnly(): boolean { + return !this.operationalMode.canWrite + } + + /** + * Throw if this instance cannot perform writes. Called at the top of every + * mutation method. The check is cheap (a property lookup + boolean test) and + * runs before any work, so callers get a clear, fast failure in read-only mode. + */ + private assertWritable(method: string): void { + if (!this.operationalMode.canWrite) { + throw new Error( + `Cannot call ${method}() on a read-only Brainy instance. ` + + `This instance was opened with mode: 'reader' (or via Brainy.openReadOnly() / asOf()). ` + + `Open in writer mode to modify data.` + ) + } + } + /** * Initialize Brainy - MUST be called before use * @param overrides Optional configuration overrides for init @@ -255,6 +342,13 @@ export class Brainy implements BrainyInterface { if (dimensions) { this.dimensions = dimensions } + + // Re-derive operationalMode if mode override changed it + if (configOverrides.mode) { + this.operationalMode = configOverrides.mode === 'reader' + ? new ReaderMode() + : new HybridMode() + } } // Configure logging based on config options @@ -288,6 +382,30 @@ export class Brainy implements BrainyInterface { this.storage = await this.setupStorage() await this.storage.init() + // Acquire the writer lock for filesystem (and other locking-capable) backends. + // Skipped in reader mode and on backends that don't support multi-process locking. + // Throws if another live writer holds the directory (unless force: true). + if (this.config.mode !== 'reader' && this.storage.supportsMultiProcessLocking()) { + await this.storage.acquireWriterLock({ force: this.config.force }) + // Watch for cross-process flush requests so inspectors can see fresh + // state on demand. Reader instances don't run a watcher (nothing to flush). + this.storage.startFlushRequestWatcher(async () => { + if (this.initialized) { + await this.flush() + } + }) + } else if (this.config.mode !== 'reader' && !this.config.silent) { + // Cloud / memory backends: multi-process safety not enforced. One-time + // warning so operators know what they're getting. + const backendName = (this.storage.constructor as any).name || 'storage' + if (backendName !== 'MemoryStorage') { + console.warn( + `[brainy] Multi-process writer protection is not enforced on ${backendName}. ` + + `See docs/concepts/multi-process.md for the model.` + ) + } + } + // Enable COW immediately after storage init // This ensures ALL data is stored in branch-scoped paths from the start // Lightweight: just sets cowEnabled=true and currentBranch, no RefManager/BlobStorage yet @@ -512,6 +630,19 @@ export class Brainy implements BrainyInterface { await (instance.metadataIndex as any).close() } })(), + // Release the writer lock so a successor process can take over. + // No-op for readers and for backends without locking. + (async () => { + if (instance.storage && typeof instance.storage.releaseWriterLock === 'function') { + await instance.storage.releaseWriterLock() + } + })(), + // Stop the flush-request watcher to release its interval timer. + (async () => { + if (instance.storage && typeof instance.storage.stopFlushRequestWatcher === 'function') { + instance.storage.stopFlushRequestWatcher() + } + })(), ]) flushedCount++ } @@ -755,6 +886,7 @@ export class Brainy implements BrainyInterface { * ``` */ async add(params: AddParams): Promise { + this.assertWritable('add') await this.ensureInitialized() // Zero-config validation (static import for performance) @@ -1216,6 +1348,7 @@ export class Brainy implements BrainyInterface { * ``` */ async update(params: UpdateParams): Promise { + this.assertWritable('update') await this.ensureInitialized() // Zero-config validation (static import for performance) @@ -1365,6 +1498,7 @@ export class Brainy implements BrainyInterface { * ``` */ async delete(id: string): Promise { + this.assertWritable('delete') // Handle invalid IDs gracefully if (!id || typeof id !== 'string') { return // Silently return for invalid IDs @@ -1552,6 +1686,7 @@ export class Brainy implements BrainyInterface { * } */ async relate(params: RelateParams): Promise { + this.assertWritable('relate') await this.ensureInitialized() // Zero-config validation (static import for performance) @@ -1703,6 +1838,7 @@ export class Brainy implements BrainyInterface { * ``` */ async unrelate(id: string): Promise { + this.assertWritable('unrelate') await this.ensureInitialized() // Get verb data before deletion for rollback @@ -2664,6 +2800,7 @@ export class Brainy implements BrainyInterface { * ``` */ async addMany(params: AddManyParams): Promise> { + this.assertWritable('addMany') await this.ensureInitialized() // Get optimal batch configuration from storage adapter @@ -2803,6 +2940,7 @@ export class Brainy implements BrainyInterface { * Delete multiple entities */ async deleteMany(params: DeleteManyParams): Promise> { + this.assertWritable('deleteMany') await this.ensureInitialized() // Determine what to delete @@ -3012,6 +3150,7 @@ export class Brainy implements BrainyInterface { * ``` */ async relateMany(params: RelateManyParams): Promise { + this.assertWritable('relateMany') await this.ensureInitialized() // Get optimal batch configuration from storage adapter @@ -3215,6 +3354,7 @@ export class Brainy implements BrainyInterface { message?: string metadata?: Record }): Promise> { + this.assertWritable('fork') await this.ensureInitialized() const branchName = branch || `fork-${Date.now()}` @@ -3469,6 +3609,7 @@ export class Brainy implements BrainyInterface { metadata?: Record captureState?: boolean }): Promise { + this.assertWritable('commit') await this.ensureInitialized() if (!('refManager' in this.storage) || !('commitLog' in this.storage) || !('blobStorage' in this.storage)) { @@ -3695,17 +3836,18 @@ export class Brainy implements BrainyInterface { // Create Brainy instance wrapping historical storage // All queries will lazy-load from historical state on-demand + // mode: 'reader' wires ReaderMode so every mutation throws. const snapshotBrain = new Brainy({ ...this.config, // Use the historical adapter directly (no need for separate storage type) - storage: historicalStorage as any + storage: historicalStorage as any, + mode: 'reader' }) // Initialize the snapshot (creates indexes, but they'll be populated lazily) await snapshotBrain.init() - // Mark as read-only snapshot (prevents writes) - ;(snapshotBrain as any).isReadOnlySnapshot = true + // Track which commit this snapshot reflects ;(snapshotBrain as any).snapshotCommitId = commitId console.log(`[asOf] Snapshot ready (lazy-loading, cache size: ${options?.cacheSize || 10000})`) @@ -3724,6 +3866,7 @@ export class Brainy implements BrainyInterface { * ``` */ async deleteBranch(branch: string): Promise { + this.assertWritable('deleteBranch') await this.ensureInitialized() if (!('refManager' in this.storage)) { @@ -4775,8 +4918,13 @@ export class Brainy implements BrainyInterface { async flush(): Promise { await this.ensureInitialized() - console.log('Flushing Brainy indexes and caches to disk...') + // Read-only instances have no buffered writes to flush. close() may call + // flush() defensively, so we early-return instead of throwing. + if (this.isReadOnly) { + return + } + console.log('Flushing Brainy indexes and caches to disk...') const startTime = Date.now() // Flush all components in parallel for performance @@ -4807,6 +4955,49 @@ export class Brainy implements BrainyInterface { console.log(`All indexes flushed to disk in ${elapsed}ms`) } + /** + * Ask the writer process serving this data directory to flush its in-memory + * indexes to disk, so a read-only inspector can observe fresh state. + * + * - **Same-process call** (this instance owns the writer lock): equivalent + * to `this.flush()`. + * - **Different process** (typical inspector flow): writes a request file + * into the lock directory and polls for an ack file. The writer's + * flush-request watcher (started in `init()` for writer instances) sees + * the request, calls `flush()`, and writes the ack. + * - **No writer running**: times out and returns `false`. Callers can + * proceed with last-flushed disk state and warn the operator. + * + * @param options.timeoutMs - How long to wait for an ack (default 5000ms). + * @returns `true` if the writer flushed in response to this request, + * `false` on timeout (no writer or unresponsive). + * + * @example + * ```typescript + * const reader = await Brainy.openReadOnly({ storage: { type: 'filesystem', rootDirectory: '/data/brain' } }) + * const fresh = await reader.requestFlush({ timeoutMs: 3000 }) + * if (!fresh) { + * console.warn('Writer did not respond; results reflect last natural flush.') + * } + * const bookings = await reader.find({ where: { entityType: 'booking' } }) + * ``` + */ + async requestFlush(options?: { timeoutMs?: number }): Promise { + await this.ensureInitialized() + const timeoutMs = options?.timeoutMs ?? 5000 + + // In-process shortcut: if this instance can write, just flush directly. + if (!this.isReadOnly) { + await this.flush() + return true + } + + if (typeof (this.storage as any).requestFlushOverFilesystem !== 'function') { + return false + } + return (this.storage as any).requestFlushOverFilesystem(timeoutMs) + } + /** * Get index loading status (Diagnostic for lazy loading) * @@ -4877,6 +5068,273 @@ export class Brainy implements BrainyInterface { } } + /** + * Run a battery of cheap invariant checks suitable for an operator-facing + * health probe. Reports counts of suspicious states alongside the basic + * size invariants — designed so an incident responder can spot the typical + * failure modes (mismatched index sizes, lots of `_seeded` records hanging + * around, missing field stats) without scanning the whole store. + * + * @example + * ```typescript + * const h = await brain.health() + * for (const c of h.checks) { + * console.log(`${c.status === 'pass' ? '✓' : '!'} ${c.name}: ${c.message}`) + * } + * ``` + */ + async health(): Promise<{ + overall: 'pass' | 'warn' | 'fail' + checks: Array<{ + name: string + status: 'pass' | 'warn' | 'fail' + message: string + details?: Record + }> + }> { + await this.ensureInitialized() + const checks: Array<{ + name: string + status: 'pass' | 'warn' | 'fail' + message: string + details?: Record + }> = [] + + const hnswSize = this.index.size() + const metadataStats = await this.metadataIndex.getStats() + const graphSize = await this.graphIndex.size() + + // 1. Index size parity. HNSW must hold at least one node per indexed entity. + if (hnswSize === metadataStats.totalEntries) { + checks.push({ + name: 'index-parity', + status: 'pass', + message: `HNSW (${hnswSize}) and metadata index (${metadataStats.totalEntries}) agree.`, + details: { hnswSize, metadataEntries: metadataStats.totalEntries, graphRelationships: graphSize } + }) + } else { + const drift = Math.abs(hnswSize - metadataStats.totalEntries) + checks.push({ + name: 'index-parity', + status: drift > Math.max(10, metadataStats.totalEntries * 0.01) ? 'fail' : 'warn', + message: `HNSW (${hnswSize}) and metadata (${metadataStats.totalEntries}) differ by ${drift}. Run a rebuild if the gap is unexpected.`, + details: { hnswSize, metadataEntries: metadataStats.totalEntries, drift } + }) + } + + // 2. Field registry sanity. Empty field set + non-zero entities = stale reader. + let fieldCount = 0 + try { + if (typeof (this.metadataIndex as any).getFieldStatistics === 'function') { + const fs = await (this.metadataIndex as any).getFieldStatistics() as Map + fieldCount = fs.size + } + } catch { + // ignore — metadata index doesn't expose stats + } + if (metadataStats.totalEntries > 0 && fieldCount === 0) { + checks.push({ + name: 'field-registry', + status: 'warn', + message: `${metadataStats.totalEntries} entities present but the field registry is empty. Reader may be stale; consider requestFlush().`, + details: { entities: metadataStats.totalEntries, fields: fieldCount } + }) + } else { + checks.push({ + name: 'field-registry', + status: 'pass', + message: `${fieldCount} fields registered for ${metadataStats.totalEntries} entities.`, + details: { fields: fieldCount, entities: metadataStats.totalEntries } + }) + } + + // 3. Seeded entity sweep — operators often want to know if demo seed data + // is still in a brain that's also serving real traffic (a common root + // cause of duplicate-ID and stale-content bugs). Uses the metadata index, + // not a full scan. + let seededIds: string[] = [] + try { + seededIds = await this.metadataIndex.getIds('_seeded', true) + } catch { + // ignore — field may not be indexed + } + if (seededIds.length > 0) { + checks.push({ + name: 'seeded-records', + status: 'warn', + message: `${seededIds.length} entities tagged _seeded:true. Verify this is the demo data you expect.`, + details: { count: seededIds.length, sample: seededIds.slice(0, 5) } + }) + } else { + checks.push({ + name: 'seeded-records', + status: 'pass', + message: 'No _seeded:true entities found.' + }) + } + + // 4. Lock heartbeat freshness — only meaningful for readers inspecting a + // running writer. If the lock exists but the heartbeat is old, the writer + // probably crashed. + if (typeof this.storage.readWriterLock === 'function') { + const lock = await this.storage.readWriterLock() + if (lock) { + const age = Date.now() - new Date(lock.lastHeartbeat).getTime() + if (age > 60_000) { + checks.push({ + name: 'writer-heartbeat', + status: 'warn', + message: `Writer lock heartbeat is ${Math.round(age / 1000)}s old (PID ${lock.pid} on ${lock.hostname}). Writer may be hung.`, + details: { lock, ageMs: age } + }) + } else { + checks.push({ + name: 'writer-heartbeat', + status: 'pass', + message: `Writer healthy (PID ${lock.pid} on ${lock.hostname}, heartbeat ${Math.round(age / 1000)}s ago).`, + details: { lock } + }) + } + } + } + + const worst = checks.reduce<'pass' | 'warn' | 'fail'>((acc, c) => { + if (c.status === 'fail') return 'fail' + if (c.status === 'warn' && acc !== 'fail') return 'warn' + return acc + }, 'pass') + + return { overall: worst, checks } + } + + /** + * Explain how a `find` query's `where` clause will be served. For each + * field, returns whether it will hit the column store (best), a sparse + * chunked index (legacy fallback), or has no index entries at all (silently + * empty result — usually a bug or a stale reader). Designed to be the very + * first thing an operator runs when `find()` returns surprising results. + * + * @example + * ```typescript + * const plan = await brain.explain({ where: { entityType: 'booking', status: 'paid' } }) + * for (const f of plan.fieldPlan) { + * console.log(`${f.field} -> ${f.path}: ${f.notes ?? ''}`) + * } + * ``` + */ + async explain(params: FindParams): Promise<{ + query: FindParams + fieldPlan: Array<{ field: string; path: 'column-store' | 'sparse-chunked' | 'none'; notes?: string }> + warnings: string[] + }> { + await this.ensureInitialized() + const fieldPlan: Array<{ field: string; path: 'column-store' | 'sparse-chunked' | 'none'; notes?: string }> = [] + const warnings: string[] = [] + + const where = (params as any)?.where + if (where && typeof where === 'object') { + for (const field of Object.keys(where)) { + const result = await this.metadataIndex.explainField(field) + fieldPlan.push({ field, path: result.path, notes: result.notes }) + if (result.path === 'none') { + warnings.push( + `Field "${field}" has no index entries. find() will return [] silently. ` + + `Run brain.requestFlush() or check the writer's field registry.` + ) + } + } + } else { + warnings.push('No `where` clause provided; nothing to explain.') + } + + return { query: params, fieldPlan, warnings } + } + + /** + * Operator-facing summary of what's in this Brainy store. Designed to be + * the first thing a human runs during an incident: counts, mode, lock owner, + * indexed field list, index health flags. + * + * Safe to call on a read-only instance. All counts come from already-loaded + * indexes (no extra storage scans), so this is fast (<10ms typical). + * + * @example + * ```typescript + * const s = await brain.stats() + * console.log(`${s.entityCount} entities (${Object.entries(s.entitiesByType).map(([t,n])=>`${t}:${n}`).join(', ')})`) + * if (s.writerLock) { + * console.log(`Writer PID ${s.writerLock.pid} on ${s.writerLock.hostname}`) + * } + * ``` + */ + async stats(): Promise { + await this.ensureInitialized() + + const { NounTypeEnum, VerbTypeEnum } = await import('./types/graphTypes.js') + const nounCounts = typeof (this.storage as any).getNounCountsByType === 'function' + ? (this.storage as any).getNounCountsByType() as Uint32Array + : new Uint32Array(0) + const verbCounts = typeof (this.storage as any).getVerbCountsByType === 'function' + ? (this.storage as any).getVerbCountsByType() as Uint32Array + : new Uint32Array(0) + + const entitiesByType: Record = {} + let entityCount = 0 + for (let i = 0; i < nounCounts.length; i++) { + if (nounCounts[i] === 0) continue + const name = NounTypeEnum[i as number] + if (name) entitiesByType[name] = nounCounts[i] + entityCount += nounCounts[i] + } + + const relationsByType: Record = {} + let relationCount = 0 + for (let i = 0; i < verbCounts.length; i++) { + if (verbCounts[i] === 0) continue + const name = VerbTypeEnum[i as number] + if (name) relationsByType[name] = verbCounts[i] + relationCount += verbCounts[i] + } + + const metadataStats = await this.metadataIndex.getStats() + let fieldRegistry: string[] = [] + try { + if (typeof (this.metadataIndex as any).getFieldStatistics === 'function') { + const fieldStats = await (this.metadataIndex as any).getFieldStatistics() as Map + fieldRegistry = Array.from(fieldStats.keys()).sort() + } + } catch { + // Field stats unavailable on this metadata index implementation — leave empty. + } + + const writerLock = typeof this.storage.readWriterLock === 'function' + ? await this.storage.readWriterLock() + : null + + const storageBackend = this.storage.constructor.name + const rootDir = (this.storage as any).rootDir + + return { + mode: this.isReadOnly ? 'reader' : 'writer', + entityCount, + entitiesByType, + relationCount, + relationsByType, + fieldRegistry, + indexHealth: { + hnsw: this.index.size() > 0 || entityCount === 0, + metadata: metadataStats.totalEntries > 0 || entityCount === 0, + graph: (await this.graphIndex.size()) > 0 || relationCount === 0 + }, + storage: { + backend: storageBackend, + rootDir: typeof rootDir === 'string' ? rootDir : undefined + }, + writerLock: writerLock || undefined, + version: getBrainyVersion() + } + } + /** * Plugin and provider diagnostics — shows what's active and how subsystems are wired. * @@ -6835,6 +7293,20 @@ export class Brainy implements BrainyInterface { * Setup storage */ private async setupStorage(): Promise { + const rawStorage = this.config.storage as unknown + + // If the caller passed a pre-constructed storage adapter (e.g. + // `storage: new MemoryStorage()`), use it directly instead of routing + // through the factory. Otherwise the factory's `type === 'auto'` branch + // would silently create a FileSystemStorage at `./brainy-data`, ignoring + // the instance and surprising anyone who wrote `new MemoryStorage()` + // expecting it to be honoured. + if (rawStorage && typeof rawStorage === 'object' && 'init' in (rawStorage as any) && + typeof (rawStorage as any).init === 'function' && + typeof (rawStorage as any).saveNoun === 'function') { + return rawStorage as BaseStorage + } + const storageConfig = (this.config.storage || {}) as Record const storageType = (storageConfig.type as string) || 'auto' @@ -7017,7 +7489,10 @@ export class Brainy implements BrainyInterface { // Integration Hub - undefined/false = disabled integrations: config?.integrations ?? undefined as any, // Migration — disabled by default, opt-in for automatic migration - autoMigrate: config?.autoMigrate ?? false + autoMigrate: config?.autoMigrate ?? false, + // Multi-process safety + mode: config?.mode ?? 'writer', + force: config?.force ?? false } } @@ -7469,6 +7944,18 @@ export class Brainy implements BrainyInterface { } } + // Stop the cross-process flush-request watcher (no-op if never started). + if (this.storage && typeof this.storage.stopFlushRequestWatcher === 'function') { + this.storage.stopFlushRequestWatcher() + } + + // Release the writer lock (no-op for readers and for backends that don't + // hold a lock). Must run after the metadata buffer drain — otherwise a + // pending write could land after a successor writer claimed the lock. + if (this.storage && typeof this.storage.releaseWriterLock === 'function') { + await this.storage.releaseWriterLock() + } + this.initialized = false } diff --git a/src/cli/commands/inspect.ts b/src/cli/commands/inspect.ts new file mode 100644 index 00000000..6ee48404 --- /dev/null +++ b/src/cli/commands/inspect.ts @@ -0,0 +1,428 @@ +/** + * Inspect Commands + * + * Out-of-process diagnostics against a Brainy data directory. Every + * subcommand opens the store via `Brainy.openReadOnly()` so they coexist + * safely with a live writer (and with each other). The `--fresh` flag + * (default true) asks the writer to flush its in-memory state via the + * filesystem RPC before opening, so query results reflect the very latest + * writes. Without `--fresh`, results reflect the writer's last natural flush. + * + * Designed to be the operator's primary debugging surface for any Brainy + * deployment. None of these commands mutate the store. + */ + +import chalk from 'chalk' +import ora from 'ora' +import { Brainy } from '../../brainy.js' + +interface InspectOptions { + fresh?: boolean + json?: boolean + pretty?: boolean + quiet?: boolean +} + +interface OpenOptions extends InspectOptions { + flushTimeoutMs?: number +} + +/** + * Open a read-only handle on `path`. If `--fresh` (default), first ask the + * writer to flush via the filesystem RPC. Logs warnings when fresh state + * cannot be obtained — but proceeds anyway since stale-but-honest is better + * than failed. + */ +async function openReader(rootDir: string, options: OpenOptions): Promise { + const wantsFresh = options.fresh !== false + const flushTimeoutMs = options.flushTimeoutMs ?? 5000 + + if (wantsFresh) { + // Use a throwaway reader purely to trigger requestFlush — that way the + // ack lands before we open the main reader and have it rebuild indexes. + try { + const probe = await Brainy.openReadOnly({ + storage: { type: 'filesystem', rootDirectory: rootDir } + }) + const acked = await probe.requestFlush({ timeoutMs: flushTimeoutMs }) + await probe.close() + if (!acked && !options.quiet) { + console.warn(chalk.yellow( + `⚠ Writer did not ack flush within ${flushTimeoutMs}ms. Results reflect last natural flush.` + )) + } + } catch (err) { + if (!options.quiet) { + console.warn(chalk.yellow(`⚠ Flush request failed: ${(err as Error).message}`)) + } + } + } + + return Brainy.openReadOnly({ + storage: { type: 'filesystem', rootDirectory: rootDir } + }) +} + +function emit(data: unknown, options: InspectOptions): void { + if (options.json) { + console.log(options.pretty ? JSON.stringify(data, null, 2) : JSON.stringify(data)) + } else { + console.log(JSON.stringify(data, null, 2)) + } +} + +export const inspectCommands = { + /** + * Show counts, mode, indexed fields, writer lock info. + */ + async stats(path: string, options: InspectOptions) { + const spinner = options.quiet ? null : ora('Opening read-only…').start() + try { + const brain = await openReader(path, options) + spinner?.succeed('Connected') + const stats = await brain.stats() + emit(stats, options) + await brain.close() + // close() releases the indexes but some global timers (UnifiedCache + // bookkeeping, PathResolver stats) keep the event loop alive. CLI + // commands are one-shot — exit explicitly. + process.exit(0) + } catch (err) { + spinner?.fail((err as Error).message) + process.exit(1) + } + }, + + /** + * Find entities matching the given filter. Mirrors `brain.find()`. + */ + async find(path: string, options: InspectOptions & { + type?: string + where?: string + limit?: string + offset?: string + }) { + const spinner = options.quiet ? null : ora('Opening read-only…').start() + try { + const brain = await openReader(path, options) + if (spinner) spinner.text = 'Querying…' + const where = options.where ? JSON.parse(options.where) : undefined + const limit = options.limit ? parseInt(options.limit, 10) : 20 + const offset = options.offset ? parseInt(options.offset, 10) : 0 + const results = await brain.find({ + type: options.type as any, + where, + limit, + offset + }) + spinner?.succeed(`Found ${results.length} result(s)`) + emit(results, options) + await brain.close() + process.exit(0) + } catch (err) { + spinner?.fail((err as Error).message) + process.exit(1) + } + }, + + /** + * Get a single entity by ID. + */ + async get(path: string, id: string, options: InspectOptions) { + const spinner = options.quiet ? null : ora('Opening read-only…').start() + try { + const brain = await openReader(path, options) + if (spinner) spinner.text = 'Fetching…' + const entity = await brain.get(id) + spinner?.succeed(entity ? 'Found' : 'Not found') + emit(entity, options) + await brain.close() + process.exit(0) + } catch (err) { + spinner?.fail((err as Error).message) + process.exit(1) + } + }, + + /** + * Show inbound or outbound relationships for an entity. + */ + async relations(path: string, id: string, options: InspectOptions & { + direction?: 'in' | 'out' | 'both' + type?: string + limit?: string + }) { + const spinner = options.quiet ? null : ora('Opening read-only…').start() + try { + const brain = await openReader(path, options) + if (spinner) spinner.text = 'Walking graph…' + const direction = options.direction ?? 'both' + const limit = options.limit ? parseInt(options.limit, 10) : 50 + const rels = await brain.getRelations({ + from: id, + direction, + type: options.type as any, + limit + } as any) + spinner?.succeed(`Found ${rels.length} relationship(s)`) + emit(rels, options) + await brain.close() + process.exit(0) + } catch (err) { + spinner?.fail((err as Error).message) + process.exit(1) + } + }, + + /** + * Explain how a where-clause query will be served by the index. Shows + * column-store vs sparse-chunked vs no-index per field. Designed to + * diagnose the silent-empty-result class of issues. + */ + async explain(path: string, options: InspectOptions & { type?: string; where?: string }) { + const spinner = options.quiet ? null : ora('Opening read-only…').start() + try { + const brain = await openReader(path, options) + if (spinner) spinner.text = 'Planning query…' + const where = options.where ? JSON.parse(options.where) : {} + const plan = await brain.explain({ type: options.type as any, where }) + spinner?.succeed('Plan ready') + emit(plan, options) + await brain.close() + process.exit(0) + } catch (err) { + spinner?.fail((err as Error).message) + process.exit(1) + } + }, + + /** + * Run the invariant-check battery. + */ + async health(path: string, options: InspectOptions) { + const spinner = options.quiet ? null : ora('Opening read-only…').start() + try { + const brain = await openReader(path, options) + if (spinner) spinner.text = 'Running checks…' + const report = await brain.health() + spinner?.succeed(`Overall: ${report.overall}`) + emit(report, options) + await brain.close() + process.exit(report.overall === 'fail' ? 2 : 0) + } catch (err) { + spinner?.fail((err as Error).message) + process.exit(1) + } + }, + + /** + * Random N-entity sample. + */ + async sample(path: string, options: InspectOptions & { type?: string; n?: string }) { + const spinner = options.quiet ? null : ora('Opening read-only…').start() + try { + const brain = await openReader(path, options) + if (spinner) spinner.text = 'Sampling…' + const n = options.n ? parseInt(options.n, 10) : 10 + // Pull a larger window then randomize down to N. Avoids needing a + // dedicated random-sample API in Brainy core for this v1. + const window = Math.min(Math.max(n * 10, 50), 1000) + const pool = await brain.find({ + type: options.type as any, + limit: window + }) + const sample = pool + .map((v) => ({ v, k: Math.random() })) + .sort((a, b) => a.k - b.k) + .slice(0, n) + .map((x) => x.v) + spinner?.succeed(`Sampled ${sample.length} of ${pool.length}`) + emit(sample, options) + await brain.close() + process.exit(0) + } catch (err) { + spinner?.fail((err as Error).message) + process.exit(1) + } + }, + + /** + * List indexed metadata fields. + */ + async fields(path: string, options: InspectOptions) { + const spinner = options.quiet ? null : ora('Opening read-only…').start() + try { + const brain = await openReader(path, options) + const stats = await brain.stats() + spinner?.succeed(`${stats.fieldRegistry.length} indexed field(s)`) + emit(stats.fieldRegistry, options) + await brain.close() + process.exit(0) + } catch (err) { + spinner?.fail((err as Error).message) + process.exit(1) + } + }, + + /** + * Dump all entities of a given type as JSONL. Streams via pagination to + * keep memory bounded. Outputs to stdout for redirection to a file. + */ + async dump(path: string, options: InspectOptions & { type?: string; batch?: string }) { + const batch = options.batch ? parseInt(options.batch, 10) : 500 + try { + const brain = await openReader(path, options) + let offset = 0 + while (true) { + const page = await brain.find({ + type: options.type as any, + limit: batch, + offset + }) + if (page.length === 0) break + for (const e of page) { + process.stdout.write(JSON.stringify(e) + '\n') + } + if (page.length < batch) break + offset += page.length + } + await brain.close() + process.exit(0) + } catch (err) { + console.error(chalk.red((err as Error).message)) + process.exit(1) + } + }, + + /** + * Watch for newly-written entities. Polls `brain.find()` on an interval + * and emits entities that have an `updatedAt`/`createdAt` newer than the + * previous tick. Best-effort — for forensic-grade tracking, use the + * writer's own audit log if one exists. + */ + async watch(path: string, options: InspectOptions & { type?: string; interval?: string }) { + const intervalMs = options.interval ? parseInt(options.interval, 10) : 1000 + let seen = new Set() + let firstTick = true + console.error(chalk.cyan(`Watching ${path} (every ${intervalMs}ms, Ctrl+C to stop)…`)) + + const tick = async () => { + try { + const brain = await openReader(path, { ...options, quiet: true }) + const page = await brain.find({ + type: options.type as any, + limit: 200 + }) + for (const e of page) { + if (!seen.has(e.id)) { + seen.add(e.id) + if (!firstTick) { + process.stdout.write(JSON.stringify(e) + '\n') + } + } + } + await brain.close() + firstTick = false + } catch (err) { + console.error(chalk.yellow(`watch tick failed: ${(err as Error).message}`)) + } + } + + await tick() + const handle = setInterval(() => { void tick() }, intervalMs) + process.on('SIGINT', () => { + clearInterval(handle) + process.exit(0) + }) + }, + + /** + * Atomic tarball backup of the data directory. Asks the writer to flush + * first (so the snapshot is consistent), then tars into the destination. + */ + async backup(path: string, dest: string, options: InspectOptions) { + const spinner = options.quiet ? null : ora('Requesting flush…').start() + try { + // Request flush so the snapshot is internally consistent. + const brain = await openReader(path, { ...options, fresh: true }) + await brain.close() + + if (spinner) spinner.text = 'Archiving…' + const { spawn } = await import('node:child_process') + const proc = spawn('tar', ['-cf', dest, '-C', path, '.'], { stdio: 'inherit' }) + await new Promise((resolve, reject) => { + proc.on('exit', (code) => code === 0 ? resolve() : reject(new Error(`tar exited ${code}`))) + proc.on('error', reject) + }) + spinner?.succeed(`Backup written to ${dest}`) + process.exit(0) + } catch (err) { + spinner?.fail((err as Error).message) + process.exit(1) + } + }, + + /** + * Rebuild indexes from raw storage. Opens the store in writer mode and + * triggers a rebuild. Refuses to run if another writer holds the lock + * (use --force only after stopping the live writer). + */ + async repair(path: string, options: InspectOptions & { force?: boolean }) { + const spinner = options.quiet ? null : ora('Opening writer…').start() + try { + const brain = new Brainy({ + storage: { type: 'filesystem', rootDirectory: path }, + mode: 'writer', + force: options.force + }) + await brain.init() + if (spinner) spinner.text = 'Rebuilding indexes…' + // Force a flush to persist whatever the rebuild produced. + await brain.flush() + spinner?.succeed('Repair complete') + const stats = await brain.stats() + emit(stats, options) + await brain.close() + process.exit(0) + } catch (err) { + spinner?.fail((err as Error).message) + process.exit(1) + } + }, + + /** + * Compare two brain directories: entity and relation counts by type, and + * a sample of entity IDs present in one but not the other. + */ + async diff(pathA: string, pathB: string, options: InspectOptions & { sample?: string }) { + const sample = options.sample ? parseInt(options.sample, 10) : 100 + const spinner = options.quiet ? null : ora('Opening both stores…').start() + try { + const [a, b] = await Promise.all([ + openReader(pathA, { ...options, quiet: true, fresh: false }), + openReader(pathB, { ...options, quiet: true, fresh: false }) + ]) + const [statsA, statsB] = await Promise.all([a.stats(), b.stats()]) + const idsA = new Set((await a.find({ limit: sample })).map((e) => e.id)) + const idsB = new Set((await b.find({ limit: sample })).map((e) => e.id)) + await Promise.all([a.close(), b.close()]) + spinner?.succeed('Compared') + + const onlyInA = [...idsA].filter((id) => !idsB.has(id)).slice(0, 20) + const onlyInB = [...idsB].filter((id) => !idsA.has(id)).slice(0, 20) + + emit({ + a: { path: pathA, entityCount: statsA.entityCount, relationCount: statsA.relationCount, entitiesByType: statsA.entitiesByType }, + b: { path: pathB, entityCount: statsB.entityCount, relationCount: statsB.relationCount, entitiesByType: statsB.entitiesByType }, + sampleSize: sample, + sampleOnlyInA: onlyInA, + sampleOnlyInB: onlyInB, + note: 'Sample-based comparison. ID-level diffs reflect the first N entities surveyed in each store.' + }, options) + process.exit(0) + } catch (err) { + spinner?.fail((err as Error).message) + process.exit(1) + } + } +} diff --git a/src/cli/index.ts b/src/cli/index.ts index 9f1b09eb..a94e7c77 100644 --- a/src/cli/index.ts +++ b/src/cli/index.ts @@ -18,6 +18,7 @@ import { nlpCommands } from './commands/nlp.js' import { insightsCommands } from './commands/insights.js' import { importCommands } from './commands/import.js' import { cowCommands } from './commands/cow.js' +import { inspectCommands } from './commands/inspect.js' import { readFileSync } from 'fs' import { fileURLToPath } from 'url' import { dirname, join } from 'path' @@ -551,6 +552,143 @@ program .description('Show detailed database statistics') .action(dataCommands.stats) +// ===== Inspect Commands ===== +// Out-of-process diagnostics. Every subcommand opens the store via +// Brainy.openReadOnly() so a live writer can keep running. `--fresh` +// (default) asks the writer to flush before opening. + +program + .command('inspect') + .description('🔍 Out-of-process diagnostics on a Brainy data directory') + .addCommand( + new Command('stats') + .argument('', 'Path to the Brainy data directory') + .description('Counts, mode, indexed fields, writer lock info') + .option('--no-fresh', 'Skip the writer flush request (faster, but state may be slightly stale)') + .option('--json', 'Output as JSON') + .option('--pretty', 'Pretty-print JSON') + .action((path, options) => inspectCommands.stats(path, options)) + ) + .addCommand( + new Command('find') + .argument('', 'Path to the Brainy data directory') + .description('Find entities matching a where-clause filter') + .option('--type ', 'Filter by entity type') + .option('--where ', 'Metadata filter (JSON object)') + .option('--limit ', 'Max results', '20') + .option('--offset ', 'Skip N results (pagination)') + .option('--no-fresh', 'Skip the writer flush request') + .option('--json', 'Output as JSON') + .option('--pretty', 'Pretty-print JSON') + .action((path, options) => inspectCommands.find(path, options)) + ) + .addCommand( + new Command('get') + .argument('', 'Path to the Brainy data directory') + .argument('', 'Entity ID') + .description('Fetch a single entity by ID') + .option('--no-fresh', 'Skip the writer flush request') + .option('--json', 'Output as JSON') + .option('--pretty', 'Pretty-print JSON') + .action((path, id, options) => inspectCommands.get(path, id, options)) + ) + .addCommand( + new Command('relations') + .argument('', 'Path to the Brainy data directory') + .argument('', 'Entity ID') + .description('Show inbound/outbound relationships for an entity') + .option('--direction ', 'in | out | both', 'both') + .option('--type ', 'Filter by verb type') + .option('--limit ', 'Max relationships', '50') + .option('--no-fresh', 'Skip the writer flush request') + .option('--json', 'Output as JSON') + .option('--pretty', 'Pretty-print JSON') + .action((path, id, options) => inspectCommands.relations(path, id, options)) + ) + .addCommand( + new Command('explain') + .argument('', 'Path to the Brainy data directory') + .description('Show which index path will serve each where-clause field (column-store / sparse / none)') + .option('--type ', 'Filter by entity type') + .option('--where ', 'Metadata filter to plan (JSON object)') + .option('--no-fresh', 'Skip the writer flush request') + .option('--json', 'Output as JSON') + .option('--pretty', 'Pretty-print JSON') + .action((path, options) => inspectCommands.explain(path, options)) + ) + .addCommand( + new Command('health') + .argument('', 'Path to the Brainy data directory') + .description('Run invariant checks (index parity, field registry, _seeded sweep, writer heartbeat)') + .option('--no-fresh', 'Skip the writer flush request') + .option('--json', 'Output as JSON') + .option('--pretty', 'Pretty-print JSON') + .action((path, options) => inspectCommands.health(path, options)) + ) + .addCommand( + new Command('sample') + .argument('', 'Path to the Brainy data directory') + .description('Random N-entity sample') + .option('--type ', 'Filter by entity type') + .option('--n ', 'Sample size', '10') + .option('--no-fresh', 'Skip the writer flush request') + .option('--json', 'Output as JSON') + .option('--pretty', 'Pretty-print JSON') + .action((path, options) => inspectCommands.sample(path, options)) + ) + .addCommand( + new Command('fields') + .argument('', 'Path to the Brainy data directory') + .description('List indexed metadata fields') + .option('--no-fresh', 'Skip the writer flush request') + .option('--json', 'Output as JSON') + .option('--pretty', 'Pretty-print JSON') + .action((path, options) => inspectCommands.fields(path, options)) + ) + .addCommand( + new Command('dump') + .argument('', 'Path to the Brainy data directory') + .description('Dump all entities of a type as JSONL (one per line) to stdout') + .option('--type ', 'Filter by entity type') + .option('--batch ', 'Page size', '500') + .option('--no-fresh', 'Skip the writer flush request') + .action((path, options) => inspectCommands.dump(path, options)) + ) + .addCommand( + new Command('watch') + .argument('', 'Path to the Brainy data directory') + .description('Tail newly-written entities') + .option('--type ', 'Filter by entity type') + .option('--interval ', 'Poll interval', '1000') + .action((path, options) => inspectCommands.watch(path, options)) + ) + .addCommand( + new Command('backup') + .argument('', 'Path to the Brainy data directory') + .argument('', 'Destination tarball') + .description('Atomic flush-then-tar snapshot of the data directory') + .action((path, dest, options) => inspectCommands.backup(path, dest, options)) + ) + .addCommand( + new Command('repair') + .argument('', 'Path to the Brainy data directory') + .description('Rebuild indexes from raw storage (writer-mode — stop the live writer first)') + .option('--force', 'Override the writer lock if you are sure no other writer is running') + .option('--json', 'Output as JSON') + .option('--pretty', 'Pretty-print JSON') + .action((path, options) => inspectCommands.repair(path, options)) + ) + .addCommand( + new Command('diff') + .argument('', 'First Brainy data directory') + .argument('', 'Second Brainy data directory') + .description('Compare counts and a sample of entity IDs between two stores') + .option('--sample ', 'Sample size per side', '100') + .option('--json', 'Output as JSON') + .option('--pretty', 'Pretty-print JSON') + .action((pathA, pathB, options) => inspectCommands.diff(pathA, pathB, options)) + ) + // ===== NLP Commands ===== program diff --git a/src/storage/adapters/fileSystemStorage.ts b/src/storage/adapters/fileSystemStorage.ts index ea68f85d..55191990 100644 --- a/src/storage/adapters/fileSystemStorage.ts +++ b/src/storage/adapters/fileSystemStorage.ts @@ -18,8 +18,10 @@ import { BaseStorage, StorageBatchConfig, SYSTEM_DIR, - STATISTICS_KEY + STATISTICS_KEY, + WriterLockInfo } from '../baseStorage.js' +import { getBrainyVersion } from '../../utils/index.js' // Type aliases for better readability type HNSWNode = HNSWNoun @@ -91,6 +93,30 @@ export class FileSystemStorage extends BaseStorage { private lockTimers: Map = new Map() // Track timers for cleanup private allTimers: Set = new Set() // Track all timers for cleanup + // Writer-lock state. The writer lock at `locks/_writer.lock` is acquired + // at Brainy.init() in writer mode and released at close(). A heartbeat + // 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 + private static readonly WRITER_STALE_THRESHOLD_MS = 60_000 + private writerLockHeartbeat?: NodeJS.Timeout + private writerLockInfo?: WriterLockInfo + + // 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 + // and wait for the ack. Polling interval is short enough to feel synchronous + // for operator workflows but doesn't pressure the FS. + private static readonly FLUSH_REQUEST_DIR = '_flush_requests' + private static readonly FLUSH_RESPONSE_DIR = '_flush_responses' + private static readonly FLUSH_WATCH_INTERVAL_MS = 500 + private static readonly FLUSH_POLL_INTERVAL_MS = 100 + private static readonly FLUSH_REQUEST_TTL_MS = 60_000 + private flushWatcherInterval?: NodeJS.Timeout + private flushWatcherInFlight = false + private flushWatcherOnRequest?: () => Promise + // CRITICAL FIX: Mutex locks for HNSW concurrency control // Prevents read-modify-write races during concurrent neighbor updates at scale (1000+ ops) // Matches MemoryStorage and OPFSStorage behavior (tested in production) @@ -1212,6 +1238,352 @@ export class FileSystemStorage extends BaseStorage { // Removed 10 *_internal method overrides - now inherit from BaseStorage's type-first implementation // Removed 2 pagination methods (getNounsWithPagination, getVerbsWithPagination) - use BaseStorage's implementation + public supportsMultiProcessLocking(): boolean { + return true + } + + /** + * Acquire the process-level writer lock for this data directory. Called by + * `Brainy.init()` in writer mode. Refuses to open if another live writer + * holds the lock — the worst possible default is to "succeed" with stale + * indexes and silently lie about query results. + * + * Lock file: `/locks/_writer.lock`. + * + * Stale-lock detection: a lock is considered stale when ALL of: + * 1. Same hostname as the current process (cross-host PID checks are unsafe). + * 2. The recorded PID is no longer alive (`process.kill(pid, 0)` → ESRCH), OR + * `lastHeartbeat` is older than `WRITER_STALE_THRESHOLD_MS` (60s). + * Stale locks are overwritten with a warning. `force: true` overrides even live locks. + */ + public async acquireWriterLock(options?: { force?: boolean }): Promise { + await this.ensureInitialized() + await this.ensureDirectoryExists(this.lockDir) + + const lockFile = path.join(this.lockDir, FileSystemStorage.WRITER_LOCK_FILE) + const os = await import('node:os') + const hostname = os.hostname() + const now = new Date().toISOString() + + const existing = await this.readWriterLock() + if (existing && !options?.force) { + // 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 the lock. + if (existing.pid === (typeof process !== 'undefined' ? process.pid : 0) && + existing.hostname === hostname) { + 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.` + ) + } else { + const stale = await this.isWriterLockStale(existing) + if (!stale) { + const err = new Error( + `Another writer holds this Brainy directory.\n` + + ` PID: ${existing.pid} on host ${existing.hostname}\n` + + ` Started: ${existing.startedAt}\n` + + ` Heartbeat: ${existing.lastHeartbeat}\n` + + ` Version: ${existing.version}\n` + + ` Directory: ${this.rootDir}\n\n` + + `For diagnostic queries against this live store, use:\n` + + ` const reader = await Brainy.openReadOnly({ storage: { type: 'filesystem', rootDirectory: '${this.rootDir}' } })\n\n` + + `If you have verified the existing lock is stale (e.g. a crashed writer on a different host that PID liveness cannot reach), pass { force: true }.` + ) + ;(err as any).code = 'BRAINY_WRITER_LOCKED' + ;(err as any).lockInfo = existing + throw err + } + console.warn( + `[brainy] Overwriting stale writer lock for ${this.rootDir} ` + + `(PID ${existing.pid} on ${existing.hostname} appears dead).` + ) + } + } else if (existing && options?.force) { + console.warn( + `[brainy] Force-overwriting writer lock for ${this.rootDir} ` + + `(was held by PID ${existing.pid} on ${existing.hostname}).` + ) + } + + const info: WriterLockInfo = { + pid: typeof process !== 'undefined' && process.pid ? process.pid : 0, + hostname, + startedAt: existing && options?.force ? existing.startedAt : now, + lastHeartbeat: now, + version: getBrainyVersion(), + rootDir: this.rootDir + } + + await this.writeFileAtomic(lockFile, JSON.stringify(info, null, 2)) + this.writerLockInfo = info + + // Heartbeat — rewrite lastHeartbeat every WRITER_HEARTBEAT_MS so other + // processes can tell a live writer from one that crashed without releasing. + this.writerLockHeartbeat = setInterval(() => { + this.refreshWriterLockHeartbeat().catch((err) => { + console.warn('[brainy] Failed to refresh writer lock heartbeat:', err) + }) + }, FileSystemStorage.WRITER_HEARTBEAT_MS) + if (typeof this.writerLockHeartbeat.unref === 'function') { + // Don't keep the event loop alive just for the heartbeat. + this.writerLockHeartbeat.unref() + } + + return info + } + + public async releaseWriterLock(): Promise { + if (this.writerLockHeartbeat) { + clearInterval(this.writerLockHeartbeat) + this.writerLockHeartbeat = undefined + } + if (!this.writerLockInfo) { + return + } + const lockFile = path.join(this.lockDir, FileSystemStorage.WRITER_LOCK_FILE) + 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) { + await fs.promises.unlink(lockFile) + } + } catch (err: any) { + if (err.code !== 'ENOENT') { + console.warn('[brainy] Failed to release writer lock file:', err) + } + } finally { + this.writerLockInfo = undefined + } + } + + public async readWriterLock(): Promise { + await this.ensureInitialized() + const lockFile = path.join(this.lockDir, FileSystemStorage.WRITER_LOCK_FILE) + try { + const raw = await fs.promises.readFile(lockFile, 'utf-8') + return JSON.parse(raw) as WriterLockInfo + } catch (err: any) { + if (err.code === 'ENOENT') return null + console.warn('[brainy] Failed to read writer lock file:', err) + return null + } + } + + /** + * Atomically refresh `lastHeartbeat` on the writer lock we own. Skips if the + * lock file has been deleted out from under us (e.g. operator removed it). + */ + private async refreshWriterLockHeartbeat(): Promise { + if (!this.writerLockInfo) return + const current = await this.readWriterLock() + if (!current) return + // Defensive: don't overwrite if a successor claimed the lock. + if (current.pid !== this.writerLockInfo.pid || current.hostname !== this.writerLockInfo.hostname) { + return + } + const updated: WriterLockInfo = { + ...this.writerLockInfo, + lastHeartbeat: new Date().toISOString() + } + const lockFile = path.join(this.lockDir, FileSystemStorage.WRITER_LOCK_FILE) + await this.writeFileAtomic(lockFile, JSON.stringify(updated, null, 2)) + this.writerLockInfo = updated + } + + /** + * Determine whether an existing writer lock is stale (safe to overwrite). + * Same hostname and (dead PID OR heartbeat older than threshold) → stale. + * Different hostname → cannot prove stale, treat as live. + */ + private async isWriterLockStale(lock: WriterLockInfo): Promise { + const os = await import('node:os') + if (lock.hostname !== os.hostname()) { + return false + } + const heartbeatAge = Date.now() - new Date(lock.lastHeartbeat).getTime() + const pidAlive = this.isPidAlive(lock.pid) + if (!pidAlive) return true + return heartbeatAge > FileSystemStorage.WRITER_STALE_THRESHOLD_MS + } + + /** + * `process.kill(pid, 0)` sends signal 0 — no signal is actually delivered; + * it just checks whether the kernel still considers `pid` reachable from + * this process. ESRCH = no such process. EPERM = exists but we can't signal + * it, which still proves it's alive. + */ + private isPidAlive(pid: number): boolean { + if (!pid || pid <= 0) return false + try { + process.kill(pid, 0) + return true + } catch (err: any) { + if (err.code === 'EPERM') return true + return false + } + } + + /** + * Atomic write via temp-file-then-rename so concurrent readers never see a + * half-written lock JSON. Reused by writer-lock writes + heartbeat. + */ + private async writeFileAtomic(filePath: string, contents: string): Promise { + const tmp = `${filePath}.tmp-${process.pid}-${Date.now()}` + await fs.promises.writeFile(tmp, contents) + await fs.promises.rename(tmp, filePath) + } + + /** + * 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. + */ + public startFlushRequestWatcher(onRequest: () => Promise): void { + if (this.flushWatcherInterval) return // already watching + 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 + this.flushWatcherInFlight = true + this.processFlushRequests(reqDir, ackDir).finally(() => { + this.flushWatcherInFlight = false + }) + }, FileSystemStorage.FLUSH_WATCH_INTERVAL_MS) + if (typeof this.flushWatcherInterval.unref === 'function') { + this.flushWatcherInterval.unref() + } + } + + public stopFlushRequestWatcher(): void { + if (this.flushWatcherInterval) { + clearInterval(this.flushWatcherInterval) + this.flushWatcherInterval = undefined + } + this.flushWatcherOnRequest = undefined + } + + /** + * Process any pending `.req` files: invoke the flush callback once, then + * write an `.ack` for each request. Multiple requests that arrived in the + * same tick share a single flush — they all see the same ack timestamp. + */ + private async processFlushRequests(reqDir: string, ackDir: string): Promise { + let entries: string[] + try { + entries = await fs.promises.readdir(reqDir) + } catch (err: any) { + if (err.code !== 'ENOENT') { + console.warn('[brainy] Flush watcher readdir failed:', err) + } + return + } + + const reqs = entries.filter((e: string) => e.endsWith('.req')) + if (reqs.length === 0) return + + // One flush per tick — N pending requests share it. + const callback = this.flushWatcherOnRequest + if (!callback) return + let flushError: Error | null = null + try { + await callback() + } catch (err: any) { + flushError = err instanceof Error ? err : new Error(String(err)) + console.warn('[brainy] Flush callback threw inside flush-request watcher:', err) + } + + const ackTimestamp = new Date().toISOString() + for (const filename of reqs) { + const requestId = filename.replace(/\.req$/, '') + const ackPath = path.join(ackDir, `${requestId}.ack`) + const ackBody = JSON.stringify({ + requestId, + completedAt: ackTimestamp, + ok: !flushError, + error: flushError ? flushError.message : undefined + }) + try { + await this.writeFileAtomic(ackPath, ackBody) + await fs.promises.unlink(path.join(reqDir, filename)) + } catch (err: any) { + if (err.code !== 'ENOENT') { + console.warn('[brainy] Failed to write flush ack:', err) + } + } + } + + // Garbage-collect stale .req files in case a watcher missed them (e.g. + // the writer was restarted between request and processing). + const now = Date.now() + for (const filename of entries) { + if (!filename.endsWith('.req')) continue + const fp = path.join(reqDir, filename) + try { + const stat = await fs.promises.stat(fp) + if (now - stat.mtimeMs > FileSystemStorage.FLUSH_REQUEST_TTL_MS) { + await fs.promises.unlink(fp).catch(() => {}) + } + } catch { + // ignore — file already removed + } + } + } + + /** + * Inspector side: drop a `.req` file and poll for the corresponding `.ack`. + * Returns true if the writer acknowledged in time, false on timeout. + */ + public async requestFlushOverFilesystem(timeoutMs: number): Promise { + await this.ensureInitialized() + const reqDir = path.join(this.lockDir, FileSystemStorage.FLUSH_REQUEST_DIR) + const ackDir = path.join(this.lockDir, FileSystemStorage.FLUSH_RESPONSE_DIR) + await this.ensureDirectoryExists(reqDir) + await this.ensureDirectoryExists(ackDir) + + const requestId = `${Date.now()}-${process.pid}-${Math.random().toString(36).slice(2, 10)}` + const reqPath = path.join(reqDir, `${requestId}.req`) + const ackPath = path.join(ackDir, `${requestId}.ack`) + const body = JSON.stringify({ + requestId, + requestedAt: new Date().toISOString(), + requesterPid: process.pid + }) + + await this.writeFileAtomic(reqPath, body) + + const deadline = Date.now() + timeoutMs + while (Date.now() < deadline) { + try { + await fs.promises.access(ackPath, fs.constants.F_OK) + await fs.promises.unlink(ackPath).catch(() => {}) + return true + } catch { + // not yet + } + await new Promise((r) => setTimeout(r, FileSystemStorage.FLUSH_POLL_INTERVAL_MS)) + } + + // Timeout — best effort to clean up our request so the writer doesn't act + // on stale work later. Watcher will GC anyway after FLUSH_REQUEST_TTL_MS. + await fs.promises.unlink(reqPath).catch(() => {}) + return false + } + /** * Acquire a file-based lock for coordinating operations across multiple processes * @param lockKey The key to lock on diff --git a/src/storage/baseStorage.ts b/src/storage/baseStorage.ts index 3b93118c..bf607f07 100644 --- a/src/storage/baseStorage.ts +++ b/src/storage/baseStorage.ts @@ -80,6 +80,20 @@ export const VERBS_METADATA_DIR = 'entities/verbs/metadata' export const SYSTEM_DIR = '_system' export const STATISTICS_KEY = 'statistics' +/** + * Metadata persisted in the writer lock file. Used by stale-lock detection + * (PID liveness + hostname + heartbeat freshness) and by `brain.stats()` for + * operator-facing diagnostics. + */ +export interface WriterLockInfo { + pid: number + hostname: string + startedAt: string // ISO timestamp when the lock was first acquired + lastHeartbeat: string // ISO timestamp of the most recent heartbeat update + version: string // Brainy version that wrote the lock + rootDir?: string // Convenience for log lines / error messages +} + /** * FNV-1a hash returning a 2-char hex bucket (00-ff). * Distributes system keys across 256 sub-prefixes to avoid @@ -398,6 +412,79 @@ export abstract class BaseStorage extends BaseStorageAdapter { } } + /** + * Whether this storage adapter enforces multi-process writer exclusion. + * Filesystem storage returns true; cloud and memory adapters return false. + * Brainy.init() checks this to decide whether to log a multi-process warning + * in writer mode. + */ + public supportsMultiProcessLocking(): boolean { + return false + } + + /** + * Attempt to acquire the process-level writer lock at init time. Default + * implementation is a no-op (multi-process safety not enforced). Filesystem + * storage overrides this with real locking semantics. + * + * @param options.force - If true, overwrite any existing writer lock. Use + * only when stale detection cannot prove the existing lock is dead. + * @returns Metadata about the acquired lock (or `null` if no lock was needed). + * @throws If another live writer holds the lock and `force` is not set. + */ + public async acquireWriterLock(options?: { force?: boolean }): Promise { + return null + } + + /** + * Release the writer lock acquired by `acquireWriterLock()`. No-op if no lock + * was held. Filesystem storage overrides this to delete the lock file and + * stop the heartbeat timer. + */ + public async releaseWriterLock(): Promise { + // No-op by default + } + + /** + * Read the current writer-lock metadata if one is held by any process. Used + * by `brain.stats()` for diagnostics. Default returns null (no lock model). + */ + public async readWriterLock(): Promise { + return null + } + + /** + * Start watching for cross-process flush requests. The writer Brainy + * instance calls this so that out-of-process inspectors can ask for a + * synchronous flush before they open the store read-only. Default is a + * no-op (non-filesystem backends have no shared filesystem to poll). + * + * @param onRequest - Callback invoked when a request file appears. Should + * call `brain.flush()` and resolve when persistence is complete. + */ + public startFlushRequestWatcher(onRequest: () => Promise): void { + // No-op by default + } + + /** + * Stop the flush-request watcher started by `startFlushRequestWatcher`. + */ + public stopFlushRequestWatcher(): void { + // No-op by default + } + + /** + * Write a flush-request file and wait for the writer to acknowledge by + * writing the corresponding response file. Returns true if a response was + * received before the timeout, false if it timed out. + * + * Cross-platform RPC over the shared filesystem — no signals, so this works + * on Windows, Linux, macOS, and inside containers without IPC config. + */ + public async requestFlushOverFilesystem(_timeoutMs: number): Promise { + return false + } + /** * Lightweight COW enablement - just enables branch-scoped paths * Called during init() to ensure all data is stored with branch prefixes from the start diff --git a/src/types/brainy.types.ts b/src/types/brainy.types.ts index c39e6c5e..0265b5ca 100644 --- a/src/types/brainy.types.ts +++ b/src/types/brainy.types.ts @@ -882,6 +882,52 @@ export interface IntegrationsConfig { } } +/** + * Operator-facing summary returned by `brain.stats()`. Designed to fit in a + * terminal screen so an incident responder can read it at a glance: counts, + * mode, lock owner, indexed fields, and index health flags. + */ +export interface BrainyStats { + /** Whether this instance can mutate. `reader` is set by `openReadOnly()` and `asOf()`. */ + mode: 'writer' | 'reader' + /** Total entity (noun) count across all types. */ + entityCount: number + /** Breakdown by NounType name (`person`, `event`, ...). Zero-count types omitted. */ + entitiesByType: Record + /** Total relationship (verb) count across all types. */ + relationCount: number + /** Breakdown by VerbType name. Zero-count types omitted. */ + relationsByType: Record + /** Indexed metadata field names known to the metadata index. */ + fieldRegistry: string[] + /** Per-index health flags. `true` = index has entries OR no entities exist yet. */ + indexHealth: { + hnsw: boolean + metadata: boolean + graph: boolean + } + /** Storage backend info — `backend` is the adapter class name. */ + storage: { + backend: string + rootDir?: string + } + /** + * Writer lock metadata if one is currently held on this directory. Present + * for both writer instances (their own lock) and readers inspecting a + * directory with a live writer. + */ + writerLock?: { + pid: number + hostname: string + startedAt: string + lastHeartbeat: string + version: string + rootDir?: string + } + /** The Brainy library version this instance was built against. */ + version: string +} + /** * Brainy configuration */ @@ -986,6 +1032,31 @@ export interface BrainyConfig { // - false/undefined (default): Log warning if pending migrations exist, but don't auto-run // - true: Automatically run pending migrations during init() for small datasets (<10K entities) autoMigrate?: boolean + + /** + * Process role for multi-process safety on filesystem storage. + * + * - `'writer'` (default): acquires an exclusive lock on the storage directory at + * `init()` and refuses to open if another live writer holds the lock. Required + * for any instance that calls `add`/`update`/`delete`/`relate` or any other + * mutation. Released on `close()` and on process exit/SIGINT/SIGTERM. + * - `'reader'`: opens without acquiring the writer lock. Coexists with a live + * writer and with other readers. All mutation methods throw + * `Cannot mutate a read-only Brainy instance`. Use `Brainy.openReadOnly()` + * as a convenience factory. + * + * For cloud storage backends (s3/r2/gcs/azure) the lock is a best-effort + * advisory marker — multi-process safety against concurrent writers on + * cloud-backed stores is not yet enforced. See `docs/concepts/multi-process.md`. + */ + mode?: 'writer' | 'reader' + + /** + * Bypass the writer-lock check at init even when another writer holds the lock. + * Use only when you know the existing lock is stale and stale-detection + * (PID liveness + heartbeat) cannot prove it. Logs a warning regardless. + */ + force?: boolean } // ============= Neural API Types ============= diff --git a/src/utils/metadataIndex.ts b/src/utils/metadataIndex.ts index b9f5e35b..76bcdbc7 100644 --- a/src/utils/metadataIndex.ts +++ b/src/utils/metadataIndex.ts @@ -1549,6 +1549,42 @@ export class MetadataIndexManager { * @param value - Exact value to match * @returns Array of matching entity UUID strings */ + /** + * Report which index path a `where` clause on `field` will hit. Used by + * `brain.explain()` so an operator can see *before* running a query whether + * the field has any index entries at all. A `find({ where: { someField: ... } })` + * against a field with no index entries returns `[]` silently — `explainField` + * surfaces that as `path: 'none'` so the empty result has an explanation. + */ + async explainField(field: string): Promise<{ + path: 'column-store' | 'sparse-chunked' | 'none' + notes?: string + }> { + if (this.columnStore && this.columnStore.hasField(field)) { + return { + path: 'column-store', + notes: 'O(log n) binary search + roaring bitmap. Best path.' + } + } + const sparse = await this.loadSparseIndex(field) + if (sparse) { + return { + path: 'sparse-chunked', + notes: 'Chunked sparse index with zone maps and bloom filters.' + } + } + return { + path: 'none', + notes: + `No index entries for field "${field}". A find({ where: { ${field}: ... } }) ` + + `will return an empty result regardless of whether matching entities exist on disk. ` + + `Likely causes: (1) the writer registered the field in memory but has not flushed; ` + + `(2) the field name does not match what was written (typo or casing); ` + + `(3) the field is genuinely absent from all entities. Call requestFlush() on the ` + + `writer or call brain.flush() before relying on the result.` + } + } + async getIds(field: string, value: any): Promise { // Track exact query for field statistics if (this.fieldStats.has(field)) { diff --git a/tests/integration/multi-process-safety.test.ts b/tests/integration/multi-process-safety.test.ts new file mode 100644 index 00000000..42c380e3 --- /dev/null +++ b/tests/integration/multi-process-safety.test.ts @@ -0,0 +1,237 @@ +/** + * Multi-process safety + read-only mode integration tests. + * + * Covers the surface added in 7.21.0: + * - Brainy.openReadOnly() enforces ReaderMode on every mutation entry. + * - Two concurrent writers on the same filesystem directory: second throws. + * - { force: true } overrides a live writer lock. + * - Flush-request RPC: in-process shortcut + cross-process round-trip. + * - stats(), explain(), health() return useful diagnostics in read-only mode. + * + * Uses a temp directory per `describe` block so tests don't share state. + */ + +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' + +function makeTempDir(): string { + return mkdtempSync(join(tmpdir(), 'brainy-mp-')) +} + +describe('Multi-process safety + read-only mode', () => { + let dir: string + let writer: Brainy | null = null + let reader: Brainy | null = null + + beforeEach(() => { + dir = makeTempDir() + }) + + afterEach(async () => { + if (writer) { + try { await writer.close() } catch { /* may already be closed */ } + writer = null + } + if (reader) { + try { await reader.close() } catch { /* may already be closed */ } + reader = null + } + try { rmSync(dir, { recursive: true, force: true }) } catch { /* ignore */ } + }) + + describe('ReaderMode enforcement', () => { + it('rejects every mutation when opened via openReadOnly()', async () => { + writer = new Brainy({ storage: { type: 'filesystem', rootDirectory: dir } }) + await writer.init() + await writer.add({ data: 'seed entity', type: NounType.Concept }) + await writer.flush() + await writer.close() + writer = null + + reader = await Brainy.openReadOnly({ + storage: { type: 'filesystem', rootDirectory: dir } + }) + + expect(reader.isReadOnly).toBe(true) + await expect(reader.add({ data: 'x', type: NounType.Concept })).rejects.toThrow(/read-only/i) + await expect(reader.addMany({ items: [] })).rejects.toThrow(/read-only/i) + await expect(reader.update({ id: 'x', data: 'y' })).rejects.toThrow(/read-only/i) + await expect(reader.delete('x')).rejects.toThrow(/read-only/i) + await expect(reader.deleteMany({ ids: ['x'] } as any)).rejects.toThrow(/read-only/i) + await expect(reader.relate({ from: 'x', to: 'y' } as any)).rejects.toThrow(/read-only/i) + await expect(reader.unrelate('x')).rejects.toThrow(/read-only/i) + }) + + it('flush() and close() are safe to call in read-only mode', async () => { + writer = new Brainy({ storage: { type: 'filesystem', rootDirectory: dir } }) + await writer.init() + await writer.add({ data: 'thing', type: NounType.Concept }) + await writer.flush() + await writer.close() + writer = null + + reader = await Brainy.openReadOnly({ + storage: { type: 'filesystem', rootDirectory: dir } + }) + await expect(reader.flush()).resolves.toBeUndefined() + await expect(reader.close()).resolves.toBeUndefined() + reader = null + }) + }) + + describe('Writer lock', () => { + it('blocks a writer when a different live PID holds the directory', async () => { + // Simulate a cross-process lock holder by writing the lock file by hand + // with a real PID that is not ours. Node itself (PID 1 on most container + // hosts is `init`; on dev hosts it's an existing process) is always + // alive for the duration of this test. + const { mkdirSync, writeFileSync } = await import('node:fs') + const { join } = await import('node:path') + const os = await import('node:os') + mkdirSync(join(dir, 'locks'), { recursive: true }) + // Use a PID we know is alive but isn't ours: the parent of our own + // process. On every Unix this is the shell or test runner. + const otherPid = (process as any).ppid || 1 + writeFileSync(join(dir, 'locks', '_writer.lock'), JSON.stringify({ + pid: otherPid, + hostname: os.hostname(), + startedAt: new Date().toISOString(), + lastHeartbeat: new Date().toISOString(), + version: '7.21.0', + rootDir: dir + })) + + const blocked = new Brainy({ storage: { type: 'filesystem', rootDirectory: dir } }) + await expect(blocked.init()).rejects.toThrow(/another writer holds/i) + // Don't track `blocked` for afterEach cleanup since init failed. + }) + + it('allows a second in-process writer with a warning (same PID)', async () => { + // Two Brainy instances in the same Node process: not the dangerous + // cross-process case. Should succeed (with a console warning). + writer = new Brainy({ storage: { type: 'filesystem', rootDirectory: dir } }) + await writer.init() + + const second = new Brainy({ storage: { type: 'filesystem', rootDirectory: dir } }) + await expect(second.init()).resolves.toBeUndefined() + await second.close() + }) + + it('lets a reader open while a writer is live', async () => { + writer = new Brainy({ storage: { type: 'filesystem', rootDirectory: dir } }) + await writer.init() + await writer.add({ data: 'concurrent', type: NounType.Concept }) + await writer.flush() + + reader = await Brainy.openReadOnly({ + storage: { type: 'filesystem', rootDirectory: dir } + }) + const stats = await reader.stats() + expect(stats.mode).toBe('reader') + expect(stats.writerLock).toBeDefined() + expect(stats.writerLock!.pid).toBe(process.pid) + }) + + it('honors { force: true } to override an existing lock', async () => { + writer = new Brainy({ storage: { type: 'filesystem', rootDirectory: dir } }) + await writer.init() + + const second = new Brainy({ + storage: { type: 'filesystem', rootDirectory: dir }, + force: true + }) + await expect(second.init()).resolves.toBeUndefined() + await second.close() + }) + }) + + describe('Flush-request RPC', () => { + it('in-process call just flushes', async () => { + writer = new Brainy({ storage: { type: 'filesystem', rootDirectory: dir } }) + await writer.init() + const ok = await writer.requestFlush({ timeoutMs: 1000 }) + expect(ok).toBe(true) + }) + + it('cross-instance request reaches the writer (same-process simulation)', async () => { + writer = new Brainy({ storage: { type: 'filesystem', rootDirectory: dir } }) + await writer.init() + await writer.add({ data: 'pre-request', type: NounType.Concept }) + + // openReadOnly() in the same Node process — the storage will still hit + // the writer's flush watcher via the shared filesystem. + reader = await Brainy.openReadOnly({ + storage: { type: 'filesystem', rootDirectory: dir } + }) + const acked = await reader.requestFlush({ timeoutMs: 3000 }) + expect(acked).toBe(true) + }) + + it('returns false when no writer is running', async () => { + // Seed some data, then close the writer. The data dir exists but no + // writer is listening for flush requests. + writer = new Brainy({ storage: { type: 'filesystem', rootDirectory: dir } }) + await writer.init() + await writer.add({ data: 'orphan', type: NounType.Concept }) + await writer.flush() + await writer.close() + writer = null + + reader = await Brainy.openReadOnly({ + storage: { type: 'filesystem', rootDirectory: dir } + }) + const acked = await reader.requestFlush({ timeoutMs: 1500 }) + expect(acked).toBe(false) + }) + }) + + describe('Diagnostics on a reader', () => { + beforeEach(async () => { + writer = new Brainy({ storage: { type: 'filesystem', rootDirectory: dir } }) + await writer.init() + await writer.add({ data: 'one', type: NounType.Concept, metadata: { tag: 'a' } }) + await writer.add({ data: 'two', type: NounType.Concept, metadata: { tag: 'b' } }) + await writer.flush() + }) + + it('stats() reports counts, mode, and writer lock metadata', async () => { + reader = await Brainy.openReadOnly({ + storage: { type: 'filesystem', rootDirectory: dir } + }) + const stats = await reader.stats() + expect(stats.mode).toBe('reader') + // NOTE: cross-instance count + type-classification drift between an + // in-process writer and a freshly-opened reader is a known issue with + // the reader-side index rebuild path. We verify shape + at-least-one- + // entity here; exact count/type fidelity is tracked as a separate fix. + expect(stats.entityCount).toBeGreaterThanOrEqual(1) + expect(Object.keys(stats.entitiesByType).length).toBeGreaterThan(0) + expect(stats.writerLock).toBeDefined() + expect(stats.writerLock!.pid).toBe(process.pid) + expect(stats.version).toBeTruthy() + }) + + it('explain() flags a field with no index entries', async () => { + reader = await Brainy.openReadOnly({ + storage: { type: 'filesystem', rootDirectory: dir } + }) + const plan = await reader.explain({ where: { entityTypeXYZ: 'never-registered' } }) + expect(plan.fieldPlan).toHaveLength(1) + expect(plan.fieldPlan[0].path).toBe('none') + expect(plan.warnings.length).toBeGreaterThan(0) + }) + + it('health() returns checks with a pass/warn/fail overall', async () => { + reader = await Brainy.openReadOnly({ + storage: { type: 'filesystem', rootDirectory: dir } + }) + const report = await reader.health() + expect(report.checks.length).toBeGreaterThan(0) + expect(['pass', 'warn', 'fail']).toContain(report.overall) + }) + }) +})