diff --git a/docs/swarm/review-classifier.md b/docs/swarm/review-classifier.md index ff3bc8bc..a3184d95 100644 --- a/docs/swarm/review-classifier.md +++ b/docs/swarm/review-classifier.md @@ -84,16 +84,6 @@ 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. diff --git a/swarm-controller/src/review_classifier/review.rs b/swarm-controller/src/review_classifier/review.rs index f84e6ef5..b49a4680 100644 --- a/swarm-controller/src/review_classifier/review.rs +++ b/swarm-controller/src/review_classifier/review.rs @@ -17,9 +17,7 @@ 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; @@ -35,18 +33,6 @@ 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 { @@ -482,12 +468,15 @@ async fn post( ) -> Result, String> { let repo = format!("{}/{}", req.owner, req.name); let event = review.event; - let mut tries = 0; - let sent = post_retrying(poster, req, head, &review, &mut tries).await; + let sent = forge::within_forge_deadline( + "review classifier review", + 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, tries, + %repo, pr = req.pr, head_sha = head.unwrap_or_default(), event, "review classifier review posted" ); Ok(url) @@ -495,74 +484,15 @@ async fn post( Some(Err(e)) => { let error = format!("{e:#}"); tracing::warn!( - %repo, pr = req.pr, %error, tries, + %repo, pr = req.pr, %error, "posting the review classifier's review failed" ); Err(format!("posting the review failed: {error}")) } - 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()) - } + None => 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>> { - 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::() { - 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, @@ -868,7 +798,6 @@ 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::*; @@ -1449,239 +1378,6 @@ 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:?}" - ); - } - - /// 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 { - 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(