From eb92e83c958cd15fc19191fbc0aca325a7a10a2e Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 17 Oct 2026 11:29:15 +0200 Subject: [PATCH] review classifier: re-post a clean verdict on a same-head re-request 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 , 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. --- docs/swarm/review-classifier.md | 19 +- .../src/review_classifier/review.rs | 332 +++++++++++++++++- .../src/review_classifier/runs.rs | 33 +- 3 files changed, 374 insertions(+), 10 deletions(-) diff --git a/docs/swarm/review-classifier.md b/docs/swarm/review-classifier.md index 10a97388..f562d969 100644 --- a/docs/swarm/review-classifier.md +++ b/docs/swarm/review-classifier.md @@ -98,6 +98,15 @@ 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. +Requesting the account again on a head it already reviewed depends on that +head's newest recorded review, as the [run store](#run-store-and-retention) +holds it. When that review approved or requested changes, nothing runs: the +account posts the same verdict and result table again, opening with +"Already reviewed ``, verdict unchanged." When it had no clean +verdict, or its post failed, the request starts a fresh run. A head with no +recorded review, such as one whose runs have dropped out of the newest 100, or +a run store the controller can't read, gets a fresh run too. + The re-request also retires the account's earlier review. The forge clears the `official` flag on a reviewer's earlier reviews when it's requested again, and only official reviews count toward the required approvals or @@ -111,7 +120,8 @@ default (see [Swarm-level settings](#swarm-level-settings)). Up to eight more wait in the controller's memory, a bound that isn't configurable, so a controller restart drops them, and nothing gets posted for them. Waiting reviews start in no guaranteed order. A request for a PR and head that's -already running or waiting starts nothing. A request with every running place +already running, waiting or re-posting starts nothing, and the head stays +taken until the controller writes its record. A request with every running place and all eight waiting places taken starts no run: the account posts a `**No verdict:**` comment saying so. @@ -556,8 +566,11 @@ The route answers **201** with `branch`, `path` (the file written) and `swarm-controller` records every run when it ends: a try-out when the classifier's reply arrives, a requested review once the controller has posted its review or the post has failed, and a request turned away because the -queue was full once its comment is on the forge or the post has failed. A review request for a head that's already -running or waiting starts nothing and isn't recorded. A run still going isn't +queue was full once its comment is on the forge or the post has failed. The +controller records a re-posted verdict the same way, with the outcome and answer of +the run it repeats and that run's id in its `detail`. A review request for a +head that's already running, waiting or re-posting starts nothing and isn't +recorded. A run still going isn't in the list, and a controller restart drops a run in progress. The newest 100 runs are one JSON array under the key `recent` in the diff --git a/swarm-controller/src/review_classifier/review.rs b/swarm-controller/src/review_classifier/review.rs index 9dc03680..f9ec0a3a 100644 --- a/swarm-controller/src/review_classifier/review.rs +++ b/swarm-controller/src/review_classifier/review.rs @@ -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, /// 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( @@ -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 { + 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 { + 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, + fake: Arc, + gate: Gate, + poster: Arc, + doc: Arc, + 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> { + 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, 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( diff --git a/swarm-controller/src/review_classifier/runs.rs b/swarm-controller/src/review_classifier/runs.rs index c7803f67..eb6c3864 100644 --- a/swarm-controller/src/review_classifier/runs.rs +++ b/swarm-controller/src/review_classifier/runs.rs @@ -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, /// 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> { + 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, ProblemDetails> { let read = match self { Self::Live(client) => {