hyperhive/swarm-controller/src/builder.rs

//! The `BuildAgent` node: send a config PR's head to the swarm builder over the
//! queue, turn its answer into the [`CONTEXT`] commit status on that head, and
//! record the build, `nix`'s tail with it, in [`runs`]. The status links to the
//! swarm UI's builds page for that build.
//!
//! Nothing is pushed, and the closure the builder made stays in its host's
//! store. A red status blocks nothing; the merge deploys as before. The build
//! that does gate a deploy is `meta_build`'s.
//!
//! Builds share one `SwarmResourceKind::Builder` slot, so one runs at a time,
//! and a newer push replaces its PR's build while that build is still
//! pending (`queue_build`).

pub(crate) mod runs;

use std::future::Future;

use forgejo_api::structs::CommitStatusState;
use hive_jobq::scheduler::Outcome;
use swarm_wire::builder::{BuildAgent, REPLY_DEADLINE, Reply, SUBJECT};

use crate::agent_renewal::unix_now;
use crate::forge::{self, CONFIG_ORG};
use runs::{BuildOutcome, BuildRecord};

/// The commit status context the check writes.
pub const CONTEXT: &str = "swarm-builder";

/// Where a build is sent: the queue in production, a fake in the tests.
/// `Err` is a build that never reached a builder or never came back.
pub(crate) trait BuildSink {
    fn build(&self, request: Vec<u8>) -> impl Future<Output = Result<Reply, String>> + Send;
}

impl BuildSink for async_nats::Client {
    async fn build(&self, request: Vec<u8>) -> Result<Reply, String> {
        use async_nats::RequestErrorKind;

        let sent = async_nats::Request::new()
            .payload(request.into())
            .timeout(Some(REPLY_DEADLINE));
        match self.send_request(SUBJECT, sent).await {
            Ok(msg) => serde_json::from_slice(&msg.payload)
                .map_err(|e| format!("the builder's reply did not decode: {e}")),
            Err(e) => Err(match e.kind() {
                RequestErrorKind::NoResponders => "no swarm builder is listening: \
                     deploy.builder.enable is set on no host, or its service is not connected \
                     to the swarm queue"
                    .to_owned(),
                RequestErrorKind::TimedOut => format!(
                    "the swarm builder sent no answer within {} minutes",
                    REPLY_DEADLINE.as_secs() / 60
                ),
                _ => format!("sending the build to the swarm builder failed: {e}"),
            }),
        }
    }
}

/// Where a status is written: the forge in production, a fake in the tests.
pub(crate) trait StatusSink {
    fn set(
        &self,
        agent: &str,
        rev: &str,
        state: CommitStatusState,
        description: &str,
        target_url: Option<&str>,
    ) -> impl Future<Output = anyhow::Result<()>> + Send;
}

impl StatusSink for forge::Client {
    async fn set(
        &self,
        agent: &str,
        rev: &str,
        state: CommitStatusState,
        description: &str,
        target_url: Option<&str>,
    ) -> anyhow::Result<()> {
        forge::within_forge_deadline(
            "build_agent status",
            self.create_commit_status(
                CONFIG_ORG,
                agent,
                rev,
                forge::CommitStatus {
                    context: CONTEXT,
                    state,
                    description,
                    target_url,
                },
            ),
        )
        .await
        .unwrap_or_else(|| Err(anyhow::anyhow!("no answer from the forge")))
    }
}

