swarm-queue-client: re-read a held OIDC client secret when the IdP refuses it

hive-c0re (`hive-<name>`), swarm-controller and swarm-review-classifier
read their queue client secret from the store once and held it as
`ClientSecret::Value` for the life of the process. After authelia
re-minted a secret and swarm-secret-publish stored the new one, every
later token mint presented the old value and was refused until the
daemon restarted.

`ClientSecret::Stored` replaces `Value`: a `StoredSecret` holds the
secret in memory with a caller-supplied read callback, so the crate
takes no store dependency. When the token endpoint answers 400 or 401,
`mint_token` (and its blocking sibling, used by the controller's OTLP
exporters) re-reads it and, if the store holds a different secret,
mints once more with it. One re-read runs at a time and is spawned so
it outlives a caller that gives up. A re-read that fails or finds the
refused value delays the next one, 5 s doubling to 5 min, so a secret
that stays wrong costs one store read per delay. Logs name the client
id and never the secret.

swarm-nats-auth's `oidc_secret` was the template; its re-read machinery
now sits on `StoredSecret`, so the tree has one copy. The responder
gains the same re-read backoff.

The controller keeps its `IDENTITY` OnceLock: it now holds a
`QueueConfig` whose secret is re-read, and the lazily built vcs-metrics
exporter reads the identity from it.

agent_renewal.rs's module doc said a re-minted agent queue secret only
reaches the agent at its next start; hive-agent reads the store on every
connection attempt after its first, so the doc now says that.

Part of #5017 (slice Q).

swarm-queue-client/src/lib.rs

@@ -402,6 +402,8 @@ pub mod subagent_term;
 /// same shapes.
 pub mod review_classifier;
 
+pub mod stored_secret;
+
 /// Only the fields this needs; authelia returns several.
 #[derive(serde::Deserialize)]
 struct TokenResponse {
@@ -489,28 +491,36 @@ pub struct QueueConfig {
     pub ca_file: Option<PathBuf>,
 }
 
-/// Where a [`QueueConfig`]'s client secret comes from.
-#[derive(Clone)]
+/// Where a [`QueueConfig`]'s client secret comes from. `Debug` never prints
+/// a held secret.
+#[derive(Clone, Debug)]
 pub enum ClientSecret {
     /// A file holding the plaintext, read at every token mint so a rotated
     /// file takes effect without a restart.
     File(PathBuf),
-    /// The plaintext itself, held only in this process's memory.
-    Value(String),
-}
-
-/// Redacts [`ClientSecret::Value`]: a [`QueueConfig`] is `Debug`, and printing
-/// one must not put the secret in a log.
-impl std::fmt::Debug for ClientSecret {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        match self {
-            Self::File(path) => f.debug_tuple("File").field(path).finish(),
-            Self::Value(_) => f.write_str("Value(<redacted>)"),
-        }
-    }
+    /// The plaintext, held only in this process's memory and read again
+    /// when the token endpoint refuses it.
+    Stored(stored_secret::StoredSecret),
 }
 
 impl ClientSecret {
+    /// The secret to read again after `error`: a held one the token endpoint
+    /// refused as a credential (400 or 401). A 429 or a 5xx says nothing
+    /// about the secret.
+    fn refused(&self, error: &Error) -> Option<&stored_secret::StoredSecret> {
+        match (self, error) {
+            (Self::Stored(stored), Error::TokenRefused { status, .. })
+                if matches!(
+                    *status,
+                    reqwest::StatusCode::BAD_REQUEST | reqwest::StatusCode::UNAUTHORIZED
+                ) =>
+            {
+                Some(stored)
+            }
+            _ => None,
+        }
+    }
+
     async fn read(&self) -> Result<String, Error> {
         match self {
             Self::File(path) => {
@@ -521,7 +531,7 @@ impl ClientSecret {
                         source,
                     })
             }
-            Self::Value(value) => Ok(value.clone()),
+            Self::Stored(stored) => Ok(stored.current()),
         }
     }
 
@@ -533,7 +543,7 @@ impl ClientSecret {
                     source,
                 })
             }
