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

review classifier: retry a review post for long enough to outlast a swarm-bao restart

Browse source 
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.
This commit is contained in:
atlas 2026-10-15 09:38:02 +02:00 • committed by mara
commit 9f0fb71fee
2 changed files with 140 additions and 25 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -36,13 +36,15 @@ use crate::forge::{self, classifier_account::LOGIN, classifier_account::Review};
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] = [
/// 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.
@ -481,11 +483,7 @@ async fn post(
let repo = format!("{}/{}", req.owner, req.name);
let event = review.event;
let mut tries = 0;
let sent = forge::within_forge_deadline(
"review classifier review",
post_retrying(poster, req, head, &review, &mut tries),
)
.await;
let sent = post_retrying(poster, req, head, &review, &mut tries).await;
match sent {
Some(Ok(url)) => {
tracing::info!(
@ -502,35 +500,46 @@ async fn post(
);
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`, and post it again after each of [`POST_RETRY_WAITS`] while
/// the post fails with a [`transient`] error. `tries` counts the posts made.
/// 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,
) -> anyhow::Result<Option<String>> {
) -> Option<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 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 Err(e);
return sent;
};
tracing::info!(
repo = %format!("{}/{}", req.owner, req.name), pr = req.pr,
error = %format!("{e:#}"), tries, wait_s = wait.as_secs(),
%error, tries, wait_s = wait.as_secs(),
"posting the review classifier's review failed; trying again"
);
tokio::time::sleep(*wait).await;
@ -1524,6 +1533,109 @@ mod tests {
);
}
/// 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 {