review classifier: retry a review post for long enough to outlast a swarm-bao restart

The retries covered about a minute, and the swarm-bao restart that lost
#5286's review lasted 3m19s. Each try now gets its own forge deadline,
a try with no answer by then is retried like a refused connection, and
the waits are 5s, 15s, 45s, 90s and 150s: six tries over about five
minutes.

The docs also said to request the review again after a post-failed run.
That does nothing while the account's earlier request is still pending,
so they now say to remove the request and then request it again.

docs/swarm/review-classifier.md

@@ -84,6 +84,16 @@ as a reviewer on a PR starts a run:
 4. The controller posts the result on the PR as the service account's
    review, on the commit it read.
 
+When a post can't reach the secret store or the forge, because the
+connection fails or a try gets no answer within two minutes, the controller
+tries it up to six times over about five minutes, enough to outlast a
+swarm-bao restart, with the same verdict. It doesn't retry any other failure,
+such as the forge refusing the review. A post that still fails ends the run
+as `post-failed`, and nothing appears on the PR. The account's review
+request is still pending then, and the forge sends no event for a request
+that's already standing: remove the service account's review request, then
+request it again to start a fresh run.
+
 A verdict belongs to the commit it ran on. A push starts no run. Once the
 PR's head moves, requesting the service account as a reviewer again starts a
 fresh run on the new head.

swarm-controller/src/review_classifier/review.rs

@@ -17,7 +17,9 @@ use std::collections::HashSet;
 use std::fmt::Write as _;
 use std::future::Future;
 use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
+use std::time::Duration;
 
+use forgejo_api::ForgejoError;
 use serde::Deserialize;
 use swarm_queue_client::review_classifier::{Overall, Reply, Row, TryOut, Verdict};
 use tokio::sync::Notify;
@@ -33,6 +35,18 @@ use crate::forge::{self, classifier_account::LOGIN, classifier_account::Review};
 /// The most review runs waiting for a slot.
 const MAX_WAITING_REVIEWS: usize = 8;
 
+/// The waits before each retry of a post that failed with a [`transient`]
+/// error or got no answer: six tries spread over about five minutes, sized to
+/// outlast a swarm-bao restart (about 3-4 minutes observed on 2026-10-14).
+/// Each try has its own [`forge::FORGE_DEADLINE`].
+const POST_RETRY_WAITS: [Duration; 5] = [
+    Duration::from_secs(5),
+    Duration::from_secs(15),
+    Duration::from_secs(45),
+    Duration::from_secs(90),
+    Duration::from_secs(150),
+];
+
 /// The forge's review events.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum Event {
