David Snelling
dea3ec2031
Some checks failed
CI / Node 22 (push) Successful in 12m26s
CI / Node 24 (push) Successful in 12m19s
CI / Bun (latest) (push) Successful in 12m28s
CI / Integration + conformance (Node 22) (push) Failing after 17m12s
The single-flight gate queued its follow-up as `leader.catch().then(() => this.flush())`. That waiter is settled ONLY by resolving the very promise the leader is being awaited through, so the moment anything inside a flush body awaits flush(), the promise graph closes on itself and nobody resolves — an unbounded hang, not a slow flush, presenting exactly like a bulk write timing out. No current call site awaits a flush from inside one, so this is a latent cycle rather than an observed one; the gate should not depend on that staying true. The queue is now a bare deferred. The leader's finally opens the gate and PROMOTES the waiter to a new leader, settling the deferred from that run; the finally returns nothing, so the leader never awaits its own follower. Every exit runs the same promotion — the leader resolving, the leader rejecting, the promoted run rejecting — so a queued caller is settled exactly once on every path, and a synchronous failure starting the promoted run is reported to the waiter instead of thrown into the leader's finally. close() drains both handles. tests/unit/brainy/flush-single-flight.test.ts pins the invariant on each path that must settle a waiter: many callers during one flush all resolve within a bound (one body, one follow-up, peak concurrency 1); a REJECTING leader still runs and settles the queued waiter; a rejecting follow-up settles its waiter and leaves the gate open; and the leader returns without waiting for a deliberately slower follower.
405 lines
17 KiB
TypeScript
405 lines
17 KiB
TypeScript
/**
|
|
* @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
|
|
_flushQueued: 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._flushQueued, '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._flushQueued).toBeNull()
|
|
|
|
inner.metadataIndex.flush = metaFlush
|
|
await brain.close()
|
|
}, 120_000)
|
|
})
|