hyperhive/swarm-controller/src/builder.rs

//! The `BuildAgent` node: send a config PR's head to the swarm builder over the
//! queue, and turn its answer into the [`CONTEXT`] commit status on that head.
//!
//! The status is the check's only record besides the node itself: 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.
//!
//! 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`).

use std::future::Future;

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

use crate::forge::{self, CONFIG_ORG};

/// 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,
    ) -> impl Future<Output = anyhow::Result<()>> + Send;
}

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

/// Build `job` and write its status. 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>,
    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 set = |state, description: String| {
        let job = &job;
        async move {
            statuses
                .set(&job.agent, &job.rev, state, &description)
                .await
                .map_err(|e| format!("{e:#}"))
        }
    };
    let Some(queue) = queue else {
        let error = "no swarm queue is configured on this host, so no build can be sent";
        return settle(set(CommitStatusState::Error, error.to_owned()).await, error);
    };
    if let Err(e) = set(
        CommitStatusState::Pending,
        "building on the swarm builder".to_owned(),
    )
    .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");
            match set(CommitStatusState::Success, out_path).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);
            settle(
                set(CommitStatusState::Failure, last.to_owned()).await,
                &format!("the build failed: {error_tail}"),
            )
        }
        Err(error) => settle(set(CommitStatusState::Error, error.clone()).await, &error),
    }
}

/// 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::Mutex;

    use super::*;

    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)>>);

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

    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), 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]
        );
    }

    #[tokio::test]
    async fn a_failed_build_goes_red_with_its_last_line_and_fails_the_node() {
        let statuses = FakeStatuses::default();
        let queue = FakeQueue(Ok(Reply::Failed {
            error_tail: "evaluating\nerror: boom\n".to_owned(),
        }));
        let outcome = run(Some(&queue), Some(&statuses), job()).await;
        assert!(matches!(&outcome, Outcome::Failed(e) if e.contains("error: boom")));
        let written = statuses.0.into_inner().unwrap();
        assert_eq!(
            written.last(),
            Some(&(CommitStatusState::Failure, "error: boom".to_owned()))
        );
    }

    /// 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,
            ) -> 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), at("new")).await;
        run(Some(&ByRevQueue), Some(&statuses), 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>, 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_queue_client::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));
    }
}