From d6aa83146a046578c78a023d299bfb0dbbf262fb Mon Sep 17 00:00:00 2001 From: atlas Date: Wed, 21 Oct 2026 22:21:24 +0200 Subject: [PATCH] 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 d7de8f52..223b0586 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 db51bc73..40a56120 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 12a035e8..a5434dd2 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 65467170..9c84556a 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 454bc822..f19ae13b 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 6cbeb30c..f294dde1 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 c1c817b5..5a09cc2e 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;