diff --git a/Cargo.lock b/Cargo.lock index 85d2b862..adf97684 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4853,6 +4853,7 @@ dependencies = [ "tokio", "tower", "tracing", + "tracing-subscriber", "url", "utoipa", "utoipa-axum", diff --git a/docs/swarm/review-classifier.md b/docs/swarm/review-classifier.md index e6be7dcb..4e8b86e9 100644 --- a/docs/swarm/review-classifier.md +++ b/docs/swarm/review-classifier.md @@ -294,8 +294,8 @@ The page posts the run to `swarm-controller` (`POST and sends it to the classifier as one queue request on `$SWARM.review-classifier.try-out`. The page then reads the result back (`GET /api/review-classifier/try-out/{id}`): a run can take longer than a request -stays open. The controller keeps the result in memory for an hour; there is no -run store. +stays open. The controller keeps the result in memory for an hour, and the +finished run goes into the [run store](#run-store-and-retention). `GET /api/review-classifier/rules?repo=owner/name` returns the repo's own rules, for the page to load into its editor; a run posts them like typed @@ -332,6 +332,60 @@ adds the classification to the repo's config, with you, the swarm UI user, as commit author. The page then links you to the forge's new-PR page for that branch, where you log in as usual and open the PR yourself. +## Run store and retention + +`swarm-controller` records every run when it ends: a try-out when the +classifier's reply arrives, a requested review once the controller has posted +its review or the post has failed, and a request turned away because the +queue was full once its comment is on the forge or the post has failed. A review request for a head that's already +running or waiting starts nothing and isn't recorded. A run still going isn't +in the list, and a controller restart drops a run in progress. + +The newest 100 runs are one JSON array under the key `recent` in the +`review-classifier-runs` KV bucket on the swarm queue, which only the +controller reads or writes. The controller creates the bucket on first use and +adds each run with a compare-and-set write, so the list survives a controller +restart or a move to another host. Recording the 101st run drops the oldest; +there's no time-based retention. The controller logs a failed write, and it +changes nothing on the forge. + +`GET /api/review-classifier/runs` returns the list, newest first. Like the +try-out GET, the gateway's `admins` rule gates it, and the controller checks no +header. Each run has: + +- `id`: a try-out's is the id its POST answered with +- `trigger`: `try-out` or `review` +- `repo` (`owner/name`), `pr`, and `requester`: the swarm UI user who started + the try-out, or the forge user who requested the review +- `head_sha`: the commit the run read, absent when the PR wasn't read, as for + an unreadable PR or a full queue +- `started_unix` and `finished_unix`: a try-out starts at its POST, a review + once it has a run place +- `outcome`: `approved`, `changes-requested`, `no-verdict`, `queue-full` or + `post-failed` for a review; `done` or `failed` for a try-out +- `detail`: the reason, for every outcome but `approved`, + `changes-requested` and `done`. For `post-failed` it's the run's own reason, + when it has one, then why the post failed. +- `rows` and `overall`, as the try-out GET returns them, when the classifier + answered +- `review_url`: the posted review, when the forge names its URL + +A run holds no diff, no commit text and no token count or cost. The route +answers 503 when the controller has no swarm queue or can't reach the bucket, +and 500 when it can't read the list, both as problem+json. + +Each recorded run is also one `info` event from `swarm-controller` with the +message `review classifier run finished`. The event carries the same fields, +with `rows` as JSON. It reaches VictoriaLogs through the host journal, where +each field arrives as `F_` and the name in capitals (`F_OUTCOME`, `F_REPO`), so +a run whose write failed is still there: + +```console +$ swarm-logs query '_time:7d _SYSTEMD_UNIT:="swarm-controller.service" _msg:="review classifier run finished" | fields _time, F_TRIGGER, F_REPO, F_PR, F_OUTCOME, F_DETAIL, F_REVIEW_URL' --format json +``` + +Add a filter such as `F_OUTCOME:="post-failed"` before the `|` to narrow it. + ## Observability and cost The swarm UI shows no token counts, cache counts or cost, on any page. On the @@ -357,8 +411,6 @@ These get settled, and filled in here, as the work reaches them: - **Cache strategy.** Fully independent sessions per classification, or one warm session and then the rest in parallel. To measure through the cache token counts before choosing. -- **Run store and retention.** Where run records live, how long they stay, - and how the recent-runs page reads them. - **Oversized diffs.** What happens when the diff doesn't fit the model's context. Lockfiles already collapse to one line each, as in `hive-forge diff`. - **Budget.** Any per-run or per-day spend limit on top of the concurrency diff --git a/swarm-controller/Cargo.toml b/swarm-controller/Cargo.toml index 82d53077..e99d1243 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -138,6 +138,8 @@ x509-cert.workspace = true tokio = { workspace = true, features = ["test-util"] } # `ServiceExt::oneshot`, to drive a route through the real router in tests. tower = { version = "0.5", features = ["util"] } +# A layer that captures `review_classifier::runs`'s log event in a test. +tracing-subscriber.workspace = true [lints] workspace = true diff --git a/swarm-controller/README.md b/swarm-controller/README.md index 4c7e12d4..6af0b1f3 100644 --- a/swarm-controller/README.md +++ b/swarm-controller/README.md @@ -57,6 +57,12 @@ The swarm's control plane. The swarm UI and `swarmctl` are its clients. (`GET`/`PUT`/`DELETE /api/agents/{name}/limits`); the hive running the agent reads it, and each cap replaces the hive's own answer for that field → [`docs/scheduler/coordinator.md`](../docs/scheduler/coordinator.md#container-resource-limits). +- **Review classifier** — try-outs from the swarm UI and reviews on review + requests, and the newest 100 finished runs, one document in the + `review-classifier-runs` KV bucket, which this daemon alone writes + (`POST /api/review-classifier/try-out`, `GET /api/review-classifier/try-out/{id}`, + `GET /api/review-classifier/rules`, `GET /api/review-classifier/runs`) → + [`docs/swarm/review-classifier.md`](../docs/swarm/review-classifier.md). - **Relays** — each agent's terminal and turn-state header as SSE, agent icons, the cross-repo issue report, and the UI's quick links. diff --git a/swarm-controller/src/agent_limits.rs b/swarm-controller/src/agent_limits.rs index 6e2fe84b..cad865e0 100644 --- a/swarm-controller/src/agent_limits.rs +++ b/swarm-controller/src/agent_limits.rs @@ -459,6 +459,7 @@ mod tests { accounts: crate::linked_accounts::Accounts::Fake(Arc::default()), reviews: Arc::default(), try_outs: crate::review_classifier::TryOuts::default(), + runs: crate::review_classifier::runs::Runs::Live(None), } } diff --git a/swarm-controller/src/forge/classifier_account.rs b/swarm-controller/src/forge/classifier_account.rs index 38999014..79813ae7 100644 --- a/swarm-controller/src/forge/classifier_account.rs +++ b/swarm-controller/src/forge/classifier_account.rs @@ -109,7 +109,7 @@ impl Client { /// Post `review` on PR `number` of `owner/repo` as [`LOGIN`], on /// `commit` when one is given. Reads the token from the store on every - /// call. + /// call. The posted review's URL, when the forge names one. /// /// # Errors /// When the store holds no token yet, or the forge refuses the review. @@ -120,7 +120,7 @@ impl Client { number: i64, commit: Option<&str>, review: Review, - ) -> Result<()> { + ) -> Result> { let path = forge_token_path()?; let store = crate::store::connect() .await @@ -146,10 +146,8 @@ impl Client { }, ) .await - .with_context(|| { - format!("posting {LOGIN}'s review on PR {number} of {owner}/{repo}") - })?; - Ok(()) + .with_context(|| format!("posting {LOGIN}'s review on PR {number} of {owner}/{repo}")) + .map(|posted| posted.html_url.map(String::from)) } /// The stored token at `path` and every token the forge lists for diff --git a/swarm-controller/src/forge_account.rs b/swarm-controller/src/forge_account.rs index 06596ac3..f639fa76 100644 --- a/swarm-controller/src/forge_account.rs +++ b/swarm-controller/src/forge_account.rs @@ -166,6 +166,7 @@ mod tests { accounts: Accounts::Fake(Arc::clone(store)), try_outs: super::super::review_classifier::TryOuts::default(), reviews: std::sync::Arc::default(), + runs: crate::review_classifier::runs::Runs::Live(None), } } diff --git a/swarm-controller/src/github_account.rs b/swarm-controller/src/github_account.rs index 7d55b5ac..79f98e7d 100644 --- a/swarm-controller/src/github_account.rs +++ b/swarm-controller/src/github_account.rs @@ -136,6 +136,7 @@ mod tests { accounts: Accounts::Fake(Arc::clone(store)), try_outs: super::super::review_classifier::TryOuts::default(), reviews: std::sync::Arc::default(), + runs: crate::review_classifier::runs::Runs::Live(None), } } diff --git a/swarm-controller/src/linked_accounts.rs b/swarm-controller/src/linked_accounts.rs index 30c39c39..78cd4f08 100644 --- a/swarm-controller/src/linked_accounts.rs +++ b/swarm-controller/src/linked_accounts.rs @@ -971,6 +971,7 @@ pub(crate) mod tests { accounts: super::Accounts::Fake(Arc::clone(store)), try_outs: super::super::review_classifier::TryOuts::default(), reviews: std::sync::Arc::default(), + runs: crate::review_classifier::runs::Runs::Live(None), } } diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index cde67d31..3e823c96 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -898,6 +898,8 @@ struct AppState { try_outs: review_classifier::TryOuts, /// The requested reviews running or waiting. reviews: Arc, + /// The review classifier's finished runs. + runs: review_classifier::runs::Runs, } /// Env var the controller's NixOS module sets from @@ -3253,6 +3255,7 @@ async fn main() -> Result<()> { // what `policy::render_agent` grants today reaches them only from here. read_policy::ensure_agent_policies().await; + let runs = review_classifier::runs::Runs::Live(status.as_ref().map(|s| s.queue_client())); let state = AppState { hives: Arc::new(hives), links: Arc::new(load_links()), @@ -3271,6 +3274,7 @@ async fn main() -> Result<()> { accounts: linked_accounts::Accounts::Live, try_outs: review_classifier::TryOuts::default(), reviews: Arc::default(), + runs, }; let app = build_app(state); @@ -3346,6 +3350,7 @@ fn build_app(state: AppState) -> axum::Router { .routes(routes!(review_classifier::post_try_out)) .routes(routes!(review_classifier::get_try_out)) .routes(routes!(review_classifier::get_rules)) + .routes(routes!(review_classifier::runs::get_runs)) .routes(routes!(webhook::post_webhook_forge)) .split_for_parts(); // Just the JSON, not the UI — Swagger UI itself is nginx-hosted from @@ -3467,6 +3472,7 @@ mod tests { accounts: super::linked_accounts::Accounts::Fake(std::sync::Arc::default()), try_outs: super::review_classifier::TryOuts::default(), reviews: std::sync::Arc::default(), + runs: super::review_classifier::runs::Runs::Live(None), }; (state, sched) } diff --git a/swarm-controller/src/matrix_account.rs b/swarm-controller/src/matrix_account.rs index 8dee8714..19258645 100644 --- a/swarm-controller/src/matrix_account.rs +++ b/swarm-controller/src/matrix_account.rs @@ -580,6 +580,7 @@ mod tests { accounts: Accounts::Fake(Arc::clone(store)), try_outs: super::super::review_classifier::TryOuts::default(), reviews: std::sync::Arc::default(), + runs: crate::review_classifier::runs::Runs::Live(None), } } diff --git a/swarm-controller/src/review_classifier.rs b/swarm-controller/src/review_classifier.rs index 006784d9..5a308741 100644 --- a/swarm-controller/src/review_classifier.rs +++ b/swarm-controller/src/review_classifier.rs @@ -8,7 +8,8 @@ //! the review classifier on [`SUBJECT`] and answers 202 with an id at once: //! one model call over a big diff can outlast the 60 seconds the gateway //! holds a request open. The reply is kept in memory, where the GET reads it -//! until [`RESULT_TTL`] after it arrived. Nothing reaches the forge. +//! until [`RESULT_TTL`] after it arrived, and recorded in [`runs`]. Nothing +//! reaches the forge. //! //! The rules are read with the same token on every request, with no cache. //! @@ -38,6 +39,7 @@ use crate::forge::{self, DefaultBranchFiles, PullForReview}; use crate::{AppState, error_problem}; pub(crate) mod review; +pub(crate) mod runs; /// How long a finished run's result stays readable. pub(crate) const RESULT_TTL: Duration = Duration::from_hours(1); @@ -342,6 +344,7 @@ pub(crate) async fn post_try_out( &state.try_outs, &*fake, (*fake).clone(), + state.runs.clone(), operator, run, )) @@ -366,7 +369,16 @@ pub(crate) async fn post_try_out( &swarm_queue_client::chain(&e), ) })?; - Box::pin(start(&state.try_outs, forge.as_ref(), queue, operator, run)).await + let runs = state.runs.clone(); + Box::pin(start( + &state.try_outs, + forge.as_ref(), + queue, + runs, + operator, + run, + )) + .await } /// Read a run's result. @@ -606,14 +618,16 @@ fn operator(headers: &HeaderMap) -> Option { } /// Read the PR, record the run and send it; the reply settles the run when -/// it arrives. +/// it arrives, and the settled run goes into `runs`. async fn start( try_outs: &TryOuts, pulls: &impl PullSource, sink: impl RunSink, + runs: runs::Runs, operator: String, run: Run, ) -> Result<(StatusCode, Json), ProblemDetails> { + let started_unix = crate::agent_renewal::unix_now(); let repo = format!("{}/{}", run.owner, run.name); let pull = forge::within_forge_deadline("review try-out", pulls.pull(&run.owner, &run.name, run.pr)) @@ -641,11 +655,38 @@ async fn start( ) })?; let id = mint_id().map_err(|e| error_problem(StatusCode::INTERNAL_SERVER_ERROR, &e))?; - try_outs.insert(id.clone(), pull.head_sha); - let (runs, settle_id) = (try_outs.clone(), id.clone()); + try_outs.insert(id.clone(), pull.head_sha.clone()); + let try_outs = try_outs.clone(); + let (settle_id, repo, pr, requester) = ( + id.clone(), + request.repo.clone(), + request.pr, + request.operator.clone(), + ); tokio::spawn(async move { let reply = sink.run(payload).await; - runs.settle(&settle_id, reply); + let (rows, overall) = runs::answer(&reply); + let (outcome, detail) = match &reply { + Reply::Done { .. } => (runs::RunOutcome::Done, None), + Reply::Failed { error } => (runs::RunOutcome::Failed, Some(error.clone())), + }; + try_outs.settle(&settle_id, reply); + let record = runs::RunRecord { + id: settle_id, + trigger: runs::Trigger::TryOut, + repo, + pr, + requester, + head_sha: Some(pull.head_sha), + started_unix, + finished_unix: crate::agent_renewal::unix_now(), + outcome, + detail, + rows, + overall, + review_url: None, + }; + runs.record(record).await; }); tracing::info!(%id, repo = %request.repo, pr = request.pr, "review try-out started"); Ok((StatusCode::ACCEPTED, Json(TryOutAccepted { id }))) @@ -877,6 +918,48 @@ pub(crate) mod tests { ); } + #[tokio::test] + async fn a_settled_try_out_is_recorded_as_done_or_failed() { + let failed = Reply::Failed { + error: "no API key".to_owned(), + }; + for (reply, outcome, detail) in [ + (done(), runs::RunOutcome::Done, None), + (failed, runs::RunOutcome::Failed, Some("no API key")), + ] { + let doc = Arc::new(runs::tests::FakeDoc::default()); + let (mut state, _) = crate::tests::state_with_roster(); + state.try_outs.fake = Some(Arc::new(Fake { + reply: reply.clone(), + files: Vec::new(), + })); + state.runs = runs::Runs::Fake(Arc::clone(&doc)); + let app = crate::build_app(state); + let (_, accepted) = post(&app, body("o/r", 7)).await; + let id = accepted["id"].as_str().expect("an id"); + settled(&app, id).await; + let mut recorded = Vec::new(); + for _ in 0..100 { + recorded = doc.runs(); + if !recorded.is_empty() { + break; + } + tokio::task::yield_now().await; + } + assert_eq!(recorded.len(), 1, "{outcome:?}"); + let run = &recorded[0]; + assert_eq!(run.id, id); + assert_eq!(run.trigger, runs::Trigger::TryOut); + assert_eq!((run.repo.as_str(), run.pr), ("o/r", 7)); + assert_eq!(run.requester, "alice"); + assert_eq!(run.head_sha.as_deref(), Some("abc123")); + assert_eq!(run.outcome, outcome); + assert_eq!(run.detail.as_deref(), detail); + assert_eq!((run.rows.clone(), run.overall), runs::answer(&reply)); + assert!(run.started_unix <= run.finished_unix); + } + } + #[tokio::test] async fn a_bad_body_is_400() { let app = app(done()); diff --git a/swarm-controller/src/review_classifier/review.rs b/swarm-controller/src/review_classifier/review.rs index 7530a564..c673ae60 100644 --- a/swarm-controller/src/review_classifier/review.rs +++ b/swarm-controller/src/review_classifier/review.rs @@ -22,8 +22,10 @@ use swarm_queue_client::review_classifier::{Overall, Reply, Row, TryOut, Verdict use tokio::sync::Semaphore; use tokio::task::JoinHandle; +use super::runs::{self, RunOutcome, RunRecord, Runs, Trigger}; use super::{FileSource, PullSource, RunSink, split_repo}; use crate::AppState; +use crate::agent_renewal::unix_now; use crate::forge::{self, classifier_account::LOGIN, classifier_account::Review}; /// The most review runs in flight at once. @@ -215,16 +217,26 @@ pub(crate) fn on_delivery(state: &AppState, body: &[u8]) { return; }; let queue = Queue(state.status.as_ref().map(|s| s.queue_client())); - start(&state.reviews, Arc::clone(&forge), queue, forge, request); + let runs = state.runs.clone(); + start( + &state.reviews, + Arc::clone(&forge), + queue, + forge, + runs, + request, + ); } /// Accept `request` into `reviews` and spawn its run, or the no-verdict -/// comment when there's no place for it. The task, if one was spawned. +/// comment when there's no place for it, and record it in `runs` when it +/// ends. The task, if one was spawned. fn start( reviews: &Arc, forge: Arc, queue: Q, poster: Arc

, + runs: Runs, request: Requested, ) -> Option> where @@ -240,8 +252,9 @@ where let Ok(_slot) = slots.acquire_owned().await else { return; }; - Box::pin(run(&*forge, &*forge, &queue, &*poster, &request)).await; + let run = Box::pin(run(&*forge, &*forge, &queue, &*poster, &request)).await; drop(accepted); + runs.record(run).await; })) } Err(Refused::Duplicate) => { @@ -257,28 +270,34 @@ where "review requested, but every run and wait place is taken; posting a no-verdict comment" ); Some(tokio::spawn(async move { - post(&*poster, &request, None, queue_full_review()).await; + let started_unix = unix_now(); + let ruling = queue_full_ruling(); + let reason = ruling.reason.clone(); + let sent = post(&*poster, &request, None, ruling.review()).await; + let ended = ended(RunOutcome::QueueFull, reason, sent); + runs.record(record(&request, None, started_unix, ended, (None, None))) + .await; })) } } } -/// The review for a request [`Reviews`] had no place for. -fn queue_full_review() -> Review { - Review { - event: no_clean_verdict_event().forge(), - body: no_verdict( - &format!( - "{MAX_CONCURRENT_REVIEWS} reviews are running and {MAX_WAITING_REVIEWS} are \ - waiting, so this request starts no run; request the review again later" - ), - &[], - ), +/// The ruling for a request [`Reviews`] had no place for. +fn queue_full_ruling() -> Ruling { + let reason = format!( + "{MAX_CONCURRENT_REVIEWS} reviews are running and {MAX_WAITING_REVIEWS} are waiting, so \ + this request starts no run; request the review again later" + ); + Ruling { + event: no_clean_verdict_event(), + body: no_verdict(&reason, &[]), + reason: Some(reason), } } /// 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, @@ -286,7 +305,7 @@ pub(crate) trait ReviewPoster { number: i64, commit: Option<&str>, review: Review, - ) -> impl Future> + Send; + ) -> impl Future>> + Send; } impl ReviewPoster for forge::Client { @@ -297,7 +316,7 @@ impl ReviewPoster for forge::Client { number: i64, commit: Option<&str>, review: Review, - ) -> anyhow::Result<()> { + ) -> anyhow::Result> { self.post_classifier_review(owner, repo, number, commit, review) .await } @@ -332,20 +351,46 @@ enum Outcome { Ran { sent: usize, reply: Reply }, } -/// Read the PR and the rules, run them, and post the review. +/// Read the PR and the rules, run them, and post the review. The finished +/// run. async fn run( files: &impl FileSource, pulls: &impl PullSource, sink: &impl RunSink, poster: &impl ReviewPoster, req: &Requested, -) { +) -> RunRecord { + let started_unix = unix_now(); let (head, outcome) = classify(files, pulls, sink, req).await; - post(poster, req, head.as_deref(), review_for(&outcome)).await; + let ruling = review_for(&outcome); + 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(_) => (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. -async fn post(poster: &impl ReviewPoster, req: &Requested, head: Option<&str>, review: Review) { +/// 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, String> { let repo = format!("{}/{}", req.owner, req.name); let event = review.event; let sent = forge::within_forge_deadline( @@ -354,15 +399,79 @@ async fn post(poster: &impl ReviewPoster, req: &Requested, head: Option<&str>, r ) .await; match sent { - Some(Ok(())) => tracing::info!( - %repo, pr = req.pr, head_sha = head.unwrap_or_default(), event, - "review classifier review posted" - ), - Some(Err(e)) => tracing::warn!( - %repo, pr = req.pr, error = %format!("{e:#}"), - "posting the review classifier's review failed" - ), - None => {} + Some(Ok(url)) => { + tracing::info!( + %repo, pr = req.pr, head_sha = head.unwrap_or_default(), event, + "review classifier review posted" + ); + Ok(url) + } + Some(Err(e)) => { + let error = format!("{e:#}"); + tracing::warn!( + %repo, pr = req.pr, %error, + "posting the review classifier's review failed" + ); + Err(format!("posting the review failed: {error}")) + } + None => Err("the forge did not answer the review's post within the deadline".to_owned()), + } +} + +/// How a run ended, as its record holds it. +struct Ended { + outcome: RunOutcome, + detail: Option, + review_url: Option, +} + +/// `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, + posted: Result, 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, + started_unix: i64, + ended: Ended, + (rows, overall): (Option>, Option), +) -> 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, } } @@ -425,17 +534,39 @@ async fn classify( (head, outcome) } -/// The review a run's outcome posts. -fn review_for(outcome: &Outcome) -> Review { - let (event, body) = match outcome { - Outcome::NotRun(reason) => (no_clean_verdict_event(), no_verdict(reason, &[])), +/// 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, +} + +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::NotRun(reason) => Ruling::no_verdict(reason.clone(), &[]), Outcome::Ran { reply: Reply::Failed { error }, .. - } => ( - no_clean_verdict_event(), - no_verdict(&format!("the run failed: {error}"), &[]), - ), + } => Ruling::no_verdict(format!("the run failed: {error}"), &[]), Outcome::Ran { sent, reply: Reply::Done { rows, overall }, @@ -446,25 +577,25 @@ fn review_for(outcome: &Outcome) -> Review { "the classifier answered {} of {sent} classifications", rows.len() ); - (no_clean_verdict_event(), no_verdict(&reason, rows)) + Ruling::no_verdict(reason, rows) } else if unanswered > 0 { let reason = format!( "{unanswered} of {} classifications got no answer", rows.len() ); - (no_clean_verdict_event(), no_verdict(&reason, rows)) + Ruling::no_verdict(reason, rows) } else { let (event, said) = match overall { Overall::Approve => (Event::Approve, "approve"), Overall::RequestChanges => (Event::RequestChanges, "request changes"), }; - (event, format!("{}\n**Overall: {said}.**\n", table(rows))) + Ruling { + event, + body: format!("{}\n**Overall: {said}.**\n", table(rows)), + reason: None, + } } } - }; - Review { - event: event.forge(), - body, } } @@ -560,6 +691,7 @@ mod tests { use swarm_queue_client::review_classifier::Polarity; use super::*; + use crate::review_classifier::runs::tests::FakeDoc; use crate::review_classifier::tests::{Fake, rules_json}; /// Every review [`run`] posted, with the commit it named. @@ -574,13 +706,15 @@ mod tests { number: i64, commit: Option<&str>, review: Review, - ) -> anyhow::Result<()> { + ) -> anyhow::Result> { self.0.lock().expect("unpoisoned").push(( format!("{owner}/{repo}#{number}"), commit.map(str::to_owned), review, )); - Ok(()) + Ok(Some(format!( + "https://forge/{owner}/{repo}/pulls/{number}#review" + ))) } } @@ -694,14 +828,14 @@ mod tests { // A run of no classifications never reaches the classifier; an // empty answer to one still has no verdict. - let review = review_for(&Outcome::Ran { + let ruling = review_for(&Outcome::Ran { sent: 0, reply: Reply::Done { rows: Vec::new(), overall: Overall::Approve, }, }); - assert_eq!(review.event, "COMMENT"); + assert_eq!(ruling.event, Event::Comment); } /// A classifier that answers approve once [`Gate::open`] lets it through. @@ -742,12 +876,15 @@ mod tests { )); let gate = Gate(Arc::new(tokio::sync::Semaphore::new(0))); let poster = Arc::new(Posted::default()); + let doc = Arc::new(FakeDoc::default()); + let store = Runs::Fake(Arc::clone(&doc)); let start_at = |head: &str| { start( &reviews, Arc::clone(&fake), gate.clone(), Arc::clone(&poster), + store.clone(), requested_at(head), ) }; @@ -764,10 +901,16 @@ mod tests { task.await.expect("ran"); } assert_eq!(poster.0.lock().expect("unpoisoned").len(), 3); + assert_eq!( + doc.runs().len(), + 3, + "a repeat that started nothing is no run" + ); // A finished run frees its head. start_at("a").expect("accepted again").await.expect("ran"); assert_eq!(poster.0.lock().expect("unpoisoned").len(), 4); + assert_eq!(doc.runs().len(), 4); } #[tokio::test] @@ -781,12 +924,15 @@ mod tests { )); let gate = Gate(Arc::new(tokio::sync::Semaphore::new(0))); let poster = Arc::new(Posted::default()); + let doc = Arc::new(FakeDoc::default()); + let store = Runs::Fake(Arc::clone(&doc)); let start_at = |head: String| { start( &reviews, Arc::clone(&fake), gate.clone(), Arc::clone(&poster), + store.clone(), requested_at(&head), ) .expect("a run or a comment") @@ -813,6 +959,22 @@ mod tests { review.body ); } + let refused = doc.runs(); + assert_eq!(refused.len(), 1); + assert_eq!(refused[0].outcome, RunOutcome::QueueFull); + assert_eq!(refused[0].head_sha, None); + assert_eq!( + refused[0].review_url.as_deref(), + Some("https://forge/o/r/pulls/7#review") + ); + assert!( + refused[0] + .detail + .as_deref() + .is_some_and(|d| d.starts_with("2 reviews are running and 8 are waiting")), + "{:?}", + refused[0].detail + ); gate.open(); for run in runs { @@ -821,6 +983,187 @@ mod tests { let all = poster.0.lock().expect("unpoisoned"); assert_eq!(all.len(), places + 1); assert!(all[1..].iter().all(|(_, _, r)| r.event == "APPROVED")); + let recorded = doc.runs(); + assert_eq!(recorded.len(), places + 1); + assert!( + recorded[..places] + .iter() + .all(|r| r.outcome == RunOutcome::Approved) + ); + } + + /// 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(&fake, &fake, &fake, poster, &request(repo)).await + } + + #[tokio::test] + async fn each_way_a_run_ends_has_its_outcome() { + let answered = |rows: Vec, 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 fake = Fake::new(failed, Vec::new()); + let run = run(&fake, &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> { + 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") + ); + } + + #[tokio::test] + async fn a_failed_record_write_still_posts_the_review() { + let fake = Arc::new(Fake::new( + Reply::Done { + rows: vec![row(None, true, true)], + overall: Overall::Approve, + }, + vec![rules_json()], + )); + let poster = Arc::new(Posted::default()); + let store = Runs::Fake(Arc::new(FakeDoc::failing())); + start( + &Arc::new(Reviews::default()), + Arc::clone(&fake), + (*fake).clone(), + Arc::clone(&poster), + store, + request("r"), + ) + .expect("accepted") + .await + .expect("ran to the end"); + let all = poster.0.lock().expect("unpoisoned"); + assert_eq!(all.len(), 1); + assert_eq!(all[0].2.event, "APPROVED"); } #[tokio::test] diff --git a/swarm-controller/src/review_classifier/runs.rs b/swarm-controller/src/review_classifier/runs.rs new file mode 100644 index 00000000..d91f2953 --- /dev/null +++ b/swarm-controller/src/review_classifier/runs.rs @@ -0,0 +1,576 @@ +//! `GET /api/review-classifier/runs`: the review classifier's finished runs, +//! try-outs and requested reviews both, newest first. +//! +//! The newest [`MAX_RUNS`] are one JSON array under [`KEY`] in [`BUCKET`], +//! which only this daemon reads or writes. Recording a run is one +//! compare-and-set on that key, as for `schedules::DEFINITIONS_KEY`; one key, +//! because listing a bucket's keys needs a consumer grant this daemon holds +//! for no bucket of its own. +//! +//! Every recorded run is also one `info` event, [`EVENT`], carrying the +//! record's fields, so a run whose write failed is still in the log store. +//! A record holds no diff, no commit text and no token count or cost. + +use std::future::Future; + +use anyhow::Context as _; +use async_nats::jetstream::kv::Store; +use axum::Json; +use axum::extract::State; +use axum::http::StatusCode; +use problem_details::ProblemDetails; +use serde::{Deserialize, Serialize}; +use swarm_queue_client::review_classifier::{Overall, Reply, Row}; +use utoipa::ToSchema; + +use crate::schedules::{read, write}; +use crate::{AppState, error_problem}; + +/// The bucket holding the runs. Agents hold no grant on it. +pub const BUCKET: &str = "review-classifier-runs"; + +/// The one key in [`BUCKET`]. +pub const KEY: &str = "recent"; + +/// How many runs the list keeps; recording one more drops the oldest. +pub(crate) const MAX_RUNS: usize = 100; + +/// How many times a losing writer re-reads and re-applies, as in +/// `schedules::MAX_ATTEMPTS`. +const MAX_ATTEMPTS: usize = 5; + +/// The message of the event [`Runs::record`] logs for every run. +pub(crate) const EVENT: &str = "review classifier run finished"; + +/// What started a run. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)] +#[serde(rename_all = "kebab-case")] +pub(crate) enum Trigger { + /// The swarm UI's try-out page. + TryOut, + /// A review request on the forge. + Review, +} + +/// How a run ended. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)] +#[serde(rename_all = "kebab-case")] +pub(crate) enum RunOutcome { + /// A review approved the PR. + Approved, + /// A review requested changes. + ChangesRequested, + /// A comment was posted: the rules did not load, the PR was unreadable, + /// the run failed, the classifier sent a different number of answers + /// than it was sent classifications, or any classification got none. + NoVerdict, + /// No run started, as every run and wait place was taken; a comment + /// saying so was posted. + QueueFull, + /// The review could not be posted. + PostFailed, + /// A try-out the classifier answered. + Done, + /// A try-out that failed. + Failed, +} + +/// One finished run. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)] +pub(crate) struct RunRecord { + /// A try-out's id is the one its POST answered with. + pub(super) id: String, + pub(super) trigger: Trigger, + /// `owner/name`. + pub(super) repo: String, + pub(super) pr: u64, + /// The swarm UI user who started a try-out, or the forge user who + /// requested the review. + pub(super) requester: String, + /// The commit the run read. Absent when the PR was not read. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(super) head_sha: Option, + /// Unix seconds: a try-out's POST, or a review's start once it had a + /// run place. + pub(super) started_unix: i64, + pub(super) finished_unix: i64, + pub(super) outcome: RunOutcome, + /// Why, for every outcome but `approved`, `changes-requested` and + /// `done`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(super) detail: Option, + /// As the try-out GET returns them, when the classifier answered. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(super) rows: Option>, + /// As the try-out GET returns it, when the classifier answered. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(super) overall: Option, + /// The posted review, when one was posted and the forge named its URL. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(super) review_url: Option, +} + +/// The rows and the overall verdict of `reply`, when the classifier +/// answered. +pub(super) fn answer(reply: &Reply) -> (Option>, Option) { + match reply { + Reply::Done { rows, overall } => (Some(rows.clone()), Some(*overall)), + Reply::Failed { .. } => (None, None), + } +} + +/// Where the runs are kept: [`BUCKET`] in production, a fake in the tests. +pub(crate) trait RunDoc { + /// The runs, with the revision to write them back against. + fn read(&self) -> impl Future, Option)>> + Send; + /// `Ok(false)` when another write landed since `revision`. + fn write( + &self, + runs: Vec, + revision: Option, + ) -> impl Future> + Send; +} + +impl RunDoc for Store { + async fn read(&self) -> anyhow::Result<(Vec, Option)> { + read(self, KEY).await + } + + async fn write(&self, runs: Vec, revision: Option) -> anyhow::Result { + write(self, KEY, &runs, revision).await + } +} + +/// Put `run` first in `doc`, keeping the newest [`MAX_RUNS`], re-reading and +/// re-applying when another write landed in between. +async fn push(doc: &impl RunDoc, run: &RunRecord) -> anyhow::Result<()> { + for _ in 0..MAX_ATTEMPTS { + let (runs, revision) = doc.read().await?; + if doc.write(newest(run, runs), revision).await? { + return Ok(()); + } + tracing::debug!("review classifier runs: write lost a race, retrying"); + } + anyhow::bail!("gave up recording the run after {MAX_ATTEMPTS} conflicting writes") +} + +/// `runs` with `run` in front, cut to [`MAX_RUNS`]. +fn newest(run: &RunRecord, mut runs: Vec) -> Vec { + runs.insert(0, run.clone()); + runs.truncate(MAX_RUNS); + runs +} + +/// The runs' store, as [`AppState`] holds it. +#[derive(Clone)] +pub(crate) enum Runs { + /// [`BUCKET`] on the swarm queue, when this host has one. + Live(Option), + /// An in-memory document a test reads back. + #[cfg(test)] + Fake(std::sync::Arc), +} + +/// Why the runs' store could not be reached: always 503. +struct Unreachable(String); + +impl Runs { + /// Log `run` and add it to the list. A failed write is logged and + /// changes nothing else. + pub(crate) async fn record(&self, run: RunRecord) { + log(&run); + if let Err(e) = self.push(&run).await { + tracing::warn!( + id = %run.id, error = %format!("{e:#}"), + "storing a review classifier run failed; the run is in this log only" + ); + } + } + + async fn push(&self, run: &RunRecord) -> anyhow::Result<()> { + match self { + Self::Live(client) => { + let store = open(client.as_ref()).await.map_err(|Unreachable(e)| { + anyhow::anyhow!(e).context(format!("opening the {BUCKET} bucket")) + })?; + push(&store, run).await + } + #[cfg(test)] + Self::Fake(doc) => push(&**doc, run).await, + } + } + + async fn list(&self) -> Result, ProblemDetails> { + let read = match self { + Self::Live(client) => { + let store = open(client.as_ref()) + .await + .map_err(|Unreachable(e)| error_problem(StatusCode::SERVICE_UNAVAILABLE, &e))?; + RunDoc::read(&store).await + } + #[cfg(test)] + Self::Fake(doc) => doc.read().await, + }; + read.map(|(runs, _)| runs).map_err(|e| { + let detail = format!("{e:#}"); + tracing::warn!(error = %detail, "reading the review classifier runs failed"); + error_problem(StatusCode::INTERNAL_SERVER_ERROR, &detail) + }) + } +} + +/// Open [`BUCKET`], creating it if nothing has yet. +async fn open(client: Option<&async_nats::Client>) -> Result { + let client = client + .ok_or_else(|| Unreachable("no swarm queue is configured on this host".to_owned()))?; + swarm_queue_client::ensure_connected(client) + .map_err(|e| Unreachable(swarm_queue_client::chain(&e)))?; + let js = async_nats::jetstream::new(client.clone()); + if let Ok(store) = js.get_key_value(BUCKET).await { + return Ok(store); + } + js.create_key_value(async_nats::jetstream::kv::Config { + bucket: BUCKET.to_owned(), + description: "The review classifier's newest finished runs".to_owned(), + history: 1, + ..Default::default() + }) + .await + .with_context(|| format!("creating the {BUCKET} bucket")) + .map_err(|e| Unreachable(format!("{e:#}"))) +} + +/// `value` as its one serde word, as the route spells it. +fn word(value: &T) -> String { + serde_json::to_value(value) + .ok() + .and_then(|v| v.as_str().map(str::to_owned)) + .unwrap_or_default() +} + +/// The [`EVENT`] for `run`, one field per record field. `rows` is the JSON +/// the route returns for them. +fn log(run: &RunRecord) { + let rows = run + .rows + .as_ref() + .and_then(|rows| serde_json::to_string(rows).ok()); + tracing::info!( + id = %run.id, + trigger = %word(&run.trigger), + repo = %run.repo, + pr = run.pr, + requester = %run.requester, + head_sha = run.head_sha.as_deref(), + started_unix = run.started_unix, + finished_unix = run.finished_unix, + outcome = %word(&run.outcome), + detail = run.detail.as_deref(), + rows = rows.as_deref(), + overall = run.overall.as_ref().map(word).as_deref(), + review_url = run.review_url.as_deref(), + "{EVENT}" + ); +} + +/// List the review classifier's finished runs. +#[utoipa::path( + get, + path = "/api/review-classifier/runs", + responses( + (status = 200, description = "the newest 100 finished runs, try-outs and requested \ + reviews, newest first; empty when there are none", body = Vec), + (status = 503, description = "no swarm queue is configured on this host, it is not \ + connected, or the runs' bucket could not be opened (problem+json)", body = String), + (status = 500, description = "the runs could not be read (problem+json)", + body = String), + ), + tag = "review-classifier" +)] +pub(crate) async fn get_runs( + State(state): State, +) -> Result>, ProblemDetails> { + state.runs.list().await.map(Json) +} + +#[cfg(test)] +pub(crate) mod tests { + use std::sync::{Arc, Mutex}; + + use axum::body::Body; + use axum::http::Request; + use swarm_queue_client::review_classifier::{Polarity, Verdict}; + use tower::ServiceExt as _; + use tracing_subscriber::layer::SubscriberExt as _; + + use super::*; + + /// A document in memory. Each of the first `conflicts` writes loses to a + /// write that lands one run in between; a document set to `fail` fails + /// every read. + #[derive(Default)] + pub(crate) struct FakeDoc { + state: Mutex<(Vec, u64)>, + conflicts: Mutex, + fail: bool, + } + + impl FakeDoc { + pub(crate) fn failing() -> Self { + Self { + fail: true, + ..Self::default() + } + } + + fn conflicting(conflicts: usize) -> Self { + Self { + conflicts: Mutex::new(conflicts), + ..Self::default() + } + } + + pub(crate) fn runs(&self) -> Vec { + self.state.lock().expect("unpoisoned").0.clone() + } + } + + impl RunDoc for FakeDoc { + async fn read(&self) -> anyhow::Result<(Vec, Option)> { + anyhow::ensure!(!self.fail, "the store is down"); + let (runs, revision) = self.state.lock().expect("unpoisoned").clone(); + Ok((runs, Some(revision).filter(|r| *r > 0))) + } + + async fn write(&self, runs: Vec, revision: Option) -> anyhow::Result { + let mut state = self.state.lock().expect("unpoisoned"); + let mut conflicts = self.conflicts.lock().expect("unpoisoned"); + if *conflicts > 0 { + *conflicts -= 1; + let mut other = record(&format!("other-{conflicts}")); + other.trigger = Trigger::Review; + state.0.insert(0, other); + state.1 += 1; + } + if revision.unwrap_or(0) != state.1 { + return Ok(false); + } + *state = (runs, state.1 + 1); + Ok(true) + } + } + + pub(crate) fn record(id: &str) -> RunRecord { + RunRecord { + id: id.to_owned(), + trigger: Trigger::TryOut, + repo: "o/r".to_owned(), + pr: 7, + requester: "alice".to_owned(), + head_sha: Some("abc123".to_owned()), + started_unix: 1_000, + finished_unix: 1_060, + outcome: RunOutcome::Done, + detail: None, + rows: None, + overall: None, + review_url: None, + } + } + + fn ids(runs: &[RunRecord]) -> Vec<&str> { + runs.iter().map(|r| r.id.as_str()).collect() + } + + #[tokio::test] + async fn the_newest_hundred_survive_newest_first() { + let doc = FakeDoc::default(); + for i in 0..=MAX_RUNS { + push(&doc, &record(&format!("r{i}"))).await.expect("stored"); + } + let runs = doc.runs(); + assert_eq!(runs.len(), 100); + assert_eq!(runs[0].id, "r100"); + assert_eq!(runs[99].id, "r1", "r0, the oldest, is gone"); + } + + #[tokio::test] + async fn a_write_that_lost_a_race_is_redone_on_the_winner() { + let doc = FakeDoc::conflicting(2); + push(&doc, &record("mine")).await.expect("stored"); + assert_eq!(ids(&doc.runs()), ["mine", "other-0", "other-1"]); + + let doc = FakeDoc::conflicting(MAX_ATTEMPTS); + let err = push(&doc, &record("mine")).await.expect_err("gives up"); + assert!(err.to_string().contains("conflicting writes"), "{err}"); + assert!(!ids(&doc.runs()).contains(&"mine")); + } + + /// The bucket on a real server: created on first use, written, read back. + #[tokio::test] + async fn runs_round_trip_through_the_bucket() { + let nats = crate::test_nats::TestNats::start(); + let runs = Runs::Live(Some(nats.client().await)); + runs.record(record("a")).await; + runs.record(record("b")).await; + let listed = runs.list().await.expect("listed"); + assert_eq!(ids(&listed), ["b", "a"]); + } + + #[tokio::test] + async fn the_route_lists_the_runs_newest_first_and_a_down_store_is_an_error() { + let doc = Arc::new(FakeDoc::default()); + let mut reviewed = record("b"); + reviewed.trigger = Trigger::Review; + reviewed.outcome = RunOutcome::NoVerdict; + reviewed.detail = Some("the rules did not load".to_owned()); + reviewed.head_sha = None; + push(&*doc, &record("a")).await.expect("stored"); + push(&*doc, &reviewed).await.expect("stored"); + + let get = |runs: Runs| async move { + let (mut state, _) = crate::tests::state_with_roster(); + state.runs = runs; + let res = crate::build_app(state) + .oneshot( + Request::get("/api/review-classifier/runs") + .body(Body::empty()) + .expect("a request"), + ) + .await + .expect("infallible"); + let status = res.status(); + let bytes = axum::body::to_bytes(res.into_body(), usize::MAX) + .await + .expect("a body"); + let body: serde_json::Value = serde_json::from_slice(&bytes).expect("JSON"); + (status, body) + }; + let (status, body) = get(Runs::Fake(doc)).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body[0]["id"], "b"); + assert_eq!(body[0]["trigger"], "review"); + assert_eq!(body[0]["outcome"], "no-verdict"); + assert_eq!(body[0]["detail"], "the rules did not load"); + assert!(body[0].get("head_sha").is_none(), "{body}"); + assert_eq!(body[1]["trigger"], "try-out"); + assert_eq!(body[1]["outcome"], "done"); + + let (status, body) = get(Runs::Fake(Arc::new(FakeDoc::failing()))).await; + assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR, "{body}"); + let (status, body) = get(Runs::Live(None)).await; + assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE, "{body}"); + assert_eq!(body["detail"], "no swarm queue is configured on this host"); + } + + #[test] + fn a_record_carries_no_diff_commit_text_or_cost() { + let mut run = record("a"); + run.rows = Some(Vec::new()); + run.overall = Some(Overall::Approve); + run.detail = Some("why".to_owned()); + run.review_url = Some("https://forge/o/r/pulls/7#issuecomment-1".to_owned()); + let value = serde_json::to_value(&run).expect("encodes"); + let mut keys: Vec<_> = value.as_object().expect("an object").keys().collect(); + keys.sort(); + assert_eq!( + keys, + [ + "detail", + "finished_unix", + "head_sha", + "id", + "outcome", + "overall", + "pr", + "repo", + "requester", + "review_url", + "rows", + "started_unix", + "trigger", + ] + ); + } + + /// Every field of every event, by name. + #[derive(Clone, Default)] + struct Fields(Arc>>); + + impl tracing::field::Visit for Fields { + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + self.0 + .lock() + .expect("unpoisoned") + .push((field.name().to_owned(), format!("{value:?}"))); + } + + fn record_str(&mut self, field: &tracing::field::Field, value: &str) { + self.0 + .lock() + .expect("unpoisoned") + .push((field.name().to_owned(), value.to_owned())); + } + } + + impl tracing_subscriber::Layer for Fields { + fn on_event( + &self, + event: &tracing::Event<'_>, + _: tracing_subscriber::layer::Context<'_, S>, + ) { + event.record(&mut self.clone()); + } + } + + #[test] + fn the_log_event_carries_the_records_fields() { + let mut run = record("a"); + run.trigger = Trigger::Review; + run.outcome = RunOutcome::ChangesRequested; + run.rows = Some(vec![Row { + name: "datetime".to_owned(), + polarity: Polarity::Positive, + verdict: Verdict::NoMatch, + certainty: 90, + threshold: 80, + counted: true, + passed: false, + reason: "No zone.".to_owned(), + error: None, + }]); + run.overall = Some(Overall::RequestChanges); + run.review_url = Some("https://forge/o/r/pulls/7#issuecomment-1".to_owned()); + let fields = Fields::default(); + let subscriber = tracing_subscriber::registry().with(fields.clone()); + tracing::subscriber::with_default(subscriber, || log(&run)); + + let got = fields.0.lock().expect("unpoisoned").clone(); + let field = |name: &str| { + got.iter() + .find(|(n, _)| n == name) + .map_or_else(|| panic!("no {name} in {got:?}"), |(_, v)| v.clone()) + }; + assert_eq!(field("message"), EVENT); + assert_eq!(field("id"), "a"); + assert_eq!(field("trigger"), "review"); + assert_eq!(field("repo"), "o/r"); + assert_eq!(field("pr"), "7"); + assert_eq!(field("requester"), "alice"); + assert_eq!(field("head_sha"), "abc123"); + assert_eq!(field("started_unix"), "1000"); + assert_eq!(field("finished_unix"), "1060"); + assert_eq!(field("outcome"), "changes-requested"); + assert_eq!(field("overall"), "request-changes"); + assert_eq!( + field("review_url"), + "https://forge/o/r/pulls/7#issuecomment-1" + ); + let rows: Vec = serde_json::from_str(&field("rows")).expect("the rows as JSON"); + assert_eq!(Some(rows), run.rows); + assert!( + !got.iter().any(|(n, _)| n == "detail"), + "an absent field is left out: {got:?}" + ); + } +}