From e23a70e488b2d5addcae373d60048318e1c00a6a Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 15 Aug 2026 22:34:04 +0200 Subject: [PATCH] refactor(swarm-queue-client): share the connected-client precondition MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An unconnected client does not fail a JetStream request, it hangs on it: `retry_on_initial_connect` hands back a client before it is usable, and a request made in that window waits (measured: still going at 15s against a queue that refuses the credential). The controller guarded its read path against that inline. Every consumer of the queue needs the same guard, so it is not one daemon's to keep. It matters more off a request path than on one. A hung request inside a periodic task never reaches its `select!`, so the shutdown branch becomes unreachable and the task cannot be stopped at all — where a request path merely times a poll out. The test is `!= Connected`, never `== Disconnected`: a client that has never connected sits in `Pending`, so the `Disconnected` form passes it straight through to the hang it was written to prevent — which is exactly the boot-order case the guard exists for. Not feature-gated; `connection_state()` is core async-nats. --- swarm-controller/src/status.rs | 24 +++++------------------- swarm-queue-client/src/lib.rs | 26 ++++++++++++++++++++++++++ 2 files changed, 31 insertions(+), 19 deletions(-) diff --git a/swarm-controller/src/status.rs b/swarm-controller/src/status.rs index 2911752c..3b1cd1b1 100644 --- a/swarm-controller/src/status.rs +++ b/swarm-controller/src/status.rs @@ -166,25 +166,11 @@ impl StatusReader { /// Every roster hive produces a row whether or not it has ever /// reported; a reporting hive outside the roster produces one too. pub async fn view(&self, roster: &[HiveEntry], now: SystemTime) -> Result> { - // Only a CONNECTED client can be asked anything. `retry_on_initial_connect` - // means the client exists before it is usable, and a JetStream request - // made in that window does not fail — it WAITS, on every call, for - // longer than any dashboard poll should take (measured: still going at - // 15s against a queue that simply refuses the credential). - // - // Testing for `!= Connected` rather than `== Disconnected` is the whole - // point: a client that has never connected once sits in `Pending`, so - // the `Disconnected` test passes it straight through to the hang it was - // written to prevent. That is exactly the case here — a controller - // whose credential is wrong from boot never reaches `Disconnected`, - // because it was never connected to begin with. - // - // Naming the state is also the better error: "not connected" is - // actionable, a timeout is not. - let state = self.client.connection_state(); - if state != async_nats::connection::State::Connected { - anyhow::bail!("not connected to the swarm queue (client state: {state:?})"); - } + // Before anything JetStream: an unconnected client does not fail a + // request, it hangs on it. The rule and the measurement behind it live + // in `swarm_queue_client::ensure_connected` — every consumer of the + // queue needs it, so it is not this daemon's to keep. + swarm_queue_client::ensure_connected(&self.client)?; let store = self.store().await?; diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 16f6f51a..d9e00fb7 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -224,6 +224,32 @@ async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result Result<()> { + let state = client.connection_state(); + if state != async_nats::connection::State::Connected { + bail!("not connected to the swarm queue (client state: {state:?})"); + } + Ok(()) +} + /// Connect to the swarm queue, minting a token for each connection attempt. pub async fn connect(cfg: QueueConfig) -> Result { // A timeout, because this client runs INSIDE the auth callback: a token