swarm-queue-client/src/lib.rs

@@ -21,12 +21,13 @@
 //!   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 presents a token
+//!   not in a timer: there is no window in which the client holds a token it
-//!   that has expired since it was minted.
+//!   minted for a previous connection.
 //!
-//! ⚠️ That second point is also a trap that cost a production incident —
+//! The alternative — mint once, pass a static `auth_token`, own the reconnect
-//! **anything in a retry path runs at the FAILURE rate.** See `CachedToken`
+//! loop — fails in the way this subsystem exists to prevent: the controller
-//! and `MAX_RECONNECT_DELAY`.
+//! keeps serving, its status data quietly stops updating, and nothing says so
+//! until someone reads a dashboard.
 
 use std::path::PathBuf;
 
@@ -149,53 +150,12 @@ pub fn chain(error: &dyn std::error::Error) -> String {
 #[cfg(feature = "kv")]
 pub mod status;
 
-/// Only the fields this needs; authelia returns several.
+/// Only the one field 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
@@ -286,7 +246,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<CachedToken, Error> {
+async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result<String, 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)
@@ -315,10 +275,7 @@ async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result<CachedT
     }
 
     let parsed: TokenResponse = serde_json::from_str(&body).map_err(Error::TokenResponse)?;
-    Ok(CachedToken {
+    Ok(parsed.access_token)
-        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.
@@ -347,13 +304,7 @@ pub fn ensure_connected(client: &async_nats::Client) -> Result<(), Error> {
     Ok(())
 }
 
-/// Connect to the swarm queue, presenting a token on each connection attempt
+/// Connect to the swarm queue, minting a token for 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
@@ -387,65 +338,20 @@ 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 mut slot = cache.lock().await;
+            let token = mint_token(&http, &cfg)
-
+                .await
-            // Reuse while there is comfortably more than the skew left. The
+                // The callback's error type carries a string, so the source
-            // freshness property this callback exists for is preserved — a
+                // chain would be lost; flatten it rather than dropping it.
-            // reconnect an hour later finds an expired entry and re-mints —
+                .map_err(|e| async_nats::AuthError::new(chain(&e)))?;
-            // 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 500ms 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
@@ -453,11 +359,8 @@ 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 runs the callback, which hands over a token
+    // each background attempt is a connection attempt, so each one mints
-    // that is still valid — minting a new one only when the cached one is
+    // its own token instead of retrying a stale one.
-    // 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