[Bug] `BrokerActor.dispatchWhenConnected` reads an empty buffer while `_drainBuffer` still awaits, so two dispatches run concurrently and the documented enqueue order is violated

[pathosDev]

Problem

BrokerActor.enqueueOutbound documents that messages are "dispatched immediately (in the order they were enqueued)", and dispatchWhenConnected carries a comment restating the mechanism: "If an earlier flush is still draining the buffer, append at the tail to preserve order." The mechanism it names does not hold, because the test it uses — this._outboundBuffer.length > 0 — is not "a flush is in progress". It is "the buffer is non-empty".

_drainBuffer shifts each envelope off the buffer before awaiting dispatchOutgoing. When it takes the last one, the buffer is empty for the entire duration of that await. Any enqueueOutbound arriving in that window sees length 0, takes the void this._dispatchOne(env) branch, and starts a second dispatchOutgoing concurrently with the first. Whichever protocol call finishes first is the one that reaches the wire first — so with a slow first message and a fast second, the second overtakes.

The same confusion breaks the recovery path. _dispatchOne's failure branch does _outboundBuffer.unshift(env) to push the failed message back "at the head", which is only correct if it was the oldest in flight. Under two concurrent dispatches it can be the newer one, and it is then re-sent ahead of an older message that is still pending or already buffered.

Every broker actor in the directory routes its publishes through this method, so this is one defect in the shared base, not five.

Evidence

The contract, src/io/broker/BrokerActor.ts:378-395:

src/io/broker/BrokerActor.ts:378-395
  /**
   * Enqueue an outbound message.  When connected, it is dispatched
   * immediately (in the order they were enqueued); when disconnected
   * or connecting, it is buffered.  Returns true if buffered or sent,
   * false if the message was dropped (overflow / not-connected with
   * `outboundBuffer: 0`).
   */
  protected enqueueOutbound(payload: P): boolean {
    const env: OutboundEnvelope<P> = { payload, enqueuedAt: Date.now() };

    // Dispatch on connection state with compile-time exhaustiveness:
    // adding a new state to `ConnectionState` forces every site that
    // matches on it (including this one) to handle the new variant.
    return match(this._state)
      .with('connected', () => this.dispatchWhenConnected(env))
      .with('connecting', 'disconnected', 'disconnecting', () => this.bufferWhileOffline(env))
      .exhaustive();
  }

The guard that is meant to enforce it, src/io/broker/BrokerActor.ts:397-410:

src/io/broker/BrokerActor.ts:397-410
  /**
   * Connected path: dispatch the envelope now, or — if an earlier flush is
   * still draining the buffer — append at the tail to preserve order.
   */
  private dispatchWhenConnected(env: OutboundEnvelope<P>): boolean {
    // Dispatch directly.  If an earlier flush is still draining the
    // buffer, append at the tail to preserve order.
    if (this._outboundBuffer.length > 0) {
      this._outboundBuffer.push(env);
      return true;
    }
    void this._dispatchOne(env);
    return true;
  }

The shift-before-await that empties the buffer while a dispatch is still running, src/io/broker/BrokerActor.ts:602-614:

src/io/broker/BrokerActor.ts:602-614
  private async _drainBuffer(): Promise<void> {
    while (this._outboundBuffer.length > 0 && this._state === 'connected') {
      const env = this._outboundBuffer.shift()!;
      try {
        await this.dispatchOutgoing(env);
      } catch (e) {
        // Push back at the head so the message isn't lost across reconnect.
        this._outboundBuffer.unshift(env);
        this.handleConnectionLost(e instanceof Error ? e : new Error(String(e)));
        return;
      }
    }
  }

and the head-unshift that assumes single-flight, src/io/broker/BrokerActor.ts:616-623:

src/io/broker/BrokerActor.ts:616-623
  private async _dispatchOne(env: OutboundEnvelope<P>): Promise<void> {
    try {
      await this.dispatchOutgoing(env);
    } catch (e) {
      this._outboundBuffer.unshift(env);
      this.handleConnectionLost(e instanceof Error ? e : new Error(String(e)));
    }
  }

Proposal

Track the in-flight state explicitly instead of inferring it from the buffer:

• a _dispatching: boolean (or a single _drainPromise) set before the first dispatchOutgoing and cleared after the last;
• dispatchWhenConnected appends whenever _dispatching is true, regardless of buffer length, and otherwise starts the drain;
• funnel both paths through one drain loop, so there is exactly one place that awaits dispatchOutgoing and exactly one place that unshifts on failure. _dispatchOne then disappears.

That also fixes the recovery ordering for free: with a single loop, the failed envelope really is the head.

Acceptance sketch

• Two enqueueOutbound calls straddling a slow dispatchOutgoing reach the transport in enqueue order.
• At most one dispatchOutgoing is in flight at a time, asserted by a counter in a test subclass.
• A failed dispatch is re-queued ahead of nothing that was enqueued before it.
• The JSDoc on enqueueOutbound and the comment in dispatchWhenConnected describe what the code does.

Verification status

Found in the ten-lens production-readiness review of 2026-08-05 (v0.13.0) and re-verified before filing: reproduced by execution. A minimal BrokerActor subclass whose dispatchOutgoing holds the first call on a promise and appends to a wire array was spawned, connected, and given A then B:

W6-13: state=connected
W6-13: after A -> buffered=0 wire=[]
W6-13: after B (A still in flight) wire=["B"]
W6-13: final wire order = ["B","A"]  (enqueue order was ["A","B"])

buffered=0 while A was in flight is the defect in one number: the guard that is supposed to make B queue behind A sees an empty buffer and dispatches concurrently. B reached the wire while A was still pending.

Adjacent issues: none in the tracker touch BrokerActor's outbound ordering. #652 (reconnect jitter), #708 (stop during a detached reconnect) and #709 (subscribeRef never handles Terminated) are the same class, different methods.

Part of the production-readiness review batch — tracked in #913.