Watch
0
0
Fork
You've already forked hyperhive
0

swarm-controller: harden the review classifier's real reviews

Follow-ups from the review of #5142.

A `Done` answer with fewer or more rows than the classifications sent, or
none, now posts the no-verdict comment instead of the classifier's overall
verdict, so a short answer can't approve.

Requests are accepted per repo, PR and head into a set bounded at the two
running plus eight waiting. A request for a key already running or waiting
starts nothing and logs at info; a request with every place taken starts no
run and posts a no-verdict comment saying so. Before, every delivery spawned
a task waiting on the semaphore, so toggling the review request queued one
paid run per toggle.

Table cells get a zero-width space after `#`, `!`, `://` and `www.`, so a
model's reason can't render an issue or PR reference or a bare URL as a link.

The secret publisher's policy denies the classifier's forge token leaf, the
way #5139 denied Grafana's admin password, with a module-eval-bao-grants arm
pinning it. The controller posts reviews with whatever token is there, so a
publisher write picked the account reviews come from.

Closes #5143, part of #5096.
This commit is contained in:
atlas 2026-10-12 17:33:45 +02:00 • committed by mara
commit 60ea7fdc04
10 changed files with 447 additions and 50 deletions

View file

@ -165,7 +165,7 @@ mod tests {
create_gate: std::sync::Arc::default(),
accounts: Accounts::Fake(Arc::clone(store)),
try_outs: super::super::review_classifier::TryOuts::default(),
review_slots: super::super::review_slots(),
reviews: std::sync::Arc::default(),
}
}

View file

@ -135,7 +135,7 @@ mod tests {
create_gate: std::sync::Arc::default(),
accounts: Accounts::Fake(Arc::clone(store)),
try_outs: super::super::review_classifier::TryOuts::default(),
review_slots: super::super::review_slots(),
reviews: std::sync::Arc::default(),
}
}

View file

@ -970,7 +970,7 @@ pub(crate) mod tests {
create_gate: std::sync::Arc::default(),
accounts: super::Accounts::Fake(Arc::clone(store)),
try_outs: super::super::review_classifier::TryOuts::default(),
review_slots: super::super::review_slots(),
reviews: std::sync::Arc::default(),
}
}

View file

@ -893,16 +893,8 @@ struct AppState {
/// The review classifier try-outs started here, until their results
/// expire.
try_outs: review_classifier::TryOuts,
/// Held by each requested review's run, so at most
/// `review_classifier::review::MAX_CONCURRENT_REVIEWS` go at once.
review_slots: Arc<tokio::sync::Semaphore>,
}
/// A fresh [`AppState::review_slots`].
fn review_slots() -> Arc<tokio::sync::Semaphore> {
Arc::new(tokio::sync::Semaphore::new(
review_classifier::review::MAX_CONCURRENT_REVIEWS,
))
/// The requested reviews running or waiting.
reviews: Arc<review_classifier::review::Reviews>,
}
/// Env var the controller's NixOS module sets from
@ -3275,7 +3267,7 @@ async fn main() -> Result<()> {
create_gate: Arc::default(),
accounts: linked_accounts::Accounts::Live,
try_outs: review_classifier::TryOuts::default(),
review_slots: review_slots(),
reviews: Arc::default(),
};
let app = build_app(state);
@ -3464,7 +3456,7 @@ mod tests {
create_gate: std::sync::Arc::default(),
accounts: super::linked_accounts::Accounts::Fake(std::sync::Arc::default()),
try_outs: super::review_classifier::TryOuts::default(),
review_slots: super::review_slots(),
reviews: std::sync::Arc::default(),
};
(state, sched)
}

View file

@ -579,7 +579,7 @@ mod tests {
create_gate: std::sync::Arc::default(),
accounts: Accounts::Fake(Arc::clone(store)),
try_outs: super::super::review_classifier::TryOuts::default(),
review_slots: super::super::review_slots(),
reviews: std::sync::Arc::default(),
}
}

View file

