hyperhive/swarm-controller/src/builder/runs.rs

//! `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 {
    /// `<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::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");
    }
}