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

review classifier: re-post a clean verdict on a same-head re-request

Browse source 
A review request for a head whose newest recorded review approved or
requested changes now runs nothing: the classifier account posts that
verdict and table again, opening "Already reviewed <sha>, verdict
unchanged". A head whose newest review had no clean verdict, a failed
post, or no record gets a fresh run, as before.

An accepted request now keeps its key until its record is written, so a
request right after a run finishes reads that run's verdict instead of
starting a second run.

Part of #5387; the restart-loss half waits on #5219.
This commit is contained in:
atlas 2026-10-17 11:29:15 +02:00 • committed by mara
commit eb92e83c95
3 changed files with 374 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -7,6 +7,11 @@
//! on a reviewer's previous reviews when they are requested again, and only
//! official reviews count toward approvals or block a merge.
//!
//! A request for a head whose newest recorded review approved or requested
//! changes runs nothing: the account posts that verdict again, saying so. A
//! head whose newest review had no clean verdict, or that has none recorded,
//! gets a fresh run.
//!
//! A run with no clean verdict never approves: see [`no_clean_verdict_event`].
//! At most `max_concurrent_reviews` runs go at once, as the swarm's
//! [settings](super::settings) say when a run asks for its place, and at most
@ -157,9 +162,9 @@ type Key = (String, String, i64, String);
/// The review runs this controller has accepted and not yet finished.
///
/// A request whose [`Key`] is already running or waiting starts nothing. A
/// request beyond the run limit + [`MAX_WAITING_REVIEWS`] starts no run
/// either; the account posts a no-verdict comment saying so.
/// A request whose [`Key`] is already running, waiting or re-posting starts
/// nothing. A request beyond the run limit + [`MAX_WAITING_REVIEWS`] starts no
/// run either; the account posts a no-verdict comment saying so.
pub(crate) struct Reviews {
slots: Mutex<Slots>,
/// Woken when a run ends or the limit changes.
@ -189,7 +194,7 @@ impl Default for Reviews {
/// Why [`Reviews::accept`] turned a request away.
#[derive(Debug, PartialEq, Eq)]
enum Refused {
/// A run for the same key is running or waiting.
/// A run for the same key is running, waiting or re-posting.
Duplicate,
/// Every run and wait place is taken, by this many runs.
Full(usize),
@ -301,6 +306,10 @@ pub(crate) fn on_delivery(state: &AppState, body: &[u8]) {
/// comment when there's no place for it, and record it in `runs` when it
/// ends. The task, if one was spawned.
///
/// An accepted request whose head's newest review [`reposted`] covers posts
/// that verdict again instead of running. Either way the request holds its key
/// until its record is written, so a request right after it reads that record.
///
/// The run reads the settings for the limit before it waits for a place, and
/// again for the model and effort once it has one.
fn start<S, Q, P>(
@ -321,6 +330,14 @@ where
Ok(accepted) => {
let reviews = Arc::clone(reviews);
Some(tokio::spawn(async move {
if let Some(earlier) = newest_review(&runs, &request).await
&& let Some(ruling) = reposted(&earlier)
{
runs.record(repost(&*poster, &request, &earlier, ruling).await)
.await;
drop(accepted);
return;
}
match queue.settings().await {
Ok(settings) => reviews.resize(settings.max_concurrent_reviews),
Err(e) => tracing::warn!(
@ -332,14 +349,14 @@ where
let slot = reviews.slot().await;
let run = Box::pin(run(&*forge, &*forge, &queue, &*poster, &request)).await;
drop(slot);
drop(accepted);
runs.record(run).await;
drop(accepted);
}))
}
Err(Refused::Duplicate) => {
tracing::info!(
%repo, pr = request.pr, head_sha = request.head.as_str(),
"review requested while a run for this head is running or waiting; nothing more starts"
"review requested while this head is running, waiting or re-posting; nothing more starts"
);
None
}
@ -375,6 +392,77 @@ fn queue_full_ruling(taken: usize) -> Ruling {
}
}
/// The newest recorded review of the head `req` names. `None` when it names
/// none, nothing is recorded for it, or the store can't be read; each of
/// those leads to a fresh run.
async fn newest_review(runs: &Runs, req: &Requested) -> Option<RunRecord> {
if req.head.is_empty() {
return None;
}
let repo = format!("{}/{}", req.owner, req.name);
match runs
.newest_review(&repo, req.pr.unsigned_abs(), &req.head)
.await
{
Ok(found) => found,
Err(e) => {
tracing::warn!(
%repo, pr = req.pr, head_sha = req.head.as_str(), error = %format!("{e:#}"),
"reading the review classifier's earlier runs failed; running the review"
);
None
}
}
}
/// The ruling that posts `earlier`'s verdict again, when it approved or
/// requested changes. `None` for any other outcome, such as no verdict, which
/// a fresh run may improve on.
fn reposted(earlier: &RunRecord) -> Option<Ruling> {
let event = match earlier.outcome {
RunOutcome::Approved => Event::Approve,
RunOutcome::ChangesRequested => Event::RequestChanges,
_ => return None,
};
let (Some(rows), Some(overall)) = (&earlier.rows, earlier.overall) else {
return None;
};
let head = earlier.head_sha.as_deref().unwrap_or_default();
Some(Ruling {
event,
body: format!(
"Already reviewed `{head}`, verdict unchanged.\n\n{}\n{}",
table(rows),
overall_line(overall)
),
reason: Some(format!("a re-post of run {}'s verdict", earlier.id)),
})
}
/// Post `ruling`, `earlier`'s verdict, on `req`'s PR at `earlier`'s head. Its
/// record, carrying `earlier`'s answer.
async fn repost(
poster: &impl ReviewPoster,
req: &Requested,
earlier: &RunRecord,
ruling: Ruling,
) -> RunRecord {
let started_unix = unix_now();
tracing::info!(
repo = %earlier.repo, pr = req.pr, head_sha = req.head.as_str(), earlier = %earlier.id,
"review requested on a head already reviewed; posting its verdict again"
);
let reason = ruling.reason.clone();
let sent = post(poster, req, earlier.head_sha.as_deref(), ruling.review()).await;
record(
req,
earlier.head_sha.clone(),
started_unix,
ended(earlier.outcome, reason, sent),
(earlier.rows.clone(), earlier.overall),
)
}
/// Where a review is posted: the forge in production, a fake in the tests.
pub(crate) trait ReviewPoster {
/// The posted review's URL, when the forge names one.
@ -1328,6 +1416,238 @@ mod tests {
);
}
/// Record in `store` a review of o/r#7 at `abc123` that ended as
/// `outcome`, with an answer as a run that got that far keeps it.
async fn reviewed(store: &Runs, id: &str, outcome: RunOutcome) -> RunRecord {
let earlier = RunRecord {
trigger: Trigger::Review,
outcome,
rows: Some(vec![row(
None,
true,
outcome != RunOutcome::ChangesRequested,
)]),
overall: Some(match outcome {
RunOutcome::ChangesRequested => Overall::RequestChanges,
_ => Overall::Approve,
}),
..crate::review_classifier::runs::tests::record(id)
};
store.record(earlier.clone()).await;
earlier
}
/// A fake forge, a closed [`Gate`], a poster and a run store, and a
/// starter for requests at a head on o/r#7.
struct Rig {
reviews: Arc<Reviews>,
fake: Arc<Fake>,
gate: Gate,
poster: Arc<Posted>,
doc: Arc<FakeDoc>,
store: Runs,
}
impl Rig {
fn new() -> Self {
let doc = Arc::new(FakeDoc::default());
Self {
reviews: Arc::new(Reviews::default()),
fake: Arc::new(Fake::new(
Reply::Failed {
error: String::new(),
},
vec![rules_json()],
)),
gate: Gate::closed(),
poster: Arc::new(Posted::default()),
store: Runs::Fake(Arc::clone(&doc)),
doc,
}
}
fn start_at(&self, head: &str) -> Option<JoinHandle<()>> {
start(
&self.reviews,
Arc::clone(&self.fake),
self.gate.clone(),
Arc::clone(&self.poster),
self.store.clone(),
requested_at(head),
)
}
fn posted(&self) -> Vec<(String, Option<String>, Review)> {
self.poster.0.lock().expect("unpoisoned").clone()
}
}
/// Await `task`, a re-post. A task that ran instead waits on the closed
/// [`Gate`], so this fails rather than hang.
async fn reposted(task: JoinHandle<()>) {
tokio::time::timeout(Duration::from_secs(5), task)
.await
.expect("re-posted without waiting for a run place")
.expect("re-posted");
}
#[tokio::test]
async fn a_request_on_a_head_with_a_clean_verdict_posts_it_again_without_a_run() {
for (outcome, event, overall) in [
(
RunOutcome::Approved,
"APPROVED",
"**Overall: ✅ approve**\n",
),
(
RunOutcome::ChangesRequested,
"REQUEST_CHANGES",
"**Overall: ❌ request changes**\n",
),
] {
let rig = Rig::new();
let earlier = reviewed(&rig.store, "earlier", outcome).await;
reposted(rig.start_at("abc123").expect("accepted")).await;
assert_eq!(
rig.gate.reached(),
0,
"{event}: no run reached the classifier"
);
let posted = rig.posted();
assert_eq!(posted.len(), 1, "{event}");
let (target, commit, review) = &posted[0];
assert_eq!(target, "o/r#7");
assert_eq!(commit.as_deref(), Some("abc123"));
assert_eq!(review.event, event);
assert!(
review
.body
.starts_with("Already reviewed `abc123`, verdict unchanged.\n\n"),
"{}",
review.body
);
assert!(
review.body.contains("| datetime | 90 / 80 |"),
"{}",
review.body
);
assert!(review.body.ends_with(overall), "{}", review.body);
let recorded = rig.doc.runs();
assert_eq!(recorded.len(), 2, "{event}");
assert_eq!(recorded[0].outcome, outcome);
assert_eq!(recorded[0].head_sha.as_deref(), Some("abc123"));
assert_eq!(recorded[0].rows, earlier.rows);
assert_eq!(
recorded[0].detail.as_deref(),
Some("a re-post of run earlier's verdict")
);
}
}
#[tokio::test]
async fn a_request_on_a_head_without_a_clean_verdict_runs_again() {
for (case, seed) in [
("no verdict", vec![RunOutcome::NoVerdict]),
("post failed", vec![RunOutcome::PostFailed]),
(
"an approval since followed by no verdict",
vec![RunOutcome::Approved, RunOutcome::NoVerdict],
),
] {
let rig = Rig::new();
for (i, outcome) in seed.into_iter().enumerate() {
reviewed(&rig.store, &format!("earlier-{i}"), outcome).await;
}
let task = rig.start_at("abc123").expect("accepted");
rig.gate.open(1);
task.await.expect("ran");
assert_eq!(rig.gate.reached(), 1, "{case}: a fresh run");
let posted = rig.posted();
assert_eq!(posted.len(), 1, "{case}");
assert_eq!(posted[0].2.event, "APPROVED", "{case}");
assert!(
!posted[0].2.body.starts_with("Already reviewed"),
"{case}: {}",
posted[0].2.body
);
}
}
#[tokio::test]
async fn only_a_review_of_the_same_head_is_posted_again() {
let approved = reviewed(&Runs::Fake(Arc::default()), "e", RunOutcome::Approved).await;
for (case, earlier) in [
(
"a try-out",
RunRecord {
trigger: Trigger::TryOut,
..approved.clone()
},
),
(
"another head",
RunRecord {
head_sha: Some("def456".to_owned()),
..approved.clone()
},
),
(
"another PR",
RunRecord {
pr: 8,
..approved.clone()
},
),
] {
let rig = Rig::new();
rig.store.record(earlier).await;
let task = rig.start_at("abc123").expect("accepted");
rig.gate.open(1);
task.await.expect("ran");
assert_eq!(rig.gate.reached(), 1, "{case}: a fresh run");
}
}
#[tokio::test]
async fn requests_in_quick_succession_on_a_reviewed_head_start_one_run() {
// No verdict yet: the second request finds the first's run waiting.
let rig = Rig::new();
reviewed(&rig.store, "earlier", RunOutcome::NoVerdict).await;
let first = rig.start_at("abc123").expect("accepted");
settle().await;
assert!(rig.start_at("abc123").is_none(), "a second run started");
rig.gate.open(1);
first.await.expect("ran");
assert_eq!(rig.gate.reached(), 1);
assert_eq!(rig.posted().len(), 1);
// That run approved, so the next request posts its verdict again.
reposted(rig.start_at("abc123").expect("accepted")).await;
assert_eq!(rig.gate.reached(), 1, "no second run");
let posted = rig.posted();
assert_eq!(posted.len(), 2);
assert!(
posted[1].2.body.starts_with("Already reviewed"),
"{}",
posted[1].2.body
);
// A clean verdict: two requests at once post it once.
let rig = Rig::new();
reviewed(&rig.store, "earlier", RunOutcome::Approved).await;
let first = rig.start_at("abc123").expect("accepted");
assert!(rig.start_at("abc123").is_none(), "a second re-post started");
reposted(first).await;
assert_eq!(rig.gate.reached(), 0);
assert_eq!(rig.posted().len(), 1);
}
/// The record of one run against the fake forge, holding `rules`, posted
/// by `poster`.
async fn recorded_for(

33
swarm-controller/src/review_classifier/runs.rs
View file

@ -96,7 +96,8 @@ pub(crate) struct RunRecord {
pub(super) finished_unix: i64,
pub(super) outcome: RunOutcome,
/// Why, for every outcome but `approved`, `changes-requested` and
/// `done`.
/// `done`. A review that re-posts an earlier run's verdict names that
/// run here.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(super) detail: Option<String>,
/// As the try-out GET returns them, when the classifier answered.
@ -248,6 +249,36 @@ impl Runs {
}
}
/// The newest recorded review of `head` on `repo`'s PR `pr`, if any.
pub(crate) async fn newest_review(
&self,
repo: &str,
pr: u64,
head: &str,
) -> anyhow::Result<Option<RunRecord>> {
let (runs, _) = match self {
Self::Live(client) => {
let store = open(client.as_ref()).await.map_err(|Unreachable(e)| {
anyhow::anyhow!(e).context(format!("opening the {BUCKET} bucket"))
})?;
RunDoc::read(&store).await?
}
#[cfg(test)]
Self::Fake(doc) => doc.read().await?,
};
Ok(runs.into_iter().find_map(|entry| match entry {
Stored::Run(run)
if run.trigger == Trigger::Review
&& run.repo == repo
&& run.pr == pr
&& run.head_sha.as_deref() == Some(head) =>
{
Some(run)
}
_ => None,
}))
}
async fn list(&self) -> Result<Vec<RunRecord>, ProblemDetails> {
let read = match self {
Self::Live(client) => {