[Bug] `BrokerActor` returns from `_handleReconnect` without `_closeTransport()` on both terminal paths, so a broker that gives up keeps a live driver client, its sockets and its internal reconnect timers

[pathosDev]

Problem

_handleReconnect has two terminal exits — reconnect: false, and the attempt budget exhausted — and neither closes the transport. _closeTransport() runs from exactly two places: the top of _tryConnect (to clean up before the next attempt) and postStop. So when the base class gives up, _transportOpened is still true and disconnectImplementation has not run for the attempt that just failed.

What that leaves behind is subclass-specific and all of it is live: an ioredis client with its own reconnect timer still ticking, an amqplib connection and channel with their sockets and their 'error'/'close' listeners still attached, a kafkajs producer and consumer, an mqtt.js client that is itself retrying on its own schedule. The framework has decided to stop reconnecting; the driver underneath it has not been told.

The result is a broker actor that reports disconnected, publishes BrokerReconnectFailed, accepts no further work — and holds a file descriptor, a heap graph and a retry loop for the rest of the actor's life. In a system that spawns a broker actor per tenant or per shard, each terminal failure adds one.

Evidence

src/io/broker/BrokerActor.ts:566-580 — both returns leave without a teardown:

src/io/broker/BrokerActor.ts:566-580
  private _handleReconnect(cause: Error): void {
    const policy = this.options.reconnect;
    if (policy === false) return;
    const initial = policy?.initialDelayMs ?? DEFAULT_RECONNECT.initialDelayMs;
    const maxDelay = policy?.maxDelayMs ?? DEFAULT_RECONNECT.maxDelayMs;
    const factor = policy?.factor ?? DEFAULT_RECONNECT.factor;
    const maxAttempts = policy?.maxAttempts ?? DEFAULT_RECONNECT.maxAttempts;

    this._reconnectAttempt++;
    if (this._reconnectAttempt > maxAttempts) {
      this.system.eventStream.publish(new BrokerReconnectFailed(
        this.self.path.toString(), this.endpointLabel(), this._reconnectAttempt - 1, cause,
      ));
      return;
    }

The only two callers of _closeTransport, src/io/broker/BrokerActor.ts:508-519:

src/io/broker/BrokerActor.ts:508-519
    // Never re-enter connectImplementation on top of the previous
    // attempt's state.  A drop (or a connect that failed half-way)
    // leaves the subclass holding a dead client, its subscription
    // handles and its pending acks; building the new connection on top
    // of that leaks them and — because the subclass still sees its own
    // stale handles — can silently skip re-subscribing (#504).
    await this._closeTransport();

    this._state = 'connecting';
    // Set before the call, not after: a connectImplementation that
    // throws part-way through has still opened transport state.
    this._transportOpened = true;

src/io/broker/BrokerActor.ts:448-461
  override async postStop(): Promise<void> {
    this._scheduledReconnectCancel?.();
    this._scheduledReconnectCancel = null;
    // Gate on transport state, not on `_state`: after a dropped
    // connection the state machine reads `disconnected` while the
    // subclass still holds sockets, clients and pending acks — the
    // old `_state !== 'disconnected'` guard skipped teardown for
    // exactly the actors that needed it most.
    if (this._transportOpened) this._state = 'disconnecting';
    await this._closeTransport();
    this._state = 'disconnected';
    this._outboundBuffer = [];
    this._subscribers.clear();
  }

The comment on that postStop gate states the exact reasoning this issue is about — "a dropped connection leaves the subclass holding sockets, clients and pending acks" — and applies it to stop but not to give-up.

Proposal

Call await this._closeTransport() on both terminal paths of _handleReconnect, before the return. _closeTransport is already documented as idempotent and never-throwing, so this is safe at both sites; it just needs _handleReconnect (and its handleConnectionLost caller) to become async, or to void the call the way _tryConnect voids _drainBuffer.

Worth considering in the same change: publish BrokerReconnectFailed after the teardown, so a subscriber that reacts by stopping the actor cannot race the close.

Acceptance sketch

• disconnectImplementation runs when reconnect: false and the connect attempt fails.
• disconnectImplementation runs when the maxAttempts budget is exhausted.
• A test subclass counts connectImplementation and disconnectImplementation calls and asserts they balance on both terminal paths.
• _transportOpened is false after a terminal reconnect failure.

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 counting both hooks, with a connectImplementation that always throws:

W6-14: connectImplementation called 2x, disconnectImplementation called 1x, state=disconnected
W6-14: reconnect:false -> connect 1x, disconnect 0x  (transport left open: true)

With maxAttempts: 1, two connect attempts ran and only one teardown — the single _closeTransport at the top of the second attempt. The transport opened by the final, failing attempt is never closed. With reconnect: false there is no teardown at all.

Adjacent issues: #708 is the inverse ordering — stopping the actor while a detached reconnect is inside connectImplementation, so postStop's teardown races a connection that is still being built. This issue is the case where no stop happens at all and the actor simply gives up. They share _closeTransport and should probably be fixed together, but neither subsumes the other.

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