From d6aa83146a046578c78a023d299bfb0dbbf262fb Mon Sep 17 00:00:00 2001 From: atlas Date: Wed, 21 Oct 2026 22:21:24 +0200 Subject: [PATCH 1/4] review classifier: queued review requests survive a restart and are never refused --- docs/swarm/review-classifier.md | 50 +- swarm-controller/README.md | 5 +- swarm-controller/src/main.rs | 5 + .../src/review_classifier/review.rs | 1325 +++++++++++------ .../src/review_classifier/runs.rs | 10 +- swarm-nats-auth/src/policy.rs | 70 + swarm-wire/src/review_classifier.rs | 16 +- 7 files changed, 1024 insertions(+), 457 deletions(-) diff --git a/docs/swarm/review-classifier.md b/docs/swarm/review-classifier.md index d7de8f525..223b05868 100644 --- a/docs/swarm/review-classifier.md +++ b/docs/swarm/review-classifier.md @@ -118,14 +118,24 @@ the earlier one. A repo whose branch protection also dismisses stale approvals ( on an older commit even without a re-request. How many requested reviews run at once is a swarm-wide setting, two by -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, 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. +default (see [Swarm-level settings](#swarm-level-settings)). The controller +turns no request away. It puts each one in the `review-requests` work-queue +stream on the swarm queue, one subject per PR, and the stream keeps one +message per subject: a newer request for a PR replaces one still waiting. +The controller takes a request once a run place is free and acks it after +writing the run's record. While a run goes, the controller tells the queue +every 20 seconds that the request is still in hand. A request it took and +never acked, because the controller stopped first, comes back within a +minute, and the controller that's running then reviews it. A restart +therefore repeats a run that was going. When it can't queue a request, the +account posts a `**No verdict:**` comment saying so. + +When the controller takes a request, it reads the PR first. It drops a +request whose head is no longer the PR's head, and posts and records nothing +for it. +A request for a PR whose review is still going waits in the queue until that +review ends, so a PR never has two runs at once. A request that re-posts an +earlier verdict waits for a run place too, and frees it before it posts. ## Components @@ -206,7 +216,7 @@ replaces it. runtime's own default. On the ACP runtime it applies only where the agent offers it - `max_concurrent_reviews`: 1 to 8, how many requested reviews run at once. - Try-outs don't count toward it. Up to eight more reviews wait for a place + Try-outs don't count toward it. The rest wait in the queue for a place - `runtime`: `api-key` or `acp` (see [Runtimes](#runtimes)) - `acp`: the provider the ACP runtime runs on, or `null`. It stays stored while the runtime is `api-key`, so switching back and forth needs no @@ -574,13 +584,12 @@ 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. 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. +its review or the post has failed, and a request the controller couldn't queue +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 dropped request, whose head the +PR has moved on from, isn't recorded. A run still going isn't in the list; a +controller restart stops it, and the queue hands its request back. The newest 100 runs are one JSON array under the key `recent` in the `review-classifier-runs` KV bucket on the swarm queue, which only the @@ -602,11 +611,12 @@ header. Each run has: - `repo` (`owner/name`), `pr`, and `requester`: the swarm UI user who started the try-out, or the forge user who requested the review - `head_sha`: the commit the run read, absent when the PR wasn't read, as for - an unreadable PR or a full queue + an unreadable PR or a request the controller couldn't queue - `started_unix` and `finished_unix`: a try-out starts at its POST, a review - once it has a run place -- `outcome`: `approved`, `changes-requested`, `no-verdict`, `queue-full` or - `post-failed` for a review; `done` or `failed` for a try-out + once it has a run place and has read the PR +- `outcome`: `approved`, `changes-requested`, `no-verdict` or `post-failed` + for a review, or `queue-full` on a record written before the request queue; + `done` or `failed` for a try-out - `detail`: the reason, for every outcome but `approved`, `changes-requested` and `done`. For `post-failed` it's the run's own reason, when it has one, then why the post failed. diff --git a/swarm-controller/README.md b/swarm-controller/README.md index db51bc73f..40a561204 100644 --- a/swarm-controller/README.md +++ b/swarm-controller/README.md @@ -69,8 +69,9 @@ The swarm's control plane. The swarm UI and `swarmctl` are its clients. the harness to write when no file exists β†’ [`docs/swarm/README.md`](../docs/swarm/README.md#an-agents-agentsmd). - **Review classifier** β€” try-outs from the swarm UI and reviews on review - requests, and the newest 100 finished runs, one document in the - `review-classifier-runs` KV bucket, which this daemon alone writes + requests, the requested reviews not yet finished, in the `review-requests` + work-queue stream, and the newest 100 finished runs, one document in the + `review-classifier-runs` KV bucket, both of which this daemon alone writes (`POST /api/review-classifier/try-out`, `GET /api/review-classifier/try-out/{id}`, `GET /api/review-classifier/rules`, `GET /api/review-classifier/runs`), and the try-out page's branch for a PR, pushed as the classifier's forge account diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 12a035e83..a5434dd2b 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -235,7 +235,12 @@ async fn main() -> Result<()> { reviews: Arc::default(), alerts: alerts::from_env(), }; + serve(listener, state).await +} +/// Start taking the queued review requests, then serve `state` on `listener`. +async fn serve(listener: tokio::net::UnixListener, state: AppState) -> Result<()> { + review_classifier::review::spawn_queue(&state); axum::serve(listener, build_app(state)) .await .context("serving swarm-controller") diff --git a/swarm-controller/src/review_classifier/review.rs b/swarm-controller/src/review_classifier/review.rs index 654671708..9c84556a8 100644 --- a/swarm-controller/src/review_classifier/review.rs +++ b/swarm-controller/src/review_classifier/review.rs @@ -13,10 +13,14 @@ //! 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 -//! [`MAX_WAITING_REVIEWS`] more wait here, in memory, so a restart drops them. -//! See [`Reviews`] for what happens to a request beyond that. +//! +//! Each accepted request waits in [`REQUEST_STREAM`], a work queue on the +//! swarm queue with one subject per PR that keeps one message per subject, so +//! a newer request for a PR replaces one still waiting. The controller takes +//! a request only once fewer than `max_concurrent_reviews` runs go, as the +//! swarm's [settings](super::settings) say, and acks it once its record is +//! written. A request a controller took and never acked, because it stopped +//! first, is delivered again. See [`consume`] for the rest. use std::collections::HashSet; use std::fmt::Write as _; @@ -24,23 +28,25 @@ use std::future::Future; use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; use std::time::Duration; +use anyhow::Context as _; +use async_nats::jetstream::{self, AckKind, consumer::PullConsumer}; use forgejo_api::ForgejoError; -use serde::Deserialize; -use swarm_wire::review_classifier::{Overall, Reply, Row, TryOut, Verdict}; +use futures_util::StreamExt as _; +use serde::{Deserialize, Serialize}; +use swarm_wire::review_classifier::{ + Overall, REQUEST_CONSUMER, REQUEST_STREAM, REQUEST_SUBJECT_PREFIX, Reply, Row, TryOut, Verdict, +}; use tokio::sync::Notify; -use tokio::task::JoinHandle; +use tokio::task::JoinSet; use super::runs::{self, RunOutcome, RunRecord, Runs, Trigger}; use super::settings::{Settings, SettingsSource}; use super::{FileSource, PullSource, RunSink, split_repo}; use crate::agent_renewal::unix_now; use crate::forge::pull_diff::CONTEXT_LINES; -use crate::forge::{self, classifier_account::LOGIN, classifier_account::Review}; +use crate::forge::{self, PullForReview, classifier_account::LOGIN, classifier_account::Review}; use crate::routes::AppState; -/// 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). @@ -53,6 +59,29 @@ const POST_RETRY_WAITS: [Duration; 5] = [ Duration::from_secs(150), ]; +/// The queue's timing in production. +const TIMING: Timing = Timing { + ack_wait: Duration::from_mins(1), + pull_wait: Duration::from_secs(30), + busy_retry: Duration::from_secs(30), +}; + +/// How the controller paces [`REQUEST_STREAM`]. +#[derive(Debug, Clone, Copy)] +struct Timing { + /// The consumer's ack wait: how long a taken request may go without an + /// ack or a progress ack before the queue delivers it again. A request in + /// hand gets a progress ack every third of it, so a run of any length + /// keeps its request. + ack_wait: Duration, + /// How long one pull waits for a request; also the wait before opening + /// the consumer again after that failed. + pull_wait: Duration, + /// How long a request for a PR whose review is going waits before the + /// queue delivers it again. + busy_retry: Duration, +} + /// The forge's review events. #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum Event { @@ -78,8 +107,8 @@ fn no_clean_verdict_event() -> Event { Event::Comment } -/// A review request for [`LOGIN`]. -#[derive(Debug, Clone, PartialEq, Eq)] +/// A review request for [`LOGIN`], as it waits in [`REQUEST_STREAM`]. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub(crate) struct Requested { owner: String, name: String, @@ -147,29 +176,48 @@ impl Requested { }) } - fn key(&self) -> Key { - ( - self.owner.clone(), - self.name.clone(), - self.pr, - self.head.clone(), + fn pr_key(&self) -> PrKey { + (self.owner.clone(), self.name.clone(), self.pr) + } + + /// This PR's subject in [`REQUEST_STREAM`]. + fn subject(&self) -> String { + format!( + "{REQUEST_SUBJECT_PREFIX}.{}.{}.{}", + token(&self.owner), + token(&self.name), + self.pr ) } } -/// Repo owner, repo name, PR number and the head the request named. -type Key = (String, String, i64, String); +/// `text` as one subject token: ASCII letters, digits and `-` as they are, +/// every other byte as `_` and its two hex digits. So a `.` in a repo name +/// can't split it, and two names never share a token. +fn token(text: &str) -> String { + let mut out = String::with_capacity(text.len()); + for b in text.bytes() { + if b.is_ascii_alphanumeric() || b == b'-' { + out.push(char::from(b)); + } else { + let _ = write!(out, "_{b:02x}"); + } + } + out +} -/// The review runs this controller has accepted and not yet finished. +/// Repo owner, repo name and PR number. +type PrKey = (String, String, i64); + +/// The run places, and the PRs with a review going. /// -/// 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. +/// A request for a PR whose review is going waits in the queue until that +/// review ends, so one PR never has two runs at once. pub(crate) struct Reviews { slots: Mutex, /// Woken when a run ends or the limit changes. changed: Notify, - accepted: Mutex>, + going: Mutex>, } /// How many runs may go at once, and how many do. @@ -186,20 +234,11 @@ impl Default for Reviews { running: 0, }), changed: Notify::new(), - accepted: Mutex::default(), + going: Mutex::default(), } } } -/// Why [`Reviews::accept`] turned a request away. -#[derive(Debug, PartialEq, Eq)] -enum Refused { - /// 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), -} - impl Reviews { fn slots(&self) -> MutexGuard<'_, Slots> { self.slots.lock().unwrap_or_else(PoisonError::into_inner) @@ -217,18 +256,14 @@ impl Reviews { self.slots().limit } - fn accept(self: &Arc, key: Key) -> Result { - let mut accepted = self.accepted.lock().unwrap_or_else(PoisonError::into_inner); - if accepted.contains(&key) { - return Err(Refused::Duplicate); - } - if accepted.len() >= self.slots().limit + MAX_WAITING_REVIEWS { - return Err(Refused::Full(accepted.len())); - } - accepted.insert(key.clone()); - Ok(Accepted { + /// `req`'s PR, marked as having a review going, or `None` when one is + /// going already. + fn claim(self: &Arc, req: &Requested) -> Option { + let pr = req.pr_key(); + let mut going = self.going.lock().unwrap_or_else(PoisonError::into_inner); + going.insert(pr.clone()).then(|| Claim { reviews: Arc::clone(self), - key, + pr, }) } @@ -250,19 +285,19 @@ impl Reviews { } } -/// One accepted run; dropping it frees its key and its place. -struct Accepted { +/// One PR's review going; dropping it lets the PR's next request run. +struct Claim { reviews: Arc, - key: Key, + pr: PrKey, } -impl Drop for Accepted { +impl Drop for Claim { fn drop(&mut self) { self.reviews - .accepted + .going .lock() .unwrap_or_else(PoisonError::into_inner) - .remove(&self.key); + .remove(&self.pr); } } @@ -276,8 +311,8 @@ impl Drop for Slot { } } -/// Start a review run for a verified `review-request` delivery that asks for -/// [`LOGIN`], and return at once. +/// Queue the review request in a verified `review-request` delivery that +/// asks for [`LOGIN`], and return at once. pub(crate) fn on_delivery(state: &AppState, body: &[u8]) { let Some(request) = Requested::parse(body) else { return; @@ -290,105 +325,361 @@ pub(crate) fn on_delivery(state: &AppState, body: &[u8]) { ); return; }; - let queue = Queue(state.status.as_ref().map(|s| s.queue_client())); + let client = state.status.as_ref().map(|s| s.queue_client()); let runs = state.runs.clone(); - start( - &state.reviews, - Arc::clone(&forge), - queue, - forge, - runs, - request, - ); + tokio::spawn(async move { accept(client.as_ref(), &*forge, &runs, &request).await }); } -/// Accept `request` into `reviews` and spawn its run, or the no-verdict -/// comment when there's no place for it, and record it in `runs` when it -/// ends. The task, if one was spawned. +/// Put `request` in [`REQUEST_STREAM`]. When that fails, the account posts a +/// no-verdict comment saying so, recorded in `runs`. +async fn accept( + client: Option<&async_nats::Client>, + poster: &impl ReviewPoster, + runs: &Runs, + request: &Requested, +) { + let repo = format!("{}/{}", request.owner, request.name); + let queued = match client { + Some(client) => enqueue(client, request).await, + None => Err(anyhow::anyhow!("no swarm queue is configured on this host")), + }; + let error = match queued { + Ok(()) => { + tracing::info!( + %repo, pr = request.pr, head_sha = request.head.as_str(), + "review request queued" + ); + return; + } + Err(e) => format!("{e:#}"), + }; + tracing::warn!( + %repo, pr = request.pr, head_sha = request.head.as_str(), %error, + "queueing a review request failed; posting a no-verdict comment" + ); + let started_unix = unix_now(); + let ruling = Ruling::no_verdict( + format!( + "the request could not be queued: {error}; remove the review request and request \ + it again" + ), + &[], + ); + let reason = ruling.reason.clone(); + let sent = post(poster, request, None, ruling.review()).await; + let ended = ended(RunOutcome::NoVerdict, reason, sent); + runs.record(record(request, None, started_unix, ended, (None, None))) + .await; +} + +/// Open [`REQUEST_STREAM`], creating it if nothing has yet. +async fn open_stream(client: &async_nats::Client) -> anyhow::Result { + swarm_queue_client::ensure_connected(client)?; + jetstream::new(client.clone()) + .get_or_create_stream(jetstream::stream::Config { + name: REQUEST_STREAM.to_owned(), + description: Some( + "Review requests the review classifier has accepted and not yet finished" + .to_owned(), + ), + subjects: vec![format!("{REQUEST_SUBJECT_PREFIX}.>")], + retention: jetstream::stream::RetentionPolicy::WorkQueue, + max_messages_per_subject: 1, + ..Default::default() + }) + .await + .with_context(|| format!("opening the {REQUEST_STREAM} stream")) +} + +/// Publish `request` on its PR's subject, replacing a request for the PR +/// that's still waiting, once the queue has stored it. +async fn enqueue(client: &async_nats::Client, request: &Requested) -> anyhow::Result<()> { + open_stream(client).await?; + let payload = serde_json::to_vec(request).context("encoding the review request")?; + jetstream::new(client.clone()) + .publish(request.subject(), payload.into()) + .await + .context("publishing the review request")? + .await + .context("the queue did not store the review request")?; + Ok(()) +} + +/// Open [`REQUEST_CONSUMER`] on [`REQUEST_STREAM`], creating either, with +/// `ack_wait`. +async fn open_consumer( + client: &async_nats::Client, + ack_wait: Duration, +) -> anyhow::Result { + open_stream(client) + .await? + .create_consumer(jetstream::consumer::pull::Config { + durable_name: Some(REQUEST_CONSUMER.to_owned()), + description: Some("swarm-controller's requested review runs".to_owned()), + ack_policy: jetstream::consumer::AckPolicy::Explicit, + ack_wait, + ..Default::default() + }) + .await + .with_context(|| format!("opening the {REQUEST_CONSUMER} consumer on {REQUEST_STREAM}")) +} + +/// Take the queued review requests and review each, for as long as the +/// controller runs, once this host has forge access. +pub(crate) fn spawn_queue(state: &AppState) { + let Some(client) = state.status.as_ref().map(|s| s.queue_client()) else { + tracing::warn!("no swarm queue is configured on this host; requested reviews don't run"); + return; + }; + let forge = Arc::clone(&state.forge); + let reviews = Arc::clone(&state.reviews); + let runs = state.runs.clone(); + tokio::spawn(async move { + let forge = loop { + if let Some(forge) = forge.get() { + break Arc::clone(forge); + } + tokio::time::sleep(TIMING.pull_wait).await; + }; + let queue = Queue(Some(client.clone())); + consume( + &client, + TIMING, + &reviews, + Arc::clone(&forge), + queue, + forge, + runs, + ) + .await; + }); +} + +/// Take requests from [`REQUEST_STREAM`] one at a time, each once a run +/// place is free, and review each in a task of its own. Never returns; +/// dropping it stops every review it started, and the queue delivers their +/// requests again. /// -/// 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( +/// The limit is read from the settings before each wait for a place. A +/// request that doesn't parse is dropped. One for a PR whose review is going +/// waits for [`Timing::busy_retry`] in the queue, without holding a place. +async fn consume( + client: &async_nats::Client, + timing: Timing, reviews: &Arc, forge: Arc, queue: Q, poster: Arc

, runs: Runs, - request: Requested, -) -> Option> -where +) where S: FileSource + PullSource + Send + Sync + 'static, Q: RunSink + SettingsSource, P: ReviewPoster + Send + Sync + 'static, { - let repo = format!("{}/{}", request.owner, request.name); - match reviews.accept(request.key()) { - 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!( - error = %format!("{e:#}"), - "reading the review classifier's settings failed; the run waits under \ - the last limit read" - ), - } - let slot = reviews.slot().await; - let run = Box::pin(run(&*forge, &*forge, &queue, &*poster, &request)).await; + let consumer = loop { + match open_consumer(client, timing.ack_wait).await { + Ok(consumer) => break consumer, + Err(e) => tracing::warn!( + error = %format!("{e:#}"), + "opening the review request queue failed; trying again" + ), + } + tokio::time::sleep(timing.pull_wait).await; + }; + let mut tasks = JoinSet::new(); + loop { + while let Some(ended) = tasks.try_join_next() { + if let Err(e) = ended { + tracing::error!(error = %e, "a requested review's task ended early"); + } + } + match queue.settings().await { + Ok(settings) => reviews.resize(settings.max_concurrent_reviews), + Err(e) => tracing::warn!( + error = %format!("{e:#}"), + "reading the review classifier's settings failed; the next run waits under \ + the last limit read" + ), + } + let slot = reviews.slot().await; + let message = match next(&consumer, timing.pull_wait).await { + Ok(Some(message)) => message, + Ok(None) => continue, + Err(e) => { + tracing::warn!( + error = %format!("{e:#}"), + "taking a review request from the queue failed; trying again" + ); drop(slot); - runs.record(run).await; - drop(accepted); - })) - } - Err(Refused::Duplicate) => { - tracing::info!( - %repo, pr = request.pr, head_sha = request.head.as_str(), - "review requested while this head is running, waiting or re-posting; nothing more starts" - ); - None - } - Err(Refused::Full(taken)) => { + tokio::time::sleep(timing.pull_wait).await; + continue; + } + }; + tasks.spawn(take( + message, + slot, + timing, + Arc::clone(reviews), + Arc::clone(&forge), + queue.clone(), + Arc::clone(&poster), + runs.clone(), + )); + } +} + +/// The next request in the queue, or `None` when none came within `wait`. +async fn next( + consumer: &PullConsumer, + wait: Duration, +) -> anyhow::Result> { + let mut batch = consumer + .batch() + .max_messages(1) + .expires(wait) + .messages() + .await + .context("asking the queue for a review request")?; + match batch.next().await { + Some(Ok(message)) => Ok(Some(message)), + Some(Err(e)) => Err(anyhow::anyhow!(e).context("reading a review request")), + None => Ok(None), + } +} + +/// Review the request in `message`, holding `slot`, and ack it once its +/// record is written. +#[expect( + clippy::too_many_arguments, + reason = "each is one thing a review reaches" +)] +async fn take( + message: jetstream::Message, + slot: Slot, + timing: Timing, + reviews: Arc, + forge: Arc, + queue: Q, + poster: Arc

