soulcraftlabs/open-brainy
2
0
Fork 0
Code Issues 1 Pull requests Projects Releases 340 Packages Wiki Activity Actions

perf(idle): the flush-request watch is event-driven; the heartbeat is observability

Browse source 
Three idle-burn items from the steady-state audit, and one correction.

THE FLUSH-REQUEST WATCH (the strongest of them). It readdir'd 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. In a process holding many stores that is tens of directory reads per second
on a completely idle service, plus a stale-request GC on every one of them. It now
uses fs.watch, so the arrival itself wakes it and a request is seen SOONER
than the poll saw it. Two concessions ride along, both stated in the code: a
30s safety sweep (fs.watch drops events on some network and fuse filesystems,
and the GC needs a tick of its own — two orders of magnitude fewer reads than
the poll made), and a fall back to the original 500 ms poll, narrated, on a
filesystem that cannot watch at all, because an inspector whose request is
never seen waits forever.

THE WRITER HEARTBEAT goes 10s → 60s. It is observability ONLY — staleness is
decided by pid liveness and the fence compares pid + hostname, so no decision
anywhere reads the timestamp — and at 10s it was a lock-file write every ten
seconds per brain forever, for a value nothing computes with. An operator
still sees a heartbeat inside the minute.

THE HEALTH NARRATION dedupes by CONTENT, not by the provider's generation
counter. That counter bumps on every ledger mutation and rebuild boundary, so
a provider bumping it on routine work re-emitted the same unchanged line on
every read, while one that never bumped could suppress a line whose reasons
had genuinely changed. The generation is still reported; it no longer decides
whether the line is worth saying.

CORRECTION, and it is against my own earlier claim: the idle-flush commit read
a reported idle-CPU observation (many stores, no writes, a flush every ~35s,
over a core burned) as caused by
the flush path. That does not follow — 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 the caller
is still unidentified. The clean-flush gate makes such a call free; it does not
account for it. The code comments and the idle lane now say exactly that.

Pins: tests/integration/flush-watcher-event-driven.test.ts — an idle writer
makes at most one request-directory read in 8 seconds (the old poll made ~16),
and a dropped request is still acked well inside the safety sweep.
This commit is contained in:
David Snelling 2026-08-28 11:25:47 -07:00
parent 417ddb5143
commit fb1da1c56d
4 changed files with 245 additions and 36 deletions
Download patch file Download diff file Expand all files Collapse all files

54
src/brainy.ts
View file

@ -749,12 +749,18 @@ export class Brainy<T = any> implements BrainyInterface<T> {
* 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<typeof setTimeout> | null = null
@ -851,7 +857,18 @@ export class Brainy<T = any> implements BrainyInterface<T> {
// 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<unknown, number>()
/**
* 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<unknown, string>()
constructor(config?: BrainyConfig) {
// The reserved-field write policy died with the field-addressing law:
@ -12366,11 +12383,11 @@ export class Brainy<T = any> implements BrainyInterface<T> {
// 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<T = any> implements BrainyInterface<T> {
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)
}
}

116
src/storage/adapters/fileSystemStorage.ts
View file

@ -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<void>
@ -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>): 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

94
tests/integration/flush-watcher-event-driven.test.ts Normal file
View file

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

17
tests/integration/idle-costs-nothing.test.ts
View file

@ -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