From 2e53e46d94a560b7507a783a5f405c640a3ebed4 Mon Sep 17 00:00:00 2001 From: atlas Date: Thu, 15 Oct 2026 08:57:24 +0200 Subject: [PATCH 1/2] review classifier: retry a review post that failed to reach the store or forge A review request's verdict was lost when the post failed: on 2026-10-14 the classifier approved #5286, but its post hit swarm-bao mid-restart (cert login: connection refused), the run was recorded post-failed, and nothing on the PR showed it. post() now tries the post up to four times, waiting 5s, 15s and 45s, when it fails because the connection to the secret store or the forge failed or timed out. Any other error fails at once, as before. The verdict is computed once and reposted as is. All tries share the one forge deadline the single post had, and the warn log carries the number of tries. --- docs/swarm/review-classifier.md | 7 + .../src/review_classifier/review.rs | 198 +++++++++++++++++- 2 files changed, 202 insertions(+), 3 deletions(-) diff --git a/docs/swarm/review-classifier.md b/docs/swarm/review-classifier.md index a3184d95..abc81d63 100644 --- a/docs/swarm/review-classifier.md +++ b/docs/swarm/review-classifier.md @@ -84,6 +84,13 @@ 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 times out, the controller tries it up to four times over +about a minute 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; request the review 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. diff --git a/swarm-controller/src/review_classifier/review.rs b/swarm-controller/src/review_classifier/review.rs index b49a4680..02d38f1f 100644 --- a/swarm-controller/src/review_classifier/review.rs +++ b/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,16 @@ 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: four tries over about a minute, enough to ride out a short restart +/// of the secret store or the forge. All the tries share one +/// [`forge::FORGE_DEADLINE`]. +const POST_RETRY_WAITS: [Duration; 3] = [ + Duration::from_secs(5), + Duration::from_secs(15), + Duration::from_secs(45), +]; + /// The forge's review events. #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum Event { @@ -468,15 +480,16 @@ async fn post( ) -> Result, String> { let repo = format!("{}/{}", req.owner, req.name); let event = review.event; + let mut tries = 0; let sent = forge::within_forge_deadline( "review classifier review", - poster.post(&req.owner, &req.name, req.pr, head, review), + post_retrying(poster, req, head, &review, &mut tries), ) .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,7 +497,7 @@ 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}")) @@ -493,6 +506,54 @@ async fn post( } } +/// Post `review`, and post it again after each of [`POST_RETRY_WAITS`] while +/// the post fails with a [`transient`] error. `tries` counts the posts made. +async fn post_retrying( + poster: &impl ReviewPoster, + req: &Requested, + head: Option<&str>, + review: &Review, + tries: &mut usize, +) -> anyhow::Result> { + let mut waits = POST_RETRY_WAITS.iter(); + loop { + *tries += 1; + let sent = poster + .post(&req.owner, &req.name, req.pr, head, review.clone()) + .await; + let e = match sent { + Err(e) if transient(&e) => e, + sent => return sent, + }; + let Some(wait) = waits.next() else { + return Err(e); + }; + tracing::info!( + repo = %format!("{}/{}", req.owner, req.name), pr = req.pr, + error = %format!("{e:#}"), 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::() { + Some(ForgejoError::ReqwestError(e)) => Some(e), + _ => cause.downcast_ref::(), + }; + 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 +859,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 +1440,136 @@ 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>, + tries: AtomicUsize, + posted: Posted, + } + + impl Flaky { + fn new(mut failures: Vec) -> 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> { + 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:?}" + ); + } + + #[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( From 9f0fb71feeed6e27abb2a3c7b67ae80bb0cf6b54 Mon Sep 17 00:00:00 2001 From: atlas Date: Thu, 15 Oct 2026 09:38:02 +0200 Subject: [PATCH 2/2] 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 | 11 +- .../src/review_classifier/review.rs | 154 +++++++++++++++--- 2 files changed, 140 insertions(+), 25 deletions(-) diff --git a/docs/swarm/review-classifier.md b/docs/swarm/review-classifier.md index abc81d63..ff3bc8bc 100644 --- a/docs/swarm/review-classifier.md +++ b/docs/swarm/review-classifier.md @@ -85,11 +85,14 @@ as a reviewer on a PR starts a run: review, on the commit it read. When a post can't reach the secret store or the forge, because the -connection fails or times out, the controller tries it up to four times over -about a minute with the same verdict. It doesn't retry any other failure, +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; request the review again to -start a fresh 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 diff --git a/swarm-controller/src/review_classifier/review.rs b/swarm-controller/src/review_classifier/review.rs index 02d38f1f..f84e6ef5 100644 --- a/swarm-controller/src/review_classifier/review.rs +++ b/swarm-controller/src/review_classifier/review.rs @@ -36,13 +36,15 @@ use crate::forge::{self, classifier_account::LOGIN, classifier_account::Review}; const MAX_WAITING_REVIEWS: usize = 8; /// The waits before each retry of a post that failed with a [`transient`] -/// error: four tries over about a minute, enough to ride out a short restart -/// of the secret store or the forge. All the tries share one -/// [`forge::FORGE_DEADLINE`]. -const POST_RETRY_WAITS: [Duration; 3] = [ +/// 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. @@ -481,11 +483,7 @@ async fn post( let repo = format!("{}/{}", req.owner, req.name); let event = review.event; let mut tries = 0; - let sent = forge::within_forge_deadline( - "review classifier review", - post_retrying(poster, req, head, &review, &mut tries), - ) - .await; + let sent = post_retrying(poster, req, head, &review, &mut tries).await; match sent { Some(Ok(url)) => { tracing::info!( @@ -502,35 +500,46 @@ async fn post( ); 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`, and post it again after each of [`POST_RETRY_WAITS`] while -/// the post fails with a [`transient`] error. `tries` counts the posts made. +/// 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, -) -> anyhow::Result> { +) -> Option>> { let mut waits = POST_RETRY_WAITS.iter(); loop { *tries += 1; - let sent = poster - .post(&req.owner, &req.name, req.pr, head, review.clone()) - .await; - let e = match sent { - Err(e) if transient(&e) => e, - sent => return sent, + 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 Err(e); + return sent; }; tracing::info!( repo = %format!("{}/{}", req.owner, req.name), pr = req.pr, - error = %format!("{e:#}"), tries, wait_s = wait.as_secs(), + %error, tries, wait_s = wait.as_secs(), "posting the review classifier's review failed; trying again" ); tokio::time::sleep(*wait).await; @@ -1524,6 +1533,109 @@ mod tests { ); } + /// A secret store that refuses every connection until `up`. + struct Outage { + up: tokio::time::Instant, + refusals: Mutex>, + 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> { + 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> { + 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 {