/** * @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) })