open-brainy/tests/integration/flush-watcher-event-driven.test.ts

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