queue the build of a config head that moved without a webhook build

swarm-controller/src/config_drift.rs

@@ -0,0 +1,496 @@
+//! Build the head of an agent's config repo that moved without a build.
+//!
+//! A merged config PR is built and deployed from its webhook delivery
+//! (`crate::webhook`). A delivery that never arrives leaves `main` ahead of
+//! what the agent runs, and nothing else asks for a build. Every
+//! `config_pr::POLL_INTERVAL` the config scan calls [`run`], which compares each
+//! placed agent's `agent-configs/<agent>` `main` with the config commit its
+//! latest green build used, and queues the build and deploy the webhook would
+//! have queued ([`crate::jobq::insert_merged_deploy`]) for each that differs.
+//!
+//! The config commit a build used is the pin of `agent-<agent>` in the
+//! `flake.nix` of the `internal/meta` commit its [`AgentClosure`] record names.
+//! The pin in `internal/meta`'s head is not the record: a later meta update for
+//! another agent moves every pin, built or not.
+//!
+//! Left alone:
+//! - An agent no hive places, or more than one does.
+//! - An agent with no record, or whose recorded flake pins no config commit for
+//!   it: nothing was built from its config repo to differ from.
+//! - A head whose deploy the graph holds in any state, or an agent with a deploy
+//!   in flight. A head that was tried and failed is not queued again here: the
+//!   meta timer builds it again every round (`jobq::queue_meta_update`). The
+//!   graph is in memory, so a restart tries each such head once more.
+
+use std::collections::{BTreeSet, HashMap};
+use std::future::Future;
+use std::sync::{Arc, Mutex};
+
+use anyhow::{Context as _, Result};
+use async_nats::jetstream::kv::Store;
+use swarm_wire::agent_closures::AgentClosure;
+
+use crate::forge::{self, objects::SHARED_ORG};
+use crate::jobq::{SwarmNodeKind, SwarmResourceKind};
+use crate::meta_flake::{CONFIG_BRANCH, META_REPO};
+use crate::wanted::WantedWriter;
+
+type Jobq = Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>;
+
+/// What a pass reads and does outside the process, so a test can stand in for it.
+trait World {
+    /// Every `(hive, agent)` a hive places.
+    fn placements(&self) -> impl Future<Output = Result<BTreeSet<(String, String)>>> + Send;
+    /// Every config repo's `(agent, head commit)`.
+    fn heads(&self) -> impl Future<Output = Result<Vec<(String, String)>>> + Send;
+    /// The config commit `agent`'s latest green build used; `None` when there is
+    /// no such build or it pinned none.
+    fn built(&self, agent: &str) -> impl Future<Output = Result<Option<String>>> + Send;
+    /// Whether the graph holds a deploy of `agent` at `head`, or one still
+    /// running.
+    fn attempted(&self, agent: &str, head: &str) -> bool;
+    /// Queue the build and deploy of `agent` at `head` on `hive`. Whether it was
+    /// queued.
+    fn queue(&self, hive: &str, agent: &str, head: &str) -> bool;
+}
+
+/// Whether `graph` holds a `TriggerDeploy` of `agent` that is not finished, or
+/// one at `head` in any state.
+fn attempted(
+    graph: &hive_jobq::Graph<SwarmNodeKind, SwarmResourceKind>,
+    agent: &str,
+    head: &str,
+) -> bool {
+    graph.nodes().any(|n| match &n.payload {
+        SwarmNodeKind::TriggerDeploy { agent: a, rev, .. } if a == agent => {
+            !n.state.is_terminal() || rev.as_deref() == Some(head)
+        }
+        _ => false,
+    })
+}
+
+/// One pass: each head that differs from what was built is queued once. The
+/// `(agent, head)` it queued, in order.
+///
+/// # Errors
+/// A read that failed before any agent was looked at (the placements, the
+/// heads). A failure on one agent is logged and the pass goes on.
+async fn pass(world: &impl World) -> Result<Vec<(String, String)>> {
+    let placed = world
+        .placements()
+        .await
+        .context("reading where the hives place agents")?;
+    let heads = world.heads().await.context("reading the config heads")?;
+    let mut queued = Vec::new();
+    for (agent, head) in heads {
+        let mut hives = placed.iter().filter(|(_, a)| *a == agent);
+        let (Some((hive, _)), None) = (hives.next(), hives.next()) else {
+            continue;
+        };
+        if world.attempted(&agent, &head) {
+            continue;
+        }
+        let built = match world.built(&agent).await {
+            Ok(Some(built)) => built,
+            Ok(None) => continue,
+            Err(e) => {
+                tracing::warn!(%agent, error = %format!("{e:#}"), "config drift: reading the built config commit failed");
+                continue;
+            }
+        };
+        if built == head {
+            continue;
+        }
+        tracing::info!(
+            %agent, %head, %built, %hive,
+            "config drift: agent-configs main moved to a head no build used, and no webhook build was seen; queueing it"
+        );
+        if world.queue(hive, &agent, &head) {
+            queued.push((agent, head));
+        }
+    }
+    Ok(queued)
+}
+
+/// What [`run`] needs beyond the forge client.
+pub(crate) struct Deps {
+    pub(crate) jobq: Arc<Jobq>,
+    pub(crate) wanted: Arc<WantedWriter>,
+    pub(crate) hives: Vec<String>,
+    pub(crate) queue: async_nats::Client,
+}
+
+/// One pass against the live swarm. A pass that fails is logged and the next
+/// scan tries again.
+pub(crate) async fn run(deps: &Deps, forge: &Arc<forge::Client>) {
+    let closures = match swarm_queue_client::agent_closures::open_or_create(&deps.queue).await {
+        Ok(closures) => closures,
+        Err(e) => {
+            tracing::warn!(error = %e, "config drift: opening the agent-closures bucket failed");
+            return;
+        }
+    };
+    let world = Live {
+        deps,
+        forge: Arc::clone(forge),
+        closures,
+        flakes: Mutex::default(),
+    };
+    match pass(&world).await {
+        Ok(queued) if !queued.is_empty() => {
+            tracing::info!("config drift: pass queued {} builds", queued.len());
+        }
+        Ok(_) => tracing::debug!("config drift: pass done, no drift"),
+        Err(e) => tracing::warn!(error = %format!("{e:#}"), "config drift: pass failed"),
+    }
+}
+
+/// The real world of one pass.
+struct Live<'a> {
+    deps: &'a Deps,
+    forge: Arc<forge::Client>,
+    closures: Store,
+    /// `internal/meta`'s `flake.nix` by commit, for the commits read this pass.
+    flakes: Mutex<HashMap<String, Option<String>>>,
+}
+
+impl Live<'_> {
+    /// `flake.nix` of `internal/meta` at `meta_rev`; `None` when it has none.
+    async fn flake_at(&self, meta_rev: &str) -> Result<Option<String>> {
+        if let Some(read) = self.flakes.lock().unwrap().get(meta_rev) {
+            return Ok(read.clone());
+        }
+        let flake = self
+            .forge
+            .file_at_commit(SHARED_ORG, META_REPO, meta_rev, "flake.nix")
+            .await
+            .with_context(|| {
+                format!("reading flake.nix of {SHARED_ORG}/{META_REPO} at {meta_rev}")
+            })?
+            .map(|bytes| String::from_utf8_lossy(&bytes).into_owned());
+        self.flakes
+            .lock()
+            .unwrap()
+            .insert(meta_rev.to_owned(), flake.clone());
+        Ok(flake)
+    }
+}
+
+impl World for Live<'_> {
+    async fn placements(&self) -> Result<BTreeSet<(String, String)>> {
+        crate::jobq::placements(&self.deps.wanted, &self.deps.hives).await
+    }
+
+    async fn heads(&self) -> Result<Vec<(String, String)>> {
+        self.forge.list_config_heads(CONFIG_BRANCH).await
+    }
+
+    async fn built(&self, agent: &str) -> Result<Option<String>> {
+        let (record, _) = crate::schedules::read_with(&self.closures, agent, |bytes| {
+            Ok(Some(serde_json::from_slice::<AgentClosure>(bytes)?))
+        })
+        .await?;
+        let Some(record) = record else {
+            return Ok(None);
+        };
+        let Some(flake) = self.flake_at(&record.meta_rev).await? else {
+            return Ok(None);
+        };
+        Ok(hive_meta_flake::pinned_rev(&flake, &format!("agent-{agent}")).map(str::to_owned))
+    }
+
+    fn attempted(&self, agent: &str, head: &str) -> bool {
+        attempted(
+            self.deps
+                .jobq
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .graph(),
+            agent,
+            head,
+        )
+    }
+
+    fn queue(&self, hive: &str, agent: &str, head: &str) -> bool {
+        crate::jobq::insert_merged_deploy(&self.deps.jobq, hive, agent, head, None)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::BTreeMap;
+
+    use hive_jobq::State;
+    use hive_jobq::scheduler::{Outcome, Scheduler};
+
+    use super::*;
+
+    const OLD: &str = "1111111111111111111111111111111111111111";
+    const HEAD: &str = "2222222222222222222222222222222222222222";
+    const NEWER: &str = "3333333333333333333333333333333333333333";
+
+    /// The swarm as a pass sees it. The graph is a real scheduler: what the pass
+    /// queues, and what it finds queued, go through the production code.
+    struct Fake {
+        placed: BTreeSet<(String, String)>,
+        heads: Vec<(String, String)>,
+        built: BTreeMap<String, Result<Option<String>, String>>,
+        jobq: Arc<Jobq>,
+    }
+
+    impl Fake {
+        /// `atlas` is placed on `pr1ma`, its `main` is at `HEAD` and its last
+        /// build used `OLD`.
+        fn drifted() -> Self {
+            Self {
+                placed: [("pr1ma".to_owned(), "atlas".to_owned())].into(),
+                heads: vec![("atlas".to_owned(), HEAD.to_owned())],
+                built: BTreeMap::from([("atlas".to_owned(), Ok(Some(OLD.to_owned())))]),
+                jobq: Arc::new(Mutex::new(Scheduler::new(
+                    hive_jobq::Graph::new(),
+                    hive_jobq::resources::ResourceTable::new(),
+                ))),
+            }
+        }
+
+        fn built_at(mut self, agent: &str, rev: &str) -> Self {
+            self.built
+                .insert(agent.to_owned(), Ok(Some(rev.to_owned())));
+            self
+        }
+
+        /// The `rev` of every deploy of `agent` in the graph, oldest first.
+        fn deploys(&self, agent: &str) -> Vec<(Option<String>, State)> {
+            let guard = self.jobq.lock().unwrap();
+            let mut nodes: Vec<_> = guard
+                .graph()
+                .nodes()
+                .filter_map(|n| match &n.payload {
+                    SwarmNodeKind::TriggerDeploy { agent: a, rev, .. } if a == agent => {
+                        Some((n.id, rev.clone(), n.state))
+                    }
+                    _ => None,
+                })
+                .collect();
+            nodes.sort_by_key(|(id, _, _)| *id);
+            nodes
+                .into_iter()
+                .map(|(_, rev, state)| (rev, state))
+                .collect()
+        }
+
+        /// Run every runnable node; the builds fail when `builds_fail`.
+        async fn drain(&self, builds_fail: bool) {
+            while let Some(runner) =
+                Scheduler::claim_next(&self.jobq, move |_id, kind, builder| async move {
+                    let outcome = match kind {
+                        SwarmNodeKind::BuildMeta { .. } if builds_fail => {
+                            Outcome::Failed("error: attribute missing".to_owned())
+                        }
+                        _ => Outcome::Done,
+                    };
+                    (builder, outcome)
+                })
+            {
+                runner.await.1.expect("no growth declared");
+            }
+        }
+    }
+
+    impl World for Fake {
+        async fn placements(&self) -> Result<BTreeSet<(String, String)>> {
+            Ok(self.placed.clone())
+        }
+
+        async fn heads(&self) -> Result<Vec<(String, String)>> {
+            Ok(self.heads.clone())
+        }
+
+        async fn built(&self, agent: &str) -> Result<Option<String>> {
+            match self.built.get(agent) {
+                Some(Ok(rev)) => Ok(rev.clone()),
+                Some(Err(e)) => anyhow::bail!("{e}"),
+                None => Ok(None),
+            }
+        }
+
+        fn attempted(&self, agent: &str, head: &str) -> bool {
+            attempted(self.jobq.lock().unwrap().graph(), agent, head)
+        }
+
+        fn queue(&self, hive: &str, agent: &str, head: &str) -> bool {
+            crate::jobq::insert_merged_deploy(&self.jobq, hive, agent, head, None)
+        }
+    }
+
+    fn queued(agent: &str, head: &str) -> Vec<(String, String)> {
+        vec![(agent.to_owned(), head.to_owned())]
+    }
+
+    #[tokio::test]
+    async fn a_head_that_moved_without_a_build_queues_one_build() {
+        let world = Fake::drifted();
+
+        assert_eq!(pass(&world).await.unwrap(), queued("atlas", HEAD));
+
+        assert_eq!(
+            world.deploys("atlas"),
+            [(Some(HEAD.to_owned()), State::Pending)]
+        );
+        let guard = world.jobq.lock().unwrap();
+        let hives: Vec<_> = guard
+            .graph()
+            .nodes()
+            .filter_map(|n| match &n.payload {
+                SwarmNodeKind::TriggerDeploy { hive, .. } => Some(hive.clone()),
+                _ => None,
+            })
+            .collect();
+        assert_eq!(hives, ["pr1ma"]);
+    }
+
+    #[tokio::test]
+    async fn a_head_that_was_built_queues_nothing() {
+        let world = Fake::drifted().built_at("atlas", HEAD);
+
+        assert_eq!(pass(&world).await.unwrap(), []);
+        assert_eq!(world.deploys("atlas"), []);
+    }
+
+    #[tokio::test]
+    async fn a_head_already_queued_is_not_queued_again() {
+        let world = Fake::drifted();
+        assert_eq!(pass(&world).await.unwrap(), queued("atlas", HEAD));
+
+        assert_eq!(pass(&world).await.unwrap(), []);
+
+        assert_eq!(world.deploys("atlas").len(), 1);
+    }
+
+    #[tokio::test]
+    async fn a_head_a_webhook_queued_is_not_queued_again() {
+        let world = Fake::drifted();
+        assert!(crate::jobq::insert_merged_deploy(
+            &world.jobq,
+            "pr1ma",
+            "atlas",
+            HEAD,
+            Some(7)
+        ));
+
+        assert_eq!(pass(&world).await.unwrap(), []);
+
+        assert_eq!(world.deploys("atlas").len(), 1);
+    }
+
+    #[tokio::test]
+    async fn a_head_whose_build_failed_is_not_queued_every_pass() {
+        let world = Fake::drifted();
+        assert_eq!(pass(&world).await.unwrap(), queued("atlas", HEAD));
+        world.drain(true).await;
+        assert_eq!(
+            world.deploys("atlas"),
+            [(Some(HEAD.to_owned()), State::Skipped)]
+        );
+
+        for _ in 0..3 {
+            assert_eq!(pass(&world).await.unwrap(), []);
+        }
+
+        assert_eq!(world.deploys("atlas").len(), 1);
+    }
+
+    #[tokio::test]
+    async fn a_newer_head_is_queued_after_a_failed_one() {
+        let mut world = Fake::drifted();
+        assert_eq!(pass(&world).await.unwrap(), queued("atlas", HEAD));
+        world.drain(true).await;
+        world.heads = vec![("atlas".to_owned(), NEWER.to_owned())];
+
+        assert_eq!(pass(&world).await.unwrap(), queued("atlas", NEWER));
+    }
+
+    #[tokio::test]
+    async fn a_deploy_in_flight_holds_back_a_newer_head_until_it_ends() {
+        let mut world = Fake::drifted();
+        assert_eq!(pass(&world).await.unwrap(), queued("atlas", HEAD));
+        world.heads = vec![("atlas".to_owned(), NEWER.to_owned())];
+        assert_eq!(pass(&world).await.unwrap(), []);
+
+        world.drain(false).await;
+        // The build recorded the closure from `HEAD`.
+        world
+            .built
+            .insert("atlas".to_owned(), Ok(Some(HEAD.to_owned())));
+
+        assert_eq!(pass(&world).await.unwrap(), queued("atlas", NEWER));
+    }
+
+    #[tokio::test]
+    async fn an_agent_nobody_places_or_two_hives_place_is_left_alone() {
+        let mut world = Fake::drifted();
+        world.heads = vec![
+            ("atlas".to_owned(), HEAD.to_owned()),
+            ("ghost".to_owned(), HEAD.to_owned()),
+            ("twin".to_owned(), HEAD.to_owned()),
+        ];
+        world.placed.insert(("pr1ma".to_owned(), "twin".to_owned()));
+        world.placed.insert(("hive2".to_owned(), "twin".to_owned()));
+        let world = world.built_at("ghost", OLD).built_at("twin", OLD);
+
+        assert_eq!(pass(&world).await.unwrap(), queued("atlas", HEAD));
+    }
+
+    #[tokio::test]
+    async fn an_agent_with_no_build_to_compare_is_left_alone() {
+        let mut world = Fake::drifted();
+        world.heads = vec![
+            ("atlas".to_owned(), HEAD.to_owned()),
+            ("fresh".to_owned(), HEAD.to_owned()),
+        ];
+        world
+            .placed
+            .insert(("pr1ma".to_owned(), "fresh".to_owned()));
+        world.built.insert("fresh".to_owned(), Ok(None));
+
+        assert_eq!(pass(&world).await.unwrap(), queued("atlas", HEAD));
+    }
+
+    #[tokio::test]
+    async fn an_agent_whose_build_cannot_be_read_does_not_stop_the_others() {
+        let mut world = Fake::drifted();
+        world.heads = vec![
+            ("broken".to_owned(), HEAD.to_owned()),
+            ("atlas".to_owned(), HEAD.to_owned()),
+        ];
+        world
+            .placed
+            .insert(("pr1ma".to_owned(), "broken".to_owned()));
+        world
+            .built
+            .insert("broken".to_owned(), Err("kv unreachable".to_owned()));
+
+        assert_eq!(pass(&world).await.unwrap(), queued("atlas", HEAD));
+        assert_eq!(world.deploys("broken"), []);
+    }
+
+    #[tokio::test]
+    async fn another_agents_deploy_does_not_hold_this_one_back() {
+        let mut world = Fake::drifted();
+        world.heads = vec![
+            ("atlas".to_owned(), HEAD.to_owned()),
+            ("iris".to_owned(), HEAD.to_owned()),
+        ];
+        world.placed.insert(("pr1ma".to_owned(), "iris".to_owned()));
+        let world = world.built_at("iris", OLD);
+        assert!(crate::jobq::insert_merged_deploy(
+            &world.jobq,
+            "pr1ma",
+            "iris",
+            HEAD,
+            None
+        ));
+
+        assert_eq!(pass(&world).await.unwrap(), queued("atlas", HEAD));
+    }
+}