, + runs: Runs, +) where + S: FileSource + PullSource + Send + Sync + 'static, + Q: RunSink + SettingsSource, + P: ReviewPoster + Send + Sync + 'static, +{ + let request: Requested = match serde_json::from_slice(&message.payload) { + Ok(request) => request, + Err(e) => { tracing::warn!( - %repo, pr = request.pr, head_sha = request.head.as_str(), - "review requested, but every run and wait place is taken; posting a no-verdict comment" + subject = %message.subject, error = %e, + "a queued review request did not parse; dropping it" ); - Some(tokio::spawn(async move { - let started_unix = unix_now(); - let ruling = queue_full_ruling(taken); - let reason = ruling.reason.clone(); - let sent = post(&*poster, &request, None, ruling.review()).await; - let ended = ended(RunOutcome::QueueFull, reason, sent); - runs.record(record(&request, None, started_unix, ended, (None, None))) - .await; - })) + if let Err(e) = message.ack_with(AckKind::Term).await { + tracing::warn!(error = %e, "dropping an unparsable review request failed"); + } + return; + } + }; + let Some(claim) = reviews.claim(&request) else { + drop(slot); + tracing::info!( + repo = %format!("{}/{}", request.owner, request.name), pr = request.pr, + head_sha = request.head.as_str(), + "review requested while this PR's review is going; it waits in the queue" + ); + if let Err(e) = message + .ack_with(AckKind::Nak(Some(timing.busy_retry))) + .await + { + tracing::warn!(error = %e, "putting a review request back in the queue failed"); + } + return; + }; + let work = review(&*forge, &*forge, &queue, &*poster, &runs, &request, slot); + in_progress(&message, timing.ack_wait / 3, work).await; + drop(claim); + if let Err(e) = message.ack().await { + tracing::warn!( + repo = %format!("{}/{}", request.owner, request.name), pr = request.pr, + error = %e, + "acking a finished review request failed; the queue delivers it again" + ); + } +} + +/// Run `work`, telling the queue every `every` that `message` is still in +/// hand. +async fn in_progress( + message: &jetstream::Message, + every: Duration, + work: impl Future, +) { + tokio::pin!(work); + let mut ticks = tokio::time::interval_at(tokio::time::Instant::now() + every, every); + loop { + tokio::select! { + () = &mut work => return, + _ = ticks.tick() => { + if let Err(e) = message.ack_with(AckKind::Progress).await { + tracing::warn!(error = %e, "telling the queue a review is still going failed"); + } + } } } } -/// The ruling for a request [`Reviews`] had no place for, with `taken` runs -/// running or waiting. -fn queue_full_ruling(taken: usize) -> Ruling { - let reason = format!( - "{taken} reviews are running or waiting, so this request starts no run; \ - request the review again later" - ); - Ruling { - event: no_clean_verdict_event(), - body: no_verdict(&reason, &[]), - reason: Some(reason), +/// Review `req`, holding `slot`, and record the result in `runs`. +/// +/// A request whose head is no longer the PR's is dropped: nothing is posted +/// or recorded. One whose head's newest review [`reposted`] covers frees +/// `slot` and posts that verdict again. Any other runs, reading the settings +/// for the model and effort. +async fn review( + files: &impl FileSource, + pulls: &impl PullSource, + sink: &(impl RunSink + SettingsSource), + poster: &impl ReviewPoster, + runs: &Runs, + req: &Requested, + slot: Slot, +) { + let pull = read_pull(pulls, req).await; + if let Ok(pull) = &pull + && !req.head.is_empty() + && pull.head_sha != req.head + { + tracing::info!( + repo = %format!("{}/{}", req.owner, req.name), pr = req.pr, + head_sha = req.head.as_str(), current = pull.head_sha.as_str(), + "review requested on a head the PR has moved on from; dropping the request" + ); + return; + } + if let Some(earlier) = newest_review(runs, req).await + && let Some(ruling) = reposted(&earlier) + { + drop(slot); + runs.record(repost(poster, req, &earlier, ruling).await) + .await; + return; + } + let run = Box::pin(run(files, pull, sink, poster, req)).await; + drop(slot); + runs.record(run).await; +} + +/// The PR `req` names, as the forge answers within [`forge::FORGE_DEADLINE`], +/// or why it couldn't be read. +async fn read_pull(pulls: &impl PullSource, req: &Requested) -> Result { + match forge::within_forge_deadline( + "review classifier PR read", + pulls.pull(&req.owner, &req.name, req.pr), + ) + .await + { + Some(Ok(pull)) => Ok(pull), + Some(Err(e)) => Err(format!("reading the PR failed: {e:#}")), + None => Err("the forge did not answer within the deadline".to_owned()), } } @@ -529,17 +820,17 @@ enum Outcome { Ran { sent: usize, reply: Reply }, } -/// Read the PR and the rules, run them, and post the review. The finished -/// run. +/// Read the rules, run them on `pull`, the PR as it was read, and post the +/// review. The finished run. async fn run( files: &impl FileSource, - pulls: &impl PullSource, + pull: Result, sink: &(impl RunSink + SettingsSource), poster: &impl ReviewPoster, req: &Requested, ) -> RunRecord { let started_unix = unix_now(); - let (head, narrowed, outcome) = classify(files, pulls, sink, req).await; + let (head, narrowed, outcome) = classify(files, pull, sink, req).await; let mut ruling = review_for(&outcome); if let (Some(reason), Outcome::Ran { .. }) = (narrowed, &outcome) { let _ = write!( @@ -717,35 +1008,17 @@ fn record( } /// The PR's head when it was read, why its diff is the forge's 3-line one -/// when it is, and how far the run got. +/// when it is, and how far the run got on `pull`, the PR as it was read. async fn classify( files: &impl FileSource, - pulls: &impl PullSource, + pull: Result, sink: &(impl RunSink + SettingsSource), req: &Requested, ) -> (Option, Option, Outcome) { let repo = format!("{}/{}", req.owner, req.name); - let pull = match forge::within_forge_deadline( - "review classifier PR read", - pulls.pull(&req.owner, &req.name, req.pr), - ) - .await - { - Some(Ok(pull)) => pull, - Some(Err(e)) => { - return ( - None, - None, - Outcome::NotRun(format!("reading the PR failed: {e:#}")), - ); - } - None => { - return ( - None, - None, - Outcome::NotRun("the forge did not answer within the deadline".to_owned()), - ); - } + let pull = match pull { + Ok(pull) => pull, + Err(reason) => return (None, None, Outcome::NotRun(reason)), }; let head = Some(pull.head_sha.clone()); let narrowed = pull.narrowed; @@ -988,6 +1261,8 @@ mod tests { use super::*; use crate::review_classifier::runs::tests::FakeDoc; use crate::review_classifier::tests::{Fake, rules_json}; + use crate::test_nats::TestNats; + use tokio::task::JoinHandle; /// Every review [`run`] posted, with the commit it named. #[derive(Default)] @@ -1132,6 +1407,16 @@ mod tests { } } + /// [`run`] on `req`'s PR as `forge` reads it. + async fn run_read( + forge: &(impl FileSource + PullSource), + sink: &(impl RunSink + SettingsSource), + poster: &impl ReviewPoster, + req: &Requested, + ) -> RunRecord { + run(forge, read_pull(forge, req).await, sink, poster, req).await + } + /// A JSON rules file holding `n` classifications. fn rules_of(n: usize) -> (&'static str, String) { let classifications: Vec<_> = (0..n) @@ -1164,7 +1449,7 @@ mod tests { ) -> (String, Option, Review) { let fake = Fake::new(reply, vec![rules]); let poster = Posted::default(); - run(&fake, &fake, &fake, &poster, &request(repo)).await; + run_read(&fake, &fake, &poster, &request(repo)).await; let mut all = poster.0.into_inner().expect("unpoisoned"); assert_eq!(all.len(), 1, "one review per run"); all.remove(0) @@ -1275,13 +1560,6 @@ mod tests { } } - /// Let every spawned task run until it waits on something. - async fn settle() { - for _ in 0..200 { - tokio::task::yield_now().await; - } - } - fn requested_at(head: &str) -> Requested { Requested { head: head.to_owned(), @@ -1289,133 +1567,6 @@ mod tests { } } - #[tokio::test] - async fn a_request_for_a_head_already_running_or_waiting_starts_nothing() { - let reviews = Arc::new(Reviews::default()); - let fake = Arc::new(Fake::new( - Reply::Failed { - error: String::new(), - }, - vec![rules_json()], - )); - let gate = Gate::closed(); - let poster = Arc::new(Posted::default()); - let doc = Arc::new(FakeDoc::default()); - let store = Runs::Fake(Arc::clone(&doc)); - let start_at = |head: &str| { - start( - &reviews, - Arc::clone(&fake), - gate.clone(), - Arc::clone(&poster), - store.clone(), - requested_at(head), - ) - }; - - // The first two run, the third waits; a repeat of any starts nothing. - let mut tasks: Vec<_> = ["a", "b", "c"] - .map(|h| start_at(h).expect("accepted")) - .into(); - for head in ["a", "b", "c"] { - assert!(start_at(head).is_none(), "{head}"); - } - gate.open(1); - for task in tasks.drain(..) { - task.await.expect("ran"); - } - assert_eq!(poster.0.lock().expect("unpoisoned").len(), 3); - assert_eq!( - doc.runs().len(), - 3, - "a repeat that started nothing is no run" - ); - - // A finished run frees its head. - start_at("a").expect("accepted again").await.expect("ran"); - assert_eq!(poster.0.lock().expect("unpoisoned").len(), 4); - assert_eq!(doc.runs().len(), 4); - } - - #[tokio::test] - async fn a_request_beyond_the_wait_places_posts_a_no_verdict_comment() { - let reviews = Arc::new(Reviews::default()); - let fake = Arc::new(Fake::new( - Reply::Failed { - error: String::new(), - }, - vec![rules_json()], - )); - let gate = Gate::closed(); - let poster = Arc::new(Posted::default()); - let doc = Arc::new(FakeDoc::default()); - let store = Runs::Fake(Arc::clone(&doc)); - let start_at = |head: String| { - start( - &reviews, - Arc::clone(&fake), - gate.clone(), - Arc::clone(&poster), - store.clone(), - requested_at(&head), - ) - .expect("a run or a comment") - }; - - let places = Settings::default().max_concurrent_reviews + MAX_WAITING_REVIEWS; - let runs: Vec<_> = (0..places).map(|i| start_at(format!("h{i}"))).collect(); - tokio::time::timeout(Duration::from_secs(5), start_at("over".to_owned())) - .await - .expect("refused without waiting for a place") - .expect("posted"); - { - let all = poster.0.lock().expect("unpoisoned"); - assert_eq!(all.len(), 1, "only the refusal is posted yet"); - let (target, commit, review) = &all[0]; - assert_eq!(target, "o/r#7"); - assert_eq!(*commit, None); - assert_eq!(review.event, "COMMENT"); - assert!( - review.body.starts_with( - "**No verdict:** 10 reviews are running or waiting, so this request" - ), - "{}", - review.body - ); - } - let refused = doc.runs(); - assert_eq!(refused.len(), 1); - assert_eq!(refused[0].outcome, RunOutcome::QueueFull); - assert_eq!(refused[0].head_sha, None); - assert_eq!( - refused[0].review_url.as_deref(), - Some("https://forge/o/r/pulls/7#review") - ); - assert!( - refused[0] - .detail - .as_deref() - .is_some_and(|d| d.starts_with("10 reviews are running or waiting")), - "{:?}", - refused[0].detail - ); - - gate.open(1); - for run in runs { - run.await.expect("ran"); - } - let all = poster.0.lock().expect("unpoisoned"); - assert_eq!(all.len(), places + 1); - assert!(all[1..].iter().all(|(_, _, r)| r.event == "APPROVED")); - let recorded = doc.runs(); - assert_eq!(recorded.len(), places + 1); - assert!( - recorded[..places] - .iter() - .all(|r| r.outcome == RunOutcome::Approved) - ); - } - /// 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 { @@ -1437,11 +1588,68 @@ mod tests { earlier } + /// The fake forge's files, and each PR's head on o/r: `abc123` until + /// [`Heads::move_to`] moves it. + struct Heads { + fake: Fake, + heads: Mutex>, + } + + impl Heads { + fn new() -> Self { + Self { + fake: Fake::new( + Reply::Failed { + error: String::new(), + }, + vec![rules_json()], + ), + heads: Mutex::default(), + } + } + + fn move_to(&self, pr: i64, head: &str) { + self.heads + .lock() + .expect("unpoisoned") + .insert(pr, head.to_owned()); + } + } + + impl PullSource for Heads { + async fn pull( + &self, + owner: &str, + repo: &str, + number: i64, + ) -> anyhow::Result { + anyhow::ensure!((owner, repo) == ("o", "r"), "no such repo"); + let head = self.heads.lock().expect("unpoisoned").get(&number).cloned(); + Ok(PullForReview { + head_sha: head.unwrap_or_else(|| "abc123".to_owned()), + commits: Vec::new(), + diff: String::new(), + narrowed: None, + }) + } + } + + impl FileSource for Heads { + async fn default_branch_files( + &self, + owner: &str, + repo: &str, + paths: &[&str], + ) -> anyhow::Result> { + self.fake.default_branch_files(owner, repo, paths).await + } + } + /// A fake forge, a closed [`Gate`], a poster and a run store, and a - /// starter for requests at a head on o/r#7. + /// starter for reviews of a head on o/r#7. struct Rig { reviews: Arc, - fake: Arc, + forge: Arc, gate: Gate, poster: Arc, doc: Arc, @@ -1450,15 +1658,13 @@ mod tests { impl Rig { fn new() -> Self { - let doc = Arc::new(FakeDoc::default()); + Self::with_doc(Arc::default()) + } + + fn with_doc(doc: Arc) -> Self { Self { reviews: Arc::new(Reviews::default()), - fake: Arc::new(Fake::new( - Reply::Failed { - error: String::new(), - }, - vec![rules_json()], - )), + forge: Arc::new(Heads::new()), gate: Gate::closed(), poster: Arc::new(Posted::default()), store: Runs::Fake(Arc::clone(&doc)), @@ -1466,15 +1672,18 @@ mod tests { } } - 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), - ) + /// [`review`] of a request at `head`, once it has a run place. + fn start_at(&self, head: &str) -> JoinHandle<()> { + let reviews = Arc::clone(&self.reviews); + let forge = Arc::clone(&self.forge); + let gate = self.gate.clone(); + let poster = Arc::clone(&self.poster); + let store = self.store.clone(); + let request = requested_at(head); + tokio::spawn(async move { + let slot = reviews.slot().await; + review(&*forge, &*forge, &gate, &*poster, &store, &request, slot).await; + }) } fn posted(&self) -> Vec<(String, Option, Review)> { @@ -1508,7 +1717,7 @@ mod tests { let rig = Rig::new(); let earlier = reviewed(&rig.store, "earlier", outcome).await; - reposted(rig.start_at("abc123").expect("accepted")).await; + reposted(rig.start_at("abc123")).await; assert_eq!( rig.gate.reached(), @@ -1562,7 +1771,7 @@ mod tests { reviewed(&rig.store, &format!("earlier-{i}"), outcome).await; } - let task = rig.start_at("abc123").expect("accepted"); + let task = rig.start_at("abc123"); rig.gate.open(1); task.await.expect("ran"); @@ -1607,45 +1816,373 @@ mod tests { let rig = Rig::new(); rig.store.record(earlier).await; - let task = rig.start_at("abc123").expect("accepted"); + let task = rig.start_at("abc123"); 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); + /// The queue's timing in the tests: short enough that a request no one + /// acks comes back within a test. + const TEST_TIMING: Timing = Timing { + ack_wait: Duration::from_secs(1), + pull_wait: Duration::from_millis(200), + busy_retry: Duration::from_millis(200), + }; - // 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); + /// A [`Rig`] behind a real queue: requests go in through [`accept`], and + /// [`consume`] takes them. + struct Queued { + _nats: TestNats, + client: async_nats::Client, + rig: Rig, + } + + impl Queued { + async fn new() -> Self { + let nats = TestNats::start(); + let client = nats.client().await; + Self { + _nats: nats, + client, + rig: Rig::new(), + } + } + + /// Queue a request for o/r#`pr` at `head`. + async fn request(&self, pr: i64, head: &str) { + let request = Requested { + pr, + ..requested_at(head) + }; + accept( + Some(&self.client), + &*self.rig.poster, + &self.rig.store, + &request, + ) + .await; + assert!(self.rig.posted().is_empty(), "queued without a comment"); + } + + /// A controller taking requests, its run places in `reviews`. + fn consume(&self, reviews: &Arc) -> JoinHandle<()> { + let client = self.client.clone(); + let reviews = Arc::clone(reviews); + let forge = Arc::clone(&self.rig.forge); + let gate = self.rig.gate.clone(); + let poster = Arc::clone(&self.rig.poster); + let store = self.rig.store.clone(); + tokio::spawn(async move { + consume( + &client, + TEST_TIMING, + &reviews, + Arc::clone(&forge), + gate, + poster, + store, + ) + .await; + }) + } + + /// How many requests the queue holds, taken or not. + async fn waiting(&self) -> u64 { + jetstream::new(self.client.clone()) + .get_stream(REQUEST_STREAM) + .await + .expect("the stream") + .info() + .await + .expect("its info") + .state + .messages + } + + async fn consumer_info(&self) -> jetstream::consumer::Info { + let js = jetstream::new(self.client.clone()); + let stream = js.get_stream(REQUEST_STREAM).await.expect("the stream"); + let mut consumer: PullConsumer = stream + .get_consumer(REQUEST_CONSUMER) + .await + .expect("the consumer"); + consumer.info().await.expect("its info").clone() + } + } + + /// Wait until `done` holds, for at most ten seconds. + async fn until(what: &str, done: impl Fn() -> bool) { + tokio::time::timeout(Duration::from_secs(10), async { + while !done() { + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .unwrap_or_else(|_| panic!("never: {what}")); + } + + #[tokio::test] + async fn a_queued_request_runs_after_a_restart() { + let q = Queued::new().await; + q.request(7, "abc123").await; + let first = q.consume(&q.rig.reviews); + until("the first controller's run started", || { + q.rig.gate.reached() == 1 + }) + .await; + first.abort(); + assert!(first.await.expect_err("stopped").is_cancelled()); + assert!(q.rig.posted().is_empty()); + + // A restarted controller has run places of its own. + let _second = q.consume(&Arc::new(Reviews::default())); + until("the request ran again", || q.rig.gate.reached() == 2).await; + q.rig.gate.open(1); + until("its review is posted", || q.rig.posted().len() == 1).await; + let posted = q.rig.posted(); + assert_eq!(posted[0].0, "o/r#7"); + assert_eq!(posted[0].2.event, "APPROVED"); + until("its record is written", || q.rig.doc.runs().len() == 1).await; + tokio::time::sleep(TEST_TIMING.ack_wait * 2).await; + assert_eq!(q.waiting().await, 0, "the finished request is acked"); + assert_eq!(q.rig.gate.reached(), 2, "no run after the ack"); + } + + #[tokio::test] + async fn a_newer_request_for_a_pr_replaces_the_waiting_one() { + let q = Queued::new().await; + q.rig.forge.move_to(7, "new"); + q.request(7, "old").await; + q.request(8, "abc123").await; + q.request(7, "new").await; + assert_eq!(q.waiting().await, 2, "one request per PR"); + + let _consumer = q.consume(&q.rig.reviews); + q.rig.gate.open(2); + until("both PRs are reviewed", || q.rig.posted().len() == 2).await; + tokio::time::sleep(TEST_TIMING.ack_wait * 2).await; + let mut posted: Vec<_> = q + .rig + .posted() + .into_iter() + .map(|(target, commit, _)| (target, commit)) + .collect(); + posted.sort(); + assert_eq!( + posted, + [ + ("o/r#7".to_owned(), Some("new".to_owned())), + ("o/r#8".to_owned(), Some("abc123".to_owned())), + ] + ); + assert_eq!(q.rig.gate.reached(), 2, "one run per PR"); + } + + #[tokio::test] + async fn a_request_for_a_head_the_pr_has_moved_on_from_is_dropped() { + let q = Queued::new().await; + q.rig.forge.move_to(7, "new"); + q.request(7, "old").await; + let _consumer = q.consume(&q.rig.reviews); + tokio::time::timeout(Duration::from_secs(10), async { + while q.waiting().await > 0 { + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .expect("the request is acked"); + assert!(q.rig.posted().is_empty()); + assert!(q.rig.doc.runs().is_empty(), "nothing is recorded"); + assert_eq!(q.rig.gate.reached(), 0); + + // Control: the PR's head runs. + q.request(7, "new").await; + q.rig.gate.open(1); + until("the head is reviewed", || q.rig.posted().len() == 1).await; + assert_eq!(q.rig.posted()[0].1.as_deref(), Some("new")); + } + + #[tokio::test] + async fn nothing_is_refused_while_every_run_place_is_taken() { + let q = Queued::new().await; + let prs = 1..=12; + for pr in prs.clone() { + q.request(pr, "abc123").await; + } + let _consumer = q.consume(&q.rig.reviews); + let limit = Settings::default().max_concurrent_reviews; + until("every place is taken", || q.rig.gate.reached() == limit).await; + tokio::time::sleep(TEST_TIMING.ack_wait * 2).await; + assert_eq!(q.rig.gate.reached(), limit, "the rest wait"); + assert!(q.rig.posted().is_empty(), "nothing is refused"); + assert_eq!(q.waiting().await, 12); + + q.rig.gate.open(12); + until("every PR is reviewed", || q.rig.posted().len() == 12).await; + let mut targets: Vec<_> = q.rig.posted().into_iter().map(|(t, ..)| t).collect(); + targets.sort(); + let mut wanted: Vec<_> = prs.map(|pr| format!("o/r#{pr}")).collect(); + wanted.sort(); + assert_eq!(targets, wanted); + assert!(q.rig.posted().iter().all(|(_, _, r)| r.event == "APPROVED")); + assert_eq!(q.rig.gate.reached(), 12); + } + + #[tokio::test] + async fn a_run_longer_than_the_ack_wait_keeps_its_request() { + let q = Queued::new().await; + q.request(7, "abc123").await; + let _consumer = q.consume(&q.rig.reviews); + until("the run started", || q.rig.gate.reached() == 1).await; + tokio::time::sleep(TEST_TIMING.ack_wait * 3).await; + let info = q.consumer_info().await; + assert_eq!(info.num_ack_pending, 1, "the request is in hand"); + assert_eq!(info.num_redelivered, 0, "and was never delivered again"); + q.rig.gate.open(1); + until("its review is posted", || q.rig.posted().len() == 1).await; + assert_eq!(q.rig.gate.reached(), 1); + } + + #[tokio::test] + async fn a_request_for_a_pr_whose_review_is_going_waits_for_it() { + let q = Queued::new().await; + q.request(7, "abc123").await; + let _consumer = q.consume(&q.rig.reviews); + until("the run started", || q.rig.gate.reached() == 1).await; + q.request(7, "abc123").await; + tokio::time::sleep(TEST_TIMING.ack_wait * 2).await; + assert_eq!(q.rig.gate.reached(), 1, "one run on the PR at once"); + + // The run approved, so the waiting request posts its verdict again. + q.rig.gate.open(1); + until("both are posted", || q.rig.posted().len() == 2).await; + let posted = q.rig.posted(); + assert_eq!(posted[0].2.event, "APPROVED"); assert!( - posted[1].2.body.starts_with("Already reviewed"), + posted[1].2.body.starts_with("Already reviewed `abc123`"), "{}", posted[1].2.body ); + assert_eq!(q.rig.gate.reached(), 1); + } - // A clean verdict: two requests at once post it once. + #[tokio::test] + async fn a_request_that_does_not_parse_is_dropped() { + let q = Queued::new().await; + q.request(8, "abc123").await; + jetstream::new(q.client.clone()) + .publish(format!("{REQUEST_SUBJECT_PREFIX}.o.r.7"), "{}".into()) + .await + .expect("published") + .await + .expect("stored"); + let _consumer = q.consume(&q.rig.reviews); + q.rig.gate.open(2); + until("the request that parses is reviewed", || { + q.rig.posted().len() == 1 + }) + .await; + tokio::time::timeout(Duration::from_secs(10), async { + while q.waiting().await > 0 { + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .expect("both leave the queue"); + assert_eq!(q.rig.posted()[0].0, "o/r#8"); + assert_eq!(q.rig.gate.reached(), 1); + } + + #[tokio::test] + async fn a_request_that_cannot_be_queued_posts_a_no_verdict_comment() { 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); + accept(None, &*rig.poster, &rig.store, &requested_at("abc123")).await; + let posted = rig.posted(); + assert_eq!(posted.len(), 1); + let (target, commit, review) = &posted[0]; + assert_eq!(target, "o/r#7"); + assert_eq!(*commit, None); + assert_eq!(review.event, "COMMENT"); + assert!( + review.body.starts_with( + "**No verdict:** the request could not be queued: no swarm queue is configured \ + on this host; remove the review request and request it again." + ), + "{}", + review.body + ); + let recorded = rig.doc.runs(); + assert_eq!(recorded.len(), 1); + assert_eq!(recorded[0].outcome, RunOutcome::NoVerdict); + assert_eq!(recorded[0].head_sha, None); + } + + #[test] + fn each_pr_has_a_subject_of_its_own() { + let at = |owner: &str, name: &str, pr| { + Requested { + owner: owner.to_owned(), + name: name.to_owned(), + pr, + ..request("r") + } + .subject() + }; + assert_eq!( + at("hyperhive", "hyperhive", 5219), + "$SWARM.review-request.hyperhive.hyperhive.5219" + ); + assert_eq!( + at("a.b", "c", 1), + "$SWARM.review-request.a_2eb.c.1", + "a dot stays inside its token" + ); + assert_ne!(at("a.b", "c", 1), at("a", "b.c", 1)); + assert_ne!(at("a_2eb", "c", 1), at("a.b", "c", 1)); + assert_ne!(at("o", "r", 1), at("o", "r", 11)); + for subject in [at("a.b", "c d*>", 1), at("ΓΌ", "r", 1)] { + assert_eq!(subject.split('.').count(), 5, "{subject}"); + assert!( + !subject.contains(['*', '>', ' ']), + "{subject} is a literal subject" + ); + } + } + + #[tokio::test] + async fn a_new_limit_applies_to_the_next_run_without_a_restart() { + let q = Queued::new().await; + let reviews = &q.rig.reviews; + let gate = &q.rig.gate; + + // The stored limit is read before a run waits, not the default 2. + gate.store(1); + q.request(1, "abc123").await; + q.request(2, "abc123").await; + let _consumer = q.consume(reviews); + until("the first run started", || gate.reached() == 1).await; + tokio::time::sleep(TEST_TIMING.ack_wait).await; + assert_eq!(gate.reached(), 1); + + // Raised, as a PUT does: the waiting run starts at once. + gate.store(3); + reviews.resize(3); + until("the waiting run started", || gate.reached() == 2).await; + q.request(3, "abc123").await; + until("a third run started", || gate.reached() == 3).await; + + // Lowered: the three keep running and the next one waits. + gate.store(1); + q.request(4, "abc123").await; + tokio::time::sleep(TEST_TIMING.ack_wait).await; + assert_eq!(gate.reached(), 3); + + gate.open(4); + until("every run is posted", || q.rig.posted().len() == 4).await; + assert_eq!(gate.reached(), 4); + until("every place is freed", || reviews.slots().running <= 1).await; } /// The record of one run against the fake forge, holding `rules`, posted @@ -1657,7 +2194,7 @@ mod tests { poster: &impl ReviewPoster, ) -> RunRecord { let fake = Fake::new(reply, vec![rules]); - run(&fake, &fake, &fake, poster, &request(repo)).await + run_read(&fake, &fake, poster, &request(repo)).await } #[tokio::test] @@ -1743,7 +2280,7 @@ mod tests { } let fake = Fake::new(failed, Vec::new()); - let run = run(&fake, &fake, &fake, &Posted::default(), &request("r")).await; + let run = run_read(&fake, &fake, &Posted::default(), &request("r")).await; assert_eq!(run.outcome, RunOutcome::NoVerdict); assert!( run.detail @@ -1867,7 +2404,7 @@ mod tests { 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; + let run = run_read(&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); @@ -1930,7 +2467,7 @@ mod tests { tries: AtomicUsize::new(0), posted: Posted::default(), }; - let run = run(&fake, &fake, &fake, &poster, &request("r")).await; + let run = run_read(&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"); @@ -1972,7 +2509,7 @@ mod tests { let started = tokio::time::Instant::now(); let run = tokio::time::timeout( Duration::from_hours(1), - run(&fake, &fake, &fake, &poster, &request("r")), + run_read(&fake, &fake, &poster, &request("r")), ) .await .expect("the hung try is abandoned"); @@ -1993,7 +2530,7 @@ mod tests { ForgejoError::UnexpectedStatusCode(StatusCode::FORBIDDEN).into(), ]); let started = tokio::time::Instant::now(); - let run = run(&fake, &fake, &fake, &poster, &request("r")).await; + let run = run_read(&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()); @@ -2013,7 +2550,7 @@ mod tests { }; let fake = Fake::new(approve, vec![rules_json()]); let poster = Flaky::new(failures); - let run = run(&fake, &fake, &fake, &poster, &request("r")).await; + let run = run_read(&fake, &fake, &poster, &request("r")).await; assert_eq!(run.outcome, RunOutcome::PostFailed); assert_eq!( poster.tries.load(Ordering::SeqCst), @@ -2030,87 +2567,15 @@ mod tests { #[tokio::test] async fn a_failed_record_write_still_posts_the_review() { - let fake = Arc::new(Fake::new( - Reply::Done { - rows: vec![row(None, true, true)], - overall: Overall::Approve, - }, - vec![rules_json()], - )); - let poster = Arc::new(Posted::default()); - let store = Runs::Fake(Arc::new(FakeDoc::failing())); - start( - &Arc::new(Reviews::default()), - Arc::clone(&fake), - (*fake).clone(), - Arc::clone(&poster), - store, - request("r"), - ) - .expect("accepted") - .await - .expect("ran to the end"); - let all = poster.0.lock().expect("unpoisoned"); + let rig = Rig::with_doc(Arc::new(FakeDoc::failing())); + let task = rig.start_at("abc123"); + rig.gate.open(1); + task.await.expect("ran to the end"); + let all = rig.posted(); assert_eq!(all.len(), 1); assert_eq!(all[0].2.event, "APPROVED"); } - #[tokio::test] - async fn a_new_limit_applies_to_the_next_run_without_a_restart() { - let reviews = Arc::new(Reviews::default()); - let fake = Arc::new(Fake::new( - Reply::Failed { - error: String::new(), - }, - vec![rules_json()], - )); - let gate = Gate::closed(); - let poster = Arc::new(Posted::default()); - let start_at = |head: &str| { - start( - &reviews, - Arc::clone(&fake), - gate.clone(), - Arc::clone(&poster), - Runs::Fake(Arc::default()), - requested_at(head), - ) - .expect("accepted") - }; - - // The stored limit is read before a run waits, not the default 2. - gate.store(1); - let mut tasks = vec![start_at("a"), start_at("b")]; - settle().await; - assert_eq!(gate.reached(), 1); - - // Raised, as a PUT does: the waiting run starts at once. - gate.store(3); - reviews.resize(3); - settle().await; - assert_eq!(gate.reached(), 2); - tasks.push(start_at("c")); - settle().await; - assert_eq!(gate.reached(), 3); - - // Lowered: the three keep running and the next one waits. - gate.store(1); - tasks.push(start_at("d")); - settle().await; - assert_eq!(gate.reached(), 3); - - gate.open(1); - for task in tasks { - tokio::time::timeout(Duration::from_secs(5), task) - .await - .expect("a finished run frees its place") - .expect("ran"); - } - assert_eq!(gate.reached(), 4); - assert_eq!(reviews.slots().running, 0, "every place is freed"); - assert_eq!(poster.0.lock().expect("unpoisoned").len(), 4); - } - #[tokio::test] async fn a_review_runs_on_the_stored_runtime_model_and_effort() { use swarm_wire::agent_inference::Effort; @@ -2140,7 +2605,7 @@ mod tests { acp: Some(acp.clone()), }; let poster = Posted::default(); - run(&fake, &fake, &fake, &poster, &request("r")).await; + run_read(&fake, &fake, &poster, &request("r")).await; let sent = fake.sent.lock().expect("unpoisoned"); assert_eq!(sent.len(), 1); assert_eq!( @@ -2163,7 +2628,7 @@ mod tests { vec![rules_json()], ); let poster = Posted::default(); - run(&fake, &fake, &Queue(None), &poster, &request("r")).await; + run_read(&fake, &Queue(None), &poster, &request("r")).await; let (_, commit, review) = poster.0.into_inner().expect("unpoisoned").remove(0); assert_eq!(review.event, "COMMENT"); assert!( @@ -2237,7 +2702,7 @@ mod tests { let fake = Fake::new(approve(), Vec::new()); let poster = Posted::default(); - run(&fake, &fake, &fake, &poster, &request("r")).await; + run_read(&fake, &fake, &poster, &request("r")).await; let (_, _, review) = poster.0.into_inner().expect("unpoisoned").remove(0); assert_eq!(review.event, "COMMENT"); assert!( @@ -2382,7 +2847,7 @@ mod tests { let mut fake = Fake::new(reply, files); fake.narrowed = reason; let poster = Posted::default(); - run(&fake, &fake, &fake, &poster, &request("r")).await; + run_read(&fake, &fake, &poster, &request("r")).await; poster.0.into_inner().expect("unpoisoned").remove(0).2.body } let reason = || Some("the PR changes 101 files".to_owned()); diff --git a/swarm-controller/src/review_classifier/runs.rs b/swarm-controller/src/review_classifier/runs.rs index 454bc8222..f19ae13b5 100644 --- a/swarm-controller/src/review_classifier/runs.rs +++ b/swarm-controller/src/review_classifier/runs.rs @@ -60,12 +60,14 @@ pub(crate) enum RunOutcome { Approved, /// A review requested changes. ChangesRequested, - /// A comment was posted: the rules did not load, the PR was unreadable, - /// the run failed, the classifier sent a different number of answers - /// than it was sent classifications, or any classification got none. + /// A comment was posted: the request could not be queued, the rules did + /// not load, the PR was unreadable, the run failed, the classifier sent a + /// different number of answers than it was sent classifications, or any + /// classification got none. NoVerdict, /// No run started, as every run and wait place was taken; a comment - /// saying so was posted. + /// saying so was posted. This build records none; it is here so that a + /// stored record of it still reads. QueueFull, /// The review could not be posted. PostFailed, diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index 6cbeb30c9..f294dde13 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -664,6 +664,25 @@ impl Policy { subjects } + /// The review requests the controller queues and takes + /// (`swarm_wire::review_classifier::REQUEST_STREAM`): publishing one on + /// its PR's subject, creating the controller's durable consumer, pulling + /// from it and acking what it pulled. `STREAM.INFO.*` and + /// `STREAM.CREATE.*` open the stream. The controller is the stream's one + /// publisher, so no other grant may name its subjects. A refused publish + /// reaches the client as a timeout. + fn review_request_subjects() -> [String; 4] { + use swarm_wire::review_classifier::{ + REQUEST_CONSUMER, REQUEST_STREAM, REQUEST_SUBJECT_PREFIX, + }; + [ + format!("{REQUEST_SUBJECT_PREFIX}.>"), + format!("$JS.API.CONSUMER.CREATE.{REQUEST_STREAM}.{REQUEST_CONSUMER}"), + format!("$JS.API.CONSUMER.MSG.NEXT.{REQUEST_STREAM}.{REQUEST_CONSUMER}"), + format!("$JS.ACK.{REQUEST_STREAM}.{REQUEST_CONSUMER}.>"), + ] + } + fn reader_subjects(&self) -> Vec { let stream = self.stream(); let mut subjects = Self::jetstream_minimum().to_vec(); @@ -774,6 +793,7 @@ impl Policy { swarm_wire::subagent_term::consumer_create_grant(), "$KV.*.>".to_owned(), ]); + subjects.extend(Self::review_request_subjects()); // The per-agent status bucket, read-side only. Same five subjects as // the hive-status block above and for the same measured reasons β€” // including BOTH `CONSUMER.CREATE` forms, since `keys()` builds an @@ -1594,6 +1614,56 @@ mod tests { } } + /// Queueing a review request, taking it through the controller's durable + /// consumer and acking it are the reader's, and nobody else's. + #[test] + fn only_the_reader_may_queue_and_take_review_requests() { + use swarm_wire::review_classifier::{ + REQUEST_CONSUMER, REQUEST_STREAM, REQUEST_SUBJECT_PREFIX, + }; + + let subjects = [ + format!("{REQUEST_SUBJECT_PREFIX}.hyperhive.hyperhive.5219"), + format!("$JS.API.STREAM.INFO.{REQUEST_STREAM}"), + format!("$JS.API.STREAM.CREATE.{REQUEST_STREAM}"), + format!("$JS.API.CONSUMER.CREATE.{REQUEST_STREAM}.{REQUEST_CONSUMER}"), + format!("$JS.API.CONSUMER.MSG.NEXT.{REQUEST_STREAM}.{REQUEST_CONSUMER}"), + format!("$JS.ACK.{REQUEST_STREAM}.{REQUEST_CONSUMER}.1.4.4.1761000000000000000.0"), + ]; + let p = with_classifier() + .with_agent_token_subjects(vec!["$SWARM.term.{agent}".to_owned()]) + .expect("a per-agent template is valid"); + let reader = p.permissions("swarm-controller").expect("the reader"); + for subject in &subjects { + assert!( + reader.publish.iter().any(|g| covers(g, subject)), + "the reader may not publish {subject}" + ); + } + for (who, grant) in [ + ("a hive", p.permissions("hive-alpha").expect("a hive")), + ( + "the classifier", + p.permissions(CLASSIFIER).expect("admitted"), + ), + ( + "an agent", + p.agent_token_permissions("atlas").expect("configured"), + ), + ] { + for subject in &subjects { + assert!( + !grant.publish.iter().any(|g| covers(g, subject)), + "{who} may publish {subject}: {:?}", + grant.publish + ); + } + } + // The consumer grants name the one consumer. + let other = format!("$JS.API.CONSUMER.MSG.NEXT.{REQUEST_STREAM}.another"); + assert!(!reader.publish.iter().any(|g| covers(g, &other))); + } + const BUILDER: &str = "swarm-builder"; #[test] diff --git a/swarm-wire/src/review_classifier.rs b/swarm-wire/src/review_classifier.rs index c1c817b57..5a09cc2e9 100644 --- a/swarm-wire/src/review_classifier.rs +++ b/swarm-wire/src/review_classifier.rs @@ -1,6 +1,7 @@ //! The review classifier's runs: the subject swarm-controller sends a run //! on, and what travels each way. A try-out and a requested review are the -//! same request. +//! same request. Also the stream the controller queues requested reviews in, +//! which the auth-callout responder grants to the controller alone. //! //! Three crates name these: the controller sends the request, the classifier //! answers it, and the auth-callout responder grants the one subject to each. @@ -20,6 +21,19 @@ pub const SUBJECT: &str = "$SWARM.review-classifier.try-out"; /// the requests instead of running each one twice. pub const QUEUE_GROUP: &str = "review-classifier"; +/// The work-queue stream holding the review requests the controller has +/// accepted and not yet finished. The controller is its one publisher and its +/// one consumer. +pub const REQUEST_STREAM: &str = "review-requests"; + +/// The subjects of [`REQUEST_STREAM`] start with this; each PR's request is +/// one subject below it. +pub const REQUEST_SUBJECT_PREFIX: &str = "$SWARM.review-request"; + +/// The durable consumer on [`REQUEST_STREAM`] the controller takes requests +/// from. +pub const REQUEST_CONSUMER: &str = "swarm-controller"; + /// The most classifications one run carries. pub const MAX_CLASSIFICATIONS: usize = 10; From efbcc3dce689cd03082073655858581e3ac0ce9a Mon Sep 17 00:00:00 2001 From: atlas Date: Thu, 22 Oct 2026 06:52:35 +0200 Subject: [PATCH 2/4] review classifier: group the queue consumer's handles into one struct take and consume carried the forge, the classifier queue, the review poster and the run store as four separate parameters, which took take to eight and needed a too_many_arguments expectation. They travel together everywhere they are used, so they are now one Handles value, cloned per taken request. take has five parameters and no lint attribute. No behaviour change. --- .../src/review_classifier/review.rs | 78 ++++++++++--------- 1 file changed, 43 insertions(+), 35 deletions(-) diff --git a/swarm-controller/src/review_classifier/review.rs b/swarm-controller/src/review_classifier/review.rs index 9c84556a8..f5d549bfc 100644 --- a/swarm-controller/src/review_classifier/review.rs +++ b/swarm-controller/src/review_classifier/review.rs @@ -442,19 +442,37 @@ pub(crate) fn spawn_queue(state: &AppState) { tokio::time::sleep(TIMING.pull_wait).await; }; let queue = Queue(Some(client.clone())); - consume( - &client, - TIMING, - &reviews, - Arc::clone(&forge), + let handles = Handles { + forge: Arc::clone(&forge), queue, - forge, + poster: forge, runs, - ) - .await; + }; + consume(&client, TIMING, &reviews, handles).await; }); } +/// What a review reaches outside the queue: the forge it reads the PR and +/// rules from, the classifier it runs on, the poster for its review, and the +/// store that records it. +struct Handles { + forge: Arc, + queue: Q, + poster: Arc

