[Security] Stopping a broker actor while a detached reconnect attempt is awaiting `connectImplementation()` leaves a fully live broker connection and an infinite reconnect loop attached to the terminated actor

[pathosDev]

Component: src/io/broker/BrokerActor.ts
Severity (assessment): MEDIUM

_scheduleReconnect runs _tryConnect on the system scheduler, deliberately detached from the mailbox. postStop cancels only the pending timer handle; it has no way to abort an already-running _tryConnect, and BrokerActor carries no stopped flag that _tryConnect checks. If the timer has already fired, postStop runs _closeTransport() against a subclass that has not yet assigned its client handle (a no-op), then the in-flight connectImplementation() resolves, sets _state = 'connected', and the terminated actor now owns a live connection plus an active handleConnectionLost → _handleReconnect → _scheduleReconnect cycle with maxAttempts: Infinity.

Exploit walkthrough

Preconditions: the broker is down or flapping (so the actor is in a reconnect cycle) and the application stops that actor — a rolling restart, a supervisor Directive.Stop, or an operator swapping configuration.

1. Broker goes down. handleConnectionLost → _handleReconnect → _scheduleReconnect(delay). With defaults, delay grows to 30 s, so the window in which a timer is armed or firing is essentially permanent.
2. The scheduler fires; _tryConnect() starts and blocks on await this.connectImplementation() (a TCP connect + protocol handshake — tens to thousands of ms).
3. During that await, ref.stop() is processed on the mailbox. postStop runs: _scheduledReconnectCancel?.() cancels an already-fired timer (no effect), _transportOpened is true so _state = 'disconnecting', _closeTransport() flips _transportOpened = false and calls disconnectImplementation() — which returns immediately because the subclass's client/consumer/producer fields are still null. _state = 'disconnected'.
4. The broker comes back and connectImplementation() resolves. For MqttActor it assigns this.client, registers 'message'/'error'/'close' handlers and issues SUBSCRIBE for every registry pattern. For KafkaActor it connects a producer and joins the consumer group and starts consumer.run. _state = 'connected'; BrokerConnected is published for an actor that no longer exists.
5. Nothing will ever tear this down: postStop has already run, and _closeTransport is now a permanent no-op because _transportOpened === false.
6. When that connection later drops, handleConnectionLost sees _state === 'connected', publishes BrokerDisconnected, and re-enters _handleReconnect → _scheduleReconnect — an infinite background reconnect loop attached to a dead actor, emitting an event stream entry per attempt, for the remaining life of the process.

Impact: leaked sockets, keep-alive timers, and — for Kafka — a live consumer-group member that keeps holding partition assignments, so the intended handover to the replacement actor never completes and the group is stuck in a rebalance/eviction cycle. Inbound messages are delivered to a dead mailbox as dead letters.

Evidence — src/io/broker/BrokerActor.ts

src/io/broker/BrokerActor.ts:588-598 — reconnect is explicitly detached from the mailbox:

  private _scheduleReconnect(delayMs: number): void {
    this._scheduledReconnectCancel?.();
    const reconnect = (): void => { void this._tryConnect(); };
    // Use the system scheduler (not the actor TimerScheduler): reconnect
    // is detached from the message pipeline — it should not queue behind
    // user commands.  Cancel-handle is tracked for postStop teardown.
    const handle = this.system.scheduler.scheduleOnceFunction(delayMs, reconnect);
    this._scheduledReconnectCancel = (): void => { handle.cancel(); };
  }

src/io/broker/BrokerActor.ts:448-461 — teardown cancels a timer, not a running attempt, and there is no stopped flag set:

  override async postStop(): Promise<void> {
    this._scheduledReconnectCancel?.();
    this._scheduledReconnectCancel = null;
    if (this._transportOpened) this._state = 'disconnecting';
    await this._closeTransport();
    this._state = 'disconnected';
    this._outboundBuffer = [];
    this._subscribers.clear();
  }

