open-brainy/tests/integration/beforeexit-never-closes.test.ts

/**
 * @module tests/integration/beforeexit-never-closes
 * @description A DRAINED EVENT LOOP IS NOT A SHUTDOWN.
 *
 * MEASURED on the 11.1 rehearsal lane, against a copy of a real store. The
 * `beforeExit` listener had been wired to the SIGNAL path — the path whose job
 * is to `close()` every live brain — so after the heal phase the log printed
 *
 *   "Shutdown signal received - flushing pending data..."
 *   "Flushed successfully (1 instance)"
 *
 * with no signal ever sent, and the script's very next `add()` threw
 *
 *   "Brainy instance is not initialized: it was closed via close().
 *    Create a new instance."
 *
 * Node emits `'beforeExit'` whenever the event loop has no REF'd work left.
 * That is not "the process is ending" — it is a state a perfectly healthy
 * script reaches, because this engine unref's its idle and cadence timers
 * ("an idle brain costs nothing"), so a script awaiting anything those timers
 * drive is, for that instant, a process with no ref'd work and an open brain.
 * The engine closed a live brain out from under a running script.
 *
 * The contract pinned here:
 *   (1) `'beforeExit'` firing while a brain is open closes NOTHING: the brain
 *       is still open, `add()` and `find()` still work, the writer lock is
 *       still held, and the process still exits 0 on its own afterwards.
 *   (2) The pass DOES persist derived state — a non-closing `flush()` ran —
 *       and it wrote no clean-shutdown marker and no clean-close record: those
 *       are `close()`'s word about itself, and no close happened.
 *   (3) The signal path is untouched: SIGTERM still closes through `close()`
 *       (pinned by tests/integration/shutdown-single-owner.test.ts, re-run
 *       with this change).
 */

import { describe, it, expect, beforeEach, afterEach } from 'vitest'
import { mkdtempSync, rmSync, existsSync, readFileSync, writeFileSync } from 'node:fs'
import { spawn } from 'node:child_process'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { Brainy } from '../../src/brainy.js'
import { NounType } from '../../src/types/graphTypes.js'

const REPO_ROOT = process.cwd()
const TSX = join(REPO_ROOT, 'node_modules', '.bin', 'tsx')
const BRAINY_SRC = join(REPO_ROOT, 'src', 'brainy.ts')

function makeTempDir(prefix: string): string {
  return mkdtempSync(join(tmpdir(), prefix))
}

/** The writer lock itself — present for as long as this process owns the store. */
const writerLockPath = (dir: string) => join(dir, 'locks', '_writer.lock')
/** The clean-close record — written by `releaseWriterLock()`, i.e. by close(). */
const closeRecordPath = (dir: string) => join(dir, 'locks', '_writer.close')
/**
 * The generation store's clean-shutdown marker — written by
 * `generationStore.close()` alone, reached only from `close()`. (Raw objects
 * are gzipped on disk, so both spellings are accepted.)
 */
const cleanShutdownWritten = (dir: string) =>
  existsSync(join(dir, '_system', 'clean-shutdown.json.gz')) ||
  existsSync(join(dir, '_system', 'clean-shutdown.json'))

/**
 * Write a child script and run it under tsx to completion, collecting stdout
 * and stderr and the exit code. (A file, not `tsx -e`: the eval form compiles
 * to CommonJS, which has no top-level await.)
 */
function runChild(
  scriptDir: string,
  body: string
): Promise<{ code: number | null; out: string }> {
  const scriptPath = join(scriptDir, 'child-process.mts')
  writeFileSync(scriptPath, body)
  // The child is an ORDINARY consumer process, so it runs the real embedding
  // pipeline: this suite's deterministic-embedder switch is inherited through
  // the environment, and under it `find()` self-retrieval returns nothing —
  // which would make the read half of this pin vacuous. (That property is the
  // deterministic embedder's, not this change's: it reproduces in a plain
  // script with no 'beforeExit' involved.)
  const env = { ...process.env }
  delete env.BRAINY_DETERMINISTIC_EMBEDDINGS
  const child = spawn(TSX, [scriptPath], {
    cwd: REPO_ROOT,
    stdio: ['ignore', 'pipe', 'pipe'],
    env
  })
  let out = ''
  child.stdout?.on('data', (d) => { out += String(d) })
  child.stderr?.on('data', (d) => { out += String(d) })
  return new Promise((resolvePromise) => {
    child.on('exit', (code) => resolvePromise({ code, out }))
  })
}

