test(shutdown): pin one owner per brain — real processes, real signals

Four pins in real child processes under real SIGTERM, following the
writer-lock-clean-close spawn pattern:

(a) A host owner registered on SIGTERM closes two brains while the engine's
    hooks are live: exactly one close entered and one close body run per brain,
    the writer lock given up exactly ONCE per brain, the handler announcing
    that it stepped aside, no "Writer fence lost", no failed instance, both
    durability markers written, exit 0, and both reopens adopting rather than
    folding. The release count is the discriminating assertion — against the
    old handler it reads {a: 2, b: 2}, one release from the owner's close and
    one from the handler's own finally.
(b) No host owner: the engine's handler closes every instance by the same
    path — one close each, markers written, clean exit, clean reopen.
(c) Two concurrent close() callers share one promise (by identity) and one
    execution; a third call after they settle runs nothing.
(d) Eight kicks during a running flush — five through the cadence door, three
    direct — arm exactly ONE follow-up: two flush bodies total, and the
    concurrency high-water mark stays at 1.

The counts come out of the child through a file written synchronously on the
way out: the engine calls process.exit(0) when it is the sole shutdown owner,
and a console.log to a pipe can be dropped by that exit.

tests/integration/shutdown-single-owner.test.ts

@@ -0,0 +1,405 @@
+/**
+ * @module tests/integration/shutdown-single-owner
+ * @description ONE SHUTDOWN, ONE OWNER.
+ *
+ * MEASURED IN PRODUCTION. A host that owns its own shutdown — one SIGTERM
+ * listener calling `close()` on every pooled store — ran head-on into the
+ * engine's own signal handler, which iterated every live instance, flushed its
+ * components in parallel, and released its writer lock in a `finally`. Two
+ * teardowns of the same brain at the same moment. The log shape:
+ *
+ *   "Shutdown signal received - flushing pending data..."   (SIGTERM)
+ *   ...148 seconds of silence...
+ *   "Flushed successfully (1 instance)"
+ *   ...the host's pool close of that same store returns 1s later
+ *
+ * 149s for the one store with engine work in flight, against 24s for its six
+ * idle siblings. The same race in a local reproduction printed
+ * `Failed to flush one Brainy instance on shutdown: Writer fence lost … the
+ * lock file is gone` — the handler observing a lock the close it was racing
+ * had already released.
+ *
+ * The contract pinned here:
+ *   (a) A host owner and the engine's hooks both live: EXACTLY ONE close runs
+ *       per brain, no fence is lost, both durability markers are written, the
+ *       process exits 0, and the reopen adopts rather than folding.
+ *   (b) No host owner: the engine's handler closes every instance by the same
+ *       `close()` path — markers written, clean exit.
+ *   (c) `close()` is idempotent and re-entrant: concurrent callers share ONE
+ *       execution and all of them settle.
+ *   (d) Flush is single-flight: N kicks during a running flush arm exactly one
+ *       follow-up, and two flush bodies never overlap.
+ */
+
+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's clean-close record — written by `releaseWriterLock()`. */
+const closeRecordPath = (dir: string) => join(dir, 'locks', '_writer.close')
+/**
+ * The generation store's clean-shutdown marker — the adopt-vs-fold gate.
+ * (`FileSystemStorage` gzips raw objects, so the file on disk carries `.gz`;
+ * both spellings are accepted so the pin survives a compression change.)
+ */
+const cleanShutdownWritten = (dir: string) =>
+  existsSync(join(dir, '_system', 'clean-shutdown.json.gz')) ||
+  existsSync(join(dir, '_system', 'clean-shutdown.json'))
+
+/**
+ * Write a child script and start it under tsx, in its OWN process group so a
+ * group-wide signal reaches the grandchild that actually holds the writer
+ * lock. (A file, not `tsx -e`: the eval form compiles to CommonJS, which has
+ * no top-level await.)
+ */
+function startChild(scriptDir: string, body: string): ReturnType<typeof spawn> {
+  const scriptPath = join(scriptDir, 'child-process.mts')
+  writeFileSync(scriptPath, body)
+  return spawn(TSX, [scriptPath], {
+    cwd: REPO_ROOT,
+    stdio: ['ignore', 'pipe', 'pipe'],
+    detached: true
+  })
+}
+
+/** Start a child and resolve once it prints READY, collecting all its output. */
+function startAndAwaitReady(
+  scriptDir: string,
+  body: string
+): Promise<{ child: ReturnType<typeof spawn>; output: () => string }> {
+  const child = startChild(scriptDir, body)
+  let out = ''
+  child.stdout?.on('data', (d) => { out += String(d) })
+  child.stderr?.on('data', (d) => { out += String(d) })
+  return new Promise((resolvePromise, rejectPromise) => {
+    const timer = setTimeout(
+      () => rejectPromise(new Error(`child never became READY:\n${out}`)),
+      120_000
+    )
+    child.stdout?.on('data', () => {
+      if (out.includes('READY')) {
+        clearTimeout(timer)
+        resolvePromise({ child, output: () => out })
+      }
+    })
+    child.on('exit', (code) => {
+      clearTimeout(timer)
+      if (!out.includes('READY')) rejectPromise(new Error(`child exited ${code} before READY:\n${out}`))
+    })
+  })
+}
+
+/** Capture console.warn/error/log lines emitted while `fn` runs. */
+async function captureConsole<T>(fn: () => Promise<T>): Promise<{ result: T; lines: string[] }> {
+  const lines: string[] = []
+  const orig = { log: console.log, warn: console.warn, error: console.error }
+  const sink = (...args: unknown[]) => { lines.push(args.map((a) => String(a)).join(' ')) }
+  console.log = sink as typeof console.log
+  console.warn = sink as typeof console.warn
+  console.error = sink as typeof console.error
+  try {
+    return { result: await fn(), lines }
+  } finally {
+    console.log = orig.log
+    console.warn = orig.warn
+    console.error = orig.error
+  }
+}
+
+/**
+ * Reopen a store and assert the open ADOPTED: no crash-recovery fold, no
+ * stale-lock verdict. This is the whole point of a close having run exactly
+ * once — a fold is measured in tens of seconds on a real store.
+ */
+async function expectCleanReopen(dir: string): Promise<void> {
+  const { result, lines } = await captureConsole(async () => {
+    const next = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dir } })
+    await next.init()
+    return next
+  })
+  try {
+    expect(lines.filter((l) => /log-authority recovery|unclean shutdown detected/i.test(l))).toEqual([])
+    expect(lines.filter((l) => /Overwriting stale writer lock|appears dead/i.test(l))).toEqual([])
+  } finally {
+    await result.close()
+  }
+}
+
+/** The child's counts of closes entered and close bodies run, per brain. */
+function readResult(
+  resultPath: string,
+  out: string
+): { entries: Record<string, number>; bodies: Record<string, number>; releases: Record<string, number> } {
+  if (!existsSync(resultPath)) throw new Error(`child wrote no result file:\n${out}`)
+  return JSON.parse(readFileSync(resultPath, 'utf-8'))
+}
+
+/**
+ * The child-side instrumentation, shared by (a) and (b): count how many times
+ * `close()` is ENTERED per brain and how many times its body actually RUNS.
+ * The counting wrapper is an OWN property, so it shadows the prototype for
+ * every caller — including the engine's own signal handler, which calls
+ * `instance.close()`.
+ *
+ * `report()` writes SYNCHRONOUSLY to a file: it runs on the way out of the
+ * process (the engine's handler calls `process.exit(0)` when it is the sole
+ * shutdown owner), and a `console.log` to a pipe is asynchronous and can be
+ * dropped by that exit.
+ */
+function childCounters(resultPath: string): string {
+  return `
+  const entries = {}
+  const bodies = {}
+  const releases = {}
+  function instrument(name, brain) {
+    entries[name] = 0
+    bodies[name] = 0
+    releases[name] = 0
+    const enter = brain.close.bind(brain)
+    brain.close = () => { entries[name]++; return enter() }
+    const durable = brain.closeDurableSteps.bind(brain)
+    brain.closeDurableSteps = () => { bodies[name]++; return durable() }
+    // The writer lock is the ownership witness: the old handler released it
+    // in its own finally, on top of the owner's close doing the same.
+    const storage = brain.storage
+    const release = storage.releaseWriterLock.bind(storage)
+    storage.releaseWriterLock = () => { releases[name]++; return release() }
+  }
+  const report = () => {
+    __writeFileSync(${JSON.stringify(resultPath)}, JSON.stringify({ entries, bodies, releases }))
+  }
+`
+}
+
+describe('shutdown has exactly one owner', () => {
+  let dirA: string
+  let dirB: string
+  let scriptDir: string
+  let resultPath: string
+
+  beforeEach(() => {
+    dirA = makeTempDir('brainy-shutdown-owner-a-')
+    dirB = makeTempDir('brainy-shutdown-owner-b-')
+    scriptDir = makeTempDir('brainy-shutdown-owner-script-')
+    resultPath = join(scriptDir, 'result.json')
+  })
+
+  afterEach(() => {
+    for (const d of [dirA, dirB, scriptDir]) {
+      try { rmSync(d, { recursive: true, force: true }) } catch { /* ignore */ }
+    }
+  })
+
+  it('(a) a host owner closes both brains and the engine handler steps aside', async () => {
+    const script = `
+      import { writeFileSync as __writeFileSync } from 'node:fs'
+      import { Brainy } from ${JSON.stringify(BRAINY_SRC)}
+      const a = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: ${JSON.stringify(dirA)} } })
+      const b = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: ${JSON.stringify(dirB)} } })
+      await a.init()
+      await b.init()
+      await a.add({ data: 'row in brain a', type: 'concept' })
+      await b.add({ data: 'row in brain b', type: 'concept' })
+      ${childCounters(resultPath)}
+      instrument('a', a)
+      instrument('b', b)
+      // THE HOST'S OWN SHUTDOWN OWNER, registered after the engine's hooks —
+      // the ordinary shape: the pool was built before the signal wiring.
+      process.on('SIGTERM', async () => {
+        await Promise.all([a.close(), b.close()])
+        // Stay alive a beat so the engine's deferred handler gets its turn and
+        // has to decide what to do about two already-closed brains.
+        await new Promise((r) => setTimeout(r, 1500))
+        report()
+        process.exit(0)
+      })
+      console.log('READY')
+      setInterval(() => {}, 1000)
+    `
+    const { child, output } = await startAndAwaitReady(scriptDir, script)
+
+    process.kill(-(child.pid as number), 'SIGTERM')
+    const code = await new Promise<number | null>((r) => child.on('exit', (c) => r(c)))
+    // The tsx wrapper's exit event and the grandchild that actually held the
+    // locks are asynchronous with each other — let its last writes land.
+    await new Promise<void>((r) => setTimeout(r, 750))
+    const out = output()
+
+    // The process shut down cleanly.
+    expect(code, `child output:\n${out}`).toBe(0)
+
+    // EXACTLY ONE close per brain — entered once, body run once. A second
+    // entry would mean the engine's handler closed a brain its owner was
+    // already closing; a second body would mean close() is not single-flight.
+    const { entries, bodies, releases } = readResult(resultPath, out)
+    expect(entries).toEqual({ a: 1, b: 1 })
+    expect(bodies).toEqual({ a: 1, b: 1 })
+    // ...and the writer lock was given up exactly once per brain. This is the
+    // assertion that fails on the old handler, which released the lock in its
+    // own `finally` on top of the owner's close doing the same — two owners.
+    expect(releases).toEqual({ a: 1, b: 1 })
+
+    // The engine's handler ran (it announced the signal) and stepped aside for
+    // both brains rather than touching them. setImmediate lands in the check
+    // phase of the same loop turn, so a close that has begun cannot have
+    // finished — it is still in flight when the handler looks.
+    expect(out).toContain('Shutdown signal received')
+    expect(out).toMatch(/2 Brainy instances are already closing/)
+
+    // Nothing was taken out from under the owner, and nothing failed.
+    expect(out).not.toMatch(/Writer fence lost/i)
+    expect(out).not.toMatch(/Failed to (flush|close) one Brainy instance/i)
+
+    // Both durability markers, both brains: the writer lock's clean-close
+    // record and the generation store's clean-shutdown marker.
+    for (const dir of [dirA, dirB]) {
+      expect(existsSync(closeRecordPath(dir)), `clean-close record missing in ${dir}`).toBe(true)
+      expect(cleanShutdownWritten(dir), `clean-shutdown marker missing in ${dir}`).toBe(true)
+    }
+
+    // And the next open adopts instead of folding.
+    await expectCleanReopen(dirA)
+    await expectCleanReopen(dirB)
+  }, 240_000)
+
+  it('(b) with no host owner the engine closes every instance the same way', async () => {
+    const script = `
+      import { writeFileSync as __writeFileSync } from 'node:fs'
+      import { Brainy } from ${JSON.stringify(BRAINY_SRC)}
+      const a = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: ${JSON.stringify(dirA)} } })
+      const b = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: ${JSON.stringify(dirB)} } })
+      await a.init()
+      await b.init()
+      await a.add({ data: 'row in brain a', type: 'concept' })
+      await b.add({ data: 'row in brain b', type: 'concept' })
+      ${childCounters(resultPath)}
+      instrument('a', a)
+      instrument('b', b)
+      process.on('exit', report)
+      console.log('READY')
+      setInterval(() => {}, 1000)
+    `
+    const { child, output } = await startAndAwaitReady(scriptDir, script)
+
+    process.kill(-(child.pid as number), 'SIGTERM')
+    const code = await new Promise<number | null>((r) => child.on('exit', (c) => r(c)))
+    // The tsx wrapper's exit event and the grandchild that actually held the
+    // locks are asynchronous with each other — let its last writes land.
+    await new Promise<void>((r) => setTimeout(r, 750))
+    const out = output()
+
+    expect(code, `child output:\n${out}`).toBe(0)
+
+    // The engine owned this shutdown: one close per brain, through close().
+    const { entries, bodies, releases } = readResult(resultPath, out)
+    expect(entries).toEqual({ a: 1, b: 1 })
+    expect(bodies).toEqual({ a: 1, b: 1 })
+    expect(releases).toEqual({ a: 1, b: 1 })
+    expect(out).toContain('Shutdown signal received')
+    expect(out).toMatch(/Flushed successfully \(2 instances\)/)
+    expect(out).not.toMatch(/Writer fence lost/i)
+    expect(out).not.toMatch(/Failed to (flush|close) one Brainy instance/i)
+
+    for (const dir of [dirA, dirB]) {
+      expect(existsSync(closeRecordPath(dir)), `clean-close record missing in ${dir}`).toBe(true)
+      expect(cleanShutdownWritten(dir), `clean-shutdown marker missing in ${dir}`).toBe(true)
+    }
+
+    await expectCleanReopen(dirA)
+    await expectCleanReopen(dirB)
+  }, 240_000)
+
+  it('(c) two concurrent close() callers share ONE execution, and both settle', async () => {
+    const brain = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dirA } })
+    await brain.init()
+    await brain.add({ data: 'one row', type: NounType.Concept })
+
+    const inner = brain as unknown as { closeDurableSteps: () => Promise<void> }
+    const durable = inner.closeDurableSteps.bind(inner)
+    let bodies = 0
+    inner.closeDurableSteps = () => { bodies++; return durable() }
+
+    expect(brain.isClosing).toBe(false)
+    expect(brain.isClosed).toBe(false)
+
+    const first = brain.close()
+    // The state is observable IMMEDIATELY — a signal handler that yields a
+    // tick and comes back must not read a stale "not yet".
+    expect(brain.isClosing).toBe(true)
+    const second = brain.close()
+    expect(first === second, 'concurrent callers must share the one promise').toBe(true)
+
+    await Promise.all([first, second])
+    expect(bodies).toBe(1)
+    expect(brain.isClosed).toBe(true)
+
+    // A caller arriving after the close finished gets the same settled answer,
+    // and nothing runs again.
+    await brain.close()
+    expect(bodies).toBe(1)
+
+    expect(existsSync(closeRecordPath(dirA))).toBe(true)
+    expect(cleanShutdownWritten(dirA)).toBe(true)
+  }, 120_000)
+
+  it('(d) N kicks during a running flush arm exactly one follow-up, never a second flush', async () => {
+    const brain = new Brainy({ requireSubtype: false, storage: { type: 'filesystem', path: dirA } })
+    await brain.init()
+
+    const inner = brain as unknown as {
+      _flushBodyRuns: number
+      _flushConcurrencyPeak: number
+      _flushInFlight: Promise<void> | null
+      _flushFollowUp: Promise<void> | null
+      _persistBackgroundFlight: Promise<void> | null
+      metadataIndex: { flush: () => Promise<void> }
+      kickBackgroundFlush: (reason: 'threshold' | 'idle') => void
+    }
+
+    // Widen the flush body's window so the kicks land INSIDE it — the
+    // production shape, where two flushes overlapped 3s apart.
+    const metaFlush = inner.metadataIndex.flush.bind(inner.metadataIndex)
+    inner.metadataIndex.flush = async () => {
+      await new Promise((r) => setTimeout(r, 400))
+      return metaFlush()
+    }
+
+    await brain.add({ data: 'a write to flush', type: NounType.Concept })
+    const runsBefore = inner._flushBodyRuns
+
+    const leader = brain.flush()
+    await new Promise((r) => setTimeout(r, 50)) // the leader is inside its body
+    expect(inner._flushInFlight, 'a flush is running').not.toBeNull()
+
+    // The cadence kicks — the door named in the defect — plus direct callers
+    // (an application flush, the cross-process flush-request watcher).
+    for (let i = 0; i < 5; i++) inner.kickBackgroundFlush('threshold')
+    const direct = [brain.flush(), brain.flush(), brain.flush()]
+
+    // EXACTLY ONE follow-up is armed, however many callers arrived.
+    expect(inner._flushFollowUp, 'the eight kicks armed one follow-up').not.toBeNull()
+
+    await Promise.all([leader, ...direct, inner._persistBackgroundFlight ?? Promise.resolve()])
+
+    // One leader + one follow-up. Not nine, and never two at once.
+    expect(inner._flushBodyRuns - runsBefore).toBe(2)
+    expect(inner._flushConcurrencyPeak).toBe(1)
+    expect(inner._flushInFlight).toBeNull()
+    expect(inner._flushFollowUp).toBeNull()
+
+    inner.metadataIndex.flush = metaFlush
+    await brain.close()
+  }, 120_000)
+})