swarm-nats-auth: re-read the OIDC client secret only on a 400 or 401

The responder re-read its secret on any non-2xx introspection answer.
With the re-read backoff added in the previous commit, an IdP outage
(503s, 502s from nginx, 429s) made each re-read find the unchanged
secret, so the delay grew to minutes. A real 401 after the rotation
then landed inside that window, and the responder denied every new
NATS connect until it ran out.

`Answer::Refused` now carries the status. `OidcSecret::introspect`
re-reads only when `stored_secret::is_credential_refusal` says the
status is 400 or 401, the same predicate `ClientSecret::refused` uses on
the client path. Every other non-2xx is a denial with no re-read.

The client-path test for a 429 now runs the mint against a loopback
endpoint answering 429 and 503, instead of calling `refused()` directly.
swarm-queue-client's README no longer names `ClientSecret::Value`.

Part of #5017 (slice Q).

swarm-nats-auth/src/oidc_secret.rs

@@ -4,18 +4,19 @@
 //!
 //! The secret can change under a running responder: authelia's host
 //! re-publishes it when it is re-minted. So when the `IdP` refuses an
-//! introspection (a non-2xx), the secret is read from the store again
-//! ([`StoredSecret::after_refusal`]) and, when it changed, the introspection
-//! is tried once more. Both fit in the one request's
+//! introspection as a credential (a 400 or 401), the secret is read from the
+//! store again ([`StoredSecret::after_refusal`]) and, when it changed, the
+//! introspection is tried once more. Both fit in the one request's
 //! [`INTROSPECTION_TIMEOUT`](crate::introspect::INTROSPECTION_TIMEOUT). A
 //! re-read that does not is left running: that request is denied, and the
-//! next one introspects with what the re-read landed.
+//! next one introspects with what the re-read landed. Any other non-2xx is
+//! denied without a re-read.
 
 use std::future::Future;
 use std::sync::Arc;
 
 use anyhow::{Context, Result};
-use swarm_queue_client::stored_secret::StoredSecret;
+use swarm_queue_client::stored_secret::{StoredSecret, is_credential_refusal};
 use swarm_secret_client::client::{DEFAULT_CERT_MOUNT, Settings};
 use swarm_secret_client::path::{Kind, principal_prefix};
 use tokio::time::Instant;
@@ -89,9 +90,10 @@ impl OidcSecret {
     }
 
     /// The caller `attempt` names, introspecting with the current secret.
-    /// When the `IdP` refuses that, the secret is re-read and, if it changed
-    /// and `deadline` has not passed, `attempt` runs once more with the new
-    /// one. `attempt` bounds itself by the same `deadline`.
+    /// When the `IdP` refuses that as a credential (a 400 or 401), the secret
+    /// is re-read and, if it changed and `deadline` has not passed, `attempt`
+    /// runs once more with the new one. Any other refusal is a denial.
+    /// `attempt` bounds itself by the same `deadline`.
     ///
     /// # Errors
     /// When an attempt could not be made at all; the caller denies.
@@ -101,8 +103,10 @@ impl OidcSecret {
         Fut: Future<Output = Result<Answer>>,
     {
         let used = self.secret.current();
-        if let Answer::Caller(caller) = attempt(used.clone()).await? {
-            return Ok(caller);
+        match attempt(used.clone()).await? {
+            Answer::Caller(caller) => return Ok(caller),
+            Answer::Refused(status) if !is_credential_refusal(status) => return Ok(None),
+            Answer::Refused(_) => {}
         }
         let Ok(fresh) = tokio::time::timeout_at(deadline, self.secret.after_refusal(&used)).await
         else {
@@ -117,7 +121,7 @@ impl OidcSecret {
         };
         match attempt(fresh).await? {
             Answer::Caller(caller) => Ok(caller),
-            Answer::Refused => Ok(None),
+            Answer::Refused(_) => Ok(None),
         }
     }
 }
@@ -161,7 +165,7 @@ mod tests {
             std::future::ready(Ok(if secret == registered {
                 Answer::Caller(Some("hive-alpha".to_owned()))
             } else {
-                Answer::Refused
+                Answer::Refused(reqwest::StatusCode::UNAUTHORIZED)
             }))
         }
     }
@@ -243,6 +247,41 @@ mod tests {
         assert_eq!(calls.load(Ordering::SeqCst), 1);
     }
 
+    /// An `IdP` outage answers 503s and 429s with the new secret already in
+    /// the store: none of them reads it, so none pushes the re-read back, and
+    /// the first 401 once the `IdP` is back re-reads and is admitted.
+    #[tokio::test]
+    async fn only_a_credential_refusal_re_reads_the_secret() {
+        use reqwest::StatusCode;
+
+        let (secret, reads) = secret("old", "new", Duration::ZERO);
+        let calls = AtomicUsize::new(0);
+        for status in [
+            StatusCode::SERVICE_UNAVAILABLE,
+            StatusCode::BAD_GATEWAY,
+            StatusCode::TOO_MANY_REQUESTS,
+        ] {
+            let caller = secret
+                .introspect(budget(), |_| {
+                    calls.fetch_add(1, Ordering::SeqCst);
+                    std::future::ready(Ok(Answer::Refused(status)))
+                })
+                .await
+                .expect("answered");
+            assert_eq!(caller, None, "{status}");
+        }
+        assert_eq!(calls.load(Ordering::SeqCst), 3);
+        assert_eq!(reads.load(Ordering::SeqCst), 0);
+
+        let caller = secret
+            .introspect(budget(), idp("new", &calls))
+            .await
+            .expect("answered");
+        assert_eq!(caller.as_deref(), Some("hive-alpha"));
+        assert_eq!(calls.load(Ordering::SeqCst), 5);
+        assert_eq!(reads.load(Ordering::SeqCst), 1);
+    }
+
     /// A store slower than the budget: the request is denied at its
     /// deadline, the re-read finishes behind it, and the next request is
     /// admitted on its first attempt without reading again.