//! `GET /api/builder/runs`: the swarm builder's finished builds, newest //! first, each with what `nix` printed when it failed; and //! `GET /api/builder/runs/{id}`, one of them, which its commit status links to. //! //! The newest [`MAX_RUNS`] are one JSON array under [`KEY`] in [`BUCKET`], //! which only this daemon reads or writes, kept the way //! `review_classifier::runs` keeps the classifier's: one compare-and-set //! document, with entries this build cannot decode kept as they were. use std::future::Future; use anyhow::Context as _; use async_nats::jetstream::kv::Store; use axum::Json; use axum::extract::{Path, State}; use axum::http::StatusCode; use problem_details::ProblemDetails; use serde::{Deserialize, Serialize}; use utoipa::ToSchema; use crate::routes::{AppState, error_problem}; use crate::schedules::{read_with, write}; /// The bucket holding the builds. Agents hold no grant on it. pub const BUCKET: &str = "builder-runs"; /// The one key in [`BUCKET`]. pub const KEY: &str = "recent"; /// How many builds 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. const MAX_ATTEMPTS: usize = 5; /// How a build ended, as its commit status reads. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)] #[serde(rename_all = "kebab-case")] pub(crate) enum BuildOutcome { /// The config built: `success`. Built, /// `nix` failed or timed out: `failure`. Failed, /// No build ran or none came back: `error`. Error, } /// One finished build. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)] pub(crate) struct BuildRecord { /// `--`, the last segment /// of the link on the build's status. pub(crate) id: String, /// `agent-configs/` for a config PR's build, `internal/meta` for a /// deploy's (`meta_build`). pub(crate) repo: String, /// The config PR a build checks. A `meta_build` build has none. #[serde(default, skip_serializing_if = "Option::is_none")] pub(crate) pr: Option, /// The agent a `meta_build` build built; a PR build's is in `repo`. #[serde(default, skip_serializing_if = "Option::is_none")] pub(crate) agent: Option, /// The commit built: the PR's head, or `internal/meta`'s for a deploy's /// build. pub(crate) head_sha: String, /// Unix seconds: when the node sent the build, and when its answer came. pub(crate) started_unix: i64, pub(crate) finished_unix: i64, pub(crate) outcome: BuildOutcome, /// The system closure's store path, for `built`. #[serde(default, skip_serializing_if = "Option::is_none")] pub(crate) out_path: Option, /// The last lines `nix` printed, for `failed`. #[serde(default, skip_serializing_if = "Option::is_none")] pub(crate) error_tail: Option, /// Why no build ran or came back, for `error`. #[serde(default, skip_serializing_if = "Option::is_none")] pub(crate) detail: Option, } /// One entry of the stored list. #[derive(Debug, Clone, PartialEq, Serialize)] #[serde(untagged)] pub(crate) enum Stored { Build(BuildRecord), /// 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. Unreadable(serde_json::Value), } /// Where the builds are kept: [`BUCKET`] in production, a fake in the tests. pub(crate) trait BuildDoc { /// 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, builds: Vec, revision: Option, ) -> impl Future> + Send; } impl BuildDoc for Store { async fn read(&self) -> anyhow::Result<(Vec, Option)> { read_with(self, KEY, |bytes| Ok(decode(bytes))).await } async fn write(&self, builds: Vec, revision: Option) -> anyhow::Result { write(self, KEY, &builds, revision).await } } /// The entries in `bytes`. 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 builder runs are not decodable; reading them as none" ); return Vec::new(); } }; values .into_iter() .map(|value| match BuildRecord::deserialize(&value) { Ok(build) => Stored::Build(build), Err(e) => { tracing::warn!( bucket = BUCKET, key = KEY, error = %e, "a builder run this build cannot decode is left out of the list and kept" ); Stored::Unreadable(value) } }) .collect() } /// Put `build` first in `doc`, keeping the newest [`MAX_RUNS`], re-reading and /// re-applying when another write landed in between. async fn push(doc: &impl BuildDoc, build: &BuildRecord) -> anyhow::Result<()> { for _ in 0..MAX_ATTEMPTS { let (mut builds, revision) = doc.read().await?; builds.insert(0, Stored::Build(build.clone())); builds.truncate(MAX_RUNS); if doc.write(builds, revision).await? { return Ok(()); } tracing::debug!("builder runs: write lost a race, retrying"); } anyhow::bail!("gave up recording the build after {MAX_ATTEMPTS} conflicting writes") } /// The builds' store. #[derive(Clone)] pub(crate) enum Builds { /// [`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 builds' store could not be reached: always 503. struct Unreachable(String); impl Builds { /// Add `build` to the list. `false`, with a warning, when the write /// failed. pub(crate) async fn record(&self, build: &BuildRecord) -> bool { let pushed = match self { Self::Live(client) => match open(client.as_ref()).await { Ok(store) => push(&store, build).await, Err(Unreachable(e)) => { Err(anyhow::anyhow!(e).context(format!("opening the {BUCKET} bucket"))) } }, #[cfg(test)] Self::Fake(doc) => push(&**doc, build).await, }; match pushed { Ok(()) => true, Err(e) => { tracing::warn!( id = %build.id, repo = %build.repo, head_sha = %build.head_sha, error = %format!("{e:#}"), "storing a builder run failed; its status links to nothing" ); false } } } 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))?; BuildDoc::read(&store).await } #[cfg(test)] Self::Fake(doc) => doc.read().await, }; read.map(|(builds, _)| { builds .into_iter() .filter_map(|entry| match entry { Stored::Build(build) => Some(build), Stored::Unreadable(_) => None, }) .collect() }) .map_err(|e| { let detail = format!("{e:#}"); tracing::warn!(error = %detail, "reading the builder 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 swarm builder's newest finished builds".to_owned(), history: 1, ..Default::default() }) .await .with_context(|| format!("creating the {BUCKET} bucket")) .map_err(|e| Unreachable(format!("{e:#}"))) } /// List the swarm builder's finished builds. #[utoipa::path( get, path = "/api/builder/runs", responses( (status = 200, description = "the newest 100 finished builds, 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 builds' bucket could not be opened (problem+json)", body = String), (status = 500, description = "the builds could not be read (problem+json)", body = String), ), tag = "builder" )] pub(crate) async fn get_runs( State(state): State, ) -> Result>, ProblemDetails> { state.builds.list().await.map(Json) } /// One finished build: the link a `swarm-builder` commit status carries. #[utoipa::path( get, path = "/api/builder/runs/{id}", params(("id" = String, Path, description = "the build's id, as the list names it")), responses( (status = 200, description = "the build, with `nix`'s tail when it failed", body = BuildRecord), (status = 404, description = "no build has this id among the newest 100 \ (problem+json)", body = String), (status = 503, description = "as for the list (problem+json)", body = String), (status = 500, description = "as for the list (problem+json)", body = String), ), tag = "builder" )] pub(crate) async fn get_run( State(state): State, Path(id): Path, ) -> Result, ProblemDetails> { state .builds .list() .await? .into_iter() .find(|build| build.id == id) .map(Json) .ok_or_else(|| { error_problem( StatusCode::NOT_FOUND, "no build has this id among the newest 100", ) }) } #[cfg(test)] pub(crate) mod tests { use std::sync::{Arc, Mutex}; use axum::body::Body; use axum::http::Request; use tower::ServiceExt as _; use super::*; /// A document in memory. Each of the first `conflicts` writes loses to a /// write that lands one build 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() } } /// The decodable builds, as the list shows them. pub(crate) fn builds(&self) -> Vec { self.state .lock() .expect("unpoisoned") .0 .iter() .filter_map(|entry| match entry { Stored::Build(build) => Some(build.clone()), Stored::Unreadable(_) => None, }) .collect() } } impl BuildDoc for FakeDoc { async fn read(&self) -> anyhow::Result<(Vec, Option)> { anyhow::ensure!(!self.fail, "the store is down"); let (builds, revision) = self.state.lock().expect("unpoisoned").clone(); Ok((builds, Some(revision).filter(|r| *r > 0))) } async fn write(&self, builds: 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; state .0 .insert(0, Stored::Build(record(&format!("other-{conflicts}")))); state.1 += 1; } if revision.unwrap_or(0) != state.1 { return Ok(false); } *state = (builds, state.1 + 1); Ok(true) } } pub(crate) fn record(id: &str) -> BuildRecord { BuildRecord { id: id.to_owned(), repo: "agent-configs/atlas".to_owned(), pr: Some(3), agent: None, head_sha: "abc".to_owned(), started_unix: 1_000, finished_unix: 1_060, outcome: BuildOutcome::Failed, out_path: None, error_tail: Some("evaluating\nerror: boom".to_owned()), detail: None, } } fn ids(builds: &[BuildRecord]) -> Vec<&str> { builds.iter().map(|b| b.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 builds = doc.builds(); assert_eq!(builds.len(), 100); assert_eq!(builds[0].id, "r100"); assert_eq!(builds[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 { conflicts: Mutex::new(2), ..FakeDoc::default() }; push(&doc, &record("mine")).await.expect("stored"); assert_eq!(ids(&doc.builds()), ["mine", "other-0", "other-1"]); } /// The bucket on a real server: created on first use, written, read /// back; a record from a newer build is left out and kept in its place. #[tokio::test] async fn builds_round_trip_through_the_bucket_and_keep_an_undecodable_one() { let nats = crate::test_nats::TestNats::start(); let client = nats.client().await; let builds = Builds::Live(Some(client.clone())); assert!(builds.record(&record("a")).await); assert!(builds.record(&record("b")).await); assert_eq!(ids(&builds.list().await.expect("listed")), ["b", "a"]); let Ok(store) = open(Some(&client)).await else { panic!("the bucket did not open"); }; let mut newer = serde_json::to_value(record("newer")).expect("encodes"); newer["outcome"] = "a-later-outcome".into(); let doc = serde_json::json!([newer, record("b"), record("a")]); store .put(KEY, doc.to_string().into()) .await .expect("written"); assert!(builds.record(&record("c")).await); assert_eq!(ids(&builds.list().await.expect("listed")), ["c", "b", "a"]); let entry = store.get(KEY).await.expect("readable").expect("a document"); let after: Vec = serde_json::from_slice(&entry).expect("a JSON array"); assert_eq!(after.len(), 4, "{after:?}"); assert_eq!(after[1], newer, "kept untouched, in its place"); } #[tokio::test] async fn a_failed_write_is_reported() { assert!( !Builds::Fake(Arc::new(FakeDoc::failing())) .record(&record("a")) .await ); assert!(!Builds::Live(None).record(&record("a")).await); } #[tokio::test] async fn the_route_lists_the_builds_with_their_tail_and_a_down_store_is_an_error() { let doc = Arc::new(FakeDoc::default()); let mut built = record("a"); built.outcome = BuildOutcome::Built; built.out_path = Some("/nix/store/x".to_owned()); built.error_tail = None; push(&*doc, &built).await.expect("stored"); push(&*doc, &record("b")).await.expect("stored"); let get_at = |builds: Builds, path: &'static str| async move { let (mut state, _) = crate::routes::tests::state_with_roster(); state.builds = builds; let res = crate::startup::build_app(state) .oneshot(Request::get(path).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 get = |builds: Builds| get_at(builds, "/api/builder/runs"); let (status, body) = get(Builds::Fake(Arc::clone(&doc))).await; assert_eq!(status, StatusCode::OK, "{body}"); assert_eq!(body[0]["id"], "b"); assert_eq!(body[0]["repo"], "agent-configs/atlas"); assert_eq!(body[0]["head_sha"], "abc"); assert_eq!(body[0]["outcome"], "failed"); assert_eq!(body[0]["error_tail"], "evaluating\nerror: boom"); assert!(body[0].get("out_path").is_none(), "{body}"); assert_eq!(body[1]["outcome"], "built"); assert_eq!(body[1]["out_path"], "/nix/store/x"); assert!(body[1].get("error_tail").is_none(), "{body}"); let one = |path| get_at(Builds::Fake(Arc::clone(&doc)), path); let (status, body) = one("/api/builder/runs/b").await; assert_eq!(status, StatusCode::OK, "{body}"); assert_eq!(body["id"], "b"); assert_eq!(body["error_tail"], "evaluating\nerror: boom"); let (status, body) = one("/api/builder/runs/a").await; assert_eq!(status, StatusCode::OK, "{body}"); assert_eq!(body["outcome"], "built"); let (status, body) = one("/api/builder/runs/nope").await; assert_eq!(status, StatusCode::NOT_FOUND, "{body}"); let (status, body) = get(Builds::Fake(Arc::new(FakeDoc::failing()))).await; assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR, "{body}"); let (status, body) = get(Builds::Live(None)).await; assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE, "{body}"); assert_eq!(body["detail"], "no swarm queue is configured on this host"); } }