swarm-nats-auth: bound the OIDC secret re-read, and test the 401 that starts it

The re-read of the responder's own OIDC client secret had no timeout:
neither swarm-secret-client nor vaultrs nor reqwest sets one by default,
and while a re-read runs every later refusal joins it rather than
starting another. A store connection whose peer stays alive but silent
therefore held off rotation pickup indefinitely. The spawned re-read
now gives up after 10 s, logs that it did (without the secret), keeps
the secret in use, and frees the slot for the next refusal's re-read.

The link from a non-2xx introspection (identify_caller's
`Answer::Refused`) to that re-read had no test: every existing test
handed `OidcSecret::introspect` an `Answer` directly. A new test runs
the real `identify_caller` against a loopback IdP that answers 401 to
the old secret and admits the new one, so mapping a non-2xx to
anything but `Refused` turns it red. axum is a dev-dependency for that
stub, as in swarm-controller's forge tests.

Cargo.lock

@@ -4895,6 +4895,7 @@ version = "0.1.0"
 dependencies = [
  "anyhow",
  "async-nats",
+ "axum",
  "clap",
  "data-encoding",
  "futures-util",

swarm-nats-auth/Cargo.toml

@@ -39,6 +39,9 @@ tracing.workspace = true
 tracing-subscriber.workspace = true
 
 [dev-dependencies]
+# A loopback stand-in `IdP` for `oidc_secret`'s test of a non-2xx
+# introspection starting the re-read.
+axum.workspace = true
 # A TEST ORACLE, not part of the production path. Neither JWT this crate emits
 # is expressible through it - `Claims` has no `aud`, which the response wrapper
 # needs (the server id) and the user token needs (the account name), and

swarm-nats-auth/src/oidc_secret.rs

@@ -14,6 +14,7 @@
 
 use std::future::Future;
 use std::sync::{Arc, Mutex, RwLock};
+use std::time::Duration;
 
 use anyhow::{Context, Result};
 use futures_util::FutureExt;
@@ -92,6 +93,11 @@ impl Source for Store {
     }
 }
 
+/// The longest one re-read runs. The store client sets no timeout of its
+/// own, and while a re-read runs every refusal waits on it instead of
+/// starting another.
+const REREAD_TIMEOUT: Duration = Duration::from_secs(10);
+
 /// A re-read in progress, shared by every request waiting on it.
 type Reread = Shared<BoxFuture<'static, ()>>;
 
@@ -167,12 +173,16 @@ impl<S: Source> OidcSecret<S> {
         let source = Arc::clone(&self.source);
         let current = Arc::clone(&self.current);
         let task = tokio::spawn(async move {
-            match source.read().await {
-                Ok(secret) => *current.write().expect("unpoisoned") = secret,
-                Err(e) => tracing::warn!(
+            match tokio::time::timeout(REREAD_TIMEOUT, source.read()).await {
+                Ok(Ok(secret)) => *current.write().expect("unpoisoned") = secret,
+                Ok(Err(e)) => tracing::warn!(
                     error = format!("{e:#}"),
                     "could not re-read the OIDC client secret; keeping the one in use"
                 ),
+                Err(_) => tracing::warn!(
+                    timeout = ?REREAD_TIMEOUT,
+                    "re-reading the OIDC client secret timed out; keeping the one in use"
+                ),
             }
         });
         let handle = task.abort_handle();
@@ -394,4 +404,97 @@ mod tests {
         assert_eq!(calls.load(Ordering::SeqCst), 3);
         assert_eq!(reads.load(Ordering::SeqCst), 1);
     }
+
+    /// A re-read that stalls is given up at [`REREAD_TIMEOUT`]: the secret in
+    /// use stays, and the next refusal starts a fresh re-read rather than
+    /// waiting on the stalled one.
+    #[tokio::test(start_paused = true)]
+    async fn a_stalled_re_read_is_dropped_at_its_timeout() {
+        let store = Fake {
+            secret: Some("new"),
+            delay: REREAD_TIMEOUT * 6,
+            reads: Arc::default(),
+        };
+        let reads = Arc::clone(&store.reads);
+        let secret = OidcSecret::new(store, "old".to_owned());
+        let calls = AtomicUsize::new(0);
+
+        let first = secret
+            .introspect(budget(), idp("new", &calls))
+            .await
+            .expect("answered");
+        assert_eq!(first, None);
+
+        tokio::time::sleep(REREAD_TIMEOUT).await;
+        assert_eq!(secret.current(), "old");
+        let second = secret
+            .introspect(budget(), idp("new", &calls))
+            .await
+            .expect("answered");
+        assert_eq!(second, None);
+        assert_eq!(reads.load(Ordering::SeqCst), 2);
+    }
+
+    /// An `IdP` on a loopback port that admits as `hive-alpha` a request
+    /// whose basic auth is `swarm-nats:{registered}`, and answers any other
+    /// with a 401.
+    async fn stub_idp(registered: &str) -> String {
+        use axum::http::{HeaderMap, StatusCode, header::AUTHORIZATION};
+
+        let expected = format!(
+            "Basic {}",
+            data_encoding::BASE64.encode(format!("swarm-nats:{registered}").as_bytes())
+        );
+        let app = axum::Router::new().fallback(move |headers: HeaderMap| async move {
+            if headers.get(AUTHORIZATION).and_then(|v| v.to_str().ok()) == Some(&*expected) {
+                (
+                    StatusCode::OK,
+                    r#"{"active":true,"client_id":"hive-alpha"}"#,
+                )
+            } else {
+                (StatusCode::UNAUTHORIZED, "")
+            }
+        });
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("bind");
+        let url = format!("http://{}/introspect", listener.local_addr().expect("addr"));
+        tokio::spawn(async move { axum::serve(listener, app).await });
+        url
+    }
+
+    /// The `IdP`'s 401 to the secret held since start is what starts the
+    /// re-read: driven through the real HTTP introspection, not an [`Answer`]
+    /// handed in.
+    #[tokio::test]
+    async fn an_idp_401_re_reads_the_secret_and_admits_with_the_new_one() {
+        let url = stub_idp("new").await;
+        let http = reqwest::Client::builder()
+            .no_proxy()
+            .build()
+            .expect("client");
+        let store = Fake::holding("new");
+        let reads = Arc::clone(&store.reads);
+        let secret = OidcSecret::new(store, "old".to_owned());
+        let deadline = budget();
+        let caller = secret
+            .introspect(deadline, |own| {
+                let (http, url) = (&http, &url);
+                async move {
+                    crate::introspect::identify_caller(
+                        http,
+                        url,
+                        "swarm-nats",
+                        &own,
+                        "a-token",
+                        deadline,
+                    )
+                    .await
+                }
+            })
+            .await
+            .expect("answered");
+        assert_eq!(caller.as_deref(), Some("hive-alpha"));
+        assert_eq!(reads.load(Ordering::SeqCst), 1);
+    }
 }