refactor(swarm-queue-client): share the connected-client precondition
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.
This commit is contained in:
parent
9c1cfafeb5
commit
e23a70e488
2 changed files with 31 additions and 19 deletions
|
|
@ -224,6 +224,32 @@ async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result<String,
|
|||
Ok(parsed.access_token)
|
||||
}
|
||||
|
||||
/// Fail fast unless the client is actually connected.
|
||||
///
|
||||
/// **Call this before every `JetStream` request.** `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*, for longer than any caller
|
||||
/// should (measured: still going at 15s against a queue that simply refuses
|
||||
/// the credential). On a request path that hangs a poll; on a periodic task it
|
||||
/// hangs the task, including the shutdown branch it never reaches.
|
||||
///
|
||||
/// 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 that matters — a process 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.
|
||||
pub fn ensure_connected(client: &async_nats::Client) -> 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<async_nats::Client, Error> {
|
||||
// A timeout, because this client runs INSIDE the auth callback: a token
|
||||
|
|
|
|||
Loading…
Reference in a new issue