/// Build `job`, record it in `builds`, and write its status, linking to the
/// recorded build under `link_base` (the swarm vhost's public URL) when there is
/// one. The node fails when the build does, so the Jobs page shows red where
/// the PR does.
pub(crate) async fn run(
    queue: Option<&impl BuildSink>,
    statuses: Option<&impl StatusSink>,
    builds: &runs::Builds,
    link_base: Option<&str>,
    job: BuildAgent,
) -> Outcome {
    let Some(statuses) = statuses else {
        return Outcome::Failed(
            "no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token not in the \
             swarm secret store), so no build status can be written"
                .to_owned(),
        );
    };
    let started_unix = unix_now();
    let set = |state, description: String, link: Option<String>| {
        let job = &job;
        async move {
            statuses
                .set(&job.agent, &job.rev, state, &description, link.as_deref())
                .await
                .map_err(|e| format!("{e:#}"))
        }
    };
    let finish =
        |outcome, text: String| record(builds, link_base, &job, started_unix, outcome, text);
    let Some(queue) = queue else {
        let error = "no swarm queue is configured on this host, so no build can be sent";
        let link = finish(BuildOutcome::Error, error.to_owned()).await;
        return settle(
            set(CommitStatusState::Error, error.to_owned(), link).await,
            error,
        );
    };
    if let Err(e) = set(
        CommitStatusState::Pending,
        "building on the swarm builder".to_owned(),
        None,
    )
    .await
    {
        tracing::warn!(agent = %job.agent, rev = %job.rev, error = %e, "build_agent: the pending status was not written");
    }
    let request = match serde_json::to_vec(&job) {
        Ok(request) => request,
        Err(e) => return Outcome::Failed(format!("encoding the build: {e}")),
    };
    match queue.build(request).await {
        Ok(Reply::Built { out_path }) => {
            tracing::info!(agent = %job.agent, rev = %job.rev, %out_path, "build_agent: built");
            let link = finish(BuildOutcome::Built, out_path.clone()).await;
            match set(CommitStatusState::Success, out_path, link).await {
                Ok(()) => Outcome::Done,
                Err(e) => Outcome::Failed(format!("built, but the status was not written: {e}")),
            }
        }
        Ok(Reply::Failed { error_tail }) => {
            let last = last_line(&error_tail).to_owned();
            let link = finish(BuildOutcome::Failed, error_tail.clone()).await;
            settle(
                set(CommitStatusState::Failure, last, link).await,
                &format!("the build failed: {error_tail}"),
            )
        }
        Err(error) => {
            let link = finish(BuildOutcome::Error, error.clone()).await;
            settle(
                set(CommitStatusState::Error, error.clone(), link).await,
                &error,
            )
        }
    }
}

/// Store the build of `job` that ended as `outcome`, `text` being its store
/// path, `nix`'s tail or the reason. The link to it, when it was
/// stored and `link_base` is known.
async fn record(
    builds: &runs::Builds,
    link_base: Option<&str>,
    job: &BuildAgent,
    started_unix: i64,
    outcome: BuildOutcome,
    text: String,
) -> Option<String> {
    let mut build = BuildRecord {
        id: format!(
            "{}-{}-{started_unix}",
            job.agent,
            job.rev.get(..12).unwrap_or(&job.rev)
        ),
        repo: format!("{CONFIG_ORG}/{}", job.agent),
        pr: job.pr,
        head_sha: job.rev.clone(),
        started_unix,
        finished_unix: unix_now(),
        outcome,
        out_path: None,
        error_tail: None,
        detail: None,
    };
    *match outcome {
        BuildOutcome::Built => &mut build.out_path,
        BuildOutcome::Failed => &mut build.error_tail,
        BuildOutcome::Error => &mut build.detail,
    } = Some(text);
    if !builds.record(&build).await {
        return None;
    }
    link_base.map(|base| format!("{}/jobs?run={}", base.trim_end_matches('/'), build.id))
}

/// A failed node carrying `error`, and the status write's own failure when
/// there was one.
fn settle(written: Result<(), String>, error: &str) -> Outcome {
    match written {
        Ok(()) => Outcome::Failed(error.to_owned()),
        Err(e) => Outcome::Failed(format!("{error} (and the status was not written: {e})")),
    }
}

/// The last non-blank line of `tail`, which for `nix` is the error itself.
fn last_line(tail: &str) -> &str {
    tail.lines()
        .rev()
        .map(str::trim)
        .find(|l| !l.is_empty())
        .unwrap_or("the build failed")
}

#[cfg(test)]
mod tests {
    use std::sync::{Arc, Mutex};

    use super::runs::tests::FakeDoc;
    use super::*;

    const BASE: &str = "https://swarm.example";

    struct FakeQueue(Result<Reply, String>);

    impl BuildSink for FakeQueue {
        async fn build(&self, request: Vec<u8>) -> Result<Reply, String> {
            let sent: BuildAgent = serde_json::from_slice(&request).expect("a build request");
            assert_eq!(sent, job());
            self.0.clone()
        }
    }

    #[derive(Default)]
    struct FakeStatuses(Mutex<Vec<(CommitStatusState, String, Option<String>)>>);

    impl StatusSink for FakeStatuses {
        async fn set(
            &self,
            agent: &str,
            rev: &str,
            state: CommitStatusState,
            description: &str,
            target_url: Option<&str>,
        ) -> anyhow::Result<()> {
            assert_eq!((agent, rev), ("atlas", "abc"));
            self.0.lock().unwrap().push((
                state,
                description.to_owned(),
                target_url.map(str::to_owned),
            ));
            Ok(())
        }
    }

    fn store() -> runs::Builds {
        runs::Builds::Fake(Arc::new(FakeDoc::default()))
    }

    fn job() -> BuildAgent {
        BuildAgent {
            agent: "atlas".to_owned(),
            rev: "abc".to_owned(),
            pr: 3,
        }
    }

