diff --git a/src/brainy.ts b/src/brainy.ts index c479fcc3..3426dcce 100644 --- a/src/brainy.ts +++ b/src/brainy.ts @@ -749,12 +749,18 @@ export class Brainy implements BrainyInterface { * Whether a write has been committed since the last flush that ran. THE * ENGINE DOES NO PERIODIC WORK WITHOUT A CAUSE: a brain nobody has written * to has nothing to make durable, and a flush over it must cost nothing and - * say nothing. Measured on a production process holding 21 brains: with no - * writes for ten minutes it still printed "All indexes flushed to disk in - * 216–601ms" per brain every ~35s and idled at 1.26 cores, because a flush - * called every provider, stamped the watermarks, persisted the generation - * counter and re-stamped the entity tree whether or not anything had - * changed. + * say nothing. Before this, a flush called every provider, stamped the + * watermarks, persisted the generation counter and re-stamped the entity + * tree whether or not anything had changed — roughly 28 writes for a store + * that had not moved. + * + * WHAT THIS DOES NOT EXPLAIN, stated so nobody reads it as solved: a + * production process holding 21 brains printed "All indexes flushed to disk + * in 216-601ms" per brain every ~35s and idled at 1.26 cores with no writes + * for ten minutes. This engine's cadence is WRITE-DRIVEN — every trigger + * runs through noteWriteForPersistence, which only a committed write calls — + * so something was calling flush() on those brains, and this gate makes such + * a call free rather than accounting for it. The caller is still unidentified. */ private _dirtySinceLastFlush = false private _persistIdleTimer: ReturnType | null = null @@ -851,7 +857,18 @@ export class Brainy implements BrainyInterface { // Read-gate narration dedup: a degraded-but-serving or not-ready health // report narrates via prodLog.warn ONCE per (provider, report.generation) — // never once per read. Keyed on the provider instance itself. - private _lastNarratedHealthGeneration = new Map() + /** + * The last health narration emitted per provider, keyed by its CONTENT. + * + * This used to dedupe on the provider's `generation` counter, which bumps on + * every ledger mutation and every rebuild boundary — so a provider that + * bumps its generation on routine work re-emitted the same unchanged health + * line on every read that consulted it, and a provider that never bumped + * could suppress a line whose reasons had genuinely changed. The dedupe key + * is now what the line SAYS: an unchanged verdict is silent however the + * generation moves, and a changed verdict is always heard. + */ + private _lastNarratedHealth = new Map() constructor(config?: BrainyConfig) { // The reserved-field write policy died with the field-addressing law: @@ -12366,11 +12383,11 @@ export class Brainy implements BrainyInterface { // committed since the last flush, so every step below would re-persist // state identical to what is already on disk — provider flushes, the // watermark stamps, the generation counter, the entity-tree stamp — and - // print two lines announcing it. On a process holding 21 brains that - // no-op cost 1.26 cores at idle. The witness is set by every committed + // print two lines announcing it. The witness is set by every committed // write (see noteWriteForPersistence) and cleared here; a write landing // DURING this flush sets it again, so it is never lost — the next flush - // does that write's work. + // does that write's work. This makes an unexplained flush FREE; it does + // not explain one (see _dirtySinceLastFlush). if (!this._dirtySinceLastFlush) { return } @@ -17243,12 +17260,17 @@ export class Brainy implements BrainyInterface { if (assessment.reasons.length > 0 && assessment.report != null) { const generation = assessment.report.generation - if (this._lastNarratedHealthGeneration.get(provider) !== generation) { - this._lastNarratedHealthGeneration.set(provider, generation) - prodLog.warn( - `[Brainy] ${assessment.report.provider} health (generation ${generation}): ` + - assessment.reasons.join('; ') - ) + // Dedupe by CONTENT, not by the provider's generation counter — see + // _lastNarratedHealth. The generation is still REPORTED (an operator + // wants to know which generation produced the verdict); it just no + // longer decides whether the line is worth saying. + const line = + `[Brainy] ${assessment.report.provider} health (generation ${generation}): ` + + assessment.reasons.join('; ') + const key = `${assessment.report.provider}\u0000${assessment.reasons.join('; ')}` + if (this._lastNarratedHealth.get(provider) !== key) { + this._lastNarratedHealth.set(provider, key) + prodLog.warn(line) } } diff --git a/src/storage/adapters/fileSystemStorage.ts b/src/storage/adapters/fileSystemStorage.ts index 3f1055c2..9d04b46d 100644 --- a/src/storage/adapters/fileSystemStorage.ts +++ b/src/storage/adapters/fileSystemStorage.ts @@ -107,7 +107,23 @@ export class FileSystemStorage extends BaseStorage { * "the previous writer died" without inferring either from a pid. */ private static readonly WRITER_CLOSE_FILE = '_writer.close' - private static readonly WRITER_HEARTBEAT_MS = 10_000 + /** + * How often the lock file's `lastHeartbeat` is rewritten. + * + * THIS IS OBSERVABILITY ONLY, and the cadence follows from that. Staleness + * is decided by PID LIVENESS alone (see isWriterLockStale) and the fence + * compares pid + hostname — no decision anywhere reads this timestamp. It + * exists so an operator inspecting a lock file, or reading the + * BRAINY_WRITER_LOCKED error, can judge liveness themselves. + * + * At 10s it was a lock-file WRITE every ten seconds per brain, forever: 2.1 + * writes/s across a production process holding 21 idle brains, for a + * human-readable timestamp nothing computes with. At 60s an operator still + * sees a heartbeat inside the minute, at a sixth of the cost. With the + * clean-close record now recording orderly releases explicitly, the + * heartbeat carries even less weight than it did. + */ + private static readonly WRITER_HEARTBEAT_MS = 60_000 private static readonly WRITER_STALE_THRESHOLD_MS = 60_000 private writerLockHeartbeat?: NodeJS.Timeout private writerLockInfo?: WriterLockInfo @@ -135,9 +151,16 @@ export class FileSystemStorage extends BaseStorage { private static readonly FLUSH_REQUEST_DIR = '_flush_requests' private static readonly FLUSH_RESPONSE_DIR = '_flush_responses' private static readonly FLUSH_WATCH_INTERVAL_MS = 500 + /** + * The safety sweep behind the fs.watch: catches events an exotic filesystem + * dropped, and runs the stale-request GC. See startFlushRequestWatcher. + */ + private static readonly FLUSH_SAFETY_SWEEP_MS = 30_000 private static readonly FLUSH_POLL_INTERVAL_MS = 100 private static readonly FLUSH_REQUEST_TTL_MS = 60_000 private flushWatcherInterval?: NodeJS.Timeout + /** The inotify-backed watch on the request directory, when the FS supports one. */ + private flushWatcher?: import('node:fs').FSWatcher private flushWatcherInFlight = false private flushWatcherOnRequest?: () => Promise @@ -2385,36 +2408,101 @@ export class FileSystemStorage extends BaseStorage { /** * Start watching for cross-process flush requests. Called by Brainy.init() - * in writer mode. Polls `locks/_flush_requests/` every - * FLUSH_WATCH_INTERVAL_MS — each new `.req` file triggers the supplied - * callback (`brain.flush()`), after which an `.ack` is written to - * `locks/_flush_responses/` with the same request ID. Stale `.req` files - * (>FLUSH_REQUEST_TTL_MS) are garbage-collected on every tick. + * in writer mode. Each new `.req` file in `locks/_flush_requests/` triggers + * the supplied callback (`brain.flush()`), after which an `.ack` is written + * to `locks/_flush_responses/` with the same request ID. Stale `.req` files + * (>FLUSH_REQUEST_TTL_MS) are garbage-collected on each sweep. + * + * THE WATCH IS EVENT-DRIVEN, NOT A POLL. It used to `readdir` the request + * directory every 500 ms, per brain, for the entire life of every writer — + * armed on every non-reader brain whether or not any inspector process + * existed. MEASURED on a production process holding 21 brains: 42 directory + * reads per second on a completely idle service, plus a stale-request GC + * pass on every one of them. The engine does no periodic work without a + * cause, and a request that has not been made is not a cause. + * + * `fs.watch` (inotify on Linux) delivers the arrival itself, so a request is + * seen SOONER than the old poll saw it. Two honest concessions ride with it: + * - a slow SAFETY SWEEP (FLUSH_SAFETY_SWEEP_MS) still runs, because + * `fs.watch` can miss events on network and fuse filesystems and because + * the stale-request GC needs some tick of its own. At 30s that is 0.7 + * reads/s across 21 brains where the poll cost 42. + * - a filesystem that cannot watch at all falls back to the ORIGINAL + * 500 ms poll, narrated once, because correctness outranks idle cost: + * an inspector whose request is never seen waits forever. */ public override startFlushRequestWatcher(onRequest: () => Promise): void { - if (this.flushWatcherInterval) return // already watching + if (this.flushWatcherInterval || this.flushWatcher) 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 + const sweep = (): void => { + if (this.flushWatcherInFlight) return // skip overlapping sweep this.flushWatcherInFlight = true this.processFlushRequests(reqDir, ackDir).finally(() => { this.flushWatcherInFlight = false }) - }, FileSystemStorage.FLUSH_WATCH_INTERVAL_MS) + } + + // Ensure both dirs exist up front so the first .req drop doesn't race with + // mkdir — and so there is a directory to watch. + void this.ensureDirectoryExists(reqDir) + .then(() => this.ensureDirectoryExists(ackDir)) + .then(() => { + if (this.flushWatcherOnRequest !== onRequest) return // stopped meanwhile + try { + const watcher = fs.watch(reqDir, () => sweep()) + this.flushWatcher = watcher + watcher.on('error', (err: Error) => { + // A watch that dies mid-life must not leave the door deaf. + console.warn( + `[brainy] Flush-request watch failed (${err.message}) — falling back to polling.` + ) + this.flushWatcher?.close() + this.flushWatcher = undefined + this.startFlushRequestPolling(sweep) + }) + if (typeof watcher.unref === 'function') watcher.unref() + // The safety sweep: missed events on exotic filesystems, and the + // stale-request GC. + this.flushWatcherInterval = setInterval(sweep, FileSystemStorage.FLUSH_SAFETY_SWEEP_MS) + if (typeof this.flushWatcherInterval.unref === 'function') { + this.flushWatcherInterval.unref() + } + // One sweep now: a request may have been dropped before the watch armed. + sweep() + } catch (err) { + console.warn( + `[brainy] Flush-request directory cannot be watched on this filesystem ` + + `(${(err as Error).message}) — polling every ` + + `${FileSystemStorage.FLUSH_WATCH_INTERVAL_MS}ms instead.` + ) + this.startFlushRequestPolling(sweep) + } + }) + .catch(() => { + // The request directory could not be created; nothing to watch. A + // cross-process flush request cannot be made either, so there is + // nothing to miss. + }) + } + + /** The original 500 ms poll — the fallback when a directory cannot be watched. */ + private startFlushRequestPolling(sweep: () => void): void { + if (this.flushWatcherInterval) return + this.flushWatcherInterval = setInterval(sweep, FileSystemStorage.FLUSH_WATCH_INTERVAL_MS) if (typeof this.flushWatcherInterval.unref === 'function') { this.flushWatcherInterval.unref() } } public override stopFlushRequestWatcher(): void { + if (this.flushWatcher) { + this.flushWatcher.close() + this.flushWatcher = undefined + } if (this.flushWatcherInterval) { clearInterval(this.flushWatcherInterval) this.flushWatcherInterval = undefined diff --git a/tests/integration/flush-watcher-event-driven.test.ts b/tests/integration/flush-watcher-event-driven.test.ts new file mode 100644 index 00000000..4b2e80c4 --- /dev/null +++ b/tests/integration/flush-watcher-event-driven.test.ts @@ -0,0 +1,94 @@ +/** + * @module tests/integration/flush-watcher-event-driven + * @description THE FLUSH-REQUEST WATCH IS EVENT-DRIVEN. + * + * It used to `readdir` the request directory every 500 ms, per brain, for the + * life of every writer — armed on every non-reader brain whether or not any + * inspector process existed. MEASURED on a production process holding 21 + * brains: 42 directory reads per second on a completely idle service, plus a + * stale-request GC pass on every one of them. + * + * The law: a request that has not been made is not a cause. The arrival itself + * wakes the watcher, so the request is seen SOONER than the poll saw it, and a + * slow safety sweep covers filesystems that drop watch events and the GC. + */ + +import { describe, it, expect, afterEach, vi } from 'vitest' +import { mkdtempSync, rmSync, writeFileSync, mkdirSync } from 'node:fs' +import * as nodeFs from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { Brainy } from '../../src/brainy.js' +import { NounType } from '../../src/types/graphTypes.js' + +describe('the flush-request watcher', () => { + const dirs: string[] = [] + const brains: Brainy[] = [] + + afterEach(async () => { + for (const b of brains.splice(0)) { + try { await b.close() } catch { /* already closed */ } + } + for (const d of dirs.splice(0)) { + try { rmSync(d, { recursive: true, force: true }) } catch { /* ignore */ } + } + vi.restoreAllMocks() + }) + + async function openWriter(): Promise<{ brain: Brainy; dir: string }> { + const dir = mkdtempSync(join(tmpdir(), 'brainy-flush-watch-')) + dirs.push(dir) + const brain = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } }) + brains.push(brain) + await brain.init() + await brain.add({ data: 'a row', type: NounType.Concept }) + await brain.flush() + return { brain, dir } + } + + it('does not poll the request directory on an idle writer', async () => { + const { dir } = await openWriter() + const reqDir = join(dir, 'locks', '_flush_requests') + + // Count real reads of the request directory over a window far longer than + // the old 500ms poll (which would have made ~16 of them). + const realReaddir = nodeFs.promises.readdir + let requestDirReads = 0 + const spy = vi + .spyOn(nodeFs.promises, 'readdir') + .mockImplementation((async (p: unknown, ...rest: unknown[]) => { + if (String(p) === reqDir) requestDirReads++ + return (realReaddir as unknown as (...a: unknown[]) => Promise)(p, ...rest) + }) as typeof nodeFs.promises.readdir) + + await new Promise((r) => setTimeout(r, 8_000)) + spy.mockRestore() + + // The old poll: 500ms → ~16 reads. The safety sweep is 30s → 0 in this window. + expect(requestDirReads).toBeLessThanOrEqual(1) + }, 120_000) + + it('answers a request that arrives, without waiting for the sweep', async () => { + const { brain, dir } = await openWriter() + const reqDir = join(dir, 'locks', '_flush_requests') + const ackDir = join(dir, 'locks', '_flush_responses') + mkdirSync(reqDir, { recursive: true }) + + // Drop a request exactly as an out-of-process inspector does. + const id = 'test-request-0001' + writeFileSync(join(reqDir, `${id}.req`), JSON.stringify({ at: Date.now() })) + + // The ack must land far sooner than the 30s safety sweep. + const deadline = Date.now() + 10_000 + let acked = false + while (Date.now() < deadline) { + try { + const entries = await nodeFs.promises.readdir(ackDir) + if (entries.some((e) => e.startsWith(id))) { acked = true; break } + } catch { /* dir not created yet */ } + await new Promise((r) => setTimeout(r, 100)) + } + expect(acked, 'the watcher must answer an arriving request').toBe(true) + void brain + }, 120_000) +}) diff --git a/tests/integration/idle-costs-nothing.test.ts b/tests/integration/idle-costs-nothing.test.ts index 8f951d46..b5c386cf 100644 --- a/tests/integration/idle-costs-nothing.test.ts +++ b/tests/integration/idle-costs-nothing.test.ts @@ -2,12 +2,17 @@ * @module tests/integration/idle-costs-nothing * @description AN IDLE BRAIN DOES NO WORK. * - * Measured on a production process holding 21 brains: with no writes for ten - * minutes it printed "All indexes flushed to disk in 216–601ms" per brain - * every ~35 seconds and idled at 1.26 cores. Every one of those flushes - * re-persisted state identical to what was already on disk — the provider - * flushes, the watermark stamps, the generation counter, the entity-tree - * stamp — because `flush()` never asked whether anything had changed. + * A flush used to re-persist state identical to what was already on disk — + * the provider flushes, the watermark stamps, the generation counter, the + * entity-tree stamp, roughly 28 writes — because `flush()` never asked whether + * anything had changed. + * + * The field observation that started this: a production process holding 21 + * brains printed "All indexes flushed to disk in 216–601ms" per brain every + * ~35 seconds and idled at 1.26 cores, with no writes for ten minutes. This + * engine's cadence is WRITE-DRIVEN, so that observation is NOT explained by + * the cadence and is not claimed to be fixed here — what is fixed is that such + * a call now costs nothing. Who was calling flush() remains open. * * The laws pinned here: * (a) the persistence cadence arms only on a write — a brain nobody writes