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(