@@ -468,15 +482,12 @@ async fn post(
 ) -> Result<Option<String>, String> {
     let repo = format!("{}/{}", req.owner, req.name);
     let event = review.event;
-    let sent = forge::within_forge_deadline(
+    let mut tries = 0;
-        "review classifier review",
+    let sent = post_retrying(poster, req, head, &review, &mut tries).await;
-        poster.post(&req.owner, &req.name, req.pr, head, review),
-    )
-    .await;
     match sent {
         Some(Ok(url)) => {
             tracing::info!(
-                %repo, pr = req.pr, head_sha = head.unwrap_or_default(), event,
+                %repo, pr = req.pr, head_sha = head.unwrap_or_default(), event, tries,
                 "review classifier review posted"
             );
             Ok(url)
@@ -484,15 +495,74 @@ async fn post(
         Some(Err(e)) => {
             let error = format!("{e:#}");
             tracing::warn!(
-                %repo, pr = req.pr, %error,
+                %repo, pr = req.pr, %error, tries,
                 "posting the review classifier's review failed"
             );
             Err(format!("posting the review failed: {error}"))
         }
-        None => Err("the forge did not answer the review's post within the deadline".to_owned()),
+        None => {
+            tracing::warn!(
+                %repo, pr = req.pr, tries,
+                "posting the review classifier's review failed: no answer within the deadline"
+            );
+            Err("the forge did not answer the review's post within the deadline".to_owned())
+        }
     }
 }
 
+/// Post `review`, each try within [`forge::FORGE_DEADLINE`], and post it
+/// again after each of [`POST_RETRY_WAITS`] while a try fails with a
+/// [`transient`] error or gets no answer by the deadline. The last try's
+/// result, `None` when it got no answer; `tries` counts the tries made.
+async fn post_retrying(
+    poster: &impl ReviewPoster,
+    req: &Requested,
+    head: Option<&str>,
+    review: &Review,
+    tries: &mut usize,
+) -> Option<anyhow::Result<Option<String>>> {
+    let mut waits = POST_RETRY_WAITS.iter();
+    loop {
+        *tries += 1;
+        let sent = forge::within_forge_deadline(
+            "review classifier review",
+            poster.post(&req.owner, &req.name, req.pr, head, review.clone()),
+        )
+        .await;
+        let error = match &sent {
+            Some(Err(e)) if transient(e) => format!("{e:#}"),
+            None => "no answer within the deadline".to_owned(),
+            Some(_) => return sent,
+        };
+        let Some(wait) = waits.next() else {
+            return sent;
+        };
+        tracing::info!(
+            repo = %format!("{}/{}", req.owner, req.name), pr = req.pr,
+            %error, tries, wait_s = wait.as_secs(),
+            "posting the review classifier's review failed; trying again"
+        );
+        tokio::time::sleep(*wait).await;
+    }
+}
+
+/// Whether a post that failed with `e` may succeed when tried again: the
+/// connection to the secret store or the forge failed or timed out. An answer
+/// from either, such as a 4xx, is not transient.
+///
+/// The store's client puts its `reqwest::Error` in the chain itself; the
+/// forge's wraps it in a transparent [`ForgejoError::ReqwestError`], whose
+/// chain skips it.
+fn transient(e: &anyhow::Error) -> bool {
+    e.chain().any(|cause| {
+        let http = match cause.downcast_ref::<ForgejoError>() {
+            Some(ForgejoError::ReqwestError(e)) => Some(e),
+            _ => cause.downcast_ref::<reqwest::Error>(),
+        };
+        http.is_some_and(|e| e.is_connect() || e.is_timeout())
+    })
+}
+
 /// How a run ended, as its record holds it.
 struct Ended {
     outcome: RunOutcome,
@@ -798,6 +868,7 @@ mod tests {
     use std::sync::atomic::{AtomicUsize, Ordering};
     use std::time::Duration;
 
+    use axum::http::StatusCode;
     use swarm_queue_client::review_classifier::Polarity;
 
     use super::*;
@@ -1378,6 +1449,239 @@ mod tests {
         );
     }
 
+    /// The forge client's error for a port nothing listens on.
+    async fn refused_by_port() -> ForgejoError {
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap();
+        drop(listener);
+        let forge = forgejo_api::Forgejo::new(forgejo_api::Auth::Token("stub-token"), url).unwrap();
+        forge.user_get_current().await.expect_err("nothing listens")
+    }
+
+    /// A refused connection as the forge client reports it.
+    async fn forge_refused() -> anyhow::Error {
+        anyhow::Error::from(refused_by_port().await).context("posting the review")
+    }
+
+    /// A refused connection as the store client reports it: the
+    /// `reqwest::Error` itself in the chain, under the login's context.
+    async fn store_refused() -> anyhow::Error {
+        let ForgejoError::ReqwestError(e) = refused_by_port().await else {
+            panic!("a connect error is a reqwest error");
+        };
+        anyhow::Error::new(e)
+            .context("An error occurred with the request")
+            .context("logging in to the swarm secret store")
+    }
+
+    /// A poster that fails with each of `failures` in turn, then posts.
+    struct Flaky {
+        failures: Mutex<Vec<anyhow::Error>>,
+        tries: AtomicUsize,
+        posted: Posted,
+    }
+
+    impl Flaky {
+        fn new(mut failures: Vec<anyhow::Error>) -> Self {
+            failures.reverse();
+            Self {
+                failures: Mutex::new(failures),
+                tries: AtomicUsize::new(0),
+                posted: Posted::default(),
+            }
+        }
+    }
+
+    impl ReviewPoster for Flaky {
+        async fn post(
+            &self,
+            owner: &str,
+            repo: &str,
+            number: i64,
+            commit: Option<&str>,
+            review: Review,
+        ) -> anyhow::Result<Option<String>> {
+            self.tries.fetch_add(1, Ordering::SeqCst);
+            let failure = self.failures.lock().expect("unpoisoned").pop();
+            match failure {
+                Some(e) => Err(e),
+                None => self.posted.post(owner, repo, number, commit, review).await,
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn a_post_refused_a_connection_is_tried_again() {
+        let failures = vec![forge_refused().await, store_refused().await];
+        tokio::time::pause();
+        let approve = Reply::Done {
+            rows: vec![row(None, true, true)],
+            overall: Overall::Approve,
+        };
+        let fake = Fake::new(approve, vec![rules_json()]);
+        let poster = Flaky::new(failures);
+        let started = tokio::time::Instant::now();
+        let run = run(&fake, &fake, &fake, &poster, &request("r")).await;
+        assert_eq!(run.outcome, RunOutcome::Approved, "{:?}", run.detail);
+        assert_eq!(poster.tries.load(Ordering::SeqCst), 3);
+        assert_eq!(poster.posted.0.lock().expect("unpoisoned").len(), 1);
+        assert_eq!(fake.sent.lock().expect("unpoisoned").len(), 1, "one run");
+        let waited = started.elapsed();
+        assert!(
+            waited >= POST_RETRY_WAITS[0] + POST_RETRY_WAITS[1] && waited < POST_RETRY_WAITS[2],
+            "{waited:?}"
+        );
+    }
+
+    /// A secret store that refuses every connection until `up`.
+    struct Outage {
+        up: tokio::time::Instant,
+        refusals: Mutex<Vec<anyhow::Error>>,
+        tries: AtomicUsize,
+        posted: Posted,
+    }
+
+    impl ReviewPoster for Outage {
+        async fn post(
+            &self,
+            owner: &str,
+            repo: &str,
+            number: i64,
+            commit: Option<&str>,
+            review: Review,
+        ) -> anyhow::Result<Option<String>> {
+            self.tries.fetch_add(1, Ordering::SeqCst);
+            if tokio::time::Instant::now() < self.up {
+                return Err(self
+                    .refusals
+                    .lock()
+                    .expect("unpoisoned")
+                    .pop()
+                    .expect("a refusal per try"));
+            }
+            self.posted.post(owner, repo, number, commit, review).await
+        }
+    }
+
+    #[tokio::test]
+    async fn a_post_outlasts_a_secret_store_restart() {
+        let mut refusals = Vec::new();
+        for _ in 0..=POST_RETRY_WAITS.len() {
+            refusals.push(store_refused().await);
+        }
+        tokio::time::pause();
+        let approve = Reply::Done {
+            rows: vec![row(None, true, true)],
+            overall: Overall::Approve,
+        };
+        let fake = Fake::new(approve, vec![rules_json()]);
+        let started = tokio::time::Instant::now();
+        // swarm-bao was down 3m19s on 2026-10-14.
+        let down = Duration::from_secs(200);
+        let poster = Outage {
+            up: started + down,
+            refusals: Mutex::new(refusals),
+            tries: AtomicUsize::new(0),
+            posted: Posted::default(),
+        };
+        let run = run(&fake, &fake, &fake, &poster, &request("r")).await;
+        assert_eq!(run.outcome, RunOutcome::Approved, "{:?}", run.detail);
+        assert_eq!(poster.posted.0.lock().expect("unpoisoned").len(), 1);
+        assert_eq!(fake.sent.lock().expect("unpoisoned").len(), 1, "one run");
+        assert!(poster.tries.load(Ordering::SeqCst) > 1);
+        assert!(started.elapsed() >= down, "{:?}", started.elapsed());
+    }
+
+    /// A forge whose first post never answers.
+    #[derive(Default)]
+    struct HangsOnce {
+        tries: AtomicUsize,
+        posted: Posted,
+    }
+
+    impl ReviewPoster for HangsOnce {
+        async fn post(
+            &self,
+            owner: &str,
+            repo: &str,
+            number: i64,
+            commit: Option<&str>,
+            review: Review,
+        ) -> anyhow::Result<Option<String>> {
+            if self.tries.fetch_add(1, Ordering::SeqCst) == 0 {
+                std::future::pending::<()>().await;
+            }
+            self.posted.post(owner, repo, number, commit, review).await
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn a_post_with_no_answer_is_abandoned_at_its_deadline_and_tried_again() {
+        let approve = Reply::Done {
+            rows: vec![row(None, true, true)],
+            overall: Overall::Approve,
+        };
+        let fake = Fake::new(approve, vec![rules_json()]);
+        let poster = HangsOnce::default();
+        let started = tokio::time::Instant::now();
+        let run = tokio::time::timeout(
+            Duration::from_hours(1),
+            run(&fake, &fake, &fake, &poster, &request("r")),
+        )
+        .await
+        .expect("the hung try is abandoned");
+        assert_eq!(run.outcome, RunOutcome::Approved, "{:?}", run.detail);
+        assert_eq!(poster.tries.load(Ordering::SeqCst), 2);
+        assert_eq!(poster.posted.0.lock().expect("unpoisoned").len(), 1);
+        assert!(started.elapsed() >= forge::FORGE_DEADLINE + POST_RETRY_WAITS[0]);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn a_post_the_forge_refused_is_not_tried_again() {
+        let approve = Reply::Done {
+            rows: vec![row(None, true, true)],
+            overall: Overall::Approve,
+        };
+        let fake = Fake::new(approve, vec![rules_json()]);
+        let poster = Flaky::new(vec![
+            ForgejoError::UnexpectedStatusCode(StatusCode::FORBIDDEN).into(),
+        ]);
+        let started = tokio::time::Instant::now();
+        let run = run(&fake, &fake, &fake, &poster, &request("r")).await;
+        assert_eq!(run.outcome, RunOutcome::PostFailed);
+        assert_eq!(poster.tries.load(Ordering::SeqCst), 1);
+        assert!(poster.posted.0.lock().expect("unpoisoned").is_empty());
+        assert_eq!(started.elapsed(), Duration::ZERO);
+    }
+
+    #[tokio::test]
+    async fn a_post_refused_every_try_is_post_failed() {
+        let mut failures = Vec::new();
+        for _ in 0..=POST_RETRY_WAITS.len() {
+            failures.push(store_refused().await);
+        }
+        tokio::time::pause();
+        let approve = Reply::Done {
+            rows: vec![row(None, true, true)],
+            overall: Overall::Approve,
+        };
+        let fake = Fake::new(approve, vec![rules_json()]);
+        let poster = Flaky::new(failures);
+        let run = run(&fake, &fake, &fake, &poster, &request("r")).await;
+        assert_eq!(run.outcome, RunOutcome::PostFailed);
+        assert_eq!(
+            poster.tries.load(Ordering::SeqCst),
+            POST_RETRY_WAITS.len() + 1
+        );
+        assert!(poster.posted.0.lock().expect("unpoisoned").is_empty());
+        assert!(
+            run.detail.as_deref().is_some_and(|d| d
+                .starts_with("posting the review failed: logging in to the swarm secret store")),
+            "{:?}",
+            run.detail
+        );
+    }
+
     #[tokio::test]
     async fn a_failed_record_write_still_posts_the_review() {
         let fake = Arc::new(Fake::new(