/** * @module tests/unit/brainy/flush-single-flight * @description THE FLUSH GATE NEVER STRANDS A WAITER. * * The gate serialises flushes: one body runs, at most one waits. The failure * mode that shape invites is a promise CYCLE — a queued follow-up expressed as * `leader.then(() => this.flush())` is settled only by resolving the promise * the leader is being awaited through, so anything that awaits `flush()` from * inside a flush body closes the graph on itself and nobody ever resolves. * That is an unbounded hang, not a slow flush, and it presents exactly like a * test timing out inside a bulk write. * * The gate therefore settles its waiter from the MACHINE (a bare deferred * promoted in the leader's `finally`), never from a chain. The laws pinned * here, each on a path that must settle the waiter: * * (a) many callers during one running flush → one body, one follow-up, and * EVERY caller resolves within a bound; * (b) the leader REJECTS → its own caller rejects, and the queued caller is * still run and still settled; * (c) the promoted follow-up itself rejects → its waiter rejects (settled, * not stranded) and the gate is left open for the next flush; * (d) the leader's promise does not wait for its follower. */ import { describe, it, expect, afterEach } from 'vitest' import { Brainy } from '../../../src/brainy' import { NounType } from '../../../src/types/graphTypes' type GateInternals = { _flushInFlight: Promise | null _flushQueued: Promise | null _flushBodyRuns: number _flushConcurrencyPeak: number _flushSteps: () => Promise kickBackgroundFlush: (reason: 'threshold' | 'idle') => void } /** Fail loudly rather than hanging the suite: a stranded waiter never settles. */ function withinBound(p: Promise, ms: number, what: string): Promise { let timer: ReturnType return Promise.race([ p, new Promise((_, reject) => { timer = setTimeout(() => reject(new Error(`${what} did not settle within ${ms}ms`)), ms) }) ]).finally(() => clearTimeout(timer)) as Promise } describe('the flush gate settles every waiter', () => { const brains: Brainy[] = [] afterEach(async () => { for (const b of brains.splice(0)) { try { await b.close() } catch { /* already closed */ } } }) async function openBrain(): Promise> { const brain = new Brainy({ requireSubtype: false, storage: { type: 'memory' } }) brains.push(brain) await brain.init() await brain.add({ data: 'a write, so a flush has work', type: NounType.Thing }) return brain } it('(a) every caller arriving during one flush resolves, and only one follows', async () => { const brain = await openBrain() const inner = brain as unknown as GateInternals const realSteps = inner._flushSteps.bind(inner) inner._flushSteps = async () => { await new Promise((r) => setTimeout(r, 120)) return realSteps() } const runsBefore = inner._flushBodyRuns const leader = brain.flush() await new Promise((r) => setTimeout(r, 20)) const joiners = [brain.flush(), brain.flush(), brain.flush(), brain.flush()] for (let i = 0; i < 4; i++) inner.kickBackgroundFlush('threshold') expect(inner._flushQueued, 'exactly one waiter is queued').not.toBeNull() await withinBound(Promise.all([leader, ...joiners]), 15_000, 'the flush callers') expect(inner._flushBodyRuns - runsBefore).toBe(2) expect(inner._flushConcurrencyPeak).toBe(1) expect(inner._flushQueued).toBeNull() }) it('(b) a leader that REJECTS still runs and settles the queued waiter', async () => { const brain = await openBrain() const inner = brain as unknown as GateInternals const realSteps = inner._flushSteps.bind(inner) let call = 0 inner._flushSteps = async () => { call++ await new Promise((r) => setTimeout(r, 80)) if (call === 1) throw new Error('injected: the leader flush failed') return realSteps() } const leader = brain.flush() await new Promise((r) => setTimeout(r, 20)) const queued = brain.flush() await expect(leader).rejects.toThrow(/injected: the leader flush failed/) // The waiter is NOT collateral damage of the leader's failure: it gets its // own run, and it settles. await withinBound(queued, 15_000, 'the queued waiter after a failed leader') expect(call).toBe(2) expect(inner._flushQueued).toBeNull() expect(inner._flushInFlight).toBeNull() }) it('(c) a promoted follow-up that rejects settles its waiter and opens the gate', async () => { const brain = await openBrain() const inner = brain as unknown as GateInternals const realSteps = inner._flushSteps.bind(inner) let call = 0 inner._flushSteps = async () => { call++ await new Promise((r) => setTimeout(r, 80)) if (call === 2) throw new Error('injected: the follow-up flush failed') return realSteps() } const leader = brain.flush() await new Promise((r) => setTimeout(r, 20)) const queued = brain.flush() await withinBound(leader, 15_000, 'the leader') await withinBound( expect(queued).rejects.toThrow(/injected: the follow-up flush failed/), 15_000, 'the rejected follow-up' ) // The gate is open: a later flush still runs. inner._flushSteps = realSteps await brain.add({ data: 'another write', type: NounType.Thing }) await withinBound(brain.flush(), 15_000, 'the flush after a failed follow-up') expect(inner._flushInFlight).toBeNull() expect(inner._flushQueued).toBeNull() }) it('(d) the leader does not wait for its follower', async () => { const brain = await openBrain() const inner = brain as unknown as GateInternals const realSteps = inner._flushSteps.bind(inner) let call = 0 inner._flushSteps = async () => { call++ // The follow-up is deliberately far slower than the leader. await new Promise((r) => setTimeout(r, call === 1 ? 60 : 600)) return realSteps() } const leader = brain.flush() await new Promise((r) => setTimeout(r, 20)) const queued = brain.flush() const t0 = Date.now() await withinBound(leader, 15_000, 'the leader') const leaderWall = Date.now() - t0 // If the leader awaited its follower it could not return before the // follower's own 600ms body had run. expect(leaderWall).toBeLessThan(500) await withinBound(queued, 15_000, 'the follower') }) })