bridge waits out a slow authelia; controller retries a bridge 503

swarm-authelia-bridge/src/introspect.rs

@@ -27,11 +27,15 @@ use std::time::Duration;
 
 use serde::Deserialize;
 
-/// Generous relative to `swarm-nats-auth`'s 1.5s (that one is bounded by
+/// One attempt, no retry. authelia's introspection endpoint takes up to
-/// the NATS server's own 2s `auth_callout` timeout; an HTTP request here
+/// ~19s when its host is overloaded, and this covers that. It must stay
-/// has no such external deadline to stay under), but still bounded — a
+/// below the controller's 30s bound on a whole bridge call
-/// hung introspection call must not wedge a `CreateIdentity` job forever.
+/// (`swarm-controller`'s `auth::HTTP_TIMEOUT`), or the controller gives up
-pub const INTROSPECTION_TIMEOUT: Duration = Duration::from_secs(5);
+/// on the bridge before the bridge has answered. A timeout is still
+/// [`Verdict::Unavailable`], never an admission.
+pub const INTROSPECTION_TIMEOUT: Duration = Duration::from_secs(20);
+
+const _: () = assert!(INTROSPECTION_TIMEOUT.as_secs() < 30);
 
 /// `active` decides whether to admit; `client_id` says as whom, which is
 /// what lets one verb be scoped to a single client. Same two fields, for the
