From 3493ba4a96da5e5e50c78ce896982efde0f81ffe Mon Sep 17 00:00:00 2001 From: atlas Date: Wed, 28 Oct 2026 02:02:41 +0100 Subject: [PATCH] count suspended time when ageing the queue client's cached token The cache aged the token on Instant, which does not advance while the host is suspended, so a resumed host kept presenting an expired token. Age it on the wall clock, and drop the token when the queue refuses it. --- swarm-queue-client/src/lib.rs | 269 +++++++++++++++++++++++++++++----- 1 file changed, 234 insertions(+), 35 deletions(-) diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 357c57eb..f7fe0ec6 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -26,7 +26,7 @@ //! that has expired since it was minted. //! //! ⚠️ That second point is also a trap that cost a production incident — -//! **anything in a retry path runs at the FAILURE rate.** See `CachedToken` +//! **anything in a retry path runs at the FAILURE rate.** See `TokenCache` //! and `MAX_RECONNECT_DELAY`. use std::path::PathBuf; @@ -496,17 +496,96 @@ struct TokenResponse { expires_in: u64, } -/// A minted token and the moment it stops being usable. +/// A minted token and how long the identity provider said it lives. +struct CachedToken { + token: String, + lifetime: std::time::Duration, +} + +/// The auth callback's token cache. /// -/// Exists because the auth callback runs **per connection attempt**, not per +/// Exists because the 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, +/// +/// A token's age is read off the wall clock, not `Instant`: `Instant` is +/// `CLOCK_MONOTONIC`, which stands still while the host is suspended, so a +/// resumed host would keep presenting a token the identity provider expired. +struct TokenCache { + slot: tokio::sync::Mutex>, + now: Box std::time::SystemTime + Send + Sync>, +} + +struct MintedToken { + token: CachedToken, + minted_at: std::time::SystemTime, +} + +impl MintedToken { + /// Whether more than [`TOKEN_REFRESH_SKEW`] of the token's life is left at + /// `now`. A clock reading earlier than the mint time is never trusted: + /// counting from it would stretch the token past its lifetime. + fn usable_at(&self, now: std::time::SystemTime) -> bool { + now.duration_since(self.minted_at) + .ok() + .and_then(|age| self.token.lifetime.checked_sub(age)) + .is_some_and(|left| left > TOKEN_REFRESH_SKEW) + } +} + +impl TokenCache { + fn new(now: impl Fn() -> std::time::SystemTime + Send + Sync + 'static) -> Self { + Self { + slot: tokio::sync::Mutex::new(None), + now: Box::new(now), + } + } + + /// The cached token while it is usable, otherwise one from `mint`, which + /// is then stored. The lock is held across `mint`, so concurrent callers + /// share one request. + async fn token(&self, mint: F) -> Result + where + F: FnOnce() -> Fut, + Fut: Future>, + { + let mut slot = self.slot.lock().await; + if let Some(minted) = slot.as_ref().filter(|m| m.usable_at((self.now)())) { + return Ok(minted.token.token.clone()); + } + // Stamped before the request, so the token's age is never understated. + let minted_at = (self.now)(); + let token = mint().await?; + let value = token.token.clone(); + *slot = Some(MintedToken { token, minted_at }); + Ok(value) + } + + /// Drop the cached token when `event` is the server refusing the + /// credential. Any other event, a plain disconnect included, keeps it: a + /// reconnect that is not an auth refusal must not mint, or the retry loop + /// becomes the token-request loop this cache exists to prevent. + async fn on_event(&self, event: &async_nats::Event) { + if refuses_credential(event) { + *self.slot.lock().await = None; + } + } +} + +/// Whether `event` is the queue refusing the presented token: at CONNECT +/// (surfaced as a client error carrying the connect error's text) or later on +/// an open connection. +fn refuses_credential(event: &async_nats::Event) -> bool { + match event { + async_nats::Event::ServerError(async_nats::ServerError::AuthorizationViolation) => true, + async_nats::Event::ClientError(async_nats::ClientError::Other(text)) => { + *text == async_nats::ConnectErrorKind::AuthorizationViolation.to_string() + } + _ => false, + } } /// Re-mint this long before expiry rather than at it. A token that expires @@ -701,7 +780,7 @@ fn parse_token_response(status: reqwest::StatusCode, body: &str) -> Result usize { } /// 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. Reply -/// inboxes are under [`client_inbox_prefix`] of `cfg.client_id`. +/// and minting a fresh one only when the cached one is near expiry or the queue +/// has refused it. Reply inboxes are under [`client_inbox_prefix`] of +/// `cfg.client_id`. /// /// The rejected alternative — mint once, pass a static `auth_token`, own the /// reconnect loop — fails in the way this subsystem exists to prevent: the @@ -959,42 +1039,25 @@ pub async fn connect(cfg: QueueConfig) -> Result { // 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 cache = std::sync::Arc::new(TokenCache::new(std::time::SystemTime::now)); + let events = cache.clone(); let mut options = 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; - // 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, None, None) - .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 token = cache + .token(|| mint_token(&http, &cfg, None, None)) + .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 auth = async_nats::Auth::new(); auth.token = Some(token); @@ -1006,6 +1069,12 @@ pub async fn connect(cfg: QueueConfig) -> Result { // the cache above, a permanent 4s token-request loop against authelia. // Exponential from 500ms so a momentary blip still reconnects promptly. .reconnect_delay_callback(reconnect_delay) + // A token the server refuses is dropped, so the next attempt, which waits + // out the backoff above first, mints once rather than presenting it again. + .event_callback(move |event| { + let cache = events.clone(); + async move { cache.on_event(&event).await } + }) // 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 @@ -1491,4 +1560,134 @@ mod tests { "the requested scope must reach the form body, got: {body}" ); } + + /// A settable wall clock, in seconds since the epoch. + #[derive(Clone, Default)] + struct FakeClock(Arc); + + impl FakeClock { + fn starting_at(secs: u64) -> Self { + Self(Arc::new(std::sync::atomic::AtomicU64::new(secs))) + } + + fn set(&self, secs: u64) { + self.0.store(secs, Ordering::SeqCst); + } + + fn advance(&self, secs: u64) { + self.0.fetch_add(secs, Ordering::SeqCst); + } + + fn cache(&self) -> TokenCache { + let secs = Arc::clone(&self.0); + TokenCache::new(move || { + std::time::UNIX_EPOCH + std::time::Duration::from_secs(secs.load(Ordering::SeqCst)) + }) + } + } + + /// The token `cache` hands out, counting a mint whenever it had to ask. + /// Each mint yields `token-` with the endpoint's one-hour lifetime. + async fn take(cache: &TokenCache, mints: &AtomicUsize) -> String { + cache + .token(|| async { + let n = mints.fetch_add(1, Ordering::SeqCst) + 1; + Ok(CachedToken { + token: format!("token-{n}"), + lifetime: std::time::Duration::from_secs(3599), + }) + }) + .await + .expect("the stub mint succeeds") + } + + const START: u64 = 1_700_000_000; + + #[tokio::test] + async fn a_token_inside_its_life_is_reused() { + let clock = FakeClock::starting_at(START); + let (cache, mints) = (clock.cache(), AtomicUsize::new(0)); + + assert_eq!(take(&cache, &mints).await, "token-1"); + clock.advance(60); + assert_eq!(take(&cache, &mints).await, "token-1"); + // Just outside the refresh skew still counts as inside the life. + clock.advance(3599 - 60 - 121); + assert_eq!(take(&cache, &mints).await, "token-1"); + assert_eq!(mints.load(Ordering::SeqCst), 1); + + // Presence control: inside the skew the same cache does ask again. + clock.advance(2); + assert_eq!(take(&cache, &mints).await, "token-2"); + } + + /// A suspend moves the wall clock and leaves the monotonic one behind. + #[tokio::test] + async fn a_token_is_not_reused_after_the_clock_jumps_past_its_life() { + let clock = FakeClock::starting_at(START); + let (cache, mints) = (clock.cache(), AtomicUsize::new(0)); + + assert_eq!(take(&cache, &mints).await, "token-1"); + clock.advance(2 * 3600); + assert_eq!(take(&cache, &mints).await, "token-2"); + } + + #[tokio::test] + async fn a_clock_stepping_backwards_does_not_extend_a_token() { + let clock = FakeClock::starting_at(START); + let (cache, mints) = (clock.cache(), AtomicUsize::new(0)); + + assert_eq!(take(&cache, &mints).await, "token-1"); + clock.set(START - 3600); + assert_eq!(take(&cache, &mints).await, "token-2"); + // The new token is aged from its own mint time. + clock.advance(3599 - 121); + assert_eq!(take(&cache, &mints).await, "token-2"); + clock.advance(2); + assert_eq!(take(&cache, &mints).await, "token-3"); + } + + fn client_error(text: &str) -> async_nats::Event { + async_nats::Event::ClientError(async_nats::ClientError::Other(text.to_owned())) + } + + #[tokio::test] + async fn an_auth_refusal_drops_the_cached_token() { + for refusal in [ + // What the connect loop reports for a CONNECT the server refused. + client_error("authorization violation"), + async_nats::Event::ServerError(async_nats::ServerError::AuthorizationViolation), + ] { + let clock = FakeClock::starting_at(START); + let (cache, mints) = (clock.cache(), AtomicUsize::new(0)); + + assert_eq!(take(&cache, &mints).await, "token-1"); + cache.on_event(&refusal).await; + assert_eq!(take(&cache, &mints).await, "token-2", "{refusal}"); + // One refusal costs one mint, not one per attempt. + assert_eq!(take(&cache, &mints).await, "token-2", "{refusal}"); + } + } + + #[tokio::test] + async fn events_that_are_not_an_auth_refusal_keep_the_cached_token() { + for event in [ + async_nats::Event::Disconnected, + async_nats::Event::Connected, + async_nats::Event::Closed, + async_nats::Event::ServerError(async_nats::ServerError::Other( + "Permissions Violation for Publish to \"x\"".to_owned(), + )), + client_error("IO error"), + client_error("timed out"), + ] { + let clock = FakeClock::starting_at(START); + let (cache, mints) = (clock.cache(), AtomicUsize::new(0)); + + assert_eq!(take(&cache, &mints).await, "token-1"); + cache.on_event(&event).await; + assert_eq!(take(&cache, &mints).await, "token-1", "{event}"); + assert_eq!(mints.load(Ordering::SeqCst), 1, "{event}"); + } + } }