Watch
0
0
Fork
You've already forked hyperhive
0

Compare commits

..
Author SHA1 Message Date
atlas
9f0fb71fee 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.
2026-10-15 09:58:46 +02:00
atlas
2e53e46d94 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.
2026-10-15 09:58:46 +02:00
2 changed files with 322 additions and 8 deletions

View file

@ -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.

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,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(
"review classifier review",
poster.post(&req.owner, &req.name, req.pr, head, review),
)
.await;
let mut tries = 0;
let sent = 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,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(