diff --git a/docs/swarm/review-classifier.md b/docs/swarm/review-classifier.md index 5cc696341..d7de8f525 100644 --- a/docs/swarm/review-classifier.md +++ b/docs/swarm/review-classifier.md @@ -118,24 +118,14 @@ 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)). 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. 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. +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. ## Components @@ -216,7 +206,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. The rest wait in the queue for a place + Try-outs don't count toward it. Up to eight more reviews wait 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 @@ -584,11 +574,13 @@ 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 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 run still going isn't in the list; a -controller restart stops it, and the queue hands its request back. +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. 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 @@ -610,12 +602,11 @@ 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 request the controller couldn't queue + an unreadable PR or a full queue - `started_unix` and `finished_unix`: a try-out starts at its POST, a review - 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 + 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 - `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 40a561204..db51bc73f 100644 --- a/swarm-controller/README.md +++ b/swarm-controller/README.md @@ -69,9 +69,8 @@ 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, 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 + requests, and the newest 100 finished runs, one document in the + `review-classifier-runs` KV bucket, 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 a5434dd2b..12a035e83 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -235,12 +235,7 @@ 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 7444d393f..654671708 100644 --- a/swarm-controller/src/review_classifier/review.rs +++ b/swarm-controller/src/review_classifier/review.rs @@ -13,14 +13,10 @@ //! gets a fresh run. //! //! A run with no clean verdict never approves: see [`no_clean_verdict_event`]. -//! -//! 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. +//! 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. use std::collections::HashSet; use std::fmt::Write as _; @@ -28,25 +24,23 @@ 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 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 serde::Deserialize; +use swarm_wire::review_classifier::{Overall, Reply, Row, TryOut, Verdict}; use tokio::sync::Notify; -use tokio::task::JoinSet; +use tokio::task::JoinHandle; 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, PullForReview, classifier_account::LOGIN, classifier_account::Review}; +use crate::forge::{self, 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). @@ -59,29 +53,6 @@ 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 { @@ -107,8 +78,8 @@ fn no_clean_verdict_event() -> Event { Event::Comment } -/// A review request for [`LOGIN`], as it waits in [`REQUEST_STREAM`]. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +/// A review request for [`LOGIN`]. +#[derive(Debug, Clone, PartialEq, Eq)] pub(crate) struct Requested { owner: String, name: String, @@ -176,48 +147,29 @@ impl Requested { }) } - 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 + fn key(&self) -> Key { + ( + self.owner.clone(), + self.name.clone(), + self.pr, + self.head.clone(), ) } } -/// `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 -} +/// Repo owner, repo name, PR number and the head the request named. +type Key = (String, String, i64, String); -/// Repo owner, repo name and PR number. -type PrKey = (String, String, i64); - -/// The run places, and the PRs with a review going. +/// The review runs this controller has accepted and not yet finished. /// -/// 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. +/// 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. changed: Notify, - going: Mutex>, + accepted: Mutex>, } /// How many runs may go at once, and how many do. @@ -234,11 +186,20 @@ impl Default for Reviews { running: 0, }), changed: Notify::new(), - going: Mutex::default(), + accepted: 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) @@ -256,14 +217,18 @@ impl Reviews { self.slots().limit } - /// `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 { + 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 { reviews: Arc::clone(self), - pr, + key, }) } @@ -285,19 +250,19 @@ impl Reviews { } } -/// One PR's review going; dropping it lets the PR's next request run. -struct Claim { +/// One accepted run; dropping it frees its key and its place. +struct Accepted { reviews: Arc, - pr: PrKey, + key: Key, } -impl Drop for Claim { +impl Drop for Accepted { fn drop(&mut self) { self.reviews - .going + .accepted .lock() .unwrap_or_else(PoisonError::into_inner) - .remove(&self.pr); + .remove(&self.key); } } @@ -311,8 +276,8 @@ impl Drop for Slot { } } -/// Queue the review request in a verified `review-request` delivery that -/// asks for [`LOGIN`], and return at once. +/// Start a review run for 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; @@ -325,371 +290,105 @@ pub(crate) fn on_delivery(state: &AppState, body: &[u8]) { ); return; }; - let client = state.status.as_ref().map(|s| s.queue_client()); + let queue = Queue(state.status.as_ref().map(|s| s.queue_client())); let runs = state.runs.clone(); - tokio::spawn(async move { accept(client.as_ref(), &*forge, &runs, &request).await }); -} - -/// 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" + start( + &state.reviews, + Arc::clone(&forge), + queue, + forge, + runs, + request, ); - 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())); - let handles = Handles { - forge: Arc::clone(&forge), - queue, - poster: forge, - runs, - }; - 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 { +/// 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. +/// +/// 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( + reviews: &Arc, 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 -/// requests again. -/// -/// 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, - handles: Handles, -) where + request: Requested, +) -> Option> +where S: FileSource + PullSource + Send + Sync + 'static, Q: RunSink + SettingsSource, P: ReviewPoster + Send + Sync + 'static, { - 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 handles.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); - tokio::time::sleep(timing.pull_wait).await; - continue; - } - }; - tasks.spawn(take( - message, - slot, - timing, - Arc::clone(reviews), - handles.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. -async fn take( - message: jetstream::Message, - slot: Slot, - timing: Timing, - reviews: Arc, - handles: Handles, -) 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!( - subject = %message.subject, error = %e, - "a queued review request did not parse; dropping it" - ); - 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 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 { - 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"); + 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; + 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)) => { + 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" + ); + 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; + })) } } } -/// Review `req`, holding `slot`, and record the result in `runs`. -/// -/// 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, - sink: &(impl RunSink + SettingsSource), - poster: &impl ReviewPoster, - runs: &Runs, - req: &Requested, - slot: Slot, -) { - let pull = read_pull(pulls, req).await; - 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(), - "review requested on a head the PR has moved on from; reviewing its current head" - ); - } - if !moved - && 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()), +/// 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), } } @@ -830,17 +529,17 @@ enum Outcome { Ran { sent: usize, reply: Reply }, } -/// Read the rules, run them on `pull`, the PR as it was read, and post the -/// review. The finished run. +/// Read the PR and the rules, run them, and post the review. The finished +/// run. async fn run( files: &impl FileSource, - pull: Result, + pulls: &impl PullSource, sink: &(impl RunSink + SettingsSource), poster: &impl ReviewPoster, req: &Requested, ) -> RunRecord { let started_unix = unix_now(); - let (head, narrowed, outcome) = classify(files, pull, sink, req).await; + let (head, narrowed, outcome) = classify(files, pulls, sink, req).await; let mut ruling = review_for(&outcome); if let (Some(reason), Outcome::Ran { .. }) = (narrowed, &outcome) { let _ = write!( @@ -1018,17 +717,35 @@ 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 on `pull`, the PR as it was read. +/// when it is, and how far the run got. async fn classify( files: &impl FileSource, - pull: Result, + pulls: &impl PullSource, sink: &(impl RunSink + SettingsSource), req: &Requested, ) -> (Option, Option, Outcome) { let repo = format!("{}/{}", req.owner, req.name); - let pull = match pull { - Ok(pull) => pull, - Err(reason) => return (None, None, Outcome::NotRun(reason)), + 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 head = Some(pull.head_sha.clone()); let narrowed = pull.narrowed; @@ -1271,8 +988,6 @@ 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)] @@ -1417,16 +1132,6 @@ 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) @@ -1459,7 +1164,7 @@ mod tests { ) -> (String, Option, Review) { let fake = Fake::new(reply, vec![rules]); let poster = Posted::default(); - run_read(&fake, &fake, &poster, &request(repo)).await; + run(&fake, &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) @@ -1570,6 +1275,13 @@ 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(), @@ -1577,6 +1289,133 @@ 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 { @@ -1598,68 +1437,11 @@ 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 reviews of a head on o/r#7. + /// starter for requests at a head on o/r#7. struct Rig { reviews: Arc, - forge: Arc, + fake: Arc, gate: Gate, poster: Arc, doc: Arc, @@ -1668,13 +1450,15 @@ mod tests { impl Rig { fn new() -> Self { - Self::with_doc(Arc::default()) - } - - fn with_doc(doc: Arc) -> Self { + let doc = Arc::new(FakeDoc::default()); Self { reviews: Arc::new(Reviews::default()), - forge: Arc::new(Heads::new()), + 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)), @@ -1682,18 +1466,15 @@ mod tests { } } - /// [`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 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)> { @@ -1727,7 +1508,7 @@ mod tests { let rig = Rig::new(); let earlier = reviewed(&rig.store, "earlier", outcome).await; - reposted(rig.start_at("abc123")).await; + reposted(rig.start_at("abc123").expect("accepted")).await; assert_eq!( rig.gate.reached(), @@ -1781,7 +1562,7 @@ mod tests { reviewed(&rig.store, &format!("earlier-{i}"), outcome).await; } - let task = rig.start_at("abc123"); + let task = rig.start_at("abc123").expect("accepted"); rig.gate.open(1); task.await.expect("ran"); @@ -1826,393 +1607,45 @@ mod tests { let rig = Rig::new(); rig.store.record(earlier).await; - let task = rig.start_at("abc123"); + 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"); } } - /// 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), - }; - - /// 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 { - let handles = Handles { - forge, - queue: gate, - poster, - runs: store, - }; - consume(&client, TEST_TIMING, &reviews, handles).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()); + 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); - // 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_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); - 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(); - assert_eq!(posted[0].1.as_deref(), Some("new")); + // 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[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] - 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 `abc123`"), + posted[1].2.body.starts_with("Already reviewed"), "{}", posted[1].2.body ); - assert_eq!(q.rig.gate.reached(), 1); - } - #[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() { + // A clean verdict: two requests at once post it once. let rig = Rig::new(); - 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; + 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 @@ -2224,7 +1657,7 @@ mod tests { poster: &impl ReviewPoster, ) -> RunRecord { let fake = Fake::new(reply, vec![rules]); - run_read(&fake, &fake, poster, &request(repo)).await + run(&fake, &fake, &fake, poster, &request(repo)).await } #[tokio::test] @@ -2310,7 +1743,7 @@ mod tests { } let fake = Fake::new(failed, Vec::new()); - let run = run_read(&fake, &fake, &Posted::default(), &request("r")).await; + let run = run(&fake, &fake, &fake, &Posted::default(), &request("r")).await; assert_eq!(run.outcome, RunOutcome::NoVerdict); assert!( run.detail @@ -2434,7 +1867,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_read(&fake, &fake, &poster, &request("r")).await; + let run = run(&fake, &fake, &fake, &poster, &request("r")).await; assert_eq!(run.outcome, RunOutcome::Approved, "{:?}", run.detail); assert_eq!(poster.tries.load(Ordering::SeqCst), 3); assert_eq!(poster.posted.0.lock().expect("unpoisoned").len(), 1); @@ -2497,7 +1930,7 @@ mod tests { tries: AtomicUsize::new(0), posted: Posted::default(), }; - let run = run_read(&fake, &fake, &poster, &request("r")).await; + let run = run(&fake, &fake, &fake, &poster, &request("r")).await; assert_eq!(run.outcome, RunOutcome::Approved, "{:?}", run.detail); assert_eq!(poster.posted.0.lock().expect("unpoisoned").len(), 1); assert_eq!(fake.sent.lock().expect("unpoisoned").len(), 1, "one run"); @@ -2539,7 +1972,7 @@ mod tests { let started = tokio::time::Instant::now(); let run = tokio::time::timeout( Duration::from_hours(1), - run_read(&fake, &fake, &poster, &request("r")), + run(&fake, &fake, &fake, &poster, &request("r")), ) .await .expect("the hung try is abandoned"); @@ -2560,7 +1993,7 @@ mod tests { ForgejoError::UnexpectedStatusCode(StatusCode::FORBIDDEN).into(), ]); let started = tokio::time::Instant::now(); - let run = run_read(&fake, &fake, &poster, &request("r")).await; + let run = run(&fake, &fake, &fake, &poster, &request("r")).await; assert_eq!(run.outcome, RunOutcome::PostFailed); assert_eq!(poster.tries.load(Ordering::SeqCst), 1); assert!(poster.posted.0.lock().expect("unpoisoned").is_empty()); @@ -2580,7 +2013,7 @@ mod tests { }; let fake = Fake::new(approve, vec![rules_json()]); let poster = Flaky::new(failures); - let run = run_read(&fake, &fake, &poster, &request("r")).await; + let run = run(&fake, &fake, &fake, &poster, &request("r")).await; assert_eq!(run.outcome, RunOutcome::PostFailed); assert_eq!( poster.tries.load(Ordering::SeqCst), @@ -2597,15 +2030,87 @@ mod tests { #[tokio::test] async fn a_failed_record_write_still_posts_the_review() { - 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(); + 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"); 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; @@ -2635,7 +2140,7 @@ mod tests { acp: Some(acp.clone()), }; let poster = Posted::default(); - run_read(&fake, &fake, &poster, &request("r")).await; + run(&fake, &fake, &fake, &poster, &request("r")).await; let sent = fake.sent.lock().expect("unpoisoned"); assert_eq!(sent.len(), 1); assert_eq!( @@ -2658,7 +2163,7 @@ mod tests { vec![rules_json()], ); let poster = Posted::default(); - run_read(&fake, &Queue(None), &poster, &request("r")).await; + run(&fake, &fake, &Queue(None), &poster, &request("r")).await; let (_, commit, review) = poster.0.into_inner().expect("unpoisoned").remove(0); assert_eq!(review.event, "COMMENT"); assert!( @@ -2732,7 +2237,7 @@ mod tests { let fake = Fake::new(approve(), Vec::new()); let poster = Posted::default(); - run_read(&fake, &fake, &poster, &request("r")).await; + run(&fake, &fake, &fake, &poster, &request("r")).await; let (_, _, review) = poster.0.into_inner().expect("unpoisoned").remove(0); assert_eq!(review.event, "COMMENT"); assert!( @@ -2877,7 +2382,7 @@ mod tests { let mut fake = Fake::new(reply, files); fake.narrowed = reason; let poster = Posted::default(); - run_read(&fake, &fake, &poster, &request("r")).await; + run(&fake, &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 f19ae13b5..454bc8222 100644 --- a/swarm-controller/src/review_classifier/runs.rs +++ b/swarm-controller/src/review_classifier/runs.rs @@ -60,14 +60,12 @@ pub(crate) enum RunOutcome { Approved, /// A review requested changes. ChangesRequested, - /// 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. + /// 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. NoVerdict, /// No run started, as every run and wait place was taken; a comment - /// saying so was posted. This build records none; it is here so that a - /// stored record of it still reads. + /// saying so was posted. 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 f294dde13..6cbeb30c9 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -664,25 +664,6 @@ 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(); @@ -793,7 +774,6 @@ 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 @@ -1614,56 +1594,6 @@ 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 5a09cc2e9..c1c817b57 100644 --- a/swarm-wire/src/review_classifier.rs +++ b/swarm-wire/src/review_classifier.rs @@ -1,7 +1,6 @@ //! 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. Also the stream the controller queues requested reviews in, -//! which the auth-callout responder grants to the controller alone. +//! same request. //! //! Three crates name these: the controller sends the request, the classifier //! answers it, and the auth-callout responder grants the one subject to each. @@ -21,19 +20,6 @@ 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;