vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

review classifier: retry a review post that failed to reach the store or forge

Browse source 
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.
This commit is contained in:
atlas 2026-10-15 08:57:24 +02:00 • committed by mara
commit 2e53e46d94
2 changed files with 202 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

198
swarm-controller/src/review_classifier/review.rs
View file

@ -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<Option<String>, 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<Option<String>> {
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::<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 +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<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:?}"
);
}
#[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(