    async fn statuses_after(queue: Option<FakeQueue>) -> (Outcome, Vec<CommitStatusState>) {
        let statuses = FakeStatuses::default();
        let outcome = run(queue.as_ref(), Some(&statuses), &store(), Some(BASE), job()).await;
        let states = statuses.0.into_inner().unwrap();
        (outcome, states.into_iter().map(|(s, _, _)| s).collect())
    }

    #[tokio::test]
    async fn a_built_config_goes_pending_then_green() {
        let (outcome, states) = statuses_after(Some(FakeQueue(Ok(Reply::Built {
            out_path: "/nix/store/x".to_owned(),
        }))))
        .await;
        assert!(matches!(outcome, Outcome::Done));
        assert_eq!(
            states,
            [CommitStatusState::Pending, CommitStatusState::Success]
        );
    }

    /// The build, the doc it was recorded in, and the statuses written.
    async fn recorded(
        reply: Result<Reply, String>,
        doc: FakeDoc,
        link_base: Option<&str>,
    ) -> (
        Outcome,
        Vec<runs::BuildRecord>,
        Vec<(CommitStatusState, String, Option<String>)>,
    ) {
        let statuses = FakeStatuses::default();
        let doc = Arc::new(doc);
        let outcome = run(
            Some(&FakeQueue(reply)),
            Some(&statuses),
            &runs::Builds::Fake(Arc::clone(&doc)),
            link_base,
            job(),
        )
        .await;
        (outcome, doc.builds(), statuses.0.into_inner().unwrap())
    }

    #[tokio::test]
    async fn a_failed_build_goes_red_with_its_last_line_and_links_to_its_recorded_tail() {
        let tail = "evaluating\nerror: boom\n";
        let (outcome, builds, written) = recorded(
            Ok(Reply::Failed {
                error_tail: tail.to_owned(),
            }),
            FakeDoc::default(),
            Some("https://swarm.example/"),
        )
        .await;
        assert!(matches!(&outcome, Outcome::Failed(e) if e.contains("error: boom")));
        let [build] = builds.as_slice() else {
            panic!("one build recorded: {builds:?}")
        };
        assert_eq!(build.repo, "agent-configs/atlas");
        assert_eq!((build.pr, build.head_sha.as_str()), (3, "abc"));
        assert_eq!(build.outcome, BuildOutcome::Failed);
        assert_eq!(build.error_tail.as_deref(), Some(tail));
        assert_eq!((&build.out_path, &build.detail), (&None, &None));
        assert!(build.id.starts_with("atlas-abc-"), "{}", build.id);
        assert!(build.finished_unix >= build.started_unix);
        let link = format!("https://swarm.example/jobs?run={}", build.id);
        assert_eq!(
            written,
            [
                (
                    CommitStatusState::Pending,
                    "building on the swarm builder".to_owned(),
                    None
                ),
                (
                    CommitStatusState::Failure,
                    "error: boom".to_owned(),
                    Some(link)
                ),
            ]
        );
    }

    #[tokio::test]
    async fn a_built_config_and_a_missing_builder_are_recorded_and_linked_too() {
        let (_, builds, written) = recorded(
            Ok(Reply::Built {
                out_path: "/nix/store/x".to_owned(),
            }),
            FakeDoc::default(),
            Some(BASE),
        )
        .await;
        assert_eq!(builds[0].outcome, BuildOutcome::Built);
        assert_eq!(builds[0].out_path.as_deref(), Some("/nix/store/x"));
        assert!(written[1].2.is_some(), "{written:?}");

        let (_, builds, written) = recorded(
            Err("no swarm builder".to_owned()),
            FakeDoc::default(),
            Some(BASE),
        )
        .await;
        assert_eq!(builds[0].outcome, BuildOutcome::Error);
        assert_eq!(builds[0].detail.as_deref(), Some("no swarm builder"));
        assert_eq!(written[1].0, CommitStatusState::Error);
        assert!(written[1].2.is_some(), "{written:?}");
    }

    /// No link to a build that was not stored, nor without the swarm's URL;
    /// the status is written either way.
    #[tokio::test]
    async fn a_status_links_only_to_a_stored_build_under_a_known_url() {
        let failed = || {
            Ok(Reply::Failed {
                error_tail: "error: boom".to_owned(),
            })
        };
        let (_, builds, written) = recorded(failed(), FakeDoc::failing(), Some(BASE)).await;
        assert!(builds.is_empty());
        assert_eq!(
            written.last(),
            Some(&(CommitStatusState::Failure, "error: boom".to_owned(), None))
        );

        let (_, builds, written) = recorded(failed(), FakeDoc::default(), None).await;
        assert_eq!(builds.len(), 1, "recorded all the same");
        assert_eq!(written.last().map(|w| &w.2), Some(&None));
    }

