mara/hyperhive
mara/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

review classifier: the most classifications a run carries is a setting

Browse source 
The limit was a constant of ten in swarm-wire, so a repo with eleven rules
could not be reviewed and nothing short of a rebuild changed it. It is now
`max_classifications` in the classifier settings, beside
`max_concurrent_reviews`: 1 to 1000, ten until set, and a document stored
before the field decodes with the default. The classifier settings form in
the swarm UI gets a "rules per review" field, and the try-out's add button
stops at the saved limit instead of a hard-coded ten.

The controller checks a try-out, and a requested review's rules file, against
the setting where the run starts. The rules file itself may hold up to the
1000 the setting allows, so raising the limit needs no change to a repo.
A review whose rules exceed the limit posts a no-verdict comment saying so.

A run's reply deadline was a constant that covered ten classifications, a
minute plus three each. It is now computed from the run's own size, so a
bigger run waits as long as it needs and a default-sized one still waits 31
minutes. The classifier's answer window in the queue permissions covers the
biggest run the setting allows.
This commit is contained in:
iris 2026-11-07 10:37:52 +01:00 • committed by mara
commit cffe8895df
1 changed files with 400 additions and 101 deletions
Download patch file Download diff file Expand all files Collapse all files

147
swarm-controller/src/review_classifier.rs
View file