@@ -91,12 +95,33 @@ pub async fn introspect(
     client_id: &str,
     client_secret: &str,
     token: &str,
+) -> Verdict {
+    introspect_within(
+        http,
+        url,
+        client_id,
+        client_secret,
+        token,
+        INTROSPECTION_TIMEOUT,
+    )
+    .await
+}
+
+/// [`introspect`] with the deadline as a parameter, so a test can exercise
+/// the timeout without waiting out [`INTROSPECTION_TIMEOUT`].
+async fn introspect_within(
+    http: &reqwest::Client,
+    url: &str,
+    client_id: &str,
+    client_secret: &str,
+    token: &str,
+    timeout: Duration,
 ) -> Verdict {
     let resp = match http
         .post(url)
         .basic_auth(client_id, Some(client_secret))
         .form(&[("token", token)])
-        .timeout(INTROSPECTION_TIMEOUT)
+        .timeout(timeout)
         .send()
         .await
     {
@@ -129,6 +154,94 @@ pub async fn introspect(
 mod tests {
     use super::*;
 
+    use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
+
+    /// An HTTP server on loopback. Each connection is answered with
+    /// `status` and `body` after `delay`; a `None` delay holds the
+    /// connection open and never answers.
+    async fn serve(delay: Option<Duration>, status: u16, body: &'static str) -> String {
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let url = format!("http://{}/introspect", listener.local_addr().unwrap());
+        tokio::spawn(async move {
+            while let Ok((mut conn, _)) = listener.accept().await {
+                tokio::spawn(async move {
+                    let mut buf = vec![0u8; 8192];
+                    let _ = conn.read(&mut buf).await;
+                    let Some(delay) = delay else {
+                        std::future::pending::<()>().await;
+                        return;
+                    };
+                    tokio::time::sleep(delay).await;
+                    let resp = format!(
+                        "HTTP/1.1 {status} X\r\ncontent-type: application/json\r\n\
+                         content-length: {}\r\nconnection: close\r\n\r\n{body}",
+                        body.len()
+                    );
+                    let _ = conn.write_all(resp.as_bytes()).await;
+                });
+            }
+        });
+        url
+    }
+
+    const ACTIVE: &str = r#"{"active":true,"client_id":"c"}"#;
+
+    fn active() -> Verdict {
+        Verdict::Active {
+            client_id: Some("c".to_owned()),
+        }
+    }
+
+    async fn ask(url: &str, timeout: Duration) -> Verdict {
+        introspect_within(
+            &reqwest::Client::new(),
+            url,
+            "bridge",
+            "secret",
+            "tok",
+            timeout,
+        )
+        .await
+    }
+
+    /// Presence control for the timeout tests: the same server shape admits
+    /// when it answers in time.
+    #[tokio::test]
+    async fn an_active_answer_within_the_deadline_admits() {
+        let url = serve(Some(Duration::ZERO), 200, ACTIVE).await;
+        assert_eq!(ask(&url, Duration::from_secs(5)).await, active());
+    }
+
+    /// A slow authelia is waited for: the answer's latency does not matter
+    /// as long as it beats the deadline.
+    #[tokio::test]
+    async fn a_slow_answer_inside_the_deadline_still_admits() {
+        let url = serve(Some(Duration::from_millis(600)), 200, ACTIVE).await;
+        assert_eq!(ask(&url, Duration::from_secs(5)).await, active());
+    }
+
+    /// Fail closed: no answer by the deadline is `Unavailable`, not an
+    /// admission.
+    #[tokio::test]
+    async fn no_answer_by_the_deadline_is_unavailable() {
+        let url = serve(None, 200, "").await;
+        assert_eq!(
+            ask(&url, Duration::from_millis(300)).await,
+            Verdict::Unavailable
+        );
+    }
+
+    /// An answer that arrives after the deadline is not used, even though
+    /// it would have admitted.
+    #[tokio::test]
+    async fn an_active_answer_after_the_deadline_is_unavailable() {
+        let url = serve(Some(Duration::from_millis(900)), 200, ACTIVE).await;
+        assert_eq!(
+            ask(&url, Duration::from_millis(300)).await,
+            Verdict::Unavailable
+        );
+    }
+
     #[test]
     fn only_active_true_deserializes_to_a_grant() {
         let yes: IntrospectionResponse = serde_json::from_str(r#"{"active":true}"#).unwrap();

swarm-authelia-bridge/src/main.rs

@@ -284,8 +284,10 @@ impl Refusal {
     /// 503 rather than 500 because the overwhelmingly common cause is
     /// transient by construction: this bridge's own secret is minted on
     /// authelia's first boot and delivered by a separate unit, so the
-    /// window where it is absent closes on its own. A client retrying is
+    /// window where it is absent closes on its own; an authelia that
-    /// doing the right thing.
+    /// answers slower than [`introspect::INTROSPECTION_TIMEOUT`] is
+    /// transient too. A client retrying is doing the right thing, and
+    /// `swarm-controller` does.
     fn status(self) -> StatusCode {
         match self {
             Refusal::Caller => StatusCode::UNAUTHORIZED,

swarm-controller/src/auth.rs

@@ -127,14 +127,56 @@ impl AuthBridge {
             .await
             .context("minting a bearer token for swarm-authelia-bridge")?;
 
-        let response = self
+        post_with_retry(&self.http, &self.base_url, &token, req, RETRY_BACKOFF).await
-            .http
+    }
-            .post(format!("{}/requests", self.base_url))
+}
+
+/// Waits between attempts at one bridge call: three tries in all.
+const RETRY_BACKOFF: &[std::time::Duration] = &[
+    std::time::Duration::from_secs(1),
+    std::time::Duration::from_secs(4),
+];
+
+/// POST `req` to the bridge, retrying a `503` or a failure to connect after
+/// each wait in `backoff`.
+///
+/// `503` is the bridge saying it could not get an answer out of authelia, so
+/// the token was never evaluated; the same call usually succeeds once
+/// authelia recovers. `401` and `403` are verdicts on the token and are
+/// final, as is any other status. A timeout after the connection was made
+/// is not retried: the request may have been acted on, and a whole attempt
+/// already took [`HTTP_TIMEOUT`]. Every bridge request is safe to repeat
+/// regardless (idempotent ensure, or a read).
+async fn post_with_retry<T: serde::de::DeserializeOwned>(
+    http: &reqwest::Client,
+    base_url: &str,
+    token: &str,
+    req: &BridgeRequest,
+    backoff: &[std::time::Duration],
+) -> Result<T> {
+    let mut waits = backoff.iter();
+    loop {
+        let sent = http
+            .post(format!("{base_url}/requests"))
             .bearer_auth(token)
             .json(req)
             .send()
-            .await
+            .await;
-            .map_err(|e| {
+        let (failure, retryable) = match sent {
+            Ok(response) => {
+                let status = response.status();
+                let body = response.text().await.unwrap_or_default();
+                if status.is_success() {
+                    return serde_json::from_str(&body)
+                        .context("parsing swarm-authelia-bridge's response");
+                }
+                (
+                    anyhow::anyhow!("swarm-authelia-bridge refused the request ({status}): {body}"),
+                    status == reqwest::StatusCode::SERVICE_UNAVAILABLE,
+                )
+            }
+            Err(e) => {
+                let retryable = e.is_connect();
                 let what = if e.is_connect() && e.is_timeout() {
                     format!(
                         "calling swarm-authelia-bridge: connect timed out after {HTTP_CONNECT_TIMEOUT:?}"
@@ -144,16 +186,16 @@ impl AuthBridge {
                 } else {
                     "calling swarm-authelia-bridge".to_owned()
                 };
-                anyhow::Error::new(e).context(what)
+                (anyhow::Error::new(e).context(what), retryable)
-            })?;
+            }
-
+        };
-        let status = response.status();
+        match waits.next() {
-        let body = response.text().await.unwrap_or_default();
+            Some(wait) if retryable => {
-        if !status.is_success() {
+                tracing::warn!(error = %failure, ?wait, "swarm-authelia-bridge call failed; retrying");
-            anyhow::bail!("swarm-authelia-bridge refused the request ({status}): {body}");
+                tokio::time::sleep(*wait).await;
+            }
+            _ => return Err(failure),
         }
-
-        serde_json::from_str(&body).context("parsing swarm-authelia-bridge's response")
     }
 }
 
@@ -161,6 +203,106 @@ impl AuthBridge {
 mod tests {
     use super::*;
 
+    use std::sync::Arc;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use std::time::Duration;
+
+    const NO_WAIT: &[Duration] = &[Duration::ZERO, Duration::ZERO];
+
+    /// A bridge on loopback answering each request with the next entry of
+    /// `statuses` (the last one repeats), and counting the requests.
+    async fn bridge(statuses: &'static [u16]) -> (String, Arc<AtomicUsize>) {
+        use axum::http::StatusCode;
+        let hits = Arc::new(AtomicUsize::new(0));
+        let counter = hits.clone();
+        let app = axum::Router::new().route(
+            "/requests",
+            axum::routing::post(move || {
+                let n = counter.fetch_add(1, Ordering::SeqCst);
+                let code = statuses[n.min(statuses.len() - 1)];
+                async move {
+                    let body = if code == 200 {
+                        r#"{"status":"operators","operators":["mara"]}"#
+                    } else {
+                        r#"{"status":"error","message":"refused"}"#
+                    };
+                    (StatusCode::from_u16(code).unwrap(), body)
+                }
+            }),
+        );
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let url = format!("http://{}", listener.local_addr().unwrap());
+        tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+        (url, hits)
+    }
+
+    async fn call(url: &str, backoff: &[Duration]) -> Result<BridgeResponse> {
+        post_with_retry(
+            &reqwest::Client::new(),
+            url,
+            "tok",
+            &BridgeRequest::ListOperators,
+            backoff,
+        )
+        .await
+    }
+
+    #[tokio::test]
+    async fn a_503_is_retried_until_the_bridge_answers() {
+        let (url, hits) = bridge(&[503, 503, 200]).await;
+        let got = call(&url, NO_WAIT).await.expect("third try succeeds");
+        assert!(matches!(got, BridgeResponse::Operators { .. }), "{got:?}");
+        assert_eq!(hits.load(Ordering::SeqCst), 3);
+    }
+
+    #[tokio::test]
+    async fn a_persistent_503_fails_after_three_tries() {
+        let (url, hits) = bridge(&[503]).await;
+        let err = call(&url, NO_WAIT).await.expect_err("never recovers");
+        assert!(format!("{err:#}").contains("503"), "{err:#}");
+        assert_eq!(hits.load(Ordering::SeqCst), 3);
+    }
+
+    /// 401 and 403 are verdicts on the token: one try each.
+    #[tokio::test]
+    async fn a_401_or_403_is_final() {
+        for code in [401, 403] {
+            let (url, hits) = bridge(match code {
+                401 => &[401],
+                _ => &[403],
+            })
+            .await;
+            let err = call(&url, NO_WAIT).await.expect_err("refused");
+            assert!(format!("{err:#}").contains(&code.to_string()), "{err:#}");
+            assert_eq!(hits.load(Ordering::SeqCst), 1, "status {code}");
+        }
+    }
+
+    #[tokio::test]
+    async fn a_connection_refusal_is_retried_then_fails() {
+        // Bind then drop, so the port refuses connections.
+        let port = {
+            let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+            l.local_addr().unwrap().port()
+        };
+        let started = tokio::time::Instant::now();
+        let err = call(
+            &format!("http://127.0.0.1:{port}"),
+            &[Duration::from_millis(150); 2],
+        )
+        .await
+        .expect_err("nothing listens");
+        assert!(
+            format!("{err:#}").contains("calling swarm-authelia-bridge"),
+            "{err:#}"
+        );
+        assert!(
+            started.elapsed() >= Duration::from_millis(300),
+            "both waits were taken: {:?}",
+            started.elapsed()
+        );
+    }
+
     /// The common case: no bridge wired up for this deployment.
     #[test]
     fn absent_url_is_not_an_error() {