    /// A superseded build that finishes after the newer head's writes only on
    /// its own commit, so the newer head keeps its status.
    #[tokio::test]
    async fn a_late_superseded_build_writes_only_on_its_own_commit() {
        struct ByRevQueue;
        impl BuildSink for ByRevQueue {
            async fn build(&self, request: Vec<u8>) -> Result<Reply, String> {
                let sent: BuildAgent = serde_json::from_slice(&request).expect("a build request");
                Ok(if sent.rev == "old" {
                    Reply::Failed {
                        error_tail: "error: boom".to_owned(),
                    }
                } else {
                    Reply::Built {
                        out_path: "/nix/store/new".to_owned(),
                    }
                })
            }
        }
        #[derive(Default)]
        struct ByRev(Mutex<Vec<(String, CommitStatusState)>>);
        impl StatusSink for ByRev {
            async fn set(
                &self,
                _: &str,
                rev: &str,
                state: CommitStatusState,
                _: &str,
                _: Option<&str>,
            ) -> anyhow::Result<()> {
                self.0.lock().unwrap().push((rev.to_owned(), state));
                Ok(())
            }
        }

        let statuses = ByRev::default();
        let at = |rev: &str| BuildAgent {
            rev: rev.to_owned(),
            ..job()
        };
        run(
            Some(&ByRevQueue),
            Some(&statuses),
            &store(),
            None,
            at("new"),
        )
        .await;
        run(
            Some(&ByRevQueue),
            Some(&statuses),
            &store(),
            None,
            at("old"),
        )
        .await;
        let written = statuses.0.into_inner().unwrap();
        let last_on = |rev: &str| {
            written
                .iter()
                .rev()
                .find(|(r, _)| r == rev)
                .map(|(_, s)| *s)
        };
        assert_eq!(last_on("new"), Some(CommitStatusState::Success));
        assert_eq!(last_on("old"), Some(CommitStatusState::Failure));
    }

    #[tokio::test]
    async fn no_builder_is_an_error_status_not_a_failure() {
        let (outcome, states) =
            statuses_after(Some(FakeQueue(Err("no swarm builder".to_owned())))).await;
        assert!(matches!(outcome, Outcome::Failed(_)));
        assert_eq!(
            states,
            [CommitStatusState::Pending, CommitStatusState::Error]
        );
    }

    #[tokio::test]
    async fn no_queue_is_an_error_status() {
        let (outcome, states) = statuses_after(None).await;
        assert!(matches!(&outcome, Outcome::Failed(e) if e.contains("no swarm queue")));
        assert_eq!(states, [CommitStatusState::Error]);
    }

    #[tokio::test]
    async fn no_forge_fails_the_node_without_sending_the_build() {
        struct Unreachable;
        impl BuildSink for Unreachable {
            async fn build(&self, _: Vec<u8>) -> Result<Reply, String> {
                panic!("a build with nowhere to report must not be sent")
            }
        }
        let outcome = run(
            Some(&Unreachable),
            None::<&FakeStatuses>,
            &store(),
            None,
            job(),
        )
        .await;
        assert!(matches!(&outcome, Outcome::Failed(e) if e.contains("no forge access")));
    }

    /// The queue half against a real server: no builder listening names the
    /// option, and a listener's reply comes back decoded.
    #[tokio::test]
    async fn a_build_reaches_a_listening_builder_and_fails_without_one() {
        use futures_util::StreamExt as _;

        let nats = crate::test_nats::TestNats::start();
        let client = nats.client().await;
        let unanswered = client.build(b"{}".to_vec()).await;
        assert!(
            matches!(&unanswered, Err(e) if e.contains("deploy.builder.enable")),
            "{unanswered:?}"
        );

        let builder = nats.client().await;
        let mut requests = builder
            .queue_subscribe(SUBJECT, swarm_wire::builder::QUEUE_GROUP.to_owned())
            .await
            .expect("subscribe");
        // A message the builder sends itself and receives back proves the
        // server processed the subscription before it.
        let mut probe = builder
            .subscribe("builder-test.probe")
            .await
            .expect("subscribe");
        builder
            .publish("builder-test.probe", "".into())
            .await
            .expect("publish");
        probe.next().await.expect("the probe comes back");
        let built = Reply::Built {
            out_path: "/nix/store/x".to_owned(),
        };
        let answer = built.clone();
        tokio::spawn(async move {
            let msg = requests.next().await.expect("a request");
            let reply = serde_json::to_vec(&answer).expect("encodes");
            builder
                .publish(msg.reply.expect("a reply subject"), reply.into())
                .await
                .expect("publish");
        });
        assert_eq!(client.build(b"{}".to_vec()).await, Ok(built));
    }
}