Watch
0
0
Fork
You've already forked hyperhive
0

Compare commits

..
2 changed files with 8 additions and 322 deletions

View file

@ -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 4. The controller posts the result on the PR as the service account's
review, on the commit it read. 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 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 PR's head moves, requesting the service account as a reviewer again starts a
fresh run on the new head. fresh run on the new head.

View file

@ -17,9 +17,7 @@ use std::collections::HashSet;
use std::fmt::Write as _; use std::fmt::Write as _;
use std::future::Future; use std::future::Future;
use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use std::time::Duration;
use forgejo_api::ForgejoError;
use serde::Deserialize; use serde::Deserialize;
use swarm_queue_client::review_classifier::{Overall, Reply, Row, TryOut, Verdict}; use swarm_queue_client::review_classifier::{Overall, Reply, Row, TryOut, Verdict};
use tokio::sync::Notify; 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. /// The most review runs waiting for a slot.
const MAX_WAITING_REVIEWS: usize = 8; 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. /// The forge's review events.
#[derive(Debug, Clone, Copy, PartialEq, Eq)] #[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Event { enum Event {
@ -482,12 +468,15 @@ async fn post(
) -> Result<Option<String>, String> { ) -> Result<Option<String>, String> {
let repo = format!("{}/{}", req.owner, req.name); let repo = format!("{}/{}", req.owner, req.name);
let event = review.event; let event = review.event;
let mut tries = 0; let sent = forge::within_forge_deadline(
let sent = post_retrying(poster, req, head, &review, &mut tries).await; "review classifier review",
poster.post(&req.owner, &req.name, req.pr, head, review),
)
.await;
match sent { match sent {
Some(Ok(url)) => { Some(Ok(url)) => {
tracing::info!( 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" "review classifier review posted"
); );
Ok(url) Ok(url)
@ -495,74 +484,15 @@ async fn post(
Some(Err(e)) => { Some(Err(e)) => {
let error = format!("{e:#}"); let error = format!("{e:#}");
tracing::warn!( tracing::warn!(
%repo, pr = req.pr, %error, tries, %repo, pr = req.pr, %error,
"posting the review classifier's review failed" "posting the review classifier's review failed"
); );
Err(format!("posting the review failed: {error}")) Err(format!("posting the review failed: {error}"))
} }
None => { None => Err("the forge did not answer the review's post within the deadline".to_owned()),
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. /// How a run ended, as its record holds it.
struct Ended { struct Ended {
outcome: RunOutcome, outcome: RunOutcome,
@ -868,7 +798,6 @@ mod tests {
use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration; use std::time::Duration;
use axum::http::StatusCode;
use swarm_queue_client::review_classifier::Polarity; use swarm_queue_client::review_classifier::Polarity;
use super::*; 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<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] #[tokio::test]
async fn a_failed_record_write_still_posts_the_review() { async fn a_failed_record_write_still_posts_the_review() {
let fake = Arc::new(Fake::new( let fake = Arc::new(Fake::new(