describe('beforeExit never closes a live brain', () => {
  let dir: string
  let scriptDir: string
  let resultPath: string

  beforeEach(() => {
    dir = makeTempDir('brainy-beforeexit-')
    scriptDir = makeTempDir('brainy-beforeexit-script-')
    resultPath = join(scriptDir, 'result.json')
  })

  afterEach(() => {
    for (const d of [dir, scriptDir]) {
      try { rmSync(d, { recursive: true, force: true }) } catch { /* ignore */ }
    }
  })

  it('(1)+(2) a drained event loop flushes, closes nothing, and the script keeps working', async () => {
    /**
     * THE DRAIN, and why the script survives it. The script awaits a promise
     * that only an UNREF'd timer will resolve — the shape every engine cadence
     * timer has, and the reason a healthy script reaches a loop with no ref'd
     * work. Node emits `'beforeExit'` there, with the brain wide open.
     *
     * The engine's listener runs first (registered by `init()`, before the
     * script's). The script's own listener is both its witness — it records
     * that the emit happened, and the flush count AT that moment — and its
     * belt: it resolves the same promise, so the pin never depends on how many
     * milliseconds the engine's pass happens to keep the loop turning.
     *
     * The brain is DIRTY at the drain (one add, after a settling flush), so
     * the pass has real work to do and pin (2) is about a flush that ran, not
     * a flush that was skipped as a no-op.
     */
    const script = `
      import { writeFileSync as __writeFileSync, existsSync as __existsSync } from 'node:fs'
      import { join as __join } from 'node:path'
      import { Brainy } from ${JSON.stringify(BRAINY_SRC)}

      const DIR = ${JSON.stringify(dir)}
      const brain = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: DIR } })
      await brain.init()

      // Count every flush that RUNS on this brain. An own property shadows the
      // prototype for every caller, including the engine's own listeners.
      let flushes = 0
      const flushImpl = brain.flush.bind(brain)
      brain.flush = () => { flushes++; return flushImpl() }
      // ...and every close ENTERED. This must still be 0 after the drain.
      let closes = 0
      const closeImpl = brain.close.bind(brain)
      brain.close = () => { closes++; return closeImpl() }

      await brain.add({ data: 'written before the drain', type: 'concept' })
      await brain.flush()                                   // settle: clean brain
      await new Promise((r) => setTimeout(r, 250))          // let the cadence quiet down
      await brain.add({ data: 'the write the drain must persist', type: 'concept' })

      const flushesBeforeDrain = flushes
      let drains = 0
      let flushesAtDrain = -1
      const drained = new Promise((resolve) => {
        const t = setTimeout(resolve, 5)
        if (typeof t.unref === 'function') t.unref()
        process.on('beforeExit', () => {
          drains++
          if (flushesAtDrain === -1) flushesAtDrain = flushes
          resolve()
        })
      })
      await drained

      // GIVE THE ENGINE'S PASS ITS FULL TURN before judging it. The signal
      // path this listener used to share defers one macrotask before it
      // touches an instance, so a script that resumes on the same tick as the
      // emit would race past the damage and see an open brain that is about to
      // be closed underneath it. Wait it out (a ref'd timer — the drain has
      // already happened), then look.
      await new Promise((r) => setTimeout(r, 1000))

      // ---- The script is still running. The brain must still be its brain. ----
      const stateAtResume = {
        drains,
        flushesBeforeDrain,
        flushesAtDrain,
        closes,
        isClosed: brain.isClosed,
        isClosing: brain.isClosing,
        writerLockHeld: __existsSync(__join(DIR, 'locks', '_writer.lock')),
        cleanCloseRecord: __existsSync(__join(DIR, 'locks', '_writer.close')),
        cleanShutdownMarker:
          __existsSync(__join(DIR, '_system', 'clean-shutdown.json.gz')) ||
          __existsSync(__join(DIR, '_system', 'clean-shutdown.json'))
      }

      let addAfterDrain = null
      let addError = null
      try {
        addAfterDrain = await brain.add({ data: 'written AFTER the drained event loop', type: 'concept' })
      } catch (error) {
        addError = error instanceof Error ? error.message : String(error)
      }

      let findHits = -1
      let findError = null
      try {
        const results = await brain.find('written AFTER the drained event loop')
        findHits = results.length
      } catch (error) {
        findError = error instanceof Error ? error.message : String(error)
      }

      __writeFileSync(
        ${JSON.stringify(resultPath)},
        JSON.stringify({ ...stateAtResume, addAfterDrain, addError, findHits, findError, closesBeforeOurs: closes })
      )

      // The script ends the way a script ends: it closes its own brain, and
      // the process exits on its own because nothing is left holding the loop.
      await brain.close()
    `

    const { code, out } = await runChild(scriptDir, script)

    expect(existsSync(resultPath), `child wrote no result file:\n${out}`).toBe(true)
    const r = JSON.parse(readFileSync(resultPath, 'utf-8'))

    // The drain really happened — this test proves nothing otherwise.
    expect(r.drains, `'beforeExit' never fired:\n${out}`).toBeGreaterThanOrEqual(1)

    // (1) NOTHING WAS CLOSED. This is the regression: under 10.4.11 the pass
    // ran close() here and `addError` carried "it was closed via close()".
    expect(r.addError, `add() after the drain failed:\n${out}`).toBeNull()
    expect(r.findError, `find() after the drain failed:\n${out}`).toBeNull()
    expect(r.closes, 'the engine closed the brain on a drained event loop').toBe(0)
    expect(r.isClosed).toBe(false)
    expect(r.isClosing).toBe(false)
    expect(typeof r.addAfterDrain).toBe('string')
    expect(r.findHits, `find() returned nothing:\n${out}`).toBeGreaterThanOrEqual(1)

    // (1) The writer lock was never given up — a drained loop is not a handover.
    expect(r.writerLockHeld, 'the writer lock was released on a drained event loop').toBe(true)

    // (2) A flush RAN, and it wrote neither of close()'s markers.
    expect(
      r.flushesAtDrain,
      `the drained-loop pass ran no flush (before=${r.flushesBeforeDrain}):\n${out}`
    ).toBeGreaterThan(r.flushesBeforeDrain)
    expect(r.cleanShutdownMarker, 'the drained-loop flush stamped a clean-shutdown marker').toBe(false)
    expect(r.cleanCloseRecord, 'the drained-loop flush wrote a clean-close record').toBe(false)
    expect(out).toMatch(/All indexes flushed to disk/)

    // The narration says what happened, and never claims a shutdown.
    expect(out).toMatch(/event loop drained with 1 brain open/)
    expect(out).toMatch(/NOTHING was closed\. A drained loop is not a shutdown/)
    expect(out).not.toMatch(/Shutdown signal received/)
    expect(out).not.toMatch(/Flushed successfully/)
    expect(out).not.toMatch(/is not initialized/)

    // (1) And the process still exits 0 on its own once the script closes up.
    expect(code, `child output:\n${out}`).toBe(0)

    // The store the script left behind is clean: it closed properly at the end.
    expect(cleanShutdownWritten(dir), 'the script\'s own close() wrote no marker').toBe(true)
    expect(existsSync(closeRecordPath(dir)), 'the script\'s own close() left no clean-close record').toBe(true)
    expect(existsSync(writerLockPath(dir)), 'the writer lock outlived close()').toBe(false)
  }, 300_000)

  it('(2) the pass is repeatable and idempotent: a second drain closes nothing either', async () => {
    // In-process, so the assertions are on the object itself rather than on a
    // report: 'beforeExit' is an ordinary event, and emitting it twice must
    // leave the brain exactly as usable as it was.
    const brain = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
    await brain.init()

    const flushed: Promise<void>[] = []
    const flushImpl = brain.flush.bind(brain)
    ;(brain as unknown as { flush: () => Promise<void> }).flush = () => {
      const p = flushImpl()
      flushed.push(p)
      return p
    }

    await brain.add({ data: 'a write the drained loop must persist', type: NounType.Concept })

    for (const pass of [1, 2]) {
      const before = flushed.length
      process.emit('beforeExit', 0)
      await Promise.all(flushed.slice(before).map((p) => p.catch(() => {})))
      // Let the pass's own `finally` run (it settles a microtask after ours),
      // so the next emit is not turned away by the in-flight guard.
      await new Promise((r) => setTimeout(r, 50))

      expect(brain.isClosed, `pass ${pass} closed the brain`).toBe(false)
      expect(brain.isClosing, `pass ${pass} started a close`).toBe(false)
      expect(existsSync(writerLockPath(dir)), `pass ${pass} released the writer lock`).toBe(true)
      expect(existsSync(closeRecordPath(dir)), `pass ${pass} wrote a clean-close record`).toBe(false)
      expect(cleanShutdownWritten(dir), `pass ${pass} stamped a clean-shutdown marker`).toBe(false)

      // Still a working brain, after every pass.
      const id = await brain.add({ data: `still writable after drain ${pass}`, type: NounType.Concept })
      expect(id).toBeTruthy()
    }

    // The first pass had a dirty brain and flushed it; the second found it
    // clean and cost nothing. Either way, neither closed anything.
    expect(flushed.length).toBeGreaterThanOrEqual(2)

    await brain.close()
    expect(brain.isClosed).toBe(true)
    expect(cleanShutdownWritten(dir)).toBe(true)
  }, 300_000)
})