diff --git a/docs/agent-lifecycle/config-changes.md b/docs/agent-lifecycle/config-changes.md index 3cdaa801..3bcf8403 100644 --- a/docs/agent-lifecycle/config-changes.md +++ b/docs/agent-lifecycle/config-changes.md @@ -141,7 +141,7 @@ agent. The controller queues its update of the repo before the deploy request for a merged config commit and for a new agent, and sends the deploy only when that update succeeded and the swarm builder built the agent from the updated repo (the -[deploy gate](../swarm/builder.md#deploy-gate)). A failed build skips the deploy, and the timer builds again. +[deploy gate](../swarm/builder.md#deploy-gate)). A failed build skips the deploy, and the timer builds again. A merge whose webhook delivery never arrives gets the same update, build and deploy within five minutes, from the controller's drift pass (see the paragraph on config drift in the deploy gate section). - **Choosing the closure.** The hive doesn't choose. It deploys the closure the agent's record in `agent-closures` declares, whichever `internal/meta` commit diff --git a/docs/swarm/builder.md b/docs/swarm/builder.md index 7d4d5da3..0f370b6e 100644 --- a/docs/swarm/builder.md +++ b/docs/swarm/builder.md @@ -121,7 +121,7 @@ status has no link. can push to [a binary cache](#binary-cache). - The PR check gates nothing. A merge deploys whatever the status says. The [deploy gate](#deploy-gate) is a separate build, from `internal/meta`. -- A lost webhook delivery isn't retried: the head isn't built until the next +- A lost webhook delivery of a PR push isn't retried: the head isn't built until the next push. The jobs graph is in memory, so a controller restart drops a build in flight and leaves its status `pending`. @@ -236,6 +236,19 @@ is the same case: the old builder has no subscriber for the new subject, so the controller gets no responders at once, and the deploys go out after the builder updates. Each round adds a build and deploy node per failing agent. +A merged config PR whose webhook delivery never arrives would leave the agent +on its old config. Every five minutes, after the config-PR scan, the controller +reads `main` of each `agent-configs/` repo that has a commit and that +exactly one hive places. It compares that head with the config commit pinned for +the agent in the `flake.nix` of the `internal/meta` commit its `agent-closures` +key names, since the pin in `internal/meta`'s own head also moves for other +agents' merges. When they differ, it declares the same update, build and deploy +a merge declares, and logs at info `config drift: …`. It leaves alone an agent +with no key, a key whose flake pins no config commit for the agent, and a head +the graph already holds a deploy for, in any state: the ten-minute rounds above +build a head whose build failed again, not this pass. The graph is +in memory, so a controller restart tries each such head once more. + When the build belongs to a merged config PR and `nix` exited with an error or hit its deadline, the controller also comments once on that PR, with its forge account: the agent, the `internal/meta` commit of the first red build, `nix`'s last lines (at most 4000 diff --git a/swarm-controller/src/config_drift.rs b/swarm-controller/src/config_drift.rs new file mode 100644 index 00000000..acfdbfe2 --- /dev/null +++ b/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/` `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-` 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>; + +/// 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>> + Send; + /// Every config repo's `(agent, head commit)`. + fn heads(&self) -> impl Future>> + 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>> + 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, + 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> { + 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, + pub(crate) wanted: Arc, + pub(crate) hives: Vec, + 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) { + 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, + closures: Store, + /// `internal/meta`'s `flake.nix` by commit, for the commits read this pass. + flakes: Mutex>>, +} + +impl Live<'_> { + /// `flake.nix` of `internal/meta` at `meta_rev`; `None` when it has none. + async fn flake_at(&self, meta_rev: &str) -> Result> { + 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> { + crate::jobq::placements(&self.deps.wanted, &self.deps.hives).await + } + + async fn heads(&self) -> Result> { + self.forge.list_config_heads(CONFIG_BRANCH).await + } + + async fn built(&self, agent: &str) -> Result> { + let (record, _) = crate::schedules::read_with(&self.closures, agent, |bytes| { + Ok(Some(serde_json::from_slice::(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>>, + jobq: Arc, + } + + 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, 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> { + Ok(self.placed.clone()) + } + + async fn heads(&self) -> Result> { + Ok(self.heads.clone()) + } + + async fn built(&self, agent: &str) -> Result> { + 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)); + } +} diff --git a/swarm-controller/src/config_pr.rs b/swarm-controller/src/config_pr.rs index 70a4504b..e44082e4 100644 --- a/swarm-controller/src/config_pr.rs +++ b/swarm-controller/src/config_pr.rs @@ -17,9 +17,10 @@ //! //! [`merged`] reads the same delivery for a merge, which the webhook handler //! turns into a deploy, and [`changed_head`] reads it for a PR's new head, -//! which the handler turns into a build check (`crate::builder`). The poll -//! has no counterpart to either: a merge whose delivery is lost deploys -//! nothing, and a head whose delivery is lost is not built. +//! which the handler turns into a build check (`crate::builder`). The cache +//! poll has no counterpart to either, and a PR head whose delivery is lost is +//! not built. A merge whose delivery is lost is deployed by +//! [`crate::config_drift`], which [`spawn`] runs after each scan. use std::collections::HashMap; use std::sync::{Arc, Mutex}; @@ -267,29 +268,39 @@ impl ConfigPrCache { /// The scan runs immediately on the first tick (`tokio::time::interval`'s /// default), so the cache is populated on startup rather than staying empty /// for a full `POLL_INTERVAL` after boot. -pub fn spawn(client: Arc) -> Arc { +/// +/// With `drift`, each tick also runs [`crate::config_drift::run`] after the scan. +pub(crate) fn spawn( + client: Arc, + drift: Option, +) -> Arc { let cache = Arc::new(ConfigPrCache::new()); let cache_for_task = Arc::clone(&cache); tokio::spawn(async move { let mut ticker = tokio::time::interval(POLL_INTERVAL); loop { ticker.tick().await; - let Some(scan) = crate::forge::within_forge_deadline( + match crate::forge::within_forge_deadline( "config-pr poll", client.list_open_config_prs(), ) .await - else { - continue; - }; - match scan { - Ok(scan) => cache_for_task.replace(scan), - Err(e) => { + { + Some(Ok(scan)) => cache_for_task.replace(scan), + Some(Err(e)) => { tracing::warn!( error = %format!("{e:#}"), "config-pr poll: scan failed, cache keeps its last value" ); } + None => {} + } + if let Some(drift) = &drift { + crate::forge::within_forge_deadline( + "config drift pass", + crate::config_drift::run(drift, &client), + ) + .await; } } }); diff --git a/swarm-controller/src/forge.rs b/swarm-controller/src/forge.rs index ea78b4a0..2615003c 100644 --- a/swarm-controller/src/forge.rs +++ b/swarm-controller/src/forge.rs @@ -978,6 +978,71 @@ impl Client { Ok(out) } + /// Every config repo in [`CONFIG_ORG`] with a commit on `branch`, as + /// `(agent, head commit)`: the heads [`crate::config_drift`] compares with + /// what was built. + /// + /// Repos the listing marks `empty` are skipped without a branch call, as is + /// a repo with no `branch`. A branch read that fails is logged and skips + /// that repo only. + /// + /// # Errors + /// The repo listing failed. + pub async fn list_config_heads(&self, branch: &str) -> Result> { + let repos = self + .api + .call(|api| Box::pin(api.org_list_repos(CONFIG_ORG).all())) + .await + .with_context(|| format!("list repos in {CONFIG_ORG}"))?; + let mut heads = Vec::new(); + for repo in repos { + let Some(agent) = repo.name else { + continue; + }; + if repo.empty.unwrap_or(false) { + continue; + } + match self + .api + .call(|api| { + api.repo_get_branch(CONFIG_ORG, &agent, branch) + .into_future() + }) + .await + { + Ok(head) => { + if let Some(commit) = head.commit.and_then(|c| c.id) { + heads.push((agent, commit)); + } + } + Err(e) if is_repo_not_found(&e) => {} + Err(e) => { + tracing::warn!(%agent, error = %e, "swarm forge: reading a config head failed, skipping repo"); + } + } + } + Ok(heads) + } + + /// The bytes of `path` in `owner/repo` at `commit`; `None` when no blob is + /// there. + /// + /// # Errors + /// A forge failure; a [`ForgejoError`] 404 underneath means the repo or the + /// commit is not on the forge ([`is_repo_not_found`]). + pub async fn file_at_commit( + &self, + owner: &str, + repo: &str, + commit: &str, + path: &str, + ) -> Result>> { + let file = self + .file_at(owner, repo, commit, path, &mut HashMap::new()) + .await?; + Ok(file.map(|(_, bytes)| bytes)) + } + /// Pages a search-shaped result (`data: Option>`, no total-count /// header) to exhaustion via `fetch`, which must issue one request per /// `page` (1-based) at `page_size` and return its `data`. `repo_search`'s @@ -2844,6 +2909,57 @@ mod tests { ); } + #[tokio::test] + async fn the_config_heads_skip_empty_repos_and_repos_without_the_branch() { + use axum::http::{StatusCode, Uri}; + + let seen = Arc::new(std::sync::Mutex::new(Vec::new())); + let recorded = Arc::clone(&seen); + let app = axum::Router::new().fallback(move |uri: Uri| { + recorded.lock().unwrap().push(uri.path().to_owned()); + let (status, count, body) = match uri.path() { + "/api/v1/orgs/agent-configs/repos" => ( + 200, + "3", + r#"[{"name":"bare","empty":true},{"name":"nomain","empty":false},{"name":"full","empty":false}]"#, + ), + "/api/v1/repos/agent-configs/full/branches/main" => ( + 200, + "0", + r#"{"name":"main","commit":{"id":"abc123"}}"#, + ), + _ => (404, "0", r#"{"message":"not found"}"#), + }; + async move { + ( + StatusCode::from_u16(status).unwrap(), + [("x-total-count", count)], + body, + ) + } + }); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await }); + let api = crate::forge::api::tests::fixed(url.clone(), "stub-token"); + let client = Client { api, url }; + + let heads = client.list_config_heads("main").await.unwrap(); + + assert_eq!(heads, [("full".to_owned(), "abc123".to_owned())]); + let paths = seen.lock().unwrap(); + assert!( + paths + .iter() + .any(|p| p.ends_with("/repos/agent-configs/nomain/branches/main")) + ); + assert!( + !paths + .iter() + .any(|p| p.contains("/repos/agent-configs/bare/")) + ); + } + /// One PR commit whose parent is `mb`. const ON_MB: &str = r#"[{"sha":"c1","parents":[{"sha":"mb"}]}]"#; diff --git a/swarm-controller/src/jobq.rs b/swarm-controller/src/jobq.rs index cd21a738..5663dff2 100644 --- a/swarm-controller/src/jobq.rs +++ b/swarm-controller/src/jobq.rs @@ -1303,22 +1303,38 @@ pub(crate) async fn queue_merged_config_deploy( ); return; }; - let hive = hive.clone(); - let inserted = state - .jobq + insert_merged_deploy(&state.jobq, hive, &agent, &rev, Some(pr)); +} + +/// Insert the [`merged_deploy_job`] of `agent`'s config at `rev` on `hive`. +/// `pr` is the merged PR a failed build reports on, when there is one. Whether +/// the job was inserted. +pub(crate) fn insert_merged_deploy( + jobq: &Mutex>, + hive: &str, + agent: &str, + rev: &str, + pr: Option, +) -> bool { + let inserted = jobq .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) - .insert_job(None, |b| merged_deploy_job(b, &hive, &agent, &rev, pr)); + .insert_job(None, |b| merged_deploy_job(b, hive, agent, rev, pr)); match inserted { - Ok(_) => tracing::info!(%agent, %rev, %hive, "config-pr merge: deploy queued"), + Ok(_) => { + tracing::info!(%agent, %rev, %hive, "config deploy queued"); + true + } Err(e) => { - tracing::warn!(%agent, %rev, %hive, error = %e, "config-pr merge: queueing the deploy failed"); + tracing::warn!(%agent, %rev, %hive, error = %e, "queueing the config deploy failed"); + false } } } /// The sub-DAG a merged config PR is: the meta update, the build of the agent -/// from the updated flake, then the deploy of `rev` on `hive`. +/// from the updated flake, then the deploy of `rev` on `hive`. `pr` is the +/// merged PR, `None` for a head no PR delivery reported. /// /// `after_ok`: a hive builds only from `internal/meta`, so a deploy asked for /// after a failed update (no builder, an unreadable config repo) could only @@ -1329,13 +1345,13 @@ fn merged_deploy_job( hive: &str, agent: &str, rev: &str, - pr: u64, + pr: Option, ) -> Vec { let update_meta = b .node(SwarmNodeKind::UpdateMeta) .needs(SwarmResourceKind::Meta); let root = update_meta.guid(); - let _deploy = gated_deploy(b, hive, agent, Some(rev.to_owned()), Some(pr), Some(root)); + let _deploy = gated_deploy(b, hive, agent, Some(rev.to_owned()), pr, Some(root)); vec![root] } @@ -2501,7 +2517,7 @@ mod tests { } fn merged_dag(b: &TestBuilder) { - let _ = super::merged_deploy_job(b, "pr1ma", "atlas", "abc", 7); + let _ = super::merged_deploy_job(b, "pr1ma", "atlas", "abc", Some(7)); } fn custom_dag(b: &TestBuilder) { @@ -4215,7 +4231,9 @@ mod tests { sched .lock() .unwrap() - .insert_job(None, |b| merged_deploy_job(b, "pr1ma", "atlas", "abc", 7)) + .insert_job(None, |b| { + merged_deploy_job(b, "pr1ma", "atlas", "abc", Some(7)) + }) .expect("the graph inserts"); sched } diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 5f78dcdd..a24c5873 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -46,6 +46,7 @@ mod backend_env; mod builder; mod cache_miss; mod claude_login; +mod config_drift; mod config_pr; mod egress_allow; mod forge; diff --git a/swarm-controller/src/meta_flake.rs b/swarm-controller/src/meta_flake.rs index 4417a48e..1ad2d913 100644 --- a/swarm-controller/src/meta_flake.rs +++ b/swarm-controller/src/meta_flake.rs @@ -22,7 +22,7 @@ use swarm_wire::builder::{LOCK_REPLY_DEADLINE, LOCK_SUBJECT, LockMeta, LockReply use crate::forge::{CONFIG_ORG, DefaultBranchFiles}; /// The branch an agent's config is deployed from. -const CONFIG_BRANCH: &str = "main"; +pub(crate) const CONFIG_BRANCH: &str = "main"; /// The repo in `internal` the flake is committed to. pub(crate) const META_REPO: &str = "meta"; diff --git a/swarm-controller/src/startup.rs b/swarm-controller/src/startup.rs index 6c1ada6a..a0a7259c 100644 --- a/swarm-controller/src/startup.rs +++ b/swarm-controller/src/startup.rs @@ -30,8 +30,8 @@ use crate::routes::{self, AppState, agents, config_prs, forge_admin, hives, jobq use crate::{ agent_control, agent_custom, agent_icon, agent_input, agent_limits, agent_login, agent_permissions, agent_renewal, agent_state_stream, agent_status, agents_md, alerts, - archetypes, auth, backend_env, builder, claude_login, config_pr, egress_allow, forge, - forge_account, github_account, gitlab_account, inference, issue_report, linked_accounts, + archetypes, auth, backend_env, builder, claude_login, config_drift, config_pr, egress_allow, + forge, forge_account, github_account, gitlab_account, inference, issue_report, linked_accounts, matrix_account, message_stream, notify_flows, notify_router_api, pulls, queue_identity, review_classifier, schedules, status, subagent_term, term_stream, wanted, webhook, }; @@ -568,8 +568,14 @@ pub(crate) async fn start_forge( } }; forge::enable(url, forge_client, move |client| { + let drift = queue.clone().map(|queue| config_drift::Deps { + jobq: Arc::clone(&jobq), + wanted: Arc::new(wanted::WantedWriter::new(queue.clone())), + hives: hives.iter().map(|h| h.name.clone()).collect(), + queue, + }); spawn_forge_workers(&jobq, Arc::clone(&client), &hives, auth, queue); - config_prs.get_or_init(|| config_pr::spawn(Arc::clone(&client))); + config_prs.get_or_init(|| config_pr::spawn(Arc::clone(&client), drift)); register_swarm_webhooks(client, webhook_secret); }) .await;