fix(#3363): cache the queue token and stop retrying at four seconds forever

The auth callback minted on every invocation, and async-nats runs it per
connection ATTEMPT. Its default reconnect delay backs off exponentially
and then clamps at four seconds, permanently. So a queue that could not
connect asked authelia for a token every four seconds indefinitely, and
authelia answered 429 Too Many Requests.

That rate limit then keeps itself alive: it outlives whatever first
broke the connection, and its log volume buries the original cause.
Observed in production tonight, once a redeploy fixed the TLS trust
problem that had been hiding it.

The callback now caches the token and re-mints only within two minutes
of expiry, which preserves the property that put the mint there — a
reconnect must never present a token that expired since it was minted —
while decoupling mint rate from retry rate. The reconnect cap moves to
one minute, exponential from half a second, so a brief outage still
recovers promptly and a persistent one stops hammering the identity
provider.

The module doc and the retry comment both argued for the old shape, so
both are corrected rather than left explaining why the bug was right.

swarm-queue-client/src/lib.rs

@@ -21,13 +21,12 @@
 //!   token that was minted an hour later.
 //! - **`async-nats` re-runs an auth callback per connection attempt** (it is
 //!   handed that attempt's nonce). So the refresh belongs in the callback and
-//!   not in a timer: there is no window in which the client holds a token it
-//!   minted for a previous connection.
+//!   not in a timer: there is no window in which the client presents a token
+//!   that has expired since it was minted.
 //!
-//! The alternative — mint once, pass a static `auth_token`, own the reconnect
-//! loop — fails in the way this subsystem exists to prevent: the controller
-//! keeps serving, its status data quietly stops updating, and nothing says so
-//! until someone reads a dashboard.
+//! ⚠️ That second point is also a trap that cost a production incident —
+//! **anything in a retry path runs at the FAILURE rate.** See `CachedToken`
+//! and `MAX_RECONNECT_DELAY`.
 
 use std::path::PathBuf;
 
@@ -150,12 +149,53 @@ pub fn chain(error: &dyn std::error::Error) -> String {
 #[cfg(feature = "kv")]
 pub mod status;
 
-/// Only the one field this needs; authelia returns several.
+/// Only the fields this needs; authelia returns several.
 #[derive(serde::Deserialize)]
 struct TokenResponse {
     access_token: String,
+    /// Seconds the token stays valid — authelia says 3599 for
+    /// `client_credentials`. Read rather than assumed, because it is what
+    /// decides when the cache below stops handing the same token back, and a
+    /// hardcoded hour would silently become wrong the day the identity
+    /// provider is retuned.
+    expires_in: u64,
 }
 
+/// A minted token and the moment it stops being usable.
+///
+/// Exists because the auth callback runs **per connection attempt**, not per
+/// hour: without a cache, every reconnect mints, and a queue that cannot
+/// connect turns its retry loop into a token-request loop against the
+/// identity provider. That is not hypothetical — the first version of this
+/// crate did exactly that, and authelia answered `429 Too Many Requests`
+/// continuously until the client was changed.
+struct CachedToken {
+    token: String,
+    expires_at: std::time::Instant,
+}
+
+/// Re-mint this long before expiry rather than at it. A token that expires
+/// mid-CONNECT is refused by the server, which costs a whole reconnect cycle
+/// to discover — far more than minting slightly early costs.
+const TOKEN_REFRESH_SKEW: std::time::Duration = std::time::Duration::from_mins(2);
+
+/// Cap on how long `async-nats` waits between reconnect attempts.
+///
+/// ⚠️ Deliberately far above the library's default, and the default is the
+/// bug: `reconnect_delay_callback_default` backs off exponentially but clamps
+/// at **4 seconds** (`connector.rs`), forever. Every attempt runs the auth
+/// callback, so an unreachable queue means one token request every 4s
+/// indefinitely — measured in production as a continuous
+/// `429 Too Many Requests / temporarily_unavailable` from authelia, which then
+/// keeps *itself* alive: the rate limit outlives whatever first broke the
+/// connection, and its log volume buries the original cause.
+///
+/// The cost of raising it is slower recovery from a brief outage. That is the
+/// right trade for an infrastructure service: a minute of staleness is a
+/// dashboard reading `stale`, whereas a hot loop against the identity provider
+/// degrades every other client of it.
+const MAX_RECONNECT_DELAY: std::time::Duration = std::time::Duration::from_mins(1);
+
 /// Where the controller finds the queue and what it authenticates with.
 ///
 /// Every field comes from an environment variable the NixOS module sets, the
@@ -246,7 +286,7 @@ impl QueueConfig {
 /// authenticates as itself. Authelia refuses the `openid` scope for this grant
 /// (a machine client receives an access token and never an id-token), so no
 /// scope is requested.
-async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result<String, Error> {
+async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result<CachedToken, Error> {
     // Read per call rather than caching: the file is small, and a cached
     // secret would survive a rotation that the operator believes took effect.
     let secret = tokio::fs::read_to_string(&cfg.client_secret_file)
@@ -275,7 +315,10 @@ async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result<String,
     }
 
     let parsed: TokenResponse = serde_json::from_str(&body).map_err(Error::TokenResponse)?;
-    Ok(parsed.access_token)
+    Ok(CachedToken {
+        token: parsed.access_token,
+        expires_at: std::time::Instant::now() + std::time::Duration::from_secs(parsed.expires_in),
+    })
 }
 
 /// Fail fast unless the client is actually connected.
@@ -304,7 +347,13 @@ pub fn ensure_connected(client: &async_nats::Client) -> Result<(), Error> {
     Ok(())
 }
 
-/// Connect to the swarm queue, minting a token for each connection attempt.
+/// Connect to the swarm queue, presenting a token on each connection attempt
+/// and minting a fresh one only when the cached one is near expiry.
+///
+/// The rejected alternative — mint once, pass a static `auth_token`, own the
+/// reconnect loop — fails in the way this subsystem exists to prevent: the
+/// token expires, the controller keeps serving, its status data quietly stops
+/// updating, and nothing says so until someone reads a dashboard.
 pub async fn connect(cfg: QueueConfig) -> Result<async_nats::Client, Error> {
     // A timeout, because this client runs INSIDE the auth callback: a token
     // endpoint that accepts the connection and then never answers would hang
@@ -338,20 +387,65 @@ pub async fn connect(cfg: QueueConfig) -> Result<async_nats::Client, Error> {
     let http = builder.build().map_err(Error::HttpClient)?;
     let url = cfg.url.clone();
 
+    // Shared across every invocation of the callback below, which is the
+    // whole point: the callback fires per connection ATTEMPT, so without
+    // somewhere to remember the last token, "mint on demand" and "mint on
+    // every retry" are the same code.
+    let cache: std::sync::Arc<tokio::sync::Mutex<Option<CachedToken>>> =
+        std::sync::Arc::new(tokio::sync::Mutex::new(None));
+
     let client = async_nats::ConnectOptions::with_auth_callback(move |_nonce| {
         let http = http.clone();
         let cfg = cfg.clone();
+        let cache = cache.clone();
         async move {
-            let token = mint_token(&http, &cfg)
-                .await
-                // The callback's error type carries a string, so the source
-                // chain would be lost; flatten it rather than dropping it.
-                .map_err(|e| async_nats::AuthError::new(chain(&e)))?;
+            let mut slot = cache.lock().await;
+
+            // Reuse while there is comfortably more than the skew left. The
+            // freshness property this callback exists for is preserved — a
+            // reconnect an hour later finds an expired entry and re-mints —
+            // but a reconnect ten seconds later does NOT ask the IdP again.
+            let reuse = slot
+                .as_ref()
+                .filter(|t| {
+                    t.expires_at
+                        .saturating_duration_since(std::time::Instant::now())
+                        > TOKEN_REFRESH_SKEW
+                })
+                .map(|t| t.token.clone());
+
+            let token = if let Some(token) = reuse {
+                token
+            } else {
+                let minted = mint_token(&http, &cfg)
+                    .await
+                    // The callback's error type carries a string, so the
+                    // source chain would be lost; flatten it rather than
+                    // dropping it.
+                    .map_err(|e| async_nats::AuthError::new(chain(&e)))?;
+                let token = minted.token.clone();
+                *slot = Some(minted);
+                token
+            };
+
             let mut auth = async_nats::Auth::new();
             auth.token = Some(token);
             Ok(auth)
         }
     })
+    // See `MAX_RECONNECT_DELAY`. The library's default clamps at 4s forever,
+    // which turns an unreachable queue into a permanent 4s poll — and, before
+    // the cache above, a permanent 4s token-request loop against authelia.
+    // Exponential from ~1s so a momentary blip still reconnects promptly.
+    .reconnect_delay_callback(|attempts| {
+        let exp = u32::try_from(attempts.saturating_sub(1)).unwrap_or(u32::MAX);
+        std::cmp::min(
+            std::time::Duration::from_millis(
+                500u64.saturating_mul(2u64.saturating_pow(exp.min(8))),
+            ),
+            MAX_RECONNECT_DELAY,
+        )
+    })
     // The controller and the queue are separate units on (possibly)
     // separate hosts, and nothing orders them. Without this, a queue that
     // comes up one second later leaves the controller permanently
@@ -359,8 +453,11 @@ pub async fn connect(cfg: QueueConfig) -> Result<async_nats::Client, Error> {
     // presents as "status has been unavailable since Tuesday".
     //
     // It also composes with the callback above rather than fighting it:
-    // each background attempt is a connection attempt, so each one mints
-    // its own token instead of retrying a stale one.
+    // each background attempt runs the callback, which hands over a token
+    // that is still valid — minting a new one only when the cached one is
+    // near expiry. (Before the cache, "runs the callback" meant "mints",
+    // and this retry loop was a token-request loop. See
+    // `MAX_RECONNECT_DELAY`.)
     .retry_on_initial_connect()
     .connect(&url)
     .await