310 lines
14 KiB
TypeScript
310 lines
14 KiB
TypeScript
|
|
/**
|
||
|
|
* @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)
|
||
|
|
})
|