//! `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_with, 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), } } /// One entry of the stored list. #[derive(Debug, Clone, PartialEq, Serialize)] #[serde(untagged)] pub(crate) enum Stored { Run(RunRecord), /// A record this build cannot decode, such as one a newer build wrote. /// Left out of the list but kept, counted toward [`MAX_RUNS`] and written /// back as it was read, so a newer build reads it again after a /// roll-forward. Unreadable(serde_json::Value), } /// Where the runs are kept: [`BUCKET`] in production, a fake in the tests. pub(crate) trait RunDoc { /// The entries, 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_with(self, KEY, |bytes| Ok(decode(bytes))).await } async fn write(&self, runs: Vec, revision: Option) -> anyhow::Result { write(self, KEY, &runs, revision).await } } /// The entries in `bytes`, the stored document, with a warning when any is /// [`Stored::Unreadable`]. A document that is not a JSON array reads as no /// entries, with a warning, and the next write replaces it. fn decode(bytes: &[u8]) -> Vec { let values: Vec = match serde_json::from_slice(bytes) { Ok(values) => values, Err(e) => { tracing::warn!( bucket = BUCKET, key = KEY, error = %e, "the review classifier runs are not decodable; reading them as none" ); return Vec::new(); } }; let mut unreadable = 0_usize; let mut first_error = None; let entries = values .into_iter() .map(|value| match RunRecord::deserialize(&value) { Ok(run) => Stored::Run(run), Err(e) => { unreadable += 1; first_error.get_or_insert(e); Stored::Unreadable(value) } }) .collect(); if let Some(error) = first_error { tracing::warn!( bucket = BUCKET, key = KEY, unreadable, %error, "review classifier runs this build cannot decode are left out of the list and kept" ); } entries } /// 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`] entries. fn newest(run: &RunRecord, mut runs: Vec) -> Vec { runs.insert(0, Stored::Run(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.into_iter() .filter_map(|entry| match entry { Stored::Run(run) => Some(run), Stored::Unreadable(_) => None, }) .collect() }) .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() } } /// The decodable runs, as the list shows them. pub(crate) fn runs(&self) -> Vec { self.entries() .into_iter() .filter_map(|entry| match entry { Stored::Run(run) => Some(run), Stored::Unreadable(_) => None, }) .collect() } fn entries(&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, Stored::Run(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"]); } /// A record from a newer build among readable ones, and a document that /// is no list at all: listing and recording both keep working. The newer /// record is left out of the list and written back as it was; the /// non-list document is replaced by one this build decodes. #[tokio::test] async fn an_undecodable_record_is_kept_and_a_non_list_replaced() { let nats = crate::test_nats::TestNats::start(); let client = nats.client().await; let Ok(store) = open(Some(&client)).await else { panic!("the bucket did not open"); }; let runs = Runs::Live(Some(client)); let stored = || async { let entry = store.get(KEY).await.expect("readable").expect("a document"); serde_json::from_slice::>(&entry).expect("a JSON array") }; let mut newer = serde_json::to_value(record("newer")).expect("encodes"); newer["outcome"] = "a-later-outcome".into(); let doc = serde_json::json!([record("b"), newer, record("a")]); store .put(KEY, doc.to_string().into()) .await .expect("written"); assert_eq!(ids(&runs.list().await.expect("listed")), ["b", "a"]); runs.record(record("c")).await; assert_eq!(ids(&runs.list().await.expect("listed")), ["c", "b", "a"]); let after = stored().await; assert_eq!(after.len(), 4, "{after:?}"); assert_eq!(after[2], newer, "kept untouched, in its place"); store .put(KEY, br#"{"not":"a list"}"#.to_vec().into()) .await .expect("written"); assert!(runs.list().await.expect("listed").is_empty()); runs.record(record("d")).await; let after = stored().await; let decoded: Vec = serde_json::from_value(after.into()).expect("decodes whole"); assert_eq!(ids(&decoded), ["d"]); } #[tokio::test] async fn an_unreadable_entry_counts_toward_the_hundred() { let doc = FakeDoc::default(); { let mut state = doc.state.lock().expect("unpoisoned"); state.0 = (0..MAX_RUNS - 1) .map(|i| Stored::Run(record(&format!("r{i}")))) .collect(); state .0 .insert(50, Stored::Unreadable(serde_json::json!({"from": "later"}))); state.1 = 1; } push(&doc, &record("new")).await.expect("stored"); let entries = doc.entries(); assert_eq!(entries.len(), MAX_RUNS); assert!(entries.iter().any(|e| matches!(e, Stored::Unreadable(_)))); assert_eq!( doc.runs().len(), MAX_RUNS - 1, "the unreadable entry took a run's place" ); } #[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:?}" ); } }