swarm-controller: record builder runs with nix's tail and link the status to them

swarm-controller/src/builder/runs.rs

@@ -0,0 +1,495 @@
+//! `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::schedules::{read_with, write};
+use crate::{AppState, error_problem};
+
+/// 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 {
+    /// `<agent>-<first 12 of head_sha>-<started_unix>`, the last segment
+    /// of the link on the build's status.
+    pub(crate) id: String,
+    /// `agent-configs/<agent>`.
+    pub(crate) repo: String,
+    pub(crate) pr: u64,
+    /// The commit built.
+    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<String>,
+    /// The last lines `nix` printed, for `failed`.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(crate) error_tail: Option<String>,
+    /// Why no build ran or came back, for `error`.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(crate) detail: Option<String>,
+}
+
+/// 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<Output = anyhow::Result<(Vec<Stored>, Option<u64>)>> + Send;
+    /// `Ok(false)` when another write landed since `revision`.
+    fn write(
+        &self,
+        builds: Vec<Stored>,
+        revision: Option<u64>,
+    ) -> impl Future<Output = anyhow::Result<bool>> + Send;
+}
+
+impl BuildDoc for Store {
+    async fn read(&self) -> anyhow::Result<(Vec<Stored>, Option<u64>)> {
+        read_with(self, KEY, |bytes| Ok(decode(bytes))).await
+    }
+
+    async fn write(&self, builds: Vec<Stored>, revision: Option<u64>) -> anyhow::Result<bool> {
+        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<Stored> {
+    let values: Vec<serde_json::Value> = 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<async_nats::Client>),
+    /// An in-memory document a test reads back.
+    #[cfg(test)]
+    Fake(std::sync::Arc<tests::FakeDoc>),
+}
+
+/// 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<Vec<BuildRecord>, 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<Store, Unreachable> {
+    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<BuildRecord>),
+        (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<AppState>,
+) -> Result<Json<Vec<BuildRecord>>, 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<AppState>,
+    Path(id): Path<String>,
+) -> Result<Json<BuildRecord>, 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<Stored>, u64)>,
+        conflicts: Mutex<usize>,
+        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<BuildRecord> {
+            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<Stored>, Option<u64>)> {
+            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<Stored>, revision: Option<u64>) -> anyhow::Result<bool> {
+            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: 3,
+            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::Value> = 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::tests::state_with_roster();
+            state.builds = builds;
+            let res = crate::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");
+    }
+}