@ -32,7 +32,7 @@ use axum::http::{HeaderMap, StatusCode};
use problem_details::ProblemDetails;
use serde::{Deserialize, Serialize};
use swarm_wire::review_classifier::{
Classification, MAX_CLASSIFICATIONS, Overall, RUN_DEADLINE, Reply, Row, SUBJECT, TryOut,
CLASSIFICATION_LIMITS, Classification, Overall, Reply, Row, SUBJECT, TryOut, run_deadline,
};
use utoipa::{IntoParams, ToSchema};
@ -114,9 +114,8 @@ struct Rule {
unknown: BTreeMap<String, serde_json::Value>,
}
/// `running` lasts at most the run deadline, 31 minutes: ten
/// classifications at three minutes each, one after another, and a minute
/// more.
/// `running` lasts at most the run deadline: the classifications at three
/// minutes each, one after another, and a minute more (31 minutes for ten).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, ToSchema)]
#[serde(rename_all = "lowercase")]
pub(crate) enum RunState {
@ -177,6 +176,8 @@ pub(crate) struct TryOuts {
struct Entry {
status: TryOutStatus,
started: Instant,
/// The longest the run's reply can take: [`run_deadline`] of its size.
deadline: Duration,
settled: Option<Instant>,
}
@ -184,7 +185,7 @@ impl Entry {
/// Expired once [`RESULT_TTL`] has passed since the reply, or since the
/// latest a reply can come.
fn expired(&self, now: Instant) -> bool {
let since = self.settled.unwrap_or(self.started + RUN_DEADLINE);
let since = self.settled.unwrap_or(self.started + self.deadline);
now.saturating_duration_since(since) > RESULT_TTL
}
}
@ -196,7 +197,7 @@ impl TryOuts {
.unwrap_or_else(std::sync::PoisonError::into_inner)
}
fn insert(&self, id: String, head_sha: String) {
fn insert(&self, id: String, head_sha: String, deadline: Duration) {
let now = Instant::now();
let mut runs = self.lock();
runs.retain(|_, e| !e.expired(now));
@ -211,6 +212,7 @@ impl TryOuts {
error: None,
},
started: now,
deadline,
settled: None,
},
);
@ -270,18 +272,20 @@ impl FileSource for forge::Client {
}
}
/// Where a run is sent: the queue in production, a fake in the tests.
/// Where a run is sent: the queue in production, a fake in the tests. The
/// reply is waited for at most `deadline`, the [`run_deadline`] of the run's
/// size.
pub(crate) trait RunSink: Clone + Send + Sync + 'static {
fn run(&self, request: Vec<u8>) -> impl Future<Output = Reply> + Send;
fn run(&self, request: Vec<u8>, deadline: Duration) -> impl Future<Output = Reply> + Send;
}
impl RunSink for async_nats::Client {
async fn run(&self, request: Vec<u8>) -> Reply {
async fn run(&self, request: Vec<u8>, deadline: Duration) -> Reply {
use async_nats::RequestErrorKind;
let sent = async_nats::Request::new()
.payload(request.into())
.timeout(Some(RUN_DEADLINE));
.timeout(Some(deadline));
let error = match self.send_request(SUBJECT, sent).await {
Ok(msg) => {
return serde_json::from_slice(&msg.payload).unwrap_or_else(|e| Reply::Failed {
@ -296,7 +300,7 @@ impl RunSink for async_nats::Client {
}
RequestErrorKind::TimedOut => format!(
"the review classifier sent no answer within {} minutes",
RUN_DEADLINE.as_secs() / 60
deadline.as_secs() / 60
),
_ => format!("sending the run to the review classifier failed: {e}"),
},
@ -316,7 +320,8 @@ impl RunSink for async_nats::Client {
(status = 400, description = "the body is not a valid run: not JSON or not this \
shape (an unknown polarity, a negative or non-integer number, a threshold over \
255); a repo that is not owner/name with segments of `[A-Za-z0-9._-]` other than \
`.` and `..`; a PR number of 0; no classification or more than ten; a \
`.` and `..`; a PR number of 0; no classification or more than the settings' \
`max_classifications` (ten until set); a \
classification whose trimmed name or description is empty; two with one trimmed \
name; or a threshold over 100 (problem+json)", body = String),
(status = 401, description = "the request carries no `Remote-User`, so there is no \
@ -398,8 +403,9 @@ pub(crate) async fn post_try_out(
path = "/api/review-classifier/try-out/{id}",
params(("id" = String, Path, description = "the id the POST answered with")),
responses(
(status = 200, description = "the run: `running` for at most 31 minutes, then \
`done` or `failed`", body = TryOutStatus),
(status = 200, description = "the run: `running` for at most a minute plus three \
per classification (31 minutes for ten), then `done` or `failed`",
body = TryOutStatus),
(status = 404, description = "no run has this id: it never existed, its result \
expired an hour after it settled, or the controller restarted since the POST, \
which loses every run (problem+json)", body = String),
@ -436,7 +442,7 @@ pub(crate) async fn get_try_out(
not parse as its extension says; or it holds a key other than `classifications`, \
or a classification with a key other than `name`, `description`, `polarity` and \
`threshold`; or it fails a check the try-out POST makes: no classification or \
more than ten, an empty trimmed name or description, two with one trimmed name, \
more than 1000, an empty trimmed name or description, two with one trimmed name, \
or a threshold over 100 (problem+json)", body = String),
(status = 503, description = "this host has no forge token yet; or the forge failed \
or did not answer within two minutes (problem+json)", body = String),
@ -566,7 +572,10 @@ fn parse_rules(path: &str, bytes: &[u8]) -> Result<Vec<Classification>, String>
}
classifications.push(rule.classification);
}
check_classifications(&classifications).map_err(|e| format!("{path}: {e}"))?;
// The file may hold up to the most the setting allows; the setting itself
// is checked where a run starts.
check_classifications(&classifications, *CLASSIFICATION_LIMITS.end())
.map_err(|e| format!("{path}: {e}"))?;
Ok(classifications)
}
@ -587,7 +596,7 @@ impl Run {
.ok()
.filter(|n| *n > 0)
.ok_or_else(|| format!("{} is not a PR number", req.pr))?;
check_classifications(&req.classifications)?;
check_classifications(&req.classifications, *CLASSIFICATION_LIMITS.end())?;
Ok(Self {
owner: owner.to_owned(),
name: name.to_owned(),
@ -604,14 +613,22 @@ pub(crate) fn split_repo(repo: &str) -> Result<(&str, &str), String> {
.ok_or_else(|| format!("repo {repo:?} is not owner/name"))
}
/// What is wrong with `classifications` as one run's, if anything.
fn check_classifications(classifications: &[Classification]) -> Result<(), String> {
let count = classifications.len();
if count == 0 || count > MAX_CLASSIFICATIONS {
/// Whether a run of `count` classifications is within the `max` the settings
/// allow, or what is wrong with it.
pub(crate) fn check_count(count: usize, max: usize) -> Result<(), String> {
if count == 0 || count > max {
return Err(format!(
"{count} classifications: a run carries 1 to {MAX_CLASSIFICATIONS}"
"{count} classifications: a run carries 1 to {max}, as the review classifier's \
settings allow"
));
}
Ok(())
}
/// What is wrong with `classifications` as one run's of at most `max`, if
/// anything.
fn check_classifications(classifications: &[Classification], max: usize) -> Result<(), String> {
check_count(classifications.len(), max)?;
let mut names = HashSet::new();
for c in classifications {
let name = c.name.trim();
@ -665,6 +682,9 @@ async fn start(
settings: Settings,
) -> Result<(StatusCode, Json<TryOutAccepted>), ProblemDetails> {
let started_unix = crate::agent_renewal::unix_now();
check_count(run.classifications.len(), settings.max_classifications)
.map_err(|d| error_problem(StatusCode::BAD_REQUEST, &d))?;
let deadline = run_deadline(run.classifications.len());
let runtime = settings
.runtime()
.map_err(|e| error_problem(StatusCode::INTERNAL_SERVER_ERROR, &e))?;
@ -698,7 +718,7 @@ async fn start(
)
})?;
let id = mint_id().map_err(|e| error_problem(StatusCode::INTERNAL_SERVER_ERROR, &e))?;
try_outs.insert(id.clone(), pull.head_sha.clone());
try_outs.insert(id.clone(), pull.head_sha.clone(), deadline);
let try_outs = try_outs.clone();
let (settle_id, repo, pr, requester) = (
id.clone(),
@ -707,7 +727,7 @@ async fn start(
request.operator.clone(),
);
tokio::spawn(async move {
let reply = sink.run(payload).await;
let reply = sink.run(payload, deadline).await;
let (rows, overall) = runs::answer(&reply);
let (outcome, detail) = match &reply {
Reply::Done { .. } => (runs::RunOutcome::Done, None),
@ -910,7 +930,7 @@ pub(crate) mod tests {
}
impl RunSink for Fake {
async fn run(&self, request: Vec<u8>) -> Reply {
async fn run(&self, request: Vec<u8>, _deadline: Duration) -> Reply {
if let Ok(sent) = serde_json::from_slice(&request) {
self.sent.lock().expect("unpoisoned").push(sent);
}
@ -1032,6 +1052,7 @@ pub(crate) mod tests {
model: "opus".to_owned(),
effort: Some(Effort::High),
max_concurrent_reviews: 2,
max_classifications: 10,
runtime: settings::RuntimeKind::Acp,
acp: Some(acp.clone()),
},
@ -1051,6 +1072,42 @@ pub(crate) mod tests {
assert_eq!(sent[0].runtime, Runtime::Acp(acp));
}
#[tokio::test]
async fn a_try_out_takes_as_many_classifications_as_the_settings_allow() {
let many = |count: usize| {
let mut b: serde_json::Value = serde_json::from_str(&body("o/r", 7)).expect("json");
let one = b["classifications"][0].clone();
b["classifications"] = (0..count)
.map(|i| {
let mut c = one.clone();
c["name"] = format!("c{i}").into();
c
})
.collect();
b.to_string()
};
let (mut state, _) = crate::routes::tests::state_with_roster();
state.try_outs.fake = Some(Arc::new(Fake {
settings: Settings {
max_classifications: 12,
..Settings::default()
},
..Fake::new(done(), Vec::new())
}));
let app = crate::startup::build_app(state);
// Past the default of ten, up to the setting.
let (status, accepted) = post(&app, many(12)).await;
assert_eq!(status, StatusCode::ACCEPTED, "{accepted}");
let (status, problem) = post(&app, many(13)).await;
assert_eq!(status, StatusCode::BAD_REQUEST, "{problem}");
assert!(
problem["detail"]
.as_str()
.is_some_and(|d| d.contains("13 classifications") && d.contains("1 to 12")),
"{problem}"
);
}
#[tokio::test]
async fn a_failed_run_reports_its_error_and_no_verdict() {
let app = app(Reply::Failed {
@ -1182,7 +1239,7 @@ pub(crate) mod tests {
let nats = crate::test_nats::TestNats::start();
let client = nats.client().await;
let unanswered = client.run(b"{}".to_vec()).await;
let unanswered = client.run(b"{}".to_vec(), run_deadline(1)).await;
assert!(
matches!(&unanswered, Reply::Failed { error } if error.contains("deploy.review-classifier.enable")),
"{unanswered:?}"
@ -1216,7 +1273,7 @@ pub(crate) mod tests {
.await
.expect("publish");
});
assert_eq!(client.run(b"{}".to_vec()).await, done());
assert_eq!(client.run(b"{}".to_vec(), run_deadline(1)).await, done());
}
#[test]
@ -1225,13 +1282,16 @@ pub(crate) mod tests {
let entry = |settled: Option<Instant>| Entry {
status: TryOutStatus::settled("abc".to_owned(), done()),
started: now,
deadline: run_deadline(10),
settled,
};
let deadline = run_deadline(10);
assert!(!entry(Some(now)).expired(now + RESULT_TTL));
assert!(entry(Some(now)).expired(now + RESULT_TTL + Duration::from_secs(1)));
// A run whose reply never came stays until the latest it could have.
assert!(!entry(None).expired(now + RUN_DEADLINE + RESULT_TTL));
assert!(entry(None).expired(now + RUN_DEADLINE + RESULT_TTL + Duration::from_secs(1)));
// A run whose reply never came stays until the latest it could have,
// which is later for a bigger run.
assert!(!entry(None).expired(now + deadline + RESULT_TTL));
assert!(entry(None).expired(now + deadline + RESULT_TTL + Duration::from_secs(1)));
}
/// A rules file holding `classifications`, and `extra` beside them.
@ -1257,11 +1317,22 @@ pub(crate) mod tests {
/// A JSON rules file holding the one `datetime` classification.
pub(crate) fn rules_json() -> (&'static str, String) {
rules_json_of(1)
}
/// A rules file holding `count` classifications.
pub(crate) fn rules_json_of(count: usize) -> (&'static str, String) {
let [_, json] = RULES_PATHS;
(
json,
rules_value(&serde_json::json!([datetime()]), &[]).to_string(),
)
let rules: Vec<_> = (0..count)
.map(|i| {
let mut c = datetime();
if i > 0 {
c["name"] = format!("c{i}").into();
}
c
})
.collect();
(json, rules_value(&rules.into(), &[]).to_string())
}
/// `file` as both formats, each with its path.
@ -1356,7 +1427,7 @@ classifications:
.as_object_mut()
.expect("an object")
.remove("polarity");
let eleven: Vec<_> = (0..11)
let too_many: Vec<_> = (0..1001)
.map(|i| {
let mut c = datetime();
c["name"] = format!("c{i}").into();
@ -1381,9 +1452,9 @@ classifications:
"threshhold",
),
(
"eleven classifications",
rules_value(&eleven.into(), &[]),
"11 classifications",
"more classifications than any setting allows",
rules_value(&too_many.into(), &[]),
"1001 classifications",
),
(
"a threshold over 100",