, + runs: Runs, +} + +impl Clone for Handles { + fn clone(&self) -> Self { + Self { + forge: Arc::clone(&self.forge), + queue: self.queue.clone(), + poster: Arc::clone(&self.poster), + runs: self.runs.clone(), + } + } +} + /// Take requests from [`REQUEST_STREAM`] one at a time, each once a run /// place is free, and review each in a task of its own. Never returns; /// dropping it stops every review it started, and the queue delivers their @@ -467,10 +485,7 @@ async fn consume( client: &async_nats::Client, timing: Timing, reviews: &Arc, - forge: Arc, - queue: Q, - poster: Arc

, - runs: Runs, + handles: Handles, ) where S: FileSource + PullSource + Send + Sync + 'static, Q: RunSink + SettingsSource, @@ -493,7 +508,7 @@ async fn consume( tracing::error!(error = %e, "a requested review's task ended early"); } } - match queue.settings().await { + match handles.queue.settings().await { Ok(settings) => reviews.resize(settings.max_concurrent_reviews), Err(e) => tracing::warn!( error = %format!("{e:#}"), @@ -520,10 +535,7 @@ async fn consume( slot, timing, Arc::clone(reviews), - Arc::clone(&forge), - queue.clone(), - Arc::clone(&poster), - runs.clone(), + handles.clone(), )); } } @@ -549,19 +561,12 @@ async fn next( /// Review the request in `message`, holding `slot`, and ack it once its /// record is written. -#[expect( - clippy::too_many_arguments, - reason = "each is one thing a review reaches" -)] async fn take( message: jetstream::Message, slot: Slot, timing: Timing, reviews: Arc, - forge: Arc, - queue: Q, - poster: Arc

, - runs: Runs, + handles: Handles, ) where S: FileSource + PullSource + Send + Sync + 'static, Q: RunSink + SettingsSource, @@ -595,7 +600,13 @@ async fn take( } return; }; - let work = review(&*forge, &*forge, &queue, &*poster, &runs, &request, slot); + let Handles { + forge, + queue, + poster, + runs, + } = &handles; + let work = review(&**forge, &**forge, queue, &**poster, runs, &request, slot); in_progress(&message, timing.ack_wait / 3, work).await; drop(claim); if let Err(e) = message.ack().await { @@ -1875,16 +1886,13 @@ mod tests { let poster = Arc::clone(&self.rig.poster); let store = self.rig.store.clone(); tokio::spawn(async move { - consume( - &client, - TEST_TIMING, - &reviews, - Arc::clone(&forge), - gate, + let handles = Handles { + forge, + queue: gate, poster, - store, - ) - .await; + runs: store, + }; + consume(&client, TEST_TIMING, &reviews, handles).await; }) } From 5ebec6d09bd51d382ad9b40327bb253923e0b865 Mon Sep 17 00:00:00 2001 From: atlas Date: Thu, 22 Oct 2026 07:15:14 +0200 Subject: [PATCH 3/4] review classifier: a request for a moved head reviews the PR's current head A request whose head the PR has moved on from was dropped without a word. It now runs on the current head, as before the queue. The re-post shortcut, keyed on the request's head, applies only while the PR is still at that head, so a moved PR never re-posts an old head's verdict. --- docs/swarm/review-classifier.md | 9 +-- .../src/review_classifier/review.rs | 75 ++++++++++++------- 2 files changed, 51 insertions(+), 33 deletions(-) diff --git a/docs/swarm/review-classifier.md b/docs/swarm/review-classifier.md index 223b05868..5cc696341 100644 --- a/docs/swarm/review-classifier.md +++ b/docs/swarm/review-classifier.md @@ -130,9 +130,9 @@ minute, and the controller that's running then reviews it. A restart therefore repeats a run that was going. When it can't queue a request, the account posts a `**No verdict:**` comment saying so. -When the controller takes a request, it reads the PR first. It drops a -request whose head is no longer the PR's head, and posts and records nothing -for it. +When the controller takes a request, it reads the PR first. A request whose +head is no longer the PR's head runs on the PR's current head, and never +re-posts the old head's recorded verdict. A request for a PR whose review is still going waits in the queue until that review ends, so a PR never has two runs at once. A request that re-posts an earlier verdict waits for a run place too, and frees it before it posts. @@ -587,8 +587,7 @@ classifier's reply arrives, a requested review once the controller has posted its review or the post has failed, and a request the controller couldn't queue 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 dropped request, whose head the -PR has moved on from, isn't recorded. A run still going isn't in the list; a +repeats and that run's id in its `detail`. A run still going isn't in the list; a controller restart stops it, and the queue hands its request back. 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 f5d549bfc..4812f5d7a 100644 --- a/swarm-controller/src/review_classifier/review.rs +++ b/swarm-controller/src/review_classifier/review.rs @@ -641,10 +641,11 @@ async fn in_progress( /// Review `req`, holding `slot`, and record the result in `runs`. /// -/// A request whose head is no longer the PR's is dropped: nothing is posted -/// or recorded. One whose head's newest review [`reposted`] covers frees -/// `slot` and posts that verdict again. Any other runs, reading the settings -/// for the model and effort. +/// A request whose head is still the PR's, and whose head's newest review +/// [`reposted`] covers, frees `slot` and posts that verdict again. Any other +/// runs on the PR's current head, reading the settings for the model and +/// effort: a request for a head the PR has moved on from gets a review of the +/// new head, never the old head's verdict. async fn review( files: &impl FileSource, pulls: &impl PullSource, @@ -655,18 +656,16 @@ async fn review( slot: Slot, ) { let pull = read_pull(pulls, req).await; - if let Ok(pull) = &pull - && !req.head.is_empty() - && pull.head_sha != req.head - { + let moved = matches!(&pull, Ok(pull) if !req.head.is_empty() && pull.head_sha != req.head); + if moved { tracing::info!( repo = %format!("{}/{}", req.owner, req.name), pr = req.pr, - head_sha = req.head.as_str(), current = pull.head_sha.as_str(), - "review requested on a head the PR has moved on from; dropping the request" + head_sha = req.head.as_str(), + "review requested on a head the PR has moved on from; reviewing its current head" ); - return; } - if let Some(earlier) = newest_review(runs, req).await + if !moved + && let Some(earlier) = newest_review(runs, req).await && let Some(ruling) = reposted(&earlier) { drop(slot); @@ -1989,27 +1988,47 @@ mod tests { } #[tokio::test] - async fn a_request_for_a_head_the_pr_has_moved_on_from_is_dropped() { + async fn a_request_for_a_head_the_pr_has_moved_on_from_reviews_the_new_head() { let q = Queued::new().await; + // `old` has an approval on record; the PR has since moved to `new`. + let approved = reviewed(&q.rig.store, "earlier", RunOutcome::Approved).await; + q.rig + .store + .record(RunRecord { + id: "earlier-old".to_owned(), + head_sha: Some("old".to_owned()), + ..approved + }) + .await; q.rig.forge.move_to(7, "new"); q.request(7, "old").await; let _consumer = q.consume(&q.rig.reviews); - tokio::time::timeout(Duration::from_secs(10), async { - while q.waiting().await > 0 { - tokio::time::sleep(Duration::from_millis(20)).await; - } - }) - .await - .expect("the request is acked"); - assert!(q.rig.posted().is_empty()); - assert!(q.rig.doc.runs().is_empty(), "nothing is recorded"); - assert_eq!(q.rig.gate.reached(), 0); - - // Control: the PR's head runs. - q.request(7, "new").await; + until("the run reached the classifier", || q.rig.gate.reached() == 1).await; q.rig.gate.open(1); - until("the head is reviewed", || q.rig.posted().len() == 1).await; - assert_eq!(q.rig.posted()[0].1.as_deref(), Some("new")); + until("the new head is reviewed", || q.rig.posted().len() == 1).await; + let posted = q.rig.posted(); + assert_eq!(posted[0].1.as_deref(), Some("new")); + assert!( + !posted[0].2.body.starts_with("Already reviewed"), + "the old head's verdict was posted again: {}", + posted[0].2.body + ); + + // Control: with the PR back on `old`, the same request re-posts its + // recorded verdict without a run. + q.rig.forge.move_to(7, "old"); + accept( + Some(&q.client), + &*q.rig.poster, + &q.rig.store, + &requested_at("old"), + ) + .await; + until("the verdict is posted again", || q.rig.posted().len() == 2).await; + let posted = q.rig.posted(); + assert_eq!(posted[1].1.as_deref(), Some("old")); + assert!(posted[1].2.body.starts_with("Already reviewed `old`")); + assert_eq!(q.rig.gate.reached(), 1, "the re-post ran nothing"); } #[tokio::test] From bc4e0c23a0b04aca35bb31ffd7c402227b6b7caf Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 24 Oct 2026 03:23:17 +0200 Subject: [PATCH 4/4] treefmt --- swarm-controller/src/review_classifier/review.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/swarm-controller/src/review_classifier/review.rs b/swarm-controller/src/review_classifier/review.rs index 4812f5d7a..7444d393f 100644 --- a/swarm-controller/src/review_classifier/review.rs +++ b/swarm-controller/src/review_classifier/review.rs @@ -2003,7 +2003,10 @@ mod tests { q.rig.forge.move_to(7, "new"); q.request(7, "old").await; let _consumer = q.consume(&q.rig.reviews); - until("the run reached the classifier", || q.rig.gate.reached() == 1).await; + until("the run reached the classifier", || { + q.rig.gate.reached() == 1 + }) + .await; q.rig.gate.open(1); until("the new head is reviewed", || q.rig.posted().len() == 1).await; let posted = q.rig.posted();