fix(flush): the gate settles its waiter from the machine, never from a chain

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.

src/brainy.ts

@@ -785,9 +785,25 @@ export class Brainy<T = any> implements BrainyInterface<T> {
    * "Flushing Brainy indexes and caches to disk..." runs overlapping 3s
    * apart on one brain, their walls growing 295ms → 4.9s as they contended
    * for the same providers.
+   *
+   * THE WAITER IS SETTLED BY THE MACHINE, NEVER BY A PROMISE CHAIN. The queue
+   * is a BARE DEFERRED (`_flushQueued` plus its `_flushQueuedSettle` handles),
+   * not `leader.then(() => this.flush())`. A chained follow-up 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 graph closes on
+   * itself and NOBODY resolves — an unbounded hang, not a slow flush. Here the
+   * leader never awaits the queue: its `finally` PROMOTES the waiter to a new
+   * leader and settles the deferred from that run, and the leader's own
+   * promise settles without waiting for it. Every exit — the leader
+   * resolving, the leader REJECTING, the promoted run rejecting — runs the
+   * same promotion, so a queued caller is always settled exactly once.
    */
   private _flushInFlight: Promise<void> | null = null
-  private _flushFollowUp: Promise<void> | null = null
+  private _flushQueued: Promise<void> | null = null
+  private _flushQueuedSettle: {
+    resolve: () => void
+    reject: (error: unknown) => void
+  } | null = null
   /** Flush bodies that got past the single-flight gate (pinned by tests). */
   private _flushBodyRuns = 0
   /** Flush bodies running right now, and the high-water mark — which the
@@ -12987,29 +13003,61 @@ export class Brainy<T = any> implements BrainyInterface<T> {
     // crossed BEFORE any await, so two callers in the same tick cannot both
     // find the field empty.
     if (this._flushInFlight) {
-      if (!this._flushFollowUp) {
-        // The running flush's failure is not this follow-up's failure: it is
-        // reported to ITS caller, and the queued work still gets its turn.
-        this._flushFollowUp = this._flushInFlight
-          .catch(() => {})
-          .then(() => {
-            this._flushFollowUp = null
-            return this.flush()
-          })
+      if (!this._flushQueued) {
+        // A BARE DEFERRED, not a chain off the leader — see the field's doc.
+        // Nothing here awaits the leader, so no waiter can ever be reachable
+        // only through the promise it is itself blocking.
+        this._flushQueued = new Promise<void>((resolve, reject) => {
+          this._flushQueuedSettle = { resolve, reject }
+        })
       }
-      return this._flushFollowUp
+      return this._flushQueued
     }
+    return this.startFlushLeader()
+  }
+
+  /**
+   * @description Run one flush body as the leader and install it as
+   * `_flushInFlight`. On settle — resolved OR rejected — the gate opens and
+   * the ONE queued waiter (if any) is promoted. The `finally` callback returns
+   * nothing on purpose: a callback that returned the promoted run's promise
+   * would make the leader await its own follower.
+   * @returns The leader's own promise, settling on its own body alone.
+   */
+  private startFlushLeader(): Promise<void> {
     const run = this._runFlush()
     // `finally` and not `then`: a failed flush must still open the gate, or
     // one rejection would wedge every later flush behind a promise nobody
     // will ever settle.
-    const gated = run.finally(() => {
+    const gated: Promise<void> = run.finally(() => {
       if (this._flushInFlight === gated) this._flushInFlight = null
+      this.promoteQueuedFlush()
     })
     this._flushInFlight = gated
     return gated
   }
 
+  /**
+   * @description Promote the single queued waiter (if one is waiting) to
+   * leader and settle its deferred from that run. Never throws into the
+   * leader's `finally`: a synchronous failure starting the promoted run is
+   * reported to the waiter, which must be settled on every path.
+   * @returns Nothing.
+   */
+  private promoteQueuedFlush(): void {
+    const settle = this._flushQueuedSettle
+    if (!settle) return
+    // Clear BEFORE starting, so the promoted run's own joiners queue afresh
+    // rather than joining a deferred that is already being settled.
+    this._flushQueued = null
+    this._flushQueuedSettle = null
+    try {
+      this.startFlushLeader().then(settle.resolve, settle.reject)
+    } catch (error) {
+      settle.reject(error)
+    }
+  }
+
   /**
    * @description The flush body — everything {@link flush} promises, run
    * exactly once at a time by that method's single-flight gate. Private
@@ -20409,9 +20457,11 @@ export class Brainy<T = any> implements BrainyInterface<T> {
     // awaits its leader too, so the second pass is a no-op unless a writer
     // raced this close.
     for (let pass = 0; pass < 2; pass++) {
-      const chain = this._flushFollowUp ?? this._flushInFlight
-      if (!chain) break
-      await chain.catch(() => {})
+      const inFlight = this._flushInFlight
+      const queued = this._flushQueued
+      if (!inFlight && !queued) break
+      if (inFlight) await inFlight.catch(() => {})
+      if (queued) await queued.catch(() => {})
     }
 
     // Cancel any pending post-import background deduplication FIRST — it is a

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

@@ -362,7 +362,7 @@ describe('shutdown has exactly one owner', () => {
       _flushBodyRuns: number
       _flushConcurrencyPeak: number
       _flushInFlight: Promise<void> | null
-      _flushFollowUp: Promise<void> | null
+      _flushQueued: Promise<void> | null
       _persistBackgroundFlight: Promise<void> | null
       metadataIndex: { flush: () => Promise<void> }
       kickBackgroundFlush: (reason: 'threshold' | 'idle') => void
@@ -389,7 +389,7 @@ describe('shutdown has exactly one owner', () => {
     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()
+    expect(inner._flushQueued, 'the eight kicks armed one follow-up').not.toBeNull()
 
     await Promise.all([leader, ...direct, inner._persistBackgroundFlight ?? Promise.resolve()])
 
@@ -397,7 +397,7 @@ describe('shutdown has exactly one owner', () => {
     expect(inner._flushBodyRuns - runsBefore).toBe(2)
     expect(inner._flushConcurrencyPeak).toBe(1)
     expect(inner._flushInFlight).toBeNull()
-    expect(inner._flushFollowUp).toBeNull()
+    expect(inner._flushQueued).toBeNull()
 
     inner.metadataIndex.flush = metaFlush
     await brain.close()

tests/unit/brainy/flush-single-flight.test.ts

@@ -0,0 +1,175 @@
+/**
+ * @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<void> | null
+  _flushQueued: Promise<void> | null
+  _flushBodyRuns: number
+  _flushConcurrencyPeak: number
+  _flushSteps: () => Promise<void>
+  kickBackgroundFlush: (reason: 'threshold' | 'idle') => void
+}
+
+/** Fail loudly rather than hanging the suite: a stranded waiter never settles. */
+function withinBound<T>(p: Promise<T>, ms: number, what: string): Promise<T> {
+  let timer: ReturnType<typeof setTimeout>
+  return Promise.race([
+    p,
+    new Promise<never>((_, reject) => {
+      timer = setTimeout(() => reject(new Error(`${what} did not settle within ${ms}ms`)), ms)
+    })
+  ]).finally(() => clearTimeout(timer)) as Promise<T>
+}
+
+describe('the flush gate settles every waiter', () => {
+  const brains: Brainy<any>[] = []
+
+  afterEach(async () => {
+    for (const b of brains.splice(0)) {
+      try { await b.close() } catch { /* already closed */ }
+    }
+  })
+
+  async function openBrain(): Promise<Brainy<any>> {
+    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')
+  })
+})