-            Self::Value(value) => Ok(value.clone()),
+            Self::Stored(stored) => Ok(stored.current()),
         }
     }
 }
@@ -545,6 +555,10 @@ impl ClientSecret {
 /// (a machine client receives an access token and never an id-token), so a
 /// caller passing `None` for `scope` requests none at all — see
 /// [`token_request`] for the one scope a caller does have to ask for.
+///
+/// A [`ClientSecret::Stored`] the endpoint refuses is read again, and when
+/// that lands a different secret the request is sent once more with it; see
+/// [`stored_secret`] for how often a re-read may run.
 async fn mint_token(
     http: &reqwest::Client,
     cfg: &QueueConfig,
@@ -554,7 +568,32 @@ async fn mint_token(
     // 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 = cfg.client_secret.read().await?;
+    let refused = match mint_with(http, cfg, &secret, audience, scope).await {
+        Err(e) => e,
+        minted => return minted,
+    };
+    let Some(stored) = cfg.client_secret.refused(&refused) else {
+        return Err(refused);
+    };
+    tracing::warn!(
+        client_id = %cfg.client_id,
+        error = %chain(&refused),
+        "the token endpoint refused this client's secret; re-reading it"
+    );
+    match stored.after_refusal(&secret).await {
+        Some(fresh) => mint_with(http, cfg, &fresh, audience, scope).await,
+        None => Err(refused),
+    }
+}
 
+/// One token request presenting `secret`.
+async fn mint_with(
+    http: &reqwest::Client,
+    cfg: &QueueConfig,
+    secret: &str,
+    audience: Option<&str>,
+    scope: Option<&str>,
+) -> Result<CachedToken, Error> {
     let response = token_request(http, cfg, secret.trim(), audience, scope)
         .send()
         .await
@@ -700,13 +739,46 @@ pub fn mint_token_for_blocking(
     audience: Option<&str>,
     scope: Option<&str>,
 ) -> Result<String, Error> {
-    let http = build_blocking_http_client(cfg)?;
+    mint_token_blocking(&build_blocking_http_client(cfg)?, cfg, audience, scope)
+}
 
+/// [`mint_token`] on the blocking client.
+fn mint_token_blocking(
+    http: &reqwest::blocking::Client,
+    cfg: &QueueConfig,
+    audience: Option<&str>,
+    scope: Option<&str>,
+) -> Result<String, Error> {
     // Read per call rather than caching — see `mint_token`'s identical
     // comment on the async path; the reasoning does not change with the
     // client type.
     let secret = cfg.client_secret.read_blocking()?;
+    let refused = match mint_with_blocking(http, cfg, &secret, audience, scope) {
+        Err(e) => e,
+        minted => return minted,
+    };
+    let Some(stored) = cfg.client_secret.refused(&refused) else {
+        return Err(refused);
+    };
+    tracing::warn!(
+        client_id = %cfg.client_id,
+        error = %chain(&refused),
+        "the token endpoint refused this client's secret; re-reading it"
+    );
+    match stored.after_refusal_blocking(&secret) {
+        Some(fresh) => mint_with_blocking(http, cfg, &fresh, audience, scope),
+        None => Err(refused),
+    }
+}
 
+/// [`mint_with`] on the blocking client.
+fn mint_with_blocking(
+    http: &reqwest::blocking::Client,
+    cfg: &QueueConfig,
+    secret: &str,
+    audience: Option<&str>,
+    scope: Option<&str>,
+) -> Result<String, Error> {
     let mut form = vec![("grant_type", "client_credentials")];
     if let Some(audience) = audience {
         form.push(("audience", audience));
@@ -947,6 +1019,9 @@ where
 
 #[cfg(test)]
 mod tests {
+    use std::sync::Arc;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
     use super::*;
 
     /// The names the publishers, the relays and the queue's grant all spell.
@@ -983,24 +1058,149 @@ mod tests {
             .with_no_client_auth();
     }
 
-    #[test]
-    fn a_held_secret_is_read_back_as_itself() {
-        let secret = ClientSecret::Value("s3cret".to_owned());
-        assert_eq!(
-            secret.read_blocking().expect("a value needs no I/O"),
-            "s3cret"
-        );
-    }
-
-    #[test]
-    fn a_debug_print_of_a_config_does_not_carry_a_held_secret() {
+    #[tokio::test]
+    async fn a_debug_print_of_a_config_does_not_carry_a_held_secret() {
         let cfg = QueueConfig {
-            client_secret: ClientSecret::Value("s3cret".to_owned()),
+            client_secret: ClientSecret::Stored(
+                stored_secret::tests::Fake::holding("new").secret("hive-alpha", "s3cret"),
+            ),
             ..token_cfg()
         };
         let printed = format!("{cfg:?}");
         assert!(!printed.contains("s3cret"), "{printed}");
-        assert!(printed.contains("<redacted>"), "{printed}");
+        assert!(printed.contains("StoredSecret"), "{printed}");
+    }
+
+    /// A token endpoint on a loopback port that mints for
+    /// `hive-alpha:{registered}` and answers any other secret with a 401,
+    /// counting its requests.
+    async fn stub_idp(registered: &'static str, calls: Arc<AtomicUsize>) -> String {
+        use axum::http::{HeaderMap, StatusCode, header::AUTHORIZATION};
+
+        let expected = format!(
+            "Basic {}",
+            data_encoding::BASE64.encode(format!("hive-alpha:{registered}").as_bytes())
+        );
+        let app = axum::Router::new().fallback(move |headers: HeaderMap| async move {
+            calls.fetch_add(1, Ordering::SeqCst);
+            if headers.get(AUTHORIZATION).and_then(|v| v.to_str().ok()) == Some(&*expected) {
+                (
+                    StatusCode::OK,
+                    r#"{"access_token":"a-token","expires_in":3599}"#,
+                )
+            } else {
+                (StatusCode::UNAUTHORIZED, r#"{"error":"invalid_client"}"#)
+            }
+        });
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("bind");
+        let url = format!("http://{}/token", listener.local_addr().expect("addr"));
+        tokio::spawn(async move { axum::serve(listener, app).await });
+        url
+    }
+
+    /// A plain-HTTP loopback client: no proxy, and no platform trust store,
+    /// which the nix build sandbox does not have (see [`offline_client`]).
+    fn loopback_client() -> reqwest::Client {
+        reqwest::Client::builder()
+            .no_proxy()
+            .danger_accept_invalid_certs(true)
+            .build()
+            .expect("a client that verifies nothing needs no system trust store")
+    }
+
+    /// The rotation the re-read exists for: the token endpoint refuses the
+    /// secret held since start, the store holds the new one, and the same
+    /// mint succeeds with it.
+    #[tokio::test]
+    async fn a_refused_secret_is_re_read_and_the_mint_retried_with_the_new_one() {
+        let calls = Arc::new(AtomicUsize::new(0));
+        let store = stored_secret::tests::Fake::holding("secret-b");
+        let cfg = QueueConfig {
+            token_endpoint: stub_idp("secret-b", Arc::clone(&calls)).await,
+            client_secret: ClientSecret::Stored(store.secret("hive-alpha", "secret-a")),
+            ..token_cfg()
+        };
+        let minted = mint_token(&loopback_client(), &cfg, None, None)
+            .await
+            .expect("minted with the re-read secret");
+        assert_eq!(minted.token, "a-token");
+        assert_eq!(calls.load(Ordering::SeqCst), 2);
+        assert_eq!(store.reads(), 1);
+    }
+
+    /// A secret the store still holds as refused: the mint fails, and the
+    /// mints after it inside the backoff ask the token endpoint once each
+    /// and read nothing.
+    #[tokio::test]
+    async fn a_permanently_refused_secret_is_not_re_read_on_every_mint() {
+        let calls = Arc::new(AtomicUsize::new(0));
+        let store = stored_secret::tests::Fake::holding("secret-a");
+        let cfg = QueueConfig {
+            token_endpoint: stub_idp("secret-b", Arc::clone(&calls)).await,
+            client_secret: ClientSecret::Stored(store.secret("hive-alpha", "secret-a")),
+            ..token_cfg()
+        };
+        let http = loopback_client();
+        for _ in 0..5 {
+            let err = mint_token(&http, &cfg, None, None)
+                .await
+                .err()
+                .expect("refused");
+            assert!(
+                matches!(err, Error::TokenRefused { status, .. } if status == 401),
+                "{err}"
+            );
+        }
+        assert_eq!(calls.load(Ordering::SeqCst), 5);
+        assert_eq!(store.reads(), 1);
+    }
+
+    /// A refusal that is not about the credential reads nothing.
+    #[tokio::test]
+    async fn a_rate_limited_mint_does_not_re_read() {
+        let store = stored_secret::tests::Fake::holding("secret-b");
+        let cfg = QueueConfig {
+            client_secret: ClientSecret::Stored(store.secret("hive-alpha", "secret-a")),
+            ..token_cfg()
+        };
+        let limited = Error::TokenRefused {
+            status: reqwest::StatusCode::TOO_MANY_REQUESTS,
+            body: String::new(),
+        };
+        assert!(cfg.client_secret.refused(&limited).is_none());
+        let refused = Error::TokenRefused {
+            status: reqwest::StatusCode::UNAUTHORIZED,
+            body: String::new(),
+        };
+        assert!(cfg.client_secret.refused(&refused).is_some());
+    }
+
+    /// The blocking mint re-reads the same way, from a thread with no
+    /// runtime of its own.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn the_blocking_mint_re_reads_a_refused_secret() {
+        let calls = Arc::new(AtomicUsize::new(0));
+        let store = stored_secret::tests::Fake::holding("secret-b");
+        let cfg = QueueConfig {
+            token_endpoint: stub_idp("secret-b", Arc::clone(&calls)).await,
+            client_secret: ClientSecret::Stored(store.secret("hive-alpha", "secret-a")),
+            ..token_cfg()
+        };
+        let token = tokio::task::spawn_blocking(move || {
+            let http = reqwest::blocking::Client::builder()
+                .no_proxy()
+                .danger_accept_invalid_certs(true)
+                .build()
+                .expect("client");
+            mint_token_blocking(&http, &cfg, None, None)
+        })
+        .await
+        .expect("joined")
+        .expect("minted with the re-read secret");
+        assert_eq!(token, "a-token");
+        assert_eq!(store.reads(), 1);
     }
 
     fn token_cfg() -> QueueConfig {

swarm-queue-client/src/stored_secret.rs

@@ -0,0 +1,459 @@
+//! A client secret held in memory and read again from where it is kept when
+//! the `IdP` refuses it.
+//!
+//! The secret can change under a running process: authelia's host
+//! re-publishes it when it is re-minted. A refusal asks for a re-read
+//! ([`StoredSecret::after_refusal`]); when the read lands a different secret,
+//! the caller tries once more with it. One re-read runs at a time, and a
+//! refusal while one is running waits for it rather than starting another.
+//! The re-read is spawned, so it finishes even when every caller waiting on
+//! it has given up, and the next refusal finds what it landed.
+//!
+//! A re-read that fails, or finds the secret the `IdP` refused, delays the
+//! next one: [`FIRST_REREAD_DELAY`], doubled per such re-read up to
+//! [`MAX_REREAD_DELAY`]. A refusal inside that delay reads nothing, so a
+//! secret that stays wrong costs the store one read per delay, not one per
+//! refusal.
+
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::{Arc, Mutex, RwLock};
+use std::time::Duration;
+
+use futures_util::FutureExt as _;
+use futures_util::future::{BoxFuture, Shared};
+use tokio::task::AbortHandle;
+use tokio::time::Instant;
+
+/// What a failed read of the secret reports. `anyhow::Error` converts into it.
+pub type ReadError = Box<dyn std::error::Error + Send + Sync>;
+
+/// The longest one re-read runs. A store client may set no timeout of its
+/// own, and while a re-read runs every refusal waits on it.
+pub const REREAD_TIMEOUT: Duration = Duration::from_secs(10);
+
+/// The wait before the next re-read after one that brought no new secret.
+pub const FIRST_REREAD_DELAY: Duration = Duration::from_secs(5);
+
+/// The longest wait between two re-reads.
+pub const MAX_REREAD_DELAY: Duration = Duration::from_mins(5);
+
+type Read =
+    dyn Fn() -> Pin<Box<dyn Future<Output = Result<String, ReadError>> + Send>> + Send + Sync;
+
+/// A re-read in progress, shared by every caller waiting on it.
+type Reread = Shared<BoxFuture<'static, ()>>;
+
+/// The secret in use and how to read it again. Clones share one secret: a
+/// re-read through any of them is seen by all. `Debug` names the client and
+/// never the secret.
+#[derive(Clone)]
+pub struct StoredSecret {
+    inner: Arc<Inner>,
+}
+
+struct Inner {
+    client_id: String,
+    read: Box<Read>,
+    runtime: tokio::runtime::Handle,
+    current: RwLock<String>,
+    rereads: Mutex<Rereads>,
+}
+
+#[derive(Default)]
+struct Rereads {
+    running: Option<(AbortHandle, Reread)>,
+    /// Re-reads in a row that brought no new secret.
+    fruitless: u32,
+    /// No re-read starts before this.
+    not_before: Option<Instant>,
+}
+
+impl std::fmt::Debug for StoredSecret {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("StoredSecret")
+            .field("client_id", &self.inner.client_id)
+            .finish_non_exhaustive()
+    }
+}
+
+impl StoredSecret {
+    /// `client_id`'s secret, `initial` until `read` replaces it. `client_id`
+    /// names the client in logs.
+    ///
+    /// # Panics
+    /// Outside a tokio runtime: re-reads are spawned on the one this is made
+    /// in.
+    pub fn new<F, Fut>(client_id: impl Into<String>, initial: String, read: F) -> Self
+    where
+        F: Fn() -> Fut + Send + Sync + 'static,
+        Fut: Future<Output = Result<String, ReadError>> + Send + 'static,
+    {
+        Self {
+            inner: Arc::new(Inner {
+                client_id: client_id.into(),
+                read: Box::new(move || Box::pin(read())),
+                runtime: tokio::runtime::Handle::current(),
+                current: RwLock::new(initial),
+                rereads: Mutex::default(),
+            }),
+        }
+    }
+
+    /// The secret in use.
+    #[must_use]
+    pub fn current(&self) -> String {
+        self.inner.current.read().expect("unpoisoned").clone()
+    }
+
+    /// After the `IdP` refused `refused`: the secret to try once more, or
+    /// `None` when there is no other one to try. Waits for a re-read, at most
+    /// [`REREAD_TIMEOUT`]; a caller with a shorter budget bounds this itself.
+    pub async fn after_refusal(&self, refused: &str) -> Option<String> {
+        let current = self.current();
+        if current != refused {
+            return Some(current);
+        }
+        let Some(reread) = self.reread() else {
+            tracing::debug!(
+                client_id = %self.inner.client_id,
+                "the IdP refused this client's secret inside the re-read backoff; not re-reading"
+            );
+            return None;
+        };
+        reread.await;
+        let fresh = self.current();
+        (fresh != refused).then_some(fresh)
+    }
+
+    /// [`after_refusal`](Self::after_refusal), for a thread with no tokio
+    /// runtime of its own.
+    ///
+    /// # Panics
+    /// Inside a tokio runtime, as [`tokio::runtime::Handle::block_on`] does.
+    #[must_use]
+    pub fn after_refusal_blocking(&self, refused: &str) -> Option<String> {
+        self.inner.runtime.block_on(self.after_refusal(refused))
+    }
+
+    /// The running re-read, or a new one when none is running and the
+    /// backoff allows it.
+    fn reread(&self) -> Option<Reread> {
+        let mut rereads = self.inner.rereads.lock().expect("unpoisoned");
+        if let Some((handle, running)) = rereads.running.as_ref()
+            && !handle.is_finished()
+        {
+            return Some(running.clone());
+        }
+        if rereads.not_before.is_some_and(|at| Instant::now() < at) {
+            return None;
+        }
+        let inner = Arc::clone(&self.inner);
+        let task = self
+            .inner
+            .runtime
+            .spawn(async move { inner.reread().await });
+        let handle = task.abort_handle();
+        let client_id = self.inner.client_id.clone();
+        let shared = task
+            .map(move |done| {
+                if let Err(e) = done {
+                    tracing::warn!(
+                        %client_id,
+                        error = %e,
+                        "the client secret re-read task died; keeping the one in use"
+                    );
+                }
+            })
+            .boxed()
+            .shared();
+        rereads.running = Some((handle, shared.clone()));
+        Some(shared)
+    }
+}
+
+impl Inner {
+    /// One read of the secret, kept when it differs from the one in use. A
+    /// read that brings no new secret pushes the next one back.
+    async fn reread(&self) {
+        let before = self.current.read().expect("unpoisoned").clone();
+        let outcome = tokio::time::timeout(REREAD_TIMEOUT, (self.read)()).await;
+        let mut rereads = self.rereads.lock().expect("unpoisoned");
+        let failure = match outcome {
+            Ok(Ok(secret)) if secret != before => {
+                *self.current.write().expect("unpoisoned") = secret;
+                rereads.fruitless = 0;
+                rereads.not_before = None;
+                tracing::info!(
+                    client_id = %self.client_id,
+                    "re-read a changed client secret after the IdP refused the one in use"
+                );
+                return;
+            }
+            Ok(Ok(_)) => "the stored client secret is the one the IdP refused".to_owned(),
+            Ok(Err(e)) => format!("could not re-read the client secret: {}", crate::chain(&*e)),
+            Err(_) => format!("re-reading the client secret timed out after {REREAD_TIMEOUT:?}"),
+        };
+        rereads.fruitless = rereads.fruitless.saturating_add(1);
+        let delay = delay_after(rereads.fruitless);
+        rereads.not_before = Some(Instant::now() + delay);
+        tracing::warn!(
+            client_id = %self.client_id,
+            next_reread_in = ?delay,
+            "{failure}; keeping the one in use"
+        );
+    }
+}
+
+/// [`FIRST_REREAD_DELAY`] doubled per fruitless re-read after the first,
+/// capped at [`MAX_REREAD_DELAY`].
+fn delay_after(fruitless: u32) -> Duration {
+    let doublings = fruitless.saturating_sub(1).min(16);
+    FIRST_REREAD_DELAY
+        .saturating_mul(1 << doublings)
+        .min(MAX_REREAD_DELAY)
+}
+
+#[cfg(test)]
+pub(crate) mod tests {
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    use super::*;
+
+    /// A store holding `secret`, or failing, that takes `delay` to answer.
+    pub(crate) struct Fake {
+        pub(crate) secret: Arc<Mutex<Option<&'static str>>>,
+        pub(crate) delay: Duration,
+        pub(crate) reads: Arc<AtomicUsize>,
+    }
+
+    impl Fake {
+        pub(crate) fn holding(secret: &'static str) -> Self {
+            Self {
+                secret: Arc::new(Mutex::new(Some(secret))),
+                delay: Duration::ZERO,
+                reads: Arc::default(),
+            }
+        }
+
+        /// `client_id`'s secret, `initial` in use, re-read from this store.
+        pub(crate) fn secret(&self, client_id: &str, initial: &str) -> StoredSecret {
+            let (secret, delay, reads) = (
+                Arc::clone(&self.secret),
+                self.delay,
+                Arc::clone(&self.reads),
+            );
+            StoredSecret::new(client_id, initial.to_owned(), move || {
+                let (secret, reads) = (Arc::clone(&secret), Arc::clone(&reads));
+                async move {
+                    reads.fetch_add(1, Ordering::SeqCst);
+                    tokio::time::sleep(delay).await;
+                    let held = *secret.lock().expect("unpoisoned");
+                    held.map(str::to_owned)
+                        .ok_or_else(|| "store unreachable".into())
+                }
+            })
+        }
+
+        pub(crate) fn reads(&self) -> usize {
+            self.reads.load(Ordering::SeqCst)
+        }
+    }
+
+    thread_local! {
+        static LOG: std::cell::RefCell<Vec<u8>> = const { std::cell::RefCell::new(Vec::new()) };
+    }
+
+    /// What `tracing` records on the calling thread. The subscriber is
+    /// process-wide, because a scoped one loses to the callsite cache when
+    /// another test reaches the same event first; the buffer is per thread,
+    /// which is where a current-thread test runtime runs the spawned re-read.
+    pub(crate) struct Logs;
+
+    impl std::io::Write for Logs {
+        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+            LOG.with(|log| log.borrow_mut().extend_from_slice(buf));
+            Ok(buf.len())
+        }
+
+        fn flush(&mut self) -> std::io::Result<()> {
+            Ok(())
+        }
+    }
+
+    impl Logs {
+        /// Starts this thread's capture empty.
+        pub(crate) fn capture() {
+            static INSTALLED: std::sync::Once = std::sync::Once::new();
+            INSTALLED.call_once(|| {
+                tracing_subscriber::fmt()
+                    .with_writer(|| Logs)
+                    .with_ansi(false)
+                    .init();
+            });
+            LOG.with(|log| log.borrow_mut().clear());
+        }
+
+        pub(crate) fn text() -> String {
+            LOG.with(|log| String::from_utf8(log.borrow().clone()).expect("utf-8"))
+        }
+    }
+
+    #[tokio::test]
+    async fn a_refused_secret_is_re_read_and_the_new_one_offered() {
+        Logs::capture();
+        let store = Fake::holding("new");
+        let secret = store.secret("hive-alpha", "old");
+        assert_eq!(secret.after_refusal("old").await.as_deref(), Some("new"));
+        assert_eq!(secret.current(), "new");
+        assert_eq!(store.reads(), 1);
+
+        let text = Logs::text();
+        assert!(text.contains("re-read a changed client secret"), "{text}");
+        assert!(text.contains("hive-alpha"), "{text}");
+        assert!(!text.contains("old") && !text.contains("new"), "{text}");
+    }
+
+    /// A refusal of a secret another caller already replaced offers the
+    /// replacement without reading again.
+    #[tokio::test]
+    async fn a_refusal_of_a_replaced_secret_offers_the_replacement() {
+        let store = Fake::holding("new");
+        let secret = store.secret("hive-alpha", "old");
+        assert_eq!(secret.after_refusal("old").await.as_deref(), Some("new"));
+        assert_eq!(secret.after_refusal("old").await.as_deref(), Some("new"));
+        assert_eq!(store.reads(), 1);
+    }
+
+    /// A secret that stays wrong: the first refusal reads, refusals inside
+    /// the backoff read nothing, and the backoff doubles per fruitless read.
+    #[tokio::test(start_paused = true)]
+    async fn a_permanently_refused_secret_backs_off() {
+        Logs::capture();
+        let store = Fake::holding("old");
+        let secret = store.secret("hive-alpha", "old");
+
+        assert_eq!(secret.after_refusal("old").await, None);
+        assert_eq!(store.reads(), 1);
+        let text = Logs::text();
+        assert!(text.contains("is the one the IdP refused"), "{text}");
+        assert!(text.contains("hive-alpha"), "{text}");
+        assert!(!text.contains("\"old\""), "{text}");
+
+        for _ in 0..10 {
+            assert_eq!(secret.after_refusal("old").await, None);
+        }
+        assert_eq!(store.reads(), 1);
+
+        tokio::time::sleep(FIRST_REREAD_DELAY).await;
+        assert_eq!(secret.after_refusal("old").await, None);
+        assert_eq!(store.reads(), 2);
+
+        tokio::time::sleep(FIRST_REREAD_DELAY).await;
+        assert_eq!(secret.after_refusal("old").await, None);
+        assert_eq!(
+            store.reads(),
+            2,
+            "the second fruitless read doubles the wait"
+        );
+
+        tokio::time::sleep(FIRST_REREAD_DELAY).await;
+        assert_eq!(secret.after_refusal("old").await, None);
+        assert_eq!(store.reads(), 3);
+    }
+
+    /// Once the store holds a new secret, the first refusal past the backoff
+    /// picks it up.
+    #[tokio::test(start_paused = true)]
+    async fn a_secret_published_during_the_backoff_is_read_after_it() {
+        let store = Fake::holding("old");
+        let secret = store.secret("hive-alpha", "old");
+        assert_eq!(secret.after_refusal("old").await, None);
+
+        *store.secret.lock().expect("unpoisoned") = Some("new");
+        assert_eq!(secret.after_refusal("old").await, None);
+        tokio::time::sleep(FIRST_REREAD_DELAY).await;
+        assert_eq!(secret.after_refusal("old").await.as_deref(), Some("new"));
+        assert_eq!(store.reads(), 2);
+    }
+
+    #[tokio::test]
+    async fn a_failed_re_read_keeps_the_secret_in_use() {
+        Logs::capture();
+        let store = Fake::holding("new");
+        *store.secret.lock().expect("unpoisoned") = None;
+        let secret = store.secret("hive-alpha", "old");
+        assert_eq!(secret.after_refusal("old").await, None);
+        assert_eq!(secret.current(), "old");
+
+        let text = Logs::text();
+        assert!(
+            text.contains("could not re-read the client secret"),
+            "{text}"
+        );
+        assert!(text.contains("store unreachable"), "{text}");
+        assert!(!text.contains("is the one the IdP refused"), "{text}");
+        assert!(!text.contains("old"), "no secret in the log: {text}");
+    }
+
+    /// A refusal while a re-read is still running waits on that one rather
+    /// than starting another.
+    #[tokio::test(start_paused = true)]
+    async fn a_refusal_during_a_running_re_read_joins_it() {
+        let mut store = Fake::holding("new");
+        store.delay = Duration::from_secs(2);
+        let secret = store.secret("hive-alpha", "old");
+        let (first, second) =
+            tokio::join!(secret.after_refusal("old"), secret.after_refusal("old"));
+        assert_eq!(first.as_deref(), Some("new"));
+        assert_eq!(second.as_deref(), Some("new"));
+        assert_eq!(store.reads(), 1);
+    }
+
+    /// A caller that gives up leaves the re-read running; the next refusal
+    /// finds what it landed without reading again.
+    #[tokio::test(start_paused = true)]
+    async fn a_re_read_outlives_a_caller_that_gave_up() {
+        let mut store = Fake::holding("new");
+        store.delay = Duration::from_secs(3);
+        let secret = store.secret("hive-alpha", "old");
+        let gave_up =
+            tokio::time::timeout(Duration::from_secs(1), secret.after_refusal("old")).await;
+        assert!(gave_up.is_err());
+
+        tokio::time::sleep(Duration::from_secs(3)).await;
+        assert_eq!(secret.current(), "new");
+        assert_eq!(secret.after_refusal("old").await.as_deref(), Some("new"));
+        assert_eq!(store.reads(), 1);
+    }
+
+    /// A re-read that stalls is given up at [`REREAD_TIMEOUT`], keeping the
+    /// secret in use, and counts as fruitless.
+    #[tokio::test(start_paused = true)]
+    async fn a_stalled_re_read_is_dropped_at_its_timeout() {
+        let mut store = Fake::holding("new");
+        store.delay = REREAD_TIMEOUT * 6;
+        let secret = store.secret("hive-alpha", "old");
+        assert_eq!(secret.after_refusal("old").await, None);
+        assert_eq!(secret.current(), "old");
+
+        tokio::time::sleep(FIRST_REREAD_DELAY).await;
+        assert_eq!(secret.after_refusal("old").await, None);
+        assert_eq!(store.reads(), 2);
+    }
+
+    #[test]
+    fn the_backoff_doubles_up_to_its_cap() {
+        assert_eq!(delay_after(1), FIRST_REREAD_DELAY);
+        assert_eq!(delay_after(2), FIRST_REREAD_DELAY * 2);
+        assert_eq!(delay_after(40), MAX_REREAD_DELAY);
+    }
+
+    #[tokio::test]
+    async fn a_debug_print_does_not_carry_the_secret() {
+        let secret = Fake::holding("new").secret("hive-alpha", "s3cret");
+        let printed = format!("{secret:?}");
+        assert!(!printed.contains("s3cret"), "{printed}");
+        assert!(printed.contains("hive-alpha"), "{printed}");
+    }
+}