@ -8,22 +8,29 @@
//! official reviews count toward approvals or block a merge.
//!
//! A run with no clean verdict never approves: see [`no_clean_verdict_event`].
//! At most [`MAX_CONCURRENT_REVIEWS`] runs go at once; the rest wait here, in
//! memory, so a restart drops them.
//! At most [`MAX_CONCURRENT_REVIEWS`] runs go at once 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 _;
use std::future::Future;
use std::sync::Arc;
use std::sync::{Arc, Mutex, PoisonError};
use serde::Deserialize;
use swarm_queue_client::review_classifier::{Overall, Reply, Row, TryOut, Verdict};
use tokio::sync::Semaphore;
use tokio::task::JoinHandle;
use super::{FileSource, PullSource, RunSink, split_repo};
use crate::AppState;
use crate::forge::{self, classifier_account::LOGIN, classifier_account::Review};
/// The most review runs in flight at once.
pub(crate) const MAX_CONCURRENT_REVIEWS: usize = 2;
const MAX_CONCURRENT_REVIEWS: usize = 2;
/// The most review runs waiting for a slot.
const MAX_WAITING_REVIEWS: usize = 8;
/// The forge's review events.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
@ -56,6 +63,9 @@ pub(crate) struct Requested {
owner: String,
name: String,
pr: i64,
/// The PR's head when the request was made, as the delivery names it;
/// empty when it names none.
head: String,
/// Who asked for the review.
requester: String,
}
@ -64,11 +74,22 @@ pub(crate) struct Requested {
struct Payload {
action: String,
number: i64,
pull_request: Option<PullRequest>,
requested_reviewer: Option<User>,
repository: Repository,
sender: Option<User>,
}
#[derive(Deserialize)]
struct PullRequest {
head: Head,
}
#[derive(Deserialize)]
struct Head {
sha: String,
}
#[derive(Deserialize)]
struct User {
login: String,
@ -100,9 +121,83 @@ impl Requested {
owner: owner.to_owned(),
name: name.to_owned(),
pr: Some(payload.number).filter(|n| *n > 0)?,
head: payload.pull_request.map(|p| p.head.sha).unwrap_or_default(),
requester: payload.sender.map(|s| s.login).unwrap_or_default(),
})
}
fn key(&self) -> Key {
(
self.owner.clone(),
self.name.clone(),
self.pr,
self.head.clone(),
)
}
}
/// Repo owner, repo name, PR number and the head the request named.
type Key = (String, String, i64, String);
/// The review runs this controller has accepted and not yet finished.
///
/// A request whose [`Key`] is already running or waiting starts nothing. A
/// request beyond [`MAX_CONCURRENT_REVIEWS`] + [`MAX_WAITING_REVIEWS`] starts
/// no run either; the account posts a no-verdict comment saying so.
pub(crate) struct Reviews {
slots: Arc<Semaphore>,
accepted: Mutex<HashSet<Key>>,
}
impl Default for Reviews {
fn default() -> Self {
Self {
slots: Arc::new(Semaphore::new(MAX_CONCURRENT_REVIEWS)),
accepted: Mutex::default(),
}
}
}
/// Why [`Reviews::accept`] turned a request away.
#[derive(Debug, PartialEq, Eq)]
enum Refused {
/// A run for the same key is running or waiting.
Duplicate,
/// Every run and wait place is taken.
Full,
}
impl Reviews {
fn accept(self: &Arc<Self>, key: Key) -> Result<Accepted, Refused> {
let mut accepted = self.accepted.lock().unwrap_or_else(PoisonError::into_inner);
if accepted.contains(&key) {
return Err(Refused::Duplicate);
}
if accepted.len() >= MAX_CONCURRENT_REVIEWS + MAX_WAITING_REVIEWS {
return Err(Refused::Full);
}
accepted.insert(key.clone());
Ok(Accepted {
reviews: Arc::clone(self),
key,
})
}
}
/// One accepted run; dropping it frees its key and its place.
struct Accepted {
reviews: Arc<Reviews>,
key: Key,
}
impl Drop for Accepted {
fn drop(&mut self) {
self.reviews
.accepted
.lock()
.unwrap_or_else(PoisonError::into_inner)
.remove(&self.key);
}
}
/// Start a review run for a verified `review-request` delivery that asks for
@ -120,13 +215,66 @@ pub(crate) fn on_delivery(state: &AppState, body: &[u8]) {
return;
};
let queue = Queue(state.status.as_ref().map(|s| s.queue_client()));
let slots = Arc::clone(&state.review_slots);
tokio::spawn(async move {
let Ok(_slot) = slots.acquire_owned().await else {
return;
};
Box::pin(run(&*forge, &*forge, &queue, &*forge, &request)).await;
});
start(&state.reviews, Arc::clone(&forge), queue, forge, request);
}
/// Accept `request` into `reviews` and spawn its run, or the no-verdict
/// comment when there's no place for it. The task, if one was spawned.
fn start<S, Q, P>(
reviews: &Arc<Reviews>,
forge: Arc<S>,
queue: Q,
poster: Arc<P>,
request: Requested,
) -> Option<JoinHandle<()>>
where
S: FileSource + PullSource + Send + Sync + 'static,
Q: RunSink,
P: ReviewPoster + Send + Sync + 'static,
{
let repo = format!("{}/{}", request.owner, request.name);
match reviews.accept(request.key()) {
Ok(accepted) => {
let slots = Arc::clone(&reviews.slots);
Some(tokio::spawn(async move {
let Ok(_slot) = slots.acquire_owned().await else {
return;
};
Box::pin(run(&*forge, &*forge, &queue, &*poster, &request)).await;
drop(accepted);
}))
}
Err(Refused::Duplicate) => {
tracing::info!(
%repo, pr = request.pr, head_sha = request.head.as_str(),
"review requested while a run for this head is running or waiting; nothing more starts"
);
None
}
Err(Refused::Full) => {
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 {
post(&*poster, &request, None, queue_full_review()).await;
}))
}
}
}
/// The review for a request [`Reviews`] had no place for.
fn queue_full_review() -> Review {
Review {
event: no_clean_verdict_event().forge(),
body: no_verdict(
&format!(
"{MAX_CONCURRENT_REVIEWS} reviews are running and {MAX_WAITING_REVIEWS} are \
waiting, so this request starts no run; request the review again later"
),
&[],
),
}
}
/// Where a review is posted: the forge in production, a fake in the tests.
@ -180,7 +328,8 @@ impl RunSink for Queue {
enum Outcome {
/// Nothing reached the classifier, for this reason.
NotRun(String),
Ran(Reply),
/// The classifier answered a run of `sent` classifications.
Ran { sent: usize, reply: Reply },
}
/// Read the PR and the rules, run them, and post the review.
@ -191,18 +340,22 @@ async fn run(
poster: &impl ReviewPoster,
req: &Requested,
) {
let repo = format!("{}/{}", req.owner, req.name);
let (head, outcome) = classify(files, pulls, sink, req).await;
let review = review_for(&outcome);
post(poster, req, head.as_deref(), review_for(&outcome)).await;
}
/// Post `review` on `req`'s PR, naming `head` when there is one.
async fn post(poster: &impl ReviewPoster, req: &Requested, head: Option<&str>, review: Review) {
let repo = format!("{}/{}", req.owner, req.name);
let event = review.event;
let sent = forge::within_forge_deadline(
"review classifier review",
poster.post(&req.owner, &req.name, req.pr, head.as_deref(), review),
poster.post(&req.owner, &req.name, req.pr, head, review),
)
.await;
match sent {
Some(Ok(())) => tracing::info!(
%repo, pr = req.pr, head_sha = head.as_deref().unwrap_or_default(), event,
%repo, pr = req.pr, head_sha = head.unwrap_or_default(), event,
"review classifier review posted"
),
Some(Err(e)) => tracing::warn!(
@ -261,8 +414,12 @@ async fn classify(
diff: pull.diff,
classifications: rules.classifications,
};
let sent = request.classifications.len();
let outcome = match serde_json::to_vec(&request) {
Ok(payload) => Outcome::Ran(sink.run(payload).await),
Ok(payload) => Outcome::Ran {
sent,
reply: sink.run(payload).await,
},
Err(e) => Outcome::NotRun(format!("encoding the run failed: {e}")),
};
(head, outcome)
@ -272,13 +429,25 @@ async fn classify(
fn review_for(outcome: &Outcome) -> Review {
let (event, body) = match outcome {
Outcome::NotRun(reason) => (no_clean_verdict_event(), no_verdict(reason, &[])),
Outcome::Ran(Reply::Failed { error }) => (
Outcome::Ran {
reply: Reply::Failed { error },
..
} => (
no_clean_verdict_event(),
no_verdict(&format!("the run failed: {error}"), &[]),
),
Outcome::Ran(Reply::Done { rows, overall }) => {
Outcome::Ran {
sent,
reply: Reply::Done { rows, overall },
} => {
let unanswered = rows.iter().filter(|r| r.error.is_some()).count();
if unanswered > 0 {
if rows.is_empty() || rows.len() != *sent {
let reason = format!(
"the classifier answered {} of {sent} classifications",
rows.len()
);
(no_clean_verdict_event(), no_verdict(&reason, rows))
} else if unanswered > 0 {
let reason = format!(
"{unanswered} of {} classifications got no answer",
rows.len()
@ -350,8 +519,10 @@ fn table(rows: &[Row]) -> String {
}
/// `text` as one table cell's inline text: on one line, with every character
/// that could end the cell, start markup or raw HTML escaped, and every `@`
/// followed by a zero-width space, so no name after it is a live mention.
/// that could end the cell, start markup or raw HTML escaped, and a zero-width
/// space after every `@`, `#`, `!`, `://` and `www.`, so it carries no live
/// mention, issue or PR reference, or link. The forge links issue and PR
/// numbers and bare URLs itself, so the escapes alone don't stop those.
fn cell(text: &str) -> String {
let mut out = String::with_capacity(text.len());
for (i, word) in text.split_whitespace().enumerate() {
@ -360,14 +531,22 @@ fn cell(text: &str) -> String {
}
for c in word.chars() {
match c {
'\\' | '|' | '`' | '*' | '_' | '~' | '[' | ']' | '(' | ')' | '!' | '<' | '>'
| '&' | '#' => {
'\\' | '|' | '`' | '*' | '_' | '~' | '[' | ']' | '(' | ')' | '<' | '>' | '&' => {
out.push('\\');
out.push(c);
}
'#' | '!' => {
out.push('\\');
out.push(c);
out.push('\u{200b}');
}
'@' => out.push_str("@\u{200b}"),
c => out.push(c),
}
let tail = out.get(out.len().saturating_sub(4)..).unwrap_or_default();
if out.ends_with("://") || tail.eq_ignore_ascii_case("www.") {
out.push('\u{200b}');
}
}
}
out
@ -376,6 +555,7 @@ fn cell(text: &str) -> String {
#[cfg(test)]
mod tests {
use std::sync::Mutex;
use std::time::Duration;
use swarm_queue_client::review_classifier::Polarity;
@ -427,13 +607,42 @@ mod tests {
owner: "o".to_owned(),
name: name.to_owned(),
pr: 7,
head: "abc123".to_owned(),
requester: "alice".to_owned(),
}
}
/// A JSON rules file holding `n` classifications.
fn rules_of(n: usize) -> (&'static str, String) {
let classifications: Vec<_> = (0..n)
.map(|i| {
serde_json::json!({
"name": format!("c{i}"),
"description": "All datetimes carry time zone information.",
"polarity": "positive",
"threshold": 80,
})
})
.collect();
let (path, _) = rules_json();
(
path,
serde_json::json!({ "classifications": classifications }).to_string(),
)
}
/// What one run against the fake forge posts.
async fn posted(reply: Reply, repo: &str) -> (String, Option<String>, Review) {
let fake = Fake::new(reply, vec![rules_json()]);
posted_for(reply, repo, rules_json()).await
}
/// What one run against the fake forge, holding `rules`, posts.
async fn posted_for(
reply: Reply,
repo: &str,
rules: (&'static str, String),
) -> (String, Option<String>, Review) {
let fake = Fake::new(reply, vec![rules]);
let poster = Posted::default();
run(&fake, &fake, &fake, &poster, &request(repo)).await;
let mut all = poster.0.into_inner().expect("unpoisoned");
@ -452,7 +661,7 @@ mod tests {
rows,
overall: Overall::Approve,
};
let (_, commit, review) = posted(reply, "r").await;
let (_, commit, review) = posted_for(reply, "r", rules_of(10)).await;
assert_eq!(review.event, "COMMENT");
assert!(
review
@ -465,6 +674,155 @@ mod tests {
assert_eq!(commit.as_deref(), Some("abc123"));
}
#[tokio::test]
async fn a_short_or_empty_answer_does_not_approve() {
for (sent, answered) in [(3, 2), (1, 0), (1, 2)] {
let reply = Reply::Done {
rows: (0..answered).map(|_| row(None, true, true)).collect(),
overall: Overall::Approve,
};
let (_, _, review) = posted_for(reply, "r", rules_of(sent)).await;
assert_eq!(review.event, "COMMENT", "{answered} of {sent}");
assert!(
review.body.starts_with(&format!(
"**No verdict:** the classifier answered {answered} of {sent} classifications."
)),
"{}",
review.body
);
}
// A run of no classifications never reaches the classifier; an
// empty answer to one still has no verdict.
let review = review_for(&Outcome::Ran {
sent: 0,
reply: Reply::Done {
rows: Vec::new(),
overall: Overall::Approve,
},
});
assert_eq!(review.event, "COMMENT");
}
/// A classifier that answers approve once [`Gate::open`] lets it through.
#[derive(Clone)]
struct Gate(Arc<tokio::sync::Semaphore>);
impl Gate {
fn open(&self) {
self.0.add_permits(1);
}
}
impl RunSink for Gate {
async fn run(&self, _request: Vec<u8>) -> Reply {
let _permit = self.0.acquire().await.expect("never closed");
Reply::Done {
rows: vec![row(None, true, true)],
overall: Overall::Approve,
}
}
}
fn requested_at(head: &str) -> Requested {
Requested {
head: head.to_owned(),
..request("r")
}
}
#[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(Arc::new(tokio::sync::Semaphore::new(0)));
let poster = Arc::new(Posted::default());
let start_at = |head: &str| {
start(
&reviews,
Arc::clone(&fake),
gate.clone(),
Arc::clone(&poster),
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();
for task in tasks.drain(..) {
task.await.expect("ran");
}
assert_eq!(poster.0.lock().expect("unpoisoned").len(), 3);
// A finished run frees its head.
start_at("a").expect("accepted again").await.expect("ran");
assert_eq!(poster.0.lock().expect("unpoisoned").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(Arc::new(tokio::sync::Semaphore::new(0)));
let poster = Arc::new(Posted::default());
let start_at = |head: String| {
start(
&reviews,
Arc::clone(&fake),
gate.clone(),
Arc::clone(&poster),
requested_at(&head),
)
.expect("a run or a comment")
};
let places = 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:** 2 reviews are running and 8 are waiting"),
"{}",
review.body
);
}
gate.open();
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"));
}
#[tokio::test]
async fn a_clean_run_posts_its_verdict_on_the_head_commit() {
let (target, commit, review) = posted(
@ -548,17 +906,28 @@ mod tests {
.count();
assert_eq!(unescaped, 8, "{line}");
assert!(
line.contains("\\!\\[\\]") && line.contains("\\<b\\>"),
line.contains("\\!\u{200b}\\[\\]") && line.contains("\\<b\\>"),
"{line}"
);
}
#[test]
fn a_reason_carries_no_link_or_reference() {
let text = cell("see https://evil.example/x, WWW.evil.example, #1, o/r#3 and !4");
assert_eq!(
text,
"see https://\u{200b}evil.example/x, WWW.\u{200b}evil.example, \\#\u{200b}1, \
o/r\\#\u{200b}3 and \\!\u{200b}4"
);
}
#[test]
fn only_a_request_for_the_classifier_parses() {
let payload = |action: &str, reviewer: Option<&str>| {
serde_json::json!({
"action": action,
"number": 7,
"pull_request": { "head": { "sha": "abc123" } },
"requested_reviewer": reviewer.map(|l| serde_json::json!({ "login": l })),
"repository": { "full_name": "o/r" },
"sender": { "login": "alice" },
@ -574,6 +943,13 @@ mod tests {
Some(request("r")),
"the forge's logins are case-insensitive"
);
let mut no_head: serde_json::Value =
serde_json::from_str(&payload("review_requested", Some(LOGIN))).expect("JSON");
no_head["pull_request"] = serde_json::Value::Null;
assert_eq!(
Requested::parse(no_head.to_string().as_bytes()),
Some(requested_at(""))
);
for body in [
payload("review_request_removed", Some(LOGIN)),
payload("review_requested", Some("argus")),