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

@@ -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,
+/// A token's age is read off the wall clock, not `Instant`: `Instant` is
-    expires_at: std::time::Instant,
+/// `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<Option<MintedToken>>,
+    now: Box<dyn Fn() -> 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<F, Fut>(&self, mint: F) -> Result<String, Error>
+    where
+        F: FnOnce() -> Fut,
+        Fut: Future<Output = Result<CachedToken, Error>>,
+    {
+        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<Cache
     let parsed: TokenResponse = serde_json::from_str(body).map_err(Error::TokenResponse)?;
     Ok(CachedToken {
         token: parsed.access_token,
-        expires_at: std::time::Instant::now() + std::time::Duration::from_secs(parsed.expires_in),
+        lifetime: std::time::Duration::from_secs(parsed.expires_in),
     })
 }
 
@@ -934,8 +1013,9 @@ pub fn max_payload(client: &async_nats::Client) -> 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
+/// and minting a fresh one only when the cached one is near expiry or the queue
-/// inboxes are under [`client_inbox_prefix`] of `cfg.client_id`.
+/// 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<async_nats::Client, Error> {
     // 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>>> =
+    let cache = std::sync::Arc::new(TokenCache::new(std::time::SystemTime::now));
-        std::sync::Arc::new(tokio::sync::Mutex::new(None));
+    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
+            let token = cache
-                .as_ref()
+                .token(|| mint_token(&http, &cfg, None, None))
-                .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 mut auth = async_nats::Auth::new();
             auth.token = Some(token);
@@ -1006,6 +1069,12 @@ pub async fn connect(cfg: QueueConfig) -> Result<async_nats::Client, Error> {
     // 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<std::sync::atomic::AtomicU64>);
+
+    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-<n>` 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}");
+        }
+    }
 }