src/io/broker/BrokerActor.ts:514-529 — nothing in the connect path consults a stopped flag before or after the await:

    await this._closeTransport();
    this._state = 'connecting';
    this._transportOpened = true;
    try {
      await this.connectImplementation();
      this._state = 'connected';
      this._reconnectAttempt = 0;
      this._consecutiveFailures = 0;
      this.system.eventStream.publish(
        new BrokerConnected(this.self.path.toString(), this.endpointLabel()),
      );
      void this._drainBuffer();

src/io/broker/BrokerActor.ts:547-554 — the no-op that lets the connection escape:

  private async _closeTransport(): Promise<void> {
    if (!this._transportOpened) return;
    this._transportOpened = false;

(once postStop has flipped _transportOpened to false, the later _state = 'connected' cannot be undone by anything).

Subclass side, e.g. src/io/broker/MqttActor.ts:487-489 — teardown before the client handle exists is a no-op:

  protected async disconnectImplementation(): Promise<void> {
    if (!this.client) return;

and this.client is only assigned inside the 'connect' callback at src/io/broker/MqttActor.ts:454. KafkaActor.disconnectImplementation (src/io/broker/KafkaActor.ts:290-313) has the same shape — this.consumer/this.producer are assigned only after connect() resolves.

The actor's fields stay usable after stop: Actor's accessors are plain reads of the injected context (src/Actor.ts:26-31), so this.system.scheduler and this.self.path keep working. Scheduler only ignores timers after a system-wide shutdown() (src/Scheduler.ts:50, 113), which a single ref.stop() does not trigger.

Why the existing guard does not cover it

_closeTransport and the _transportOpened flag were clearly added for exactly this family of bug — the comment at BrokerActor.ts:451-455 explains it fixed the case where a dropped connection left the subclass holding sockets. It handles the drop case correctly; it does not handle the in-flight-connect case, because _transportOpened is set before the await and cleared by whoever calls _closeTransport first. The test suite covers both adjacent cases and neither reaches this one: tests/integration/in-process/io/broker/BrokerActor.test.ts:268 (postStop calls disconnectImplementation and clears state) stops a connected actor, and :386 (postStop tears down after a connection loss) deliberately uses reconnect: { initialDelayMs: 10_000 } so the timer is guaranteed not to have fired — the exact configuration that avoids the race.

Suggested fix

Add a private _stopped = false to BrokerActor; set it as the first statement of postStop(). Guard _tryConnect at entry (if (this._stopped) return;) and immediately after await this.connectImplementation() — on the post-await path, run await this.disconnectImplementation() directly (bypassing the _transportOpened gate, which postStop has already cleared) and return without publishing BrokerConnected or scheduling anything. Also guard handleConnectionLost and _handleReconnect on _stopped. Add a regression test that stops the actor while a fake connectImplementation promise is deliberately left pending, then resolves it and asserts disconnects incremented and no further connect attempts follow.

Verification status

Found in the second, independent whole-framework security re-audit of 2026-08-02 (v0.12.0) — a fresh pass run without reference to the first wave's findings, then triaged against the existing tracker and adjudicated by verifiers instructed to refute it.

Verifier note

Verified against source; the defect holds, with two corrections to the evidence.

Citation accuracy: every base-class line cited is correct. _scheduleReconnect at src/io/broker/BrokerActor.ts:588-598 does run void this._tryConnect() on this.system.scheduler, and the in-code comment at :593-595 states the detachment from the mailbox explicitly. postStop at :448-461 cancels only the timer handle and awaits _closeTransport(). _tryConnect at :499-540 consults nothing before or after await this.connectImplementation() at :521. _closeTransport at :547-554 early-returns on !this._transportOpened.

Guard search: I grepped src/io for _stopped|isStopped|stopped — no matches, so there is no stopped flag anywhere in the broker tree. postStop is overridden only in BrokerActor and in the unrelated GrpcServerActor. ActorCell.finalizeTermination (src/internal/ActorCell.ts:729-742) cancels this.timers (the actor-scoped TimerScheduler) before calling postStop, which is precisely the scheduler BrokerActor deliberately does not use. Scheduler.shutdown() is called from exactly one place, ActorSystem._rootTerminated (src/ActorSystem.ts:393), so a single ref.stop() leaves the system scheduler live. ActorCell never detaches the context or nulls this.actor (only _attach at :661 and :803), so this.system, this.self and this.options all keep working on the terminated instance. SimpleCancellable.cancel() (src/Scheduler.ts:41-43) is a no-op once the timeout has fired, confirming that postStop's cancel cannot abort an already-running attempt.

Reachability: the window is the duration of connectImplementation(). It is not vanishingly narrow — against a black-holed broker the TCP connect hangs for the OS timeout, and MqttActor's connect promise (MqttActor.ts:439-484) resolves only on the 'connect' event, so the whole handshake is inside the window. Once the timer callback starts, it reaches the connect await within the same macrotask, so the ordering that produces the bug is a plain interleaving, not an exotic one.

Test coverage: confirmed the two adjacent tests miss it. tests/integration/in-process/io/broker/BrokerActor.test.ts:268 stops a connected actor; the teardown test at :386 sets reconnect: { initialDelayMs: 10_000 } so the timer provably has not fired.

Corrections applied (see corrections): the KafkaActor evidence is factually wrong about when producer/consumer are assigned, which narrows the claimed Kafka consumer-group impact to the createKafkaInstance() await window; and the write-up under-states reachability, since the failure branch of the in-flight attempt (BrokerActor.ts:530-539) produces the endless zombie reconnect cycle without the broker ever coming back.

Severity: MEDIUM stands. No attacker is involved and there is no confidentiality or integrity impact; this is an availability/resource-lifecycle defect. But the leak is permanent for the process (nothing can ever call _closeTransport again after postStop), it accumulates with actor churn, and the trigger — stopping a broker actor while the broker is down — is ordinary operations. I did not inflate it to HIGH because it is not remotely triggerable and each occurrence costs one connection plus one timer loop, not unbounded amplification.

Correction applied: Two corrections to the write-up.

(1) The KafkaActor evidence is wrong. The finding states "this.consumer/this.producer are assigned only after connect() resolves". They are not: src/io/broker/KafkaActor.ts:228 assigns this.producer synchronously before await this.producer.connect() at :232, and :235 assigns this.consumer before await this.consumer.connect() at :236. The all-handles-null window for Kafka is therefore only await this.createKafkaInstance() at :227 (which includes the lazy import('kafkajs')). A stop landing after :228 does reach a real teardown in disconnectImplementation (:290-313), which also nulls this.kafka (:309); the resuming connectImplementation then throws a TypeError on this.kafka.consumer(...) at :235, which lands in the _tryConnect catch (BrokerActor.ts:530). So the specific claimed impact "a live consumer-group member keeps holding partition assignments and the handover never completes" is reachable only through the narrow :227 window, not generally. The MQTT half of the evidence is accurate as written (this.client is assigned only inside the 'connect' callback at MqttActor.ts:454, and disconnectImplementation at :487-489 short-circuits on null), so the escaped-live-connection outcome holds fully for MqttActor.

(2) The write-up under-states reachability. It presents the zombie state as requiring the broker to come back so connectImplementation() resolves. It does not: if the in-flight attempt rejects — the common case during an outage — the catch at BrokerActor.ts:530-539 calls _handleReconnect, which calls _scheduleReconnect and installs a fresh _scheduledReconnectCancel on an actor whose postStop already ran and nulled that field (:450). The endless background reconnect cycle on a terminated actor is therefore the dominant outcome of the race, independent of the success branch.

Also worth stating in the issue: the defect is in the base class, so it applies to every BrokerActor subclass (Amqp, Nats, JetStream, RedisStreams, Tcp, Udp, Sse, GrpcClient, Mqtt, Kafka), and it is scoped to a single-actor stop — ActorSystem.terminate() calls scheduler.shutdown() (ActorSystem.ts:393), after which Scheduler.scheduleOnceFunction's callback is suppressed (Scheduler.ts:50), so a full-system shutdown does not accumulate zombies beyond the one attempt already in flight.