diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 81f3098d..bf509e10 100644 --- a/swarm-queue-client/src/lib.rs +++ b/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 { +async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result { // 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 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 { // 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 { 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>> = + 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 { // 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