26c4ff89d5
On deploy: a review requested of the review classifier on a PR whose repo holds neither .hyperhive/review-classifier.yml nor .json on its default branch now posts an APPROVED review at once, saying no review rules are configured for the repo and it was approved without checks. No model call is made. Before, that case posted a "No verdict" comment. A rules file that exists but fails to load or parse still posts no verdict. The rule model is unchanged and instance-wide: every repo without a rules file is approved.
2963 lines
101 KiB
Rust
2963 lines
101 KiB
Rust
//! Real reviews: a review request for the classifier's forge account
|
|
//! ([`LOGIN`]) runs the repo's rules against the PR and posts the result as
|
|
//! that account's review, on the commit the run read.
|
|
//!
|
|
//! Only a review request starts a run. A push makes none itself: the
|
|
//! notification router's `head-classifier` flow requests the review. A fresh
|
|
//! request retires the account's earlier review: the forge clears the
|
|
//! `official` flag on a reviewer's previous reviews when they are requested
|
|
//! again, and only official reviews count toward approvals or block a merge.
|
|
//!
|
|
//! A request for a head whose newest recorded review approved or requested
|
|
//! changes runs nothing: the account posts that verdict again, saying so. A
|
|
//! head whose newest review had no clean verdict, or that has none recorded,
|
|
//! gets a fresh run.
|
|
//!
|
|
//! A run with no clean verdict never approves: see [`no_clean_verdict_event`].
|
|
//!
|
|
//! Each accepted request waits in [`REQUEST_STREAM`], a work queue on the
|
|
//! swarm queue with one subject per PR that keeps one message per subject, so
|
|
//! a newer request for a PR replaces one still waiting. The controller takes
|
|
//! a request only once fewer than `max_concurrent_reviews` runs go, as the
|
|
//! swarm's [settings](super::settings) say, and acks it once its record is
|
|
//! written. A request a controller took and never acked, because it stopped
|
|
//! first, is delivered again. See [`consume`] for the rest.
|
|
|
|
use std::collections::HashSet;
|
|
use std::fmt::Write as _;
|
|
use std::future::Future;
|
|
use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
|
|
use std::time::Duration;
|
|
|
|
use anyhow::Context as _;
|
|
use async_nats::jetstream::{self, AckKind, consumer::PullConsumer};
|
|
use forgejo_api::ForgejoError;
|
|
use futures_util::StreamExt as _;
|
|
use serde::{Deserialize, Serialize};
|
|
use swarm_wire::review_classifier::{
|
|
Overall, REQUEST_CONSUMER, REQUEST_STREAM, REQUEST_SUBJECT_PREFIX, Reply, Row, TryOut, Verdict,
|
|
};
|
|
use tokio::sync::Notify;
|
|
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, PullForReview, classifier_account::LOGIN, classifier_account::Review};
|
|
use crate::routes::AppState;
|
|
|
|
/// 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).
|
|
/// Each try has its own [`forge::FORGE_DEADLINE`].
|
|
const POST_RETRY_WAITS: [Duration; 5] = [
|
|
Duration::from_secs(5),
|
|
Duration::from_secs(15),
|
|
Duration::from_secs(45),
|
|
Duration::from_secs(90),
|
|
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 {
|
|
Approve,
|
|
RequestChanges,
|
|
Comment,
|
|
}
|
|
|
|
impl Event {
|
|
fn forge(self) -> &'static str {
|
|
match self {
|
|
Self::Approve => "APPROVED",
|
|
Self::RequestChanges => "REQUEST_CHANGES",
|
|
Self::Comment => "COMMENT",
|
|
}
|
|
}
|
|
}
|
|
|
|
/// What a run posts when it has no clean verdict: the rules didn't load, the
|
|
/// PR couldn't be read, the run failed, or any classification has no answer.
|
|
/// A comment neither approves nor blocks the merge.
|
|
fn no_clean_verdict_event() -> Event {
|
|
Event::Comment
|
|
}
|
|
|
|
/// A review request for [`LOGIN`], as it waits in [`REQUEST_STREAM`].
|
|
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
|
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,
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
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,
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
struct Repository {
|
|
full_name: String,
|
|
}
|
|
|
|
impl Requested {
|
|
/// The request in a `pull_request_review_request` delivery, when it
|
|
/// asks for [`LOGIN`]'s review. A removed request, a request for anyone
|
|
/// else and an unparsable body are `None`.
|
|
fn parse(body: &[u8]) -> Option<Self> {
|
|
let payload: Payload = match serde_json::from_slice(body) {
|
|
Ok(p) => p,
|
|
Err(e) => {
|
|
tracing::warn!(error = %e, "webhook: review-request delivery did not parse");
|
|
return None;
|
|
}
|
|
};
|
|
let reviewer = payload.requested_reviewer?;
|
|
if payload.action != "review_requested" || !reviewer.login.eq_ignore_ascii_case(LOGIN) {
|
|
return None;
|
|
}
|
|
let (owner, name) = split_repo(&payload.repository.full_name).ok()?;
|
|
Some(Self {
|
|
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 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
|
|
)
|
|
}
|
|
}
|
|
|
|
/// `text` as one subject token: ASCII letters, digits and `-` as they are,
|
|
/// every other byte as `_` and its two hex digits. So a `.` in a repo name
|
|
/// can't split it, and two names never share a token.
|
|
fn token(text: &str) -> String {
|
|
let mut out = String::with_capacity(text.len());
|
|
for b in text.bytes() {
|
|
if b.is_ascii_alphanumeric() || b == b'-' {
|
|
out.push(char::from(b));
|
|
} else {
|
|
let _ = write!(out, "_{b:02x}");
|
|
}
|
|
}
|
|
out
|
|
}
|
|
|
|
/// Repo owner, repo name and PR number.
|
|
type PrKey = (String, String, i64);
|
|
|
|
/// The run places, and the PRs with a review going.
|
|
///
|
|
/// 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<Slots>,
|
|
/// Woken when a run ends or the limit changes.
|
|
changed: Notify,
|
|
going: Mutex<HashSet<PrKey>>,
|
|
}
|
|
|
|
/// How many runs may go at once, and how many do.
|
|
struct Slots {
|
|
limit: usize,
|
|
running: usize,
|
|
}
|
|
|
|
impl Default for Reviews {
|
|
fn default() -> Self {
|
|
Self {
|
|
slots: Mutex::new(Slots {
|
|
limit: Settings::default().max_concurrent_reviews,
|
|
running: 0,
|
|
}),
|
|
changed: Notify::new(),
|
|
going: Mutex::default(),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Reviews {
|
|
fn slots(&self) -> MutexGuard<'_, Slots> {
|
|
self.slots.lock().unwrap_or_else(PoisonError::into_inner)
|
|
}
|
|
|
|
/// Let `limit` runs go at once from now on. Runs already going when it
|
|
/// shrinks finish; no new one starts until fewer than `limit` run.
|
|
pub(crate) fn resize(&self, limit: usize) {
|
|
self.slots().limit = limit;
|
|
self.changed.notify_waiters();
|
|
}
|
|
|
|
#[cfg(test)]
|
|
pub(crate) fn limit(&self) -> usize {
|
|
self.slots().limit
|
|
}
|
|
|
|
/// `req`'s PR, marked as having a review going, or `None` when one is
|
|
/// going already.
|
|
fn claim(self: &Arc<Self>, req: &Requested) -> Option<Claim> {
|
|
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),
|
|
pr,
|
|
})
|
|
}
|
|
|
|
/// A place to run in, once fewer runs go than the limit allows.
|
|
async fn slot(self: &Arc<Self>) -> Slot {
|
|
loop {
|
|
let changed = self.changed.notified();
|
|
tokio::pin!(changed);
|
|
changed.as_mut().enable();
|
|
{
|
|
let mut slots = self.slots();
|
|
if slots.running < slots.limit {
|
|
slots.running += 1;
|
|
return Slot(Arc::clone(self));
|
|
}
|
|
}
|
|
changed.await;
|
|
}
|
|
}
|
|
}
|
|
|
|
/// One PR's review going; dropping it lets the PR's next request run.
|
|
struct Claim {
|
|
reviews: Arc<Reviews>,
|
|
pr: PrKey,
|
|
}
|
|
|
|
impl Drop for Claim {
|
|
fn drop(&mut self) {
|
|
self.reviews
|
|
.going
|
|
.lock()
|
|
.unwrap_or_else(PoisonError::into_inner)
|
|
.remove(&self.pr);
|
|
}
|
|
}
|
|
|
|
/// One run holding a place; dropping it lets the next one start.
|
|
struct Slot(Arc<Reviews>);
|
|
|
|
impl Drop for Slot {
|
|
fn drop(&mut self) {
|
|
self.0.slots().running -= 1;
|
|
self.0.changed.notify_waiters();
|
|
}
|
|
}
|
|
|
|
/// 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;
|
|
};
|
|
let Some(forge) = state.forge.get().map(Arc::clone) else {
|
|
tracing::warn!(
|
|
repo = %format!("{}/{}", request.owner, request.name),
|
|
pr = request.pr,
|
|
"review requested, but this host has no forge access yet; nothing is posted"
|
|
);
|
|
return;
|
|
};
|
|
let client = state.status.as_ref().map(|s| s.queue_client());
|
|
let runs = state.runs.clone();
|
|
tokio::spawn(async move { accept(client.as_ref(), &*forge, &runs, &request).await });
|
|
}
|
|
|
|
/// Put `request` in [`REQUEST_STREAM`]. When that fails, the account posts a
|
|
/// no-verdict comment saying so, recorded in `runs`.
|
|
async fn accept(
|
|
client: Option<&async_nats::Client>,
|
|
poster: &impl ReviewPoster,
|
|
runs: &Runs,
|
|
request: &Requested,
|
|
) {
|
|
let repo = format!("{}/{}", request.owner, request.name);
|
|
let queued = match client {
|
|
Some(client) => enqueue(client, request).await,
|
|
None => Err(anyhow::anyhow!("no swarm queue is configured on this host")),
|
|
};
|
|
let error = match queued {
|
|
Ok(()) => {
|
|
tracing::info!(
|
|
%repo, pr = request.pr, head_sha = request.head.as_str(),
|
|
"review request queued"
|
|
);
|
|
return;
|
|
}
|
|
Err(e) => format!("{e:#}"),
|
|
};
|
|
tracing::warn!(
|
|
%repo, pr = request.pr, head_sha = request.head.as_str(), %error,
|
|
"queueing a review request failed; posting a no-verdict comment"
|
|
);
|
|
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<jetstream::stream::Stream> {
|
|
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<PullConsumer> {
|
|
open_stream(client)
|
|
.await?
|
|
.create_consumer(jetstream::consumer::pull::Config {
|
|
durable_name: Some(REQUEST_CONSUMER.to_owned()),
|
|
description: Some("swarm-controller's requested review runs".to_owned()),
|
|
ack_policy: jetstream::consumer::AckPolicy::Explicit,
|
|
ack_wait,
|
|
..Default::default()
|
|
})
|
|
.await
|
|
.with_context(|| format!("opening the {REQUEST_CONSUMER} consumer on {REQUEST_STREAM}"))
|
|
}
|
|
|
|
/// Take the queued review requests and review each, for as long as the
|
|
/// controller runs, once this host has forge access.
|
|
pub(crate) fn spawn_queue(state: &AppState) {
|
|
let Some(client) = state.status.as_ref().map(|s| s.queue_client()) else {
|
|
tracing::warn!("no swarm queue is configured on this host; requested reviews don't run");
|
|
return;
|
|
};
|
|
let forge = Arc::clone(&state.forge);
|
|
let reviews = Arc::clone(&state.reviews);
|
|
let runs = state.runs.clone();
|
|
tokio::spawn(async move {
|
|
let forge = loop {
|
|
if let Some(forge) = forge.get() {
|
|
break Arc::clone(forge);
|
|
}
|
|
tokio::time::sleep(TIMING.pull_wait).await;
|
|
};
|
|
let queue = Queue(Some(client.clone()));
|
|
let handles = Handles {
|
|
forge: Arc::clone(&forge),
|
|
queue,
|
|
poster: forge,
|
|
runs,
|
|
};
|
|
consume(&client, TIMING, &reviews, handles).await;
|
|
});
|
|
}
|
|
|
|
/// What a review reaches outside the queue: the forge it reads the PR and
|
|
/// rules from, the classifier it runs on, the poster for its review, and the
|
|
/// store that records it.
|
|
struct Handles<S, Q, P> {
|
|
forge: Arc<S>,
|
|
queue: Q,
|
|
poster: Arc<P>,
|
|
runs: Runs,
|
|
}
|
|
|
|
impl<S, Q: Clone, P> Clone for Handles<S, Q, P> {
|
|
fn clone(&self) -> Self {
|
|
Self {
|
|
forge: Arc::clone(&self.forge),
|
|
queue: self.queue.clone(),
|
|
poster: Arc::clone(&self.poster),
|
|
runs: self.runs.clone(),
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Take requests from [`REQUEST_STREAM`] one at a time, each once a run
|
|
/// place is free, and review each in a task of its own. Never returns;
|
|
/// dropping it stops every review it started, and the queue delivers their
|
|
/// requests again.
|
|
///
|
|
/// The limit is read from the settings before each wait for a place. A
|
|
/// request that doesn't parse is dropped. One for a PR whose review is going
|
|
/// waits for [`Timing::busy_retry`] in the queue, without holding a place.
|
|
async fn consume<S, Q, P>(
|
|
client: &async_nats::Client,
|
|
timing: Timing,
|
|
reviews: &Arc<Reviews>,
|
|
handles: Handles<S, Q, P>,
|
|
) where
|
|
S: FileSource + PullSource + Send + Sync + 'static,
|
|
Q: RunSink + SettingsSource,
|
|
P: ReviewPoster + Send + Sync + 'static,
|
|
{
|
|
let consumer = loop {
|
|
match open_consumer(client, timing.ack_wait).await {
|
|
Ok(consumer) => break consumer,
|
|
Err(e) => tracing::warn!(
|
|
error = %format!("{e:#}"),
|
|
"opening the review request queue failed; trying again"
|
|
),
|
|
}
|
|
tokio::time::sleep(timing.pull_wait).await;
|
|
};
|
|
let mut tasks = JoinSet::new();
|
|
loop {
|
|
while let Some(ended) = tasks.try_join_next() {
|
|
if let Err(e) = ended {
|
|
tracing::error!(error = %e, "a requested review's task ended early");
|
|
}
|
|
}
|
|
match handles.queue.settings().await {
|
|
Ok(settings) => reviews.resize(settings.max_concurrent_reviews),
|
|
Err(e) => tracing::warn!(
|
|
error = %format!("{e:#}"),
|
|
"reading the review classifier's settings failed; the next run waits under \
|
|
the last limit read"
|
|
),
|
|
}
|
|
let slot = reviews.slot().await;
|
|
let message = match next(&consumer, timing.pull_wait).await {
|
|
Ok(Some(message)) => message,
|
|
Ok(None) => continue,
|
|
Err(e) => {
|
|
tracing::warn!(
|
|
error = %format!("{e:#}"),
|
|
"taking a review request from the queue failed; trying again"
|
|
);
|
|
drop(slot);
|
|
tokio::time::sleep(timing.pull_wait).await;
|
|
continue;
|
|
}
|
|
};
|
|
tasks.spawn(take(
|
|
message,
|
|
slot,
|
|
timing,
|
|
Arc::clone(reviews),
|
|
handles.clone(),
|
|
));
|
|
}
|
|
}
|
|
|
|
/// The next request in the queue, or `None` when none came within `wait`.
|
|
async fn next(
|
|
consumer: &PullConsumer,
|
|
wait: Duration,
|
|
) -> anyhow::Result<Option<jetstream::Message>> {
|
|
let mut batch = consumer
|
|
.batch()
|
|
.max_messages(1)
|
|
.expires(wait)
|
|
.messages()
|
|
.await
|
|
.context("asking the queue for a review request")?;
|
|
match batch.next().await {
|
|
Some(Ok(message)) => Ok(Some(message)),
|
|
Some(Err(e)) => Err(anyhow::anyhow!(e).context("reading a review request")),
|
|
None => Ok(None),
|
|
}
|
|
}
|
|
|
|
/// Review the request in `message`, holding `slot`, and ack it once its
|
|
/// record is written.
|
|
async fn take<S, Q, P>(
|
|
message: jetstream::Message,
|
|
slot: Slot,
|
|
timing: Timing,
|
|
reviews: Arc<Reviews>,
|
|
handles: Handles<S, Q, P>,
|
|
) where
|
|
S: FileSource + PullSource + Send + Sync + 'static,
|
|
Q: RunSink + SettingsSource,
|
|
P: ReviewPoster + Send + Sync + 'static,
|
|
{
|
|
let request: Requested = match serde_json::from_slice(&message.payload) {
|
|
Ok(request) => request,
|
|
Err(e) => {
|
|
tracing::warn!(
|
|
subject = %message.subject, error = %e,
|
|
"a queued review request did not parse; dropping it"
|
|
);
|
|
if let Err(e) = message.ack_with(AckKind::Term).await {
|
|
tracing::warn!(error = %e, "dropping an unparsable review request failed");
|
|
}
|
|
return;
|
|
}
|
|
};
|
|
let Some(claim) = reviews.claim(&request) else {
|
|
drop(slot);
|
|
tracing::info!(
|
|
repo = %format!("{}/{}", request.owner, request.name), pr = request.pr,
|
|
head_sha = request.head.as_str(),
|
|
"review requested while this PR's review is going; it waits in the queue"
|
|
);
|
|
if let Err(e) = message
|
|
.ack_with(AckKind::Nak(Some(timing.busy_retry)))
|
|
.await
|
|
{
|
|
tracing::warn!(error = %e, "putting a review request back in the queue failed");
|
|
}
|
|
return;
|
|
};
|
|
let Handles {
|
|
forge,
|
|
queue,
|
|
poster,
|
|
runs,
|
|
} = &handles;
|
|
let work = review(&**forge, &**forge, queue, &**poster, runs, &request, slot);
|
|
in_progress(&message, timing.ack_wait / 3, work).await;
|
|
drop(claim);
|
|
if let Err(e) = message.ack().await {
|
|
tracing::warn!(
|
|
repo = %format!("{}/{}", request.owner, request.name), pr = request.pr,
|
|
error = %e,
|
|
"acking a finished review request failed; the queue delivers it again"
|
|
);
|
|
}
|
|
}
|
|
|
|
/// Run `work`, telling the queue every `every` that `message` is still in
|
|
/// hand.
|
|
async fn in_progress(
|
|
message: &jetstream::Message,
|
|
every: Duration,
|
|
work: impl Future<Output = ()>,
|
|
) {
|
|
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");
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Review `req`, holding `slot`, and record the result in `runs`.
|
|
///
|
|
/// A request whose head is still the PR's, and whose head's newest review
|
|
/// [`reposted`] covers, frees `slot` and posts that verdict again. Any other
|
|
/// runs on the PR's current head, reading the settings for the model and
|
|
/// effort: a request for a head the PR has moved on from gets a review of the
|
|
/// new head, never the old head's verdict.
|
|
async fn review(
|
|
files: &impl FileSource,
|
|
pulls: &impl PullSource,
|
|
sink: &(impl RunSink + SettingsSource),
|
|
poster: &impl ReviewPoster,
|
|
runs: &Runs,
|
|
req: &Requested,
|
|
slot: Slot,
|
|
) {
|
|
let pull = read_pull(pulls, req).await;
|
|
let moved = matches!(&pull, Ok(pull) if !req.head.is_empty() && pull.head_sha != req.head);
|
|
if moved {
|
|
tracing::info!(
|
|
repo = %format!("{}/{}", req.owner, req.name), pr = req.pr,
|
|
head_sha = req.head.as_str(),
|
|
"review requested on a head the PR has moved on from; reviewing its current head"
|
|
);
|
|
}
|
|
if !moved
|
|
&& let Some(earlier) = newest_review(runs, req).await
|
|
&& let Some(ruling) = reposted(&earlier)
|
|
{
|
|
drop(slot);
|
|
runs.record(repost(poster, req, &earlier, ruling).await)
|
|
.await;
|
|
return;
|
|
}
|
|
let run = Box::pin(run(files, pull, sink, poster, req)).await;
|
|
drop(slot);
|
|
runs.record(run).await;
|
|
}
|
|
|
|
/// The PR `req` names, as the forge answers within [`forge::FORGE_DEADLINE`],
|
|
/// or why it couldn't be read.
|
|
async fn read_pull(pulls: &impl PullSource, req: &Requested) -> Result<PullForReview, String> {
|
|
match forge::within_forge_deadline(
|
|
"review classifier PR read",
|
|
pulls.pull(&req.owner, &req.name, req.pr),
|
|
)
|
|
.await
|
|
{
|
|
Some(Ok(pull)) => Ok(pull),
|
|
Some(Err(e)) => Err(format!("reading the PR failed: {e:#}")),
|
|
None => Err("the forge did not answer within the deadline".to_owned()),
|
|
}
|
|
}
|
|
|
|
/// The newest recorded review of the head `req` names. `None` when it names
|
|
/// none, nothing is recorded for it, or the store can't be read; each of
|
|
/// those leads to a fresh run.
|
|
async fn newest_review(runs: &Runs, req: &Requested) -> Option<RunRecord> {
|
|
if req.head.is_empty() {
|
|
return None;
|
|
}
|
|
let repo = format!("{}/{}", req.owner, req.name);
|
|
match runs
|
|
.newest_review(&repo, req.pr.unsigned_abs(), &req.head)
|
|
.await
|
|
{
|
|
Ok(found) => found,
|
|
Err(e) => {
|
|
tracing::warn!(
|
|
%repo, pr = req.pr, head_sha = req.head.as_str(), error = %format!("{e:#}"),
|
|
"reading the review classifier's earlier runs failed; running the review"
|
|
);
|
|
None
|
|
}
|
|
}
|
|
}
|
|
|
|
/// The ruling that posts `earlier`'s verdict again, when it approved or
|
|
/// requested changes. `None` for any other outcome, such as no verdict, which
|
|
/// a fresh run may improve on.
|
|
fn reposted(earlier: &RunRecord) -> Option<Ruling> {
|
|
let event = match earlier.outcome {
|
|
RunOutcome::Approved => Event::Approve,
|
|
RunOutcome::ChangesRequested => Event::RequestChanges,
|
|
_ => return None,
|
|
};
|
|
let (Some(rows), Some(overall)) = (&earlier.rows, earlier.overall) else {
|
|
return None;
|
|
};
|
|
let head = earlier.head_sha.as_deref().unwrap_or_default();
|
|
Some(Ruling {
|
|
event,
|
|
body: format!(
|
|
"Already reviewed `{head}`, verdict unchanged.\n\n{}\n{}",
|
|
table(rows),
|
|
overall_line(overall)
|
|
),
|
|
reason: Some(format!("a re-post of run {}'s verdict", earlier.id)),
|
|
})
|
|
}
|
|
|
|
/// Post `ruling`, `earlier`'s verdict, on `req`'s PR at `earlier`'s head. Its
|
|
/// record, carrying `earlier`'s answer.
|
|
async fn repost(
|
|
poster: &impl ReviewPoster,
|
|
req: &Requested,
|
|
earlier: &RunRecord,
|
|
ruling: Ruling,
|
|
) -> RunRecord {
|
|
let started_unix = unix_now();
|
|
tracing::info!(
|
|
repo = %earlier.repo, pr = req.pr, head_sha = req.head.as_str(), earlier = %earlier.id,
|
|
"review requested on a head already reviewed; posting its verdict again"
|
|
);
|
|
let reason = ruling.reason.clone();
|
|
let sent = post(poster, req, earlier.head_sha.as_deref(), ruling.review()).await;
|
|
record(
|
|
req,
|
|
earlier.head_sha.clone(),
|
|
started_unix,
|
|
ended(earlier.outcome, reason, sent),
|
|
(earlier.rows.clone(), earlier.overall),
|
|
)
|
|
}
|
|
|
|
/// Where a review is posted: the forge in production, a fake in the tests.
|
|
pub(crate) trait ReviewPoster {
|
|
/// The posted review's URL, when the forge names one.
|
|
fn post(
|
|
&self,
|
|
owner: &str,
|
|
repo: &str,
|
|
number: i64,
|
|
commit: Option<&str>,
|
|
review: Review,
|
|
) -> impl Future<Output = anyhow::Result<Option<String>>> + Send;
|
|
}
|
|
|
|
impl ReviewPoster for forge::Client {
|
|
async fn post(
|
|
&self,
|
|
owner: &str,
|
|
repo: &str,
|
|
number: i64,
|
|
commit: Option<&str>,
|
|
review: Review,
|
|
) -> anyhow::Result<Option<String>> {
|
|
self.post_classifier_review(owner, repo, number, commit, review)
|
|
.await
|
|
}
|
|
}
|
|
|
|
/// The swarm queue, when this host has one.
|
|
#[derive(Clone)]
|
|
struct Queue(Option<async_nats::Client>);
|
|
|
|
impl RunSink for Queue {
|
|
async fn run(&self, request: Vec<u8>) -> Reply {
|
|
let Some(client) = &self.0 else {
|
|
return Reply::Failed {
|
|
error: "no swarm queue is configured on this host".to_owned(),
|
|
};
|
|
};
|
|
if let Err(e) = swarm_queue_client::ensure_connected(client) {
|
|
return Reply::Failed {
|
|
error: swarm_queue_client::chain(&e),
|
|
};
|
|
}
|
|
client.run(request).await
|
|
}
|
|
}
|
|
|
|
impl SettingsSource for Queue {
|
|
async fn settings(&self) -> anyhow::Result<Settings> {
|
|
let Some(client) = &self.0 else {
|
|
anyhow::bail!("no swarm queue is configured on this host");
|
|
};
|
|
swarm_queue_client::ensure_connected(client)?;
|
|
client.settings().await
|
|
}
|
|
}
|
|
|
|
/// How far a run got.
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
enum Outcome {
|
|
/// The repo has no rules file: nothing was asked of the classifier.
|
|
NoRules { repo: String },
|
|
/// Nothing reached the classifier, for this reason.
|
|
NotRun(String),
|
|
/// The classifier answered a run of `sent` classifications.
|
|
Ran { sent: usize, reply: Reply },
|
|
}
|
|
|
|
/// Read the rules, run them on `pull`, the PR as it was read, and post the
|
|
/// review. The finished run.
|
|
async fn run(
|
|
files: &impl FileSource,
|
|
pull: Result<PullForReview, String>,
|
|
sink: &(impl RunSink + SettingsSource),
|
|
poster: &impl ReviewPoster,
|
|
req: &Requested,
|
|
) -> RunRecord {
|
|
let started_unix = unix_now();
|
|
let (head, narrowed, outcome) = classify(files, pull, sink, req).await;
|
|
let mut ruling = review_for(&outcome);
|
|
if let (Some(reason), Outcome::Ran { .. }) = (narrowed, &outcome) {
|
|
let _ = write!(
|
|
ruling.body,
|
|
"\nThe classifier read the forge's diff, with 3 lines of context, not \
|
|
{CONTEXT_LINES}: {reason}.\n"
|
|
);
|
|
}
|
|
let posted_as = match ruling.event {
|
|
Event::Approve => RunOutcome::Approved,
|
|
Event::RequestChanges => RunOutcome::ChangesRequested,
|
|
Event::Comment => RunOutcome::NoVerdict,
|
|
};
|
|
let reason = ruling.reason.clone();
|
|
let sent = post(poster, req, head.as_deref(), ruling.review()).await;
|
|
let answer = match &outcome {
|
|
Outcome::Ran { reply, .. } => runs::answer(reply),
|
|
Outcome::NotRun(_) | Outcome::NoRules { .. } => (None, None),
|
|
};
|
|
record(
|
|
req,
|
|
head,
|
|
started_unix,
|
|
ended(posted_as, reason, sent),
|
|
answer,
|
|
)
|
|
}
|
|
|
|
/// Post `review` on `req`'s PR, naming `head` when there is one. The posted
|
|
/// review's URL when the forge names one, or why the post failed.
|
|
async fn post(
|
|
poster: &impl ReviewPoster,
|
|
req: &Requested,
|
|
head: Option<&str>,
|
|
review: Review,
|
|
) -> Result<Option<String>, String> {
|
|
let repo = format!("{}/{}", req.owner, req.name);
|
|
let event = review.event;
|
|
let mut tries = 0;
|
|
let sent = post_retrying(poster, req, head, &review, &mut tries).await;
|
|
match sent {
|
|
Some(Ok(url)) => {
|
|
tracing::info!(
|
|
%repo, pr = req.pr, head_sha = head.unwrap_or_default(), event, tries,
|
|
"review classifier review posted"
|
|
);
|
|
Ok(url)
|
|
}
|
|
Some(Err(e)) => {
|
|
let error = format!("{e:#}");
|
|
tracing::warn!(
|
|
%repo, pr = req.pr, %error, tries,
|
|
"posting the review classifier's review failed"
|
|
);
|
|
Err(format!("posting the review failed: {error}"))
|
|
}
|
|
None => {
|
|
tracing::warn!(
|
|
%repo, pr = req.pr, tries,
|
|
"posting the review classifier's review failed: no answer within the deadline"
|
|
);
|
|
Err("the forge did not answer the review's post within the deadline".to_owned())
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Post `review`, each try within [`forge::FORGE_DEADLINE`], and post it
|
|
/// again after each of [`POST_RETRY_WAITS`] while a try fails with a
|
|
/// [`transient`] error or gets no answer by the deadline. The last try's
|
|
/// result, `None` when it got no answer; `tries` counts the tries made.
|
|
async fn post_retrying(
|
|
poster: &impl ReviewPoster,
|
|
req: &Requested,
|
|
head: Option<&str>,
|
|
review: &Review,
|
|
tries: &mut usize,
|
|
) -> Option<anyhow::Result<Option<String>>> {
|
|
let mut waits = POST_RETRY_WAITS.iter();
|
|
loop {
|
|
*tries += 1;
|
|
let sent = forge::within_forge_deadline(
|
|
"review classifier review",
|
|
poster.post(&req.owner, &req.name, req.pr, head, review.clone()),
|
|
)
|
|
.await;
|
|
let error = match &sent {
|
|
Some(Err(e)) if transient(e) => format!("{e:#}"),
|
|
None => "no answer within the deadline".to_owned(),
|
|
Some(_) => return sent,
|
|
};
|
|
let Some(wait) = waits.next() else {
|
|
return sent;
|
|
};
|
|
tracing::info!(
|
|
repo = %format!("{}/{}", req.owner, req.name), pr = req.pr,
|
|
%error, tries, wait_s = wait.as_secs(),
|
|
"posting the review classifier's review failed; trying again"
|
|
);
|
|
tokio::time::sleep(*wait).await;
|
|
}
|
|
}
|
|
|
|
/// Whether a post that failed with `e` may succeed when tried again: the
|
|
/// connection to the secret store or the forge failed or timed out. An answer
|
|
/// from either, such as a 4xx, is not transient.
|
|
///
|
|
/// The store's client puts its `reqwest::Error` in the chain itself; the
|
|
/// forge's wraps it in a transparent [`ForgejoError::ReqwestError`], whose
|
|
/// chain skips it.
|
|
fn transient(e: &anyhow::Error) -> bool {
|
|
e.chain().any(|cause| {
|
|
let http = match cause.downcast_ref::<ForgejoError>() {
|
|
Some(ForgejoError::ReqwestError(e)) => Some(e),
|
|
_ => cause.downcast_ref::<reqwest::Error>(),
|
|
};
|
|
http.is_some_and(|e| e.is_connect() || e.is_timeout())
|
|
})
|
|
}
|
|
|
|
/// How a run ended, as its record holds it.
|
|
struct Ended {
|
|
outcome: RunOutcome,
|
|
detail: Option<String>,
|
|
review_url: Option<String>,
|
|
}
|
|
|
|
/// `posted_as` when the post went through, `post-failed` otherwise. `reason`
|
|
/// is why the run has no clean verdict, if it has none.
|
|
fn ended(
|
|
posted_as: RunOutcome,
|
|
reason: Option<String>,
|
|
posted: Result<Option<String>, String>,
|
|
) -> Ended {
|
|
match posted {
|
|
Ok(review_url) => Ended {
|
|
outcome: posted_as,
|
|
detail: reason,
|
|
review_url,
|
|
},
|
|
Err(error) => Ended {
|
|
outcome: RunOutcome::PostFailed,
|
|
detail: Some(match reason {
|
|
Some(reason) => format!("{reason}; {error}"),
|
|
None => error,
|
|
}),
|
|
review_url: None,
|
|
},
|
|
}
|
|
}
|
|
|
|
/// The record of a run on `req` that read `head`.
|
|
fn record(
|
|
req: &Requested,
|
|
head: Option<String>,
|
|
started_unix: i64,
|
|
ended: Ended,
|
|
(rows, overall): (Option<Vec<Row>>, Option<Overall>),
|
|
) -> RunRecord {
|
|
let repo = format!("{}/{}", req.owner, req.name);
|
|
RunRecord {
|
|
id: super::mint_id().unwrap_or_else(|_| format!("{repo}#{}@{started_unix}", req.pr)),
|
|
trigger: Trigger::Review,
|
|
repo,
|
|
pr: req.pr.unsigned_abs(),
|
|
requester: req.requester.clone(),
|
|
head_sha: head,
|
|
started_unix,
|
|
finished_unix: unix_now(),
|
|
outcome: ended.outcome,
|
|
detail: ended.detail,
|
|
rows,
|
|
overall,
|
|
review_url: ended.review_url,
|
|
}
|
|
}
|
|
|
|
/// The PR's head when it was read, why its diff is the forge's 3-line one
|
|
/// when it is, and how far the run got on `pull`, the PR as it was read.
|
|
async fn classify(
|
|
files: &impl FileSource,
|
|
pull: Result<PullForReview, String>,
|
|
sink: &(impl RunSink + SettingsSource),
|
|
req: &Requested,
|
|
) -> (Option<String>, Option<String>, Outcome) {
|
|
let repo = format!("{}/{}", req.owner, req.name);
|
|
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;
|
|
let rules = match super::load_rules(files, &req.owner, &req.name).await {
|
|
Ok(super::Loaded::Rules(rules)) => rules,
|
|
Ok(super::Loaded::Absent { .. }) => return (head, narrowed, Outcome::NoRules { repo }),
|
|
Err(problem) => {
|
|
let detail = problem.detail.unwrap_or_default();
|
|
return (
|
|
head,
|
|
narrowed,
|
|
Outcome::NotRun(format!("the rules did not load: {detail}")),
|
|
);
|
|
}
|
|
};
|
|
let settings = match sink.settings().await {
|
|
Ok(settings) => settings,
|
|
Err(e) => {
|
|
return (
|
|
head,
|
|
narrowed,
|
|
Outcome::NotRun(format!("reading the classifier's settings failed: {e:#}")),
|
|
);
|
|
}
|
|
};
|
|
let runtime = match settings.runtime() {
|
|
Ok(runtime) => runtime,
|
|
Err(e) => {
|
|
return (
|
|
head,
|
|
narrowed,
|
|
Outcome::NotRun(format!("the classifier's settings: {e}")),
|
|
);
|
|
}
|
|
};
|
|
let request = TryOut {
|
|
operator: req.requester.clone(),
|
|
repo,
|
|
pr: req.pr.unsigned_abs(),
|
|
head_sha: pull.head_sha,
|
|
commits: pull.commits,
|
|
diff: pull.diff,
|
|
classifications: rules.classifications,
|
|
model: settings.model,
|
|
effort: settings.effort,
|
|
runtime,
|
|
};
|
|
let sent = request.classifications.len();
|
|
let outcome = match serde_json::to_vec(&request) {
|
|
Ok(payload) => Outcome::Ran {
|
|
sent,
|
|
reply: sink.run(payload).await,
|
|
},
|
|
Err(e) => Outcome::NotRun(format!("encoding the run failed: {e}")),
|
|
};
|
|
(head, narrowed, outcome)
|
|
}
|
|
|
|
/// What a run posts: the event, the review body, and why it has no clean
|
|
/// verdict when it has none.
|
|
struct Ruling {
|
|
event: Event,
|
|
body: String,
|
|
reason: Option<String>,
|
|
}
|
|
|
|
impl Ruling {
|
|
fn review(self) -> Review {
|
|
Review {
|
|
event: self.event.forge(),
|
|
body: self.body,
|
|
}
|
|
}
|
|
|
|
fn no_verdict(reason: String, rows: &[Row]) -> Self {
|
|
Self {
|
|
event: no_clean_verdict_event(),
|
|
body: no_verdict(&reason, rows),
|
|
reason: Some(reason),
|
|
}
|
|
}
|
|
}
|
|
|
|
/// The ruling on a run's outcome.
|
|
fn review_for(outcome: &Outcome) -> Ruling {
|
|
match outcome {
|
|
Outcome::NoRules { repo } => Ruling {
|
|
event: Event::Approve,
|
|
body: format!("no review rules are configured for {repo}; approved without checks\n"),
|
|
reason: None,
|
|
},
|
|
Outcome::NotRun(reason) => Ruling::no_verdict(reason.clone(), &[]),
|
|
Outcome::Ran {
|
|
reply: Reply::Failed { error },
|
|
..
|
|
} => Ruling::no_verdict(format!("the run failed: {error}"), &[]),
|
|
Outcome::Ran {
|
|
sent,
|
|
reply: Reply::Done { rows, overall },
|
|
} => {
|
|
let unanswered = rows.iter().filter(|r| r.error.is_some()).count();
|
|
if rows.is_empty() || rows.len() != *sent {
|
|
let reason = format!(
|
|
"the classifier answered {} of {sent} classifications",
|
|
rows.len()
|
|
);
|
|
Ruling::no_verdict(reason, rows)
|
|
} else if unanswered > 0 {
|
|
let reason = format!(
|
|
"{unanswered} of {} classifications got no answer",
|
|
rows.len()
|
|
);
|
|
Ruling::no_verdict(reason, rows)
|
|
} else {
|
|
let event = match overall {
|
|
Overall::Approve => Event::Approve,
|
|
Overall::RequestChanges => Event::RequestChanges,
|
|
};
|
|
Ruling {
|
|
event,
|
|
body: format!("{}\n{}", table(rows), overall_line(*overall)),
|
|
reason: None,
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// A review body for a run without a clean verdict.
|
|
fn no_verdict(reason: &str, rows: &[Row]) -> String {
|
|
let mut body = format!(
|
|
"**No verdict:** {}. This review doesn't approve or request changes.\n",
|
|
cell(reason)
|
|
);
|
|
if !rows.is_empty() {
|
|
body.push('\n');
|
|
body.push_str(&table(rows));
|
|
}
|
|
body
|
|
}
|
|
|
|
/// The result table, one row per classification.
|
|
fn table(rows: &[Row]) -> String {
|
|
let mut out = String::from(
|
|
"| result | check | certainty | reason |\n\
|
|
| --- | --- | --- | --- |\n",
|
|
);
|
|
for r in rows {
|
|
let (certainty, reason) = match &r.error {
|
|
Some(error) => ("—".to_owned(), error.as_str()),
|
|
None => (r.certainty.to_string(), r.reason.as_str()),
|
|
};
|
|
let _ = writeln!(
|
|
out,
|
|
"| {} | {} | {certainty} / {} | {} |",
|
|
result(r),
|
|
cell(&r.name),
|
|
r.threshold,
|
|
cell(reason)
|
|
);
|
|
}
|
|
out
|
|
}
|
|
|
|
/// A row's result cell: the symbol for what the row did to the review, then
|
|
/// the model's verdict. ✅ is a counted row that passed its polarity, ❌ a
|
|
/// counted row that failed it, ❔ a row that doesn't count: below its
|
|
/// threshold, not applicable, or without an answer. So `match` is ✅ under a
|
|
/// positive rule and ❌ under a negative one.
|
|
fn result(r: &Row) -> &'static str {
|
|
match (r.error.is_some(), r.counted, r.passed, r.verdict) {
|
|
(true, ..) => "❔ no answer",
|
|
(_, true, true, Verdict::Match) => "✅ match",
|
|
(_, true, true, Verdict::NoMatch) => "✅ no match",
|
|
(_, true, false, Verdict::Match) => "❌ match",
|
|
(_, true, false, Verdict::NoMatch) => "❌ no match",
|
|
(_, _, _, Verdict::Match) => "❔ match",
|
|
(_, _, _, Verdict::NoMatch) => "❔ no match",
|
|
(_, _, _, Verdict::NotApplicable) => "❔ unsure",
|
|
}
|
|
}
|
|
|
|
/// The closing line of a review: the overall verdict, with its symbol.
|
|
fn overall_line(overall: Overall) -> &'static str {
|
|
match overall {
|
|
Overall::Approve => "**Overall: ✅ approve**\n",
|
|
Overall::RequestChanges => "**Overall: ❌ request changes**\n",
|
|
}
|
|
}
|
|
|
|
/// `text` as one table cell's inline text: on one line, with a backslash before
|
|
/// every character that could end the cell, start markup, math (`$`) or raw
|
|
/// HTML, 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; `#` and `!` get only the zero-width space.
|
|
///
|
|
/// `_` is escaped except between two alphanumerics, where it can't open
|
|
/// emphasis (`ci_metrics.rs`). Other punctuation, such as `(`, `)`, `.` and
|
|
/// `:`, passes through: it starts no markup inside a table cell, and the `[`
|
|
/// that would make `(` a link target is escaped.
|
|
fn cell(text: &str) -> String {
|
|
let mut out = String::with_capacity(text.len());
|
|
for (i, word) in text.split_whitespace().enumerate() {
|
|
if i > 0 {
|
|
out.push(' ');
|
|
}
|
|
let chars: Vec<char> = word.chars().collect();
|
|
for (j, &c) in chars.iter().enumerate() {
|
|
match c {
|
|
'_' if j > 0
|
|
&& chars[j - 1].is_alphanumeric()
|
|
&& chars.get(j + 1).is_some_and(|n| n.is_alphanumeric()) =>
|
|
{
|
|
out.push(c);
|
|
}
|
|
'\\' | '|' | '`' | '*' | '_' | '~' | '[' | ']' | '<' | '>' | '&' | '$' => {
|
|
out.push('\\');
|
|
out.push(c);
|
|
}
|
|
'#' | '!' => {
|
|
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
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use std::sync::Mutex;
|
|
use std::sync::atomic::{AtomicUsize, Ordering};
|
|
use std::time::Duration;
|
|
|
|
use axum::http::StatusCode;
|
|
use swarm_wire::review_classifier::Polarity;
|
|
|
|
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)]
|
|
struct Posted(Mutex<Vec<(String, Option<String>, Review)>>);
|
|
|
|
impl ReviewPoster for Posted {
|
|
async fn post(
|
|
&self,
|
|
owner: &str,
|
|
repo: &str,
|
|
number: i64,
|
|
commit: Option<&str>,
|
|
review: Review,
|
|
) -> anyhow::Result<Option<String>> {
|
|
self.0.lock().expect("unpoisoned").push((
|
|
format!("{owner}/{repo}#{number}"),
|
|
commit.map(str::to_owned),
|
|
review,
|
|
));
|
|
Ok(Some(format!(
|
|
"https://forge/{owner}/{repo}/pulls/{number}#review"
|
|
)))
|
|
}
|
|
}
|
|
|
|
fn row(error: Option<&str>, counted: bool, passed: bool) -> Row {
|
|
Row {
|
|
name: "datetime".to_owned(),
|
|
polarity: Polarity::Positive,
|
|
verdict: if error.is_some() {
|
|
Verdict::NotApplicable
|
|
} else {
|
|
Verdict::Match
|
|
},
|
|
certainty: if error.is_some() { 0 } else { 90 },
|
|
threshold: 80,
|
|
counted,
|
|
passed,
|
|
reason: String::new(),
|
|
error: error.map(str::to_owned),
|
|
}
|
|
}
|
|
|
|
/// A row as the aggregation produces it for `polarity`, `verdict`,
|
|
/// `certainty` against `threshold`.
|
|
fn answered(polarity: Polarity, verdict: Verdict, certainty: u8, threshold: u8) -> Row {
|
|
let counted = verdict != Verdict::NotApplicable && certainty >= threshold;
|
|
let wanted = match polarity {
|
|
Polarity::Positive => Verdict::Match,
|
|
Polarity::Negative => Verdict::NoMatch,
|
|
};
|
|
Row {
|
|
name: "no-ads".to_owned(),
|
|
polarity,
|
|
verdict,
|
|
certainty,
|
|
threshold,
|
|
counted,
|
|
passed: counted && verdict == wanted,
|
|
reason: "Because.".to_owned(),
|
|
error: None,
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn a_row_shows_whether_it_passed_not_the_raw_verdict() {
|
|
use Polarity::{Negative, Positive};
|
|
use Verdict::{Match, NoMatch, NotApplicable};
|
|
for (case, r, expected) in [
|
|
(
|
|
"positive match passes",
|
|
answered(Positive, Match, 85, 65),
|
|
"| ✅ match | no-ads | 85 / 65 | Because. |",
|
|
),
|
|
(
|
|
"negative match fails",
|
|
answered(Negative, Match, 85, 65),
|
|
"| ❌ match | no-ads | 85 / 65 | Because. |",
|
|
),
|
|
(
|
|
"negative no match passes",
|
|
answered(Negative, NoMatch, 97, 80),
|
|
"| ✅ no match | no-ads | 97 / 80 | Because. |",
|
|
),
|
|
(
|
|
"positive no match fails",
|
|
answered(Positive, NoMatch, 97, 80),
|
|
"| ❌ no match | no-ads | 97 / 80 | Because. |",
|
|
),
|
|
(
|
|
"not applicable is unsure",
|
|
answered(Negative, NotApplicable, 98, 70),
|
|
"| ❔ unsure | no-ads | 98 / 70 | Because. |",
|
|
),
|
|
(
|
|
"below the threshold does not count",
|
|
answered(Positive, Match, 41, 60),
|
|
"| ❔ match | no-ads | 41 / 60 | Because. |",
|
|
),
|
|
(
|
|
"no answer shows its error",
|
|
Row {
|
|
error: Some("timed out".to_owned()),
|
|
verdict: NotApplicable,
|
|
certainty: 0,
|
|
counted: false,
|
|
passed: false,
|
|
reason: String::new(),
|
|
..answered(Positive, Match, 0, 65)
|
|
},
|
|
"| ❔ no answer | no-ads | — / 65 | timed out |",
|
|
),
|
|
] {
|
|
let body = table(&[r]);
|
|
assert_eq!(
|
|
body,
|
|
format!(
|
|
"| result | check | certainty | reason |\n\
|
|
| --- | --- | --- | --- |\n{expected}\n"
|
|
),
|
|
"{case}"
|
|
);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn the_overall_line_carries_its_emoji() {
|
|
assert_eq!(overall_line(Overall::Approve), "**Overall: ✅ approve**\n");
|
|
assert_eq!(
|
|
overall_line(Overall::RequestChanges),
|
|
"**Overall: ❌ request changes**\n"
|
|
);
|
|
}
|
|
|
|
fn request(name: &str) -> Requested {
|
|
Requested {
|
|
owner: "o".to_owned(),
|
|
name: name.to_owned(),
|
|
pr: 7,
|
|
head: "abc123".to_owned(),
|
|
requester: "alice".to_owned(),
|
|
}
|
|
}
|
|
|
|
/// [`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)
|
|
.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) {
|
|
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_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)
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn nine_timeouts_and_one_pass_do_not_approve() {
|
|
let mut rows: Vec<Row> = (0..9)
|
|
.map(|_| row(Some("timed out after 3 minutes"), false, false))
|
|
.collect();
|
|
rows.push(row(None, true, true));
|
|
// The classifier's own verdict for these rows is approve.
|
|
let reply = Reply::Done {
|
|
rows,
|
|
overall: Overall::Approve,
|
|
};
|
|
let (_, commit, review) = posted_for(reply, "r", rules_of(10)).await;
|
|
assert_eq!(review.event, "COMMENT");
|
|
assert!(
|
|
review
|
|
.body
|
|
.starts_with("**No verdict:** 9 of 10 classifications got no answer."),
|
|
"{}",
|
|
review.body
|
|
);
|
|
assert!(review.body.contains("timed out after 3 minutes"));
|
|
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 ruling = review_for(&Outcome::Ran {
|
|
sent: 0,
|
|
reply: Reply::Done {
|
|
rows: Vec::new(),
|
|
overall: Overall::Approve,
|
|
},
|
|
});
|
|
assert_eq!(ruling.event, Event::Comment);
|
|
}
|
|
|
|
/// A classifier that counts the runs reaching it and answers approve once
|
|
/// [`Gate::open`] lets each through, beside a store holding `settings`.
|
|
#[derive(Clone)]
|
|
struct Gate {
|
|
permits: Arc<tokio::sync::Semaphore>,
|
|
reached: Arc<AtomicUsize>,
|
|
settings: Arc<Mutex<Settings>>,
|
|
}
|
|
|
|
impl Gate {
|
|
fn closed() -> Self {
|
|
Self {
|
|
permits: Arc::new(tokio::sync::Semaphore::new(0)),
|
|
reached: Arc::default(),
|
|
settings: Arc::default(),
|
|
}
|
|
}
|
|
|
|
fn open(&self, runs: usize) {
|
|
self.permits.add_permits(runs);
|
|
}
|
|
|
|
fn reached(&self) -> usize {
|
|
self.reached.load(Ordering::SeqCst)
|
|
}
|
|
|
|
fn store(&self, max_concurrent_reviews: usize) {
|
|
self.settings
|
|
.lock()
|
|
.expect("unpoisoned")
|
|
.max_concurrent_reviews = max_concurrent_reviews;
|
|
}
|
|
}
|
|
|
|
impl RunSink for Gate {
|
|
async fn run(&self, _request: Vec<u8>) -> Reply {
|
|
self.reached.fetch_add(1, Ordering::SeqCst);
|
|
let _permit = self.permits.acquire().await.expect("never closed");
|
|
Reply::Done {
|
|
rows: vec![row(None, true, true)],
|
|
overall: Overall::Approve,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl SettingsSource for Gate {
|
|
async fn settings(&self) -> anyhow::Result<Settings> {
|
|
Ok(self.settings.lock().expect("unpoisoned").clone())
|
|
}
|
|
}
|
|
|
|
fn requested_at(head: &str) -> Requested {
|
|
Requested {
|
|
head: head.to_owned(),
|
|
..request("r")
|
|
}
|
|
}
|
|
|
|
/// 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 {
|
|
let earlier = RunRecord {
|
|
trigger: Trigger::Review,
|
|
outcome,
|
|
rows: Some(vec![row(
|
|
None,
|
|
true,
|
|
outcome != RunOutcome::ChangesRequested,
|
|
)]),
|
|
overall: Some(match outcome {
|
|
RunOutcome::ChangesRequested => Overall::RequestChanges,
|
|
_ => Overall::Approve,
|
|
}),
|
|
..crate::review_classifier::runs::tests::record(id)
|
|
};
|
|
store.record(earlier.clone()).await;
|
|
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<std::collections::HashMap<i64, String>>,
|
|
}
|
|
|
|
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<PullForReview> {
|
|
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<Option<forge::DefaultBranchFiles>> {
|
|
self.fake.default_branch_files(owner, repo, paths).await
|
|
}
|
|
}
|
|
|
|
/// A fake forge, a closed [`Gate`], a poster and a run store, and a
|
|
/// starter for reviews of a head on o/r#7.
|
|
struct Rig {
|
|
reviews: Arc<Reviews>,
|
|
forge: Arc<Heads>,
|
|
gate: Gate,
|
|
poster: Arc<Posted>,
|
|
doc: Arc<FakeDoc>,
|
|
store: Runs,
|
|
}
|
|
|
|
impl Rig {
|
|
fn new() -> Self {
|
|
Self::with_doc(Arc::default())
|
|
}
|
|
|
|
fn with_doc(doc: Arc<FakeDoc>) -> Self {
|
|
Self {
|
|
reviews: Arc::new(Reviews::default()),
|
|
forge: Arc::new(Heads::new()),
|
|
gate: Gate::closed(),
|
|
poster: Arc::new(Posted::default()),
|
|
store: Runs::Fake(Arc::clone(&doc)),
|
|
doc,
|
|
}
|
|
}
|
|
|
|
/// [`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<String>, Review)> {
|
|
self.poster.0.lock().expect("unpoisoned").clone()
|
|
}
|
|
}
|
|
|
|
/// Await `task`, a re-post. A task that ran instead waits on the closed
|
|
/// [`Gate`], so this fails rather than hang.
|
|
async fn reposted(task: JoinHandle<()>) {
|
|
tokio::time::timeout(Duration::from_secs(5), task)
|
|
.await
|
|
.expect("re-posted without waiting for a run place")
|
|
.expect("re-posted");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_request_on_a_head_with_a_clean_verdict_posts_it_again_without_a_run() {
|
|
for (outcome, event, overall) in [
|
|
(
|
|
RunOutcome::Approved,
|
|
"APPROVED",
|
|
"**Overall: ✅ approve**\n",
|
|
),
|
|
(
|
|
RunOutcome::ChangesRequested,
|
|
"REQUEST_CHANGES",
|
|
"**Overall: ❌ request changes**\n",
|
|
),
|
|
] {
|
|
let rig = Rig::new();
|
|
let earlier = reviewed(&rig.store, "earlier", outcome).await;
|
|
|
|
reposted(rig.start_at("abc123")).await;
|
|
|
|
assert_eq!(
|
|
rig.gate.reached(),
|
|
0,
|
|
"{event}: no run reached the classifier"
|
|
);
|
|
let posted = rig.posted();
|
|
assert_eq!(posted.len(), 1, "{event}");
|
|
let (target, commit, review) = &posted[0];
|
|
assert_eq!(target, "o/r#7");
|
|
assert_eq!(commit.as_deref(), Some("abc123"));
|
|
assert_eq!(review.event, event);
|
|
assert!(
|
|
review
|
|
.body
|
|
.starts_with("Already reviewed `abc123`, verdict unchanged.\n\n"),
|
|
"{}",
|
|
review.body
|
|
);
|
|
assert!(
|
|
review.body.contains("| datetime | 90 / 80 |"),
|
|
"{}",
|
|
review.body
|
|
);
|
|
assert!(review.body.ends_with(overall), "{}", review.body);
|
|
|
|
let recorded = rig.doc.runs();
|
|
assert_eq!(recorded.len(), 2, "{event}");
|
|
assert_eq!(recorded[0].outcome, outcome);
|
|
assert_eq!(recorded[0].head_sha.as_deref(), Some("abc123"));
|
|
assert_eq!(recorded[0].rows, earlier.rows);
|
|
assert_eq!(
|
|
recorded[0].detail.as_deref(),
|
|
Some("a re-post of run earlier's verdict")
|
|
);
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_request_on_a_head_without_a_clean_verdict_runs_again() {
|
|
for (case, seed) in [
|
|
("no verdict", vec![RunOutcome::NoVerdict]),
|
|
("post failed", vec![RunOutcome::PostFailed]),
|
|
(
|
|
"an approval since followed by no verdict",
|
|
vec![RunOutcome::Approved, RunOutcome::NoVerdict],
|
|
),
|
|
] {
|
|
let rig = Rig::new();
|
|
for (i, outcome) in seed.into_iter().enumerate() {
|
|
reviewed(&rig.store, &format!("earlier-{i}"), outcome).await;
|
|
}
|
|
|
|
let task = rig.start_at("abc123");
|
|
rig.gate.open(1);
|
|
task.await.expect("ran");
|
|
|
|
assert_eq!(rig.gate.reached(), 1, "{case}: a fresh run");
|
|
let posted = rig.posted();
|
|
assert_eq!(posted.len(), 1, "{case}");
|
|
assert_eq!(posted[0].2.event, "APPROVED", "{case}");
|
|
assert!(
|
|
!posted[0].2.body.starts_with("Already reviewed"),
|
|
"{case}: {}",
|
|
posted[0].2.body
|
|
);
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn only_a_review_of_the_same_head_is_posted_again() {
|
|
let approved = reviewed(&Runs::Fake(Arc::default()), "e", RunOutcome::Approved).await;
|
|
for (case, earlier) in [
|
|
(
|
|
"a try-out",
|
|
RunRecord {
|
|
trigger: Trigger::TryOut,
|
|
..approved.clone()
|
|
},
|
|
),
|
|
(
|
|
"another head",
|
|
RunRecord {
|
|
head_sha: Some("def456".to_owned()),
|
|
..approved.clone()
|
|
},
|
|
),
|
|
(
|
|
"another PR",
|
|
RunRecord {
|
|
pr: 8,
|
|
..approved.clone()
|
|
},
|
|
),
|
|
] {
|
|
let rig = Rig::new();
|
|
rig.store.record(earlier).await;
|
|
|
|
let task = rig.start_at("abc123");
|
|
rig.gate.open(1);
|
|
task.await.expect("ran");
|
|
assert_eq!(rig.gate.reached(), 1, "{case}: a fresh run");
|
|
}
|
|
}
|
|
|
|
/// The queue's timing in the tests: short enough that a request no one
|
|
/// acks comes back within a test.
|
|
const TEST_TIMING: Timing = Timing {
|
|
ack_wait: Duration::from_secs(1),
|
|
pull_wait: Duration::from_millis(200),
|
|
busy_retry: Duration::from_millis(200),
|
|
};
|
|
|
|
/// A [`Rig`] behind a real queue: requests go in through [`accept`], and
|
|
/// [`consume`] takes them.
|
|
struct Queued {
|
|
_nats: TestNats,
|
|
client: async_nats::Client,
|
|
rig: Rig,
|
|
}
|
|
|
|
impl Queued {
|
|
async fn new() -> Self {
|
|
let nats = TestNats::start();
|
|
let client = nats.client().await;
|
|
Self {
|
|
_nats: nats,
|
|
client,
|
|
rig: Rig::new(),
|
|
}
|
|
}
|
|
|
|
/// Queue a request for o/r#`pr` at `head`.
|
|
async fn request(&self, pr: i64, head: &str) {
|
|
let request = Requested {
|
|
pr,
|
|
..requested_at(head)
|
|
};
|
|
accept(
|
|
Some(&self.client),
|
|
&*self.rig.poster,
|
|
&self.rig.store,
|
|
&request,
|
|
)
|
|
.await;
|
|
assert!(self.rig.posted().is_empty(), "queued without a comment");
|
|
}
|
|
|
|
/// A controller taking requests, its run places in `reviews`.
|
|
fn consume(&self, reviews: &Arc<Reviews>) -> JoinHandle<()> {
|
|
let client = self.client.clone();
|
|
let reviews = Arc::clone(reviews);
|
|
let forge = Arc::clone(&self.rig.forge);
|
|
let gate = self.rig.gate.clone();
|
|
let poster = Arc::clone(&self.rig.poster);
|
|
let store = self.rig.store.clone();
|
|
tokio::spawn(async move {
|
|
let handles = Handles {
|
|
forge,
|
|
queue: gate,
|
|
poster,
|
|
runs: store,
|
|
};
|
|
consume(&client, TEST_TIMING, &reviews, handles).await;
|
|
})
|
|
}
|
|
|
|
/// How many requests the queue holds, taken or not.
|
|
async fn waiting(&self) -> u64 {
|
|
jetstream::new(self.client.clone())
|
|
.get_stream(REQUEST_STREAM)
|
|
.await
|
|
.expect("the stream")
|
|
.info()
|
|
.await
|
|
.expect("its info")
|
|
.state
|
|
.messages
|
|
}
|
|
|
|
async fn consumer_info(&self) -> jetstream::consumer::Info {
|
|
let js = jetstream::new(self.client.clone());
|
|
let stream = js.get_stream(REQUEST_STREAM).await.expect("the stream");
|
|
let mut consumer: PullConsumer = stream
|
|
.get_consumer(REQUEST_CONSUMER)
|
|
.await
|
|
.expect("the consumer");
|
|
consumer.info().await.expect("its info").clone()
|
|
}
|
|
}
|
|
|
|
/// Wait until `done` holds, for at most ten seconds.
|
|
async fn until(what: &str, done: impl Fn() -> bool) {
|
|
tokio::time::timeout(Duration::from_secs(10), async {
|
|
while !done() {
|
|
tokio::time::sleep(Duration::from_millis(20)).await;
|
|
}
|
|
})
|
|
.await
|
|
.unwrap_or_else(|_| panic!("never: {what}"));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_queued_request_runs_after_a_restart() {
|
|
let q = Queued::new().await;
|
|
q.request(7, "abc123").await;
|
|
let first = q.consume(&q.rig.reviews);
|
|
until("the first controller's run started", || {
|
|
q.rig.gate.reached() == 1
|
|
})
|
|
.await;
|
|
first.abort();
|
|
assert!(first.await.expect_err("stopped").is_cancelled());
|
|
assert!(q.rig.posted().is_empty());
|
|
|
|
// A restarted controller has run places of its own.
|
|
let _second = q.consume(&Arc::new(Reviews::default()));
|
|
until("the request ran again", || q.rig.gate.reached() == 2).await;
|
|
q.rig.gate.open(1);
|
|
until("its review is posted", || q.rig.posted().len() == 1).await;
|
|
let posted = q.rig.posted();
|
|
assert_eq!(posted[0].0, "o/r#7");
|
|
assert_eq!(posted[0].2.event, "APPROVED");
|
|
until("its record is written", || q.rig.doc.runs().len() == 1).await;
|
|
tokio::time::sleep(TEST_TIMING.ack_wait * 2).await;
|
|
assert_eq!(q.waiting().await, 0, "the finished request is acked");
|
|
assert_eq!(q.rig.gate.reached(), 2, "no run after the ack");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_newer_request_for_a_pr_replaces_the_waiting_one() {
|
|
let q = Queued::new().await;
|
|
q.rig.forge.move_to(7, "new");
|
|
q.request(7, "old").await;
|
|
q.request(8, "abc123").await;
|
|
q.request(7, "new").await;
|
|
assert_eq!(q.waiting().await, 2, "one request per PR");
|
|
|
|
let _consumer = q.consume(&q.rig.reviews);
|
|
q.rig.gate.open(2);
|
|
until("both PRs are reviewed", || q.rig.posted().len() == 2).await;
|
|
tokio::time::sleep(TEST_TIMING.ack_wait * 2).await;
|
|
let mut posted: Vec<_> = q
|
|
.rig
|
|
.posted()
|
|
.into_iter()
|
|
.map(|(target, commit, _)| (target, commit))
|
|
.collect();
|
|
posted.sort();
|
|
assert_eq!(
|
|
posted,
|
|
[
|
|
("o/r#7".to_owned(), Some("new".to_owned())),
|
|
("o/r#8".to_owned(), Some("abc123".to_owned())),
|
|
]
|
|
);
|
|
assert_eq!(q.rig.gate.reached(), 2, "one run per PR");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_request_for_a_head_the_pr_has_moved_on_from_reviews_the_new_head() {
|
|
let q = Queued::new().await;
|
|
// `old` has an approval on record; the PR has since moved to `new`.
|
|
let approved = reviewed(&q.rig.store, "earlier", RunOutcome::Approved).await;
|
|
q.rig
|
|
.store
|
|
.record(RunRecord {
|
|
id: "earlier-old".to_owned(),
|
|
head_sha: Some("old".to_owned()),
|
|
..approved
|
|
})
|
|
.await;
|
|
q.rig.forge.move_to(7, "new");
|
|
q.request(7, "old").await;
|
|
let _consumer = q.consume(&q.rig.reviews);
|
|
until("the run reached the classifier", || {
|
|
q.rig.gate.reached() == 1
|
|
})
|
|
.await;
|
|
q.rig.gate.open(1);
|
|
until("the new head is reviewed", || q.rig.posted().len() == 1).await;
|
|
let posted = q.rig.posted();
|
|
assert_eq!(posted[0].1.as_deref(), Some("new"));
|
|
assert!(
|
|
!posted[0].2.body.starts_with("Already reviewed"),
|
|
"the old head's verdict was posted again: {}",
|
|
posted[0].2.body
|
|
);
|
|
|
|
// Control: with the PR back on `old`, the same request re-posts its
|
|
// recorded verdict without a run.
|
|
q.rig.forge.move_to(7, "old");
|
|
accept(
|
|
Some(&q.client),
|
|
&*q.rig.poster,
|
|
&q.rig.store,
|
|
&requested_at("old"),
|
|
)
|
|
.await;
|
|
until("the verdict is posted again", || q.rig.posted().len() == 2).await;
|
|
let posted = q.rig.posted();
|
|
assert_eq!(posted[1].1.as_deref(), Some("old"));
|
|
assert!(posted[1].2.body.starts_with("Already reviewed `old`"));
|
|
assert_eq!(q.rig.gate.reached(), 1, "the re-post ran nothing");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn nothing_is_refused_while_every_run_place_is_taken() {
|
|
let q = Queued::new().await;
|
|
let prs = 1..=12;
|
|
for pr in prs.clone() {
|
|
q.request(pr, "abc123").await;
|
|
}
|
|
let _consumer = q.consume(&q.rig.reviews);
|
|
let limit = Settings::default().max_concurrent_reviews;
|
|
until("every place is taken", || q.rig.gate.reached() == limit).await;
|
|
tokio::time::sleep(TEST_TIMING.ack_wait * 2).await;
|
|
assert_eq!(q.rig.gate.reached(), limit, "the rest wait");
|
|
assert!(q.rig.posted().is_empty(), "nothing is refused");
|
|
assert_eq!(q.waiting().await, 12);
|
|
|
|
q.rig.gate.open(12);
|
|
until("every PR is reviewed", || q.rig.posted().len() == 12).await;
|
|
let mut targets: Vec<_> = q.rig.posted().into_iter().map(|(t, ..)| t).collect();
|
|
targets.sort();
|
|
let mut wanted: Vec<_> = prs.map(|pr| format!("o/r#{pr}")).collect();
|
|
wanted.sort();
|
|
assert_eq!(targets, wanted);
|
|
assert!(q.rig.posted().iter().all(|(_, _, r)| r.event == "APPROVED"));
|
|
assert_eq!(q.rig.gate.reached(), 12);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_run_longer_than_the_ack_wait_keeps_its_request() {
|
|
let q = Queued::new().await;
|
|
q.request(7, "abc123").await;
|
|
let _consumer = q.consume(&q.rig.reviews);
|
|
until("the run started", || q.rig.gate.reached() == 1).await;
|
|
tokio::time::sleep(TEST_TIMING.ack_wait * 3).await;
|
|
let info = q.consumer_info().await;
|
|
assert_eq!(info.num_ack_pending, 1, "the request is in hand");
|
|
assert_eq!(info.num_redelivered, 0, "and was never delivered again");
|
|
q.rig.gate.open(1);
|
|
until("its review is posted", || q.rig.posted().len() == 1).await;
|
|
assert_eq!(q.rig.gate.reached(), 1);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_request_for_a_pr_whose_review_is_going_waits_for_it() {
|
|
let q = Queued::new().await;
|
|
q.request(7, "abc123").await;
|
|
let _consumer = q.consume(&q.rig.reviews);
|
|
until("the run started", || q.rig.gate.reached() == 1).await;
|
|
q.request(7, "abc123").await;
|
|
tokio::time::sleep(TEST_TIMING.ack_wait * 2).await;
|
|
assert_eq!(q.rig.gate.reached(), 1, "one run on the PR at once");
|
|
|
|
// The run approved, so the waiting request posts its verdict again.
|
|
q.rig.gate.open(1);
|
|
until("both are posted", || q.rig.posted().len() == 2).await;
|
|
let posted = q.rig.posted();
|
|
assert_eq!(posted[0].2.event, "APPROVED");
|
|
assert!(
|
|
posted[1].2.body.starts_with("Already reviewed `abc123`"),
|
|
"{}",
|
|
posted[1].2.body
|
|
);
|
|
assert_eq!(q.rig.gate.reached(), 1);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_request_that_does_not_parse_is_dropped() {
|
|
let q = Queued::new().await;
|
|
q.request(8, "abc123").await;
|
|
jetstream::new(q.client.clone())
|
|
.publish(format!("{REQUEST_SUBJECT_PREFIX}.o.r.7"), "{}".into())
|
|
.await
|
|
.expect("published")
|
|
.await
|
|
.expect("stored");
|
|
let _consumer = q.consume(&q.rig.reviews);
|
|
q.rig.gate.open(2);
|
|
until("the request that parses is reviewed", || {
|
|
q.rig.posted().len() == 1
|
|
})
|
|
.await;
|
|
tokio::time::timeout(Duration::from_secs(10), async {
|
|
while q.waiting().await > 0 {
|
|
tokio::time::sleep(Duration::from_millis(20)).await;
|
|
}
|
|
})
|
|
.await
|
|
.expect("both leave the queue");
|
|
assert_eq!(q.rig.posted()[0].0, "o/r#8");
|
|
assert_eq!(q.rig.gate.reached(), 1);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_request_that_cannot_be_queued_posts_a_no_verdict_comment() {
|
|
let rig = Rig::new();
|
|
accept(None, &*rig.poster, &rig.store, &requested_at("abc123")).await;
|
|
let posted = rig.posted();
|
|
assert_eq!(posted.len(), 1);
|
|
let (target, commit, review) = &posted[0];
|
|
assert_eq!(target, "o/r#7");
|
|
assert_eq!(*commit, None);
|
|
assert_eq!(review.event, "COMMENT");
|
|
assert!(
|
|
review.body.starts_with(
|
|
"**No verdict:** the request could not be queued: no swarm queue is configured \
|
|
on this host; remove the review request and request it again."
|
|
),
|
|
"{}",
|
|
review.body
|
|
);
|
|
let recorded = rig.doc.runs();
|
|
assert_eq!(recorded.len(), 1);
|
|
assert_eq!(recorded[0].outcome, RunOutcome::NoVerdict);
|
|
assert_eq!(recorded[0].head_sha, None);
|
|
}
|
|
|
|
#[test]
|
|
fn each_pr_has_a_subject_of_its_own() {
|
|
let at = |owner: &str, name: &str, pr| {
|
|
Requested {
|
|
owner: owner.to_owned(),
|
|
name: name.to_owned(),
|
|
pr,
|
|
..request("r")
|
|
}
|
|
.subject()
|
|
};
|
|
assert_eq!(
|
|
at("hyperhive", "hyperhive", 5219),
|
|
"$SWARM.review-request.hyperhive.hyperhive.5219"
|
|
);
|
|
assert_eq!(
|
|
at("a.b", "c", 1),
|
|
"$SWARM.review-request.a_2eb.c.1",
|
|
"a dot stays inside its token"
|
|
);
|
|
assert_ne!(at("a.b", "c", 1), at("a", "b.c", 1));
|
|
assert_ne!(at("a_2eb", "c", 1), at("a.b", "c", 1));
|
|
assert_ne!(at("o", "r", 1), at("o", "r", 11));
|
|
for subject in [at("a.b", "c d*>", 1), at("ü", "r", 1)] {
|
|
assert_eq!(subject.split('.').count(), 5, "{subject}");
|
|
assert!(
|
|
!subject.contains(['*', '>', ' ']),
|
|
"{subject} is a literal subject"
|
|
);
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_new_limit_applies_to_the_next_run_without_a_restart() {
|
|
let q = Queued::new().await;
|
|
let reviews = &q.rig.reviews;
|
|
let gate = &q.rig.gate;
|
|
|
|
// The stored limit is read before a run waits, not the default 2.
|
|
gate.store(1);
|
|
q.request(1, "abc123").await;
|
|
q.request(2, "abc123").await;
|
|
let _consumer = q.consume(reviews);
|
|
until("the first run started", || gate.reached() == 1).await;
|
|
tokio::time::sleep(TEST_TIMING.ack_wait).await;
|
|
assert_eq!(gate.reached(), 1);
|
|
|
|
// Raised, as a PUT does: the waiting run starts at once.
|
|
gate.store(3);
|
|
reviews.resize(3);
|
|
until("the waiting run started", || gate.reached() == 2).await;
|
|
q.request(3, "abc123").await;
|
|
until("a third run started", || gate.reached() == 3).await;
|
|
|
|
// Lowered: the three keep running and the next one waits.
|
|
gate.store(1);
|
|
q.request(4, "abc123").await;
|
|
tokio::time::sleep(TEST_TIMING.ack_wait).await;
|
|
assert_eq!(gate.reached(), 3);
|
|
|
|
gate.open(4);
|
|
until("every run is posted", || q.rig.posted().len() == 4).await;
|
|
assert_eq!(gate.reached(), 4);
|
|
until("every place is freed", || reviews.slots().running <= 1).await;
|
|
}
|
|
|
|
/// The record of one run against the fake forge, holding `rules`, posted
|
|
/// by `poster`.
|
|
async fn recorded_for(
|
|
reply: Reply,
|
|
repo: &str,
|
|
rules: (&'static str, String),
|
|
poster: &impl ReviewPoster,
|
|
) -> RunRecord {
|
|
let fake = Fake::new(reply, vec![rules]);
|
|
run_read(&fake, &fake, poster, &request(repo)).await
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn each_way_a_run_ends_has_its_outcome() {
|
|
let answered = |rows: Vec<Row>, overall| Reply::Done { rows, overall };
|
|
let failed = Reply::Failed {
|
|
error: "no API key".to_owned(),
|
|
};
|
|
for (case, reply, repo, rules, outcome, detail) in [
|
|
(
|
|
"approve",
|
|
answered(vec![row(None, true, true)], Overall::Approve),
|
|
"r",
|
|
rules_json(),
|
|
RunOutcome::Approved,
|
|
None,
|
|
),
|
|
(
|
|
"request changes",
|
|
answered(vec![row(None, true, false)], Overall::RequestChanges),
|
|
"r",
|
|
rules_json(),
|
|
RunOutcome::ChangesRequested,
|
|
None,
|
|
),
|
|
(
|
|
"a failed run",
|
|
failed.clone(),
|
|
"r",
|
|
rules_json(),
|
|
RunOutcome::NoVerdict,
|
|
Some("the run failed: no API key"),
|
|
),
|
|
(
|
|
"a short answer",
|
|
answered(vec![row(None, true, true)], Overall::Approve),
|
|
"r",
|
|
rules_of(2),
|
|
RunOutcome::NoVerdict,
|
|
Some("the classifier answered 1 of 2 classifications"),
|
|
),
|
|
(
|
|
"an unanswered classification",
|
|
answered(vec![row(Some("timed out"), false, false)], Overall::Approve),
|
|
"r",
|
|
rules_json(),
|
|
RunOutcome::NoVerdict,
|
|
Some("1 of 1 classifications got no answer"),
|
|
),
|
|
(
|
|
"an unreadable PR",
|
|
answered(vec![row(None, true, true)], Overall::Approve),
|
|
"elsewhere",
|
|
rules_json(),
|
|
RunOutcome::NoVerdict,
|
|
Some("reading the PR failed"),
|
|
),
|
|
] {
|
|
let run = recorded_for(reply.clone(), repo, rules, &Posted::default()).await;
|
|
assert_eq!(run.outcome, outcome, "{case}");
|
|
assert_eq!(run.trigger, Trigger::Review, "{case}");
|
|
assert_eq!(run.requester, "alice", "{case}");
|
|
match detail {
|
|
None => assert_eq!(run.detail, None, "{case}"),
|
|
Some(want) => assert!(
|
|
run.detail.as_deref().is_some_and(|d| d.starts_with(want)),
|
|
"{case}: {:?}",
|
|
run.detail
|
|
),
|
|
}
|
|
assert_eq!(
|
|
run.review_url.as_deref(),
|
|
Some(format!("https://forge/o/{repo}/pulls/7#review").as_str()),
|
|
"{case}"
|
|
);
|
|
if repo == "elsewhere" {
|
|
assert_eq!(run.head_sha, None, "{case}");
|
|
assert_eq!((run.rows, run.overall), (None, None), "{case}");
|
|
} else {
|
|
assert_eq!(run.head_sha.as_deref(), Some("abc123"), "{case}");
|
|
assert_eq!((run.rows, run.overall), runs::answer(&reply), "{case}");
|
|
}
|
|
}
|
|
|
|
let [yml, _] = crate::review_classifier::RULES_PATHS;
|
|
let fake = Fake::new(failed, vec![rules_json(), (yml, String::new())]);
|
|
let run = run_read(&fake, &fake, &Posted::default(), &request("r")).await;
|
|
assert_eq!(run.outcome, RunOutcome::NoVerdict);
|
|
assert!(
|
|
run.detail
|
|
.as_deref()
|
|
.is_some_and(|d| d.starts_with("the rules did not load")),
|
|
"{:?}",
|
|
run.detail
|
|
);
|
|
}
|
|
|
|
/// A forge that refuses every review.
|
|
struct Refusing;
|
|
|
|
impl ReviewPoster for Refusing {
|
|
async fn post(
|
|
&self,
|
|
_owner: &str,
|
|
_repo: &str,
|
|
_number: i64,
|
|
_commit: Option<&str>,
|
|
_review: Review,
|
|
) -> anyhow::Result<Option<String>> {
|
|
anyhow::bail!("the forge answered 500")
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_failed_post_is_post_failed_with_its_reason() {
|
|
let approve = Reply::Done {
|
|
rows: vec![row(None, true, true)],
|
|
overall: Overall::Approve,
|
|
};
|
|
let run = recorded_for(approve, "r", rules_json(), &Refusing).await;
|
|
assert_eq!(run.outcome, RunOutcome::PostFailed);
|
|
assert_eq!(
|
|
run.detail.as_deref(),
|
|
Some("posting the review failed: the forge answered 500")
|
|
);
|
|
assert_eq!(run.review_url, None);
|
|
|
|
let failed = Reply::Failed {
|
|
error: "no API key".to_owned(),
|
|
};
|
|
let run = recorded_for(failed, "r", rules_json(), &Refusing).await;
|
|
assert_eq!(run.outcome, RunOutcome::PostFailed);
|
|
assert_eq!(
|
|
run.detail.as_deref(),
|
|
Some("the run failed: no API key; posting the review failed: the forge answered 500")
|
|
);
|
|
}
|
|
|
|
/// The forge client's error for a port nothing listens on.
|
|
async fn refused_by_port() -> ForgejoError {
|
|
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
|
let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap();
|
|
drop(listener);
|
|
let forge = forgejo_api::Forgejo::new(forgejo_api::Auth::Token("stub-token"), url).unwrap();
|
|
forge.user_get_current().await.expect_err("nothing listens")
|
|
}
|
|
|
|
/// A refused connection as the forge client reports it.
|
|
async fn forge_refused() -> anyhow::Error {
|
|
anyhow::Error::from(refused_by_port().await).context("posting the review")
|
|
}
|
|
|
|
/// A refused connection as the store client reports it: the
|
|
/// `reqwest::Error` itself in the chain, under the login's context.
|
|
async fn store_refused() -> anyhow::Error {
|
|
let ForgejoError::ReqwestError(e) = refused_by_port().await else {
|
|
panic!("a connect error is a reqwest error");
|
|
};
|
|
anyhow::Error::new(e)
|
|
.context("An error occurred with the request")
|
|
.context("logging in to the swarm secret store")
|
|
}
|
|
|
|
/// A poster that fails with each of `failures` in turn, then posts.
|
|
struct Flaky {
|
|
failures: Mutex<Vec<anyhow::Error>>,
|
|
tries: AtomicUsize,
|
|
posted: Posted,
|
|
}
|
|
|
|
impl Flaky {
|
|
fn new(mut failures: Vec<anyhow::Error>) -> Self {
|
|
failures.reverse();
|
|
Self {
|
|
failures: Mutex::new(failures),
|
|
tries: AtomicUsize::new(0),
|
|
posted: Posted::default(),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl ReviewPoster for Flaky {
|
|
async fn post(
|
|
&self,
|
|
owner: &str,
|
|
repo: &str,
|
|
number: i64,
|
|
commit: Option<&str>,
|
|
review: Review,
|
|
) -> anyhow::Result<Option<String>> {
|
|
self.tries.fetch_add(1, Ordering::SeqCst);
|
|
let failure = self.failures.lock().expect("unpoisoned").pop();
|
|
match failure {
|
|
Some(e) => Err(e),
|
|
None => self.posted.post(owner, repo, number, commit, review).await,
|
|
}
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_post_refused_a_connection_is_tried_again() {
|
|
let failures = vec![forge_refused().await, store_refused().await];
|
|
tokio::time::pause();
|
|
let approve = Reply::Done {
|
|
rows: vec![row(None, true, true)],
|
|
overall: Overall::Approve,
|
|
};
|
|
let fake = Fake::new(approve, vec![rules_json()]);
|
|
let poster = Flaky::new(failures);
|
|
let started = tokio::time::Instant::now();
|
|
let run = run_read(&fake, &fake, &poster, &request("r")).await;
|
|
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);
|
|
assert_eq!(fake.sent.lock().expect("unpoisoned").len(), 1, "one run");
|
|
let waited = started.elapsed();
|
|
assert!(
|
|
waited >= POST_RETRY_WAITS[0] + POST_RETRY_WAITS[1] && waited < POST_RETRY_WAITS[2],
|
|
"{waited:?}"
|
|
);
|
|
}
|
|
|
|
/// A secret store that refuses every connection until `up`.
|
|
struct Outage {
|
|
up: tokio::time::Instant,
|
|
refusals: Mutex<Vec<anyhow::Error>>,
|
|
tries: AtomicUsize,
|
|
posted: Posted,
|
|
}
|
|
|
|
impl ReviewPoster for Outage {
|
|
async fn post(
|
|
&self,
|
|
owner: &str,
|
|
repo: &str,
|
|
number: i64,
|
|
commit: Option<&str>,
|
|
review: Review,
|
|
) -> anyhow::Result<Option<String>> {
|
|
self.tries.fetch_add(1, Ordering::SeqCst);
|
|
if tokio::time::Instant::now() < self.up {
|
|
return Err(self
|
|
.refusals
|
|
.lock()
|
|
.expect("unpoisoned")
|
|
.pop()
|
|
.expect("a refusal per try"));
|
|
}
|
|
self.posted.post(owner, repo, number, commit, review).await
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_post_outlasts_a_secret_store_restart() {
|
|
let mut refusals = Vec::new();
|
|
for _ in 0..=POST_RETRY_WAITS.len() {
|
|
refusals.push(store_refused().await);
|
|
}
|
|
tokio::time::pause();
|
|
let approve = Reply::Done {
|
|
rows: vec![row(None, true, true)],
|
|
overall: Overall::Approve,
|
|
};
|
|
let fake = Fake::new(approve, vec![rules_json()]);
|
|
let started = tokio::time::Instant::now();
|
|
// swarm-bao was down 3m19s on 2026-10-14.
|
|
let down = Duration::from_secs(200);
|
|
let poster = Outage {
|
|
up: started + down,
|
|
refusals: Mutex::new(refusals),
|
|
tries: AtomicUsize::new(0),
|
|
posted: Posted::default(),
|
|
};
|
|
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");
|
|
assert!(poster.tries.load(Ordering::SeqCst) > 1);
|
|
assert!(started.elapsed() >= down, "{:?}", started.elapsed());
|
|
}
|
|
|
|
/// A forge whose first post never answers.
|
|
#[derive(Default)]
|
|
struct HangsOnce {
|
|
tries: AtomicUsize,
|
|
posted: Posted,
|
|
}
|
|
|
|
impl ReviewPoster for HangsOnce {
|
|
async fn post(
|
|
&self,
|
|
owner: &str,
|
|
repo: &str,
|
|
number: i64,
|
|
commit: Option<&str>,
|
|
review: Review,
|
|
) -> anyhow::Result<Option<String>> {
|
|
if self.tries.fetch_add(1, Ordering::SeqCst) == 0 {
|
|
std::future::pending::<()>().await;
|
|
}
|
|
self.posted.post(owner, repo, number, commit, review).await
|
|
}
|
|
}
|
|
|
|
#[tokio::test(start_paused = true)]
|
|
async fn a_post_with_no_answer_is_abandoned_at_its_deadline_and_tried_again() {
|
|
let approve = Reply::Done {
|
|
rows: vec![row(None, true, true)],
|
|
overall: Overall::Approve,
|
|
};
|
|
let fake = Fake::new(approve, vec![rules_json()]);
|
|
let poster = HangsOnce::default();
|
|
let started = tokio::time::Instant::now();
|
|
let run = tokio::time::timeout(
|
|
Duration::from_hours(1),
|
|
run_read(&fake, &fake, &poster, &request("r")),
|
|
)
|
|
.await
|
|
.expect("the hung try is abandoned");
|
|
assert_eq!(run.outcome, RunOutcome::Approved, "{:?}", run.detail);
|
|
assert_eq!(poster.tries.load(Ordering::SeqCst), 2);
|
|
assert_eq!(poster.posted.0.lock().expect("unpoisoned").len(), 1);
|
|
assert!(started.elapsed() >= forge::FORGE_DEADLINE + POST_RETRY_WAITS[0]);
|
|
}
|
|
|
|
#[tokio::test(start_paused = true)]
|
|
async fn a_post_the_forge_refused_is_not_tried_again() {
|
|
let approve = Reply::Done {
|
|
rows: vec![row(None, true, true)],
|
|
overall: Overall::Approve,
|
|
};
|
|
let fake = Fake::new(approve, vec![rules_json()]);
|
|
let poster = Flaky::new(vec![
|
|
ForgejoError::UnexpectedStatusCode(StatusCode::FORBIDDEN).into(),
|
|
]);
|
|
let started = tokio::time::Instant::now();
|
|
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());
|
|
assert_eq!(started.elapsed(), Duration::ZERO);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_post_refused_every_try_is_post_failed() {
|
|
let mut failures = Vec::new();
|
|
for _ in 0..=POST_RETRY_WAITS.len() {
|
|
failures.push(store_refused().await);
|
|
}
|
|
tokio::time::pause();
|
|
let approve = Reply::Done {
|
|
rows: vec![row(None, true, true)],
|
|
overall: Overall::Approve,
|
|
};
|
|
let fake = Fake::new(approve, vec![rules_json()]);
|
|
let poster = Flaky::new(failures);
|
|
let run = run_read(&fake, &fake, &poster, &request("r")).await;
|
|
assert_eq!(run.outcome, RunOutcome::PostFailed);
|
|
assert_eq!(
|
|
poster.tries.load(Ordering::SeqCst),
|
|
POST_RETRY_WAITS.len() + 1
|
|
);
|
|
assert!(poster.posted.0.lock().expect("unpoisoned").is_empty());
|
|
assert!(
|
|
run.detail.as_deref().is_some_and(|d| d
|
|
.starts_with("posting the review failed: logging in to the swarm secret store")),
|
|
"{:?}",
|
|
run.detail
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_failed_record_write_still_posts_the_review() {
|
|
let rig = Rig::with_doc(Arc::new(FakeDoc::failing()));
|
|
let task = rig.start_at("abc123");
|
|
rig.gate.open(1);
|
|
task.await.expect("ran to the end");
|
|
let all = rig.posted();
|
|
assert_eq!(all.len(), 1);
|
|
assert_eq!(all[0].2.event, "APPROVED");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_review_runs_on_the_stored_runtime_model_and_effort() {
|
|
use swarm_wire::agent_inference::Effort;
|
|
|
|
let mut fake = Fake::new(
|
|
Reply::Done {
|
|
rows: vec![row(None, true, true)],
|
|
overall: Overall::Approve,
|
|
},
|
|
vec![rules_json()],
|
|
);
|
|
let acp = swarm_wire::review_classifier::Acp {
|
|
base_url: "https://llm.example/v1".to_owned(),
|
|
model: "qwen".to_owned(),
|
|
prices: swarm_wire::review_classifier::Prices {
|
|
input: 1.0,
|
|
output: 2.0,
|
|
cache_read: None,
|
|
cache_write: None,
|
|
},
|
|
};
|
|
fake.settings = Settings {
|
|
model: "sonnet".to_owned(),
|
|
effort: Some(Effort::Medium),
|
|
max_concurrent_reviews: 2,
|
|
runtime: super::super::settings::RuntimeKind::Acp,
|
|
acp: Some(acp.clone()),
|
|
};
|
|
let poster = Posted::default();
|
|
run_read(&fake, &fake, &poster, &request("r")).await;
|
|
let sent = fake.sent.lock().expect("unpoisoned");
|
|
assert_eq!(sent.len(), 1);
|
|
assert_eq!(
|
|
(sent[0].model.as_str(), sent[0].effort),
|
|
("sonnet", Some(Effort::Medium))
|
|
);
|
|
assert_eq!(
|
|
sent[0].runtime,
|
|
swarm_wire::review_classifier::Runtime::Acp(acp)
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn unreadable_settings_run_nothing_and_post_a_comment() {
|
|
let fake = Fake::new(
|
|
Reply::Done {
|
|
rows: vec![row(None, true, true)],
|
|
overall: Overall::Approve,
|
|
},
|
|
vec![rules_json()],
|
|
);
|
|
let poster = Posted::default();
|
|
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!(
|
|
review
|
|
.body
|
|
.contains("reading the classifier's settings failed: no swarm queue is configured"),
|
|
"{}",
|
|
review.body
|
|
);
|
|
assert_eq!(commit.as_deref(), Some("abc123"));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_clean_run_posts_its_verdict_on_the_head_commit() {
|
|
let (target, commit, review) = posted(
|
|
Reply::Done {
|
|
rows: vec![row(None, true, true)],
|
|
overall: Overall::Approve,
|
|
},
|
|
"r",
|
|
)
|
|
.await;
|
|
assert_eq!(target, "o/r#7");
|
|
assert_eq!(commit.as_deref(), Some("abc123"));
|
|
assert_eq!(review.event, "APPROVED");
|
|
assert!(
|
|
review.body.ends_with("**Overall: ✅ approve**\n"),
|
|
"{}",
|
|
review.body
|
|
);
|
|
|
|
let (_, _, review) = posted(
|
|
Reply::Done {
|
|
rows: vec![row(None, true, false)],
|
|
overall: Overall::RequestChanges,
|
|
},
|
|
"r",
|
|
)
|
|
.await;
|
|
assert_eq!(review.event, "REQUEST_CHANGES");
|
|
assert!(
|
|
review.body.contains("| ❌ match | datetime | 90 / 80 |"),
|
|
"{}",
|
|
review.body
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn no_rules_a_failed_run_or_no_pr_is_a_comment() {
|
|
let approve = || Reply::Done {
|
|
rows: vec![row(None, true, true)],
|
|
overall: Overall::Approve,
|
|
};
|
|
// `o/elsewhere` has no PR 7.
|
|
for (reply, repo, says, commit) in [
|
|
(
|
|
Reply::Failed {
|
|
error: "no API key".to_owned(),
|
|
},
|
|
"r",
|
|
"the run failed: no API key",
|
|
Some("abc123"),
|
|
),
|
|
(approve(), "elsewhere", "reading the PR failed", None),
|
|
] {
|
|
let (_, posted_commit, review) = posted(reply, repo).await;
|
|
assert_eq!(review.event, "COMMENT", "{repo}");
|
|
assert!(review.body.contains(says), "{repo}: {}", review.body);
|
|
assert_eq!(posted_commit.as_deref(), commit, "{repo}");
|
|
}
|
|
|
|
let [yml, _] = crate::review_classifier::RULES_PATHS;
|
|
let fake = Fake::new(approve(), vec![rules_json(), (yml, String::new())]);
|
|
let poster = Posted::default();
|
|
run_read(&fake, &fake, &poster, &request("r")).await;
|
|
let (_, _, review) = poster.0.into_inner().expect("unpoisoned").remove(0);
|
|
assert_eq!(review.event, "COMMENT");
|
|
assert!(
|
|
review.body.contains("the rules did not load") && review.body.contains("holds both"),
|
|
"{}",
|
|
review.body
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_repo_without_rules_is_approved_without_a_model_call() {
|
|
let fake = Fake::new(
|
|
Reply::Failed {
|
|
error: "must not run".to_owned(),
|
|
},
|
|
Vec::new(),
|
|
);
|
|
let poster = Posted::default();
|
|
let run = run_read(&fake, &fake, &poster, &request("r")).await;
|
|
assert_eq!(run.outcome, RunOutcome::Approved);
|
|
assert_eq!(run.detail, None);
|
|
assert_eq!(run.head_sha.as_deref(), Some("abc123"));
|
|
assert!(fake.sent.lock().expect("unpoisoned").is_empty());
|
|
let reviews = poster.0.into_inner().expect("unpoisoned");
|
|
assert_eq!(reviews.len(), 1);
|
|
let (pr, commit, review) = &reviews[0];
|
|
assert_eq!(pr, "o/r#7");
|
|
assert_eq!(commit.as_deref(), Some("abc123"));
|
|
assert_eq!(review.event, "APPROVED");
|
|
assert_eq!(
|
|
review.body,
|
|
"no review rules are configured for o/r; approved without checks\n"
|
|
);
|
|
|
|
// A request for the same head approves again, still without a run.
|
|
let poster = Posted::default();
|
|
let again = run_read(&fake, &fake, &poster, &request("r")).await;
|
|
assert_eq!(again.outcome, RunOutcome::Approved);
|
|
assert!(fake.sent.lock().expect("unpoisoned").is_empty());
|
|
assert_eq!(poster.0.into_inner().expect("unpoisoned").len(), 1);
|
|
|
|
// Control: a repo with rules still runs the classifier and posts its verdict.
|
|
let approve = Reply::Done {
|
|
rows: vec![row(None, true, true)],
|
|
overall: Overall::Approve,
|
|
};
|
|
let with = Fake::new(approve, vec![rules_json()]);
|
|
let poster = Posted::default();
|
|
let run = run_read(&with, &with, &poster, &request("r")).await;
|
|
assert_eq!(run.outcome, RunOutcome::Approved);
|
|
assert_eq!(with.sent.lock().expect("unpoisoned").len(), 1);
|
|
let (_, _, review) = poster.0.into_inner().expect("unpoisoned").remove(0);
|
|
assert_eq!(review.event, "APPROVED");
|
|
assert!(
|
|
review.body.contains("Overall: ✅ approve"),
|
|
"{}",
|
|
review.body
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn a_hostile_reason_mentions_nobody_and_stays_in_its_cell() {
|
|
let mut r = row(None, true, true);
|
|
r.reason = "ok | @mara please merge\n| x |  <b>`".to_owned();
|
|
let body = table(&[r]);
|
|
let line = body.lines().nth(2).expect("a data row");
|
|
assert_eq!(body.lines().count(), 3, "{body}");
|
|
assert!(!line.contains("@mara"), "{line}");
|
|
assert!(line.contains("@\u{200b}mara"), "{line}");
|
|
// Four cells: five unescaped pipes.
|
|
let unescaped = line
|
|
.char_indices()
|
|
.filter(|(i, c)| *c == '|' && (*i == 0 || line.as_bytes()[i - 1] != b'\\'))
|
|
.count();
|
|
assert_eq!(unescaped, 5, "{line}");
|
|
assert!(
|
|
line.contains("!\u{200b}\\[\\]") && line.contains("\\<b\\>"),
|
|
"{line}"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn a_reason_carries_no_live_math_delimiter() {
|
|
let mut r = row(None, true, true);
|
|
r.reason =
|
|
"uses https://${hyperhiveCfg.swarm.forge.domain}/metrics and $$x$$ | y".to_owned();
|
|
let body = table(&[r]);
|
|
let line = body.lines().nth(2).expect("a data row");
|
|
assert!(
|
|
line.contains("\\${hyperhiveCfg") && line.contains("\\$\\$x\\$\\$"),
|
|
"{line}"
|
|
);
|
|
let bare = line
|
|
.char_indices()
|
|
.filter(|(i, c)| *c == '$' && (*i == 0 || line.as_bytes()[i - 1] != b'\\'))
|
|
.count();
|
|
assert_eq!(bare, 0, "{line}");
|
|
let unescaped = line
|
|
.char_indices()
|
|
.filter(|(i, c)| *c == '|' && (*i == 0 || line.as_bytes()[i - 1] != b'\\'))
|
|
.count();
|
|
assert_eq!(unescaped, 5, "{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 a_plain_reason_reads_without_backslashes() {
|
|
assert_eq!(
|
|
cell("an added #[allow(clippy::foo)] in ci_metrics.rs (see main_loop_v2.rs)"),
|
|
"an added #\u{200b}\\[allow(clippy::foo)\\] in ci_metrics.rs (see main_loop_v2.rs)"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn an_underscore_escapes_unless_it_sits_between_alphanumerics() {
|
|
// Presence control: the edge underscores are escaped.
|
|
assert_eq!(
|
|
cell("_em_ __b__ a_ _b a__b snake_case"),
|
|
"\\_em\\_ \\_\\_b\\_\\_ a\\_ \\_b a\\_\\_b snake_case"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn a_reason_with_markup_characters_keeps_them_inert() {
|
|
assert_eq!(
|
|
cell(r"a\b `c` *d* ~e~ [f] <g> &h; i|j $k"),
|
|
r"a\\b \`c\` \*d\* \~e\~ \[f\] \<g\> \&h; i\|j \$k"
|
|
);
|
|
}
|
|
|
|
#[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" },
|
|
})
|
|
.to_string()
|
|
};
|
|
assert_eq!(
|
|
Requested::parse(payload("review_requested", Some(LOGIN)).as_bytes()),
|
|
Some(request("r"))
|
|
);
|
|
assert_eq!(
|
|
Requested::parse(payload("review_requested", Some("Review-Classifier")).as_bytes()),
|
|
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")),
|
|
payload("review_requested", None),
|
|
"{}".to_owned(),
|
|
] {
|
|
assert_eq!(Requested::parse(body.as_bytes()), None, "{body}");
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_run_on_the_forge_diff_says_so_in_its_review() {
|
|
/// The body a run posts when the PR read was narrowed for `reason`,
|
|
/// with the rules on the default branch or without them.
|
|
async fn narrowed(reason: Option<String>, rules: bool) -> String {
|
|
let reply = Reply::Done {
|
|
rows: vec![row(None, true, true)],
|
|
overall: Overall::Approve,
|
|
};
|
|
let files = if rules {
|
|
vec![rules_json()]
|
|
} else {
|
|
Vec::new()
|
|
};
|
|
let mut fake = Fake::new(reply, files);
|
|
fake.narrowed = reason;
|
|
let poster = Posted::default();
|
|
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());
|
|
let body = narrowed(reason(), true).await;
|
|
assert!(
|
|
body.ends_with(
|
|
"\nThe classifier read the forge's diff, with 3 lines of context, not 25: \
|
|
the PR changes 101 files.\n"
|
|
),
|
|
"{body}"
|
|
);
|
|
let body = narrowed(None, true).await;
|
|
assert!(!body.contains("forge's diff"), "{body}");
|
|
let body = narrowed(reason(), false).await;
|
|
assert!(body.starts_with("no review rules are configured"), "{body}");
|
|
assert!(
|
|
!body.contains("forge's diff"),
|
|
"a run that never ran read no diff: {body}"
|
|
);
|
|
}
|
|
}
|