From 22e2d30581c3d0137947e6e2c2ec24b13f63f6a3 Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 24 Oct 2026 14:29:12 +0200 Subject: [PATCH] pin internal/meta's hyperhive on the controller's rev and rebuild agents when it moves The swarm controller offers its own running rev (services.hyperhive.c0re. hyperhiveRev) as the hyperhive pin of internal/meta and takes it only when the forge compare API shows it descends from the committed pin; merges to hyperhive main no longer move the pin, and the first commit pins main's head. When a commit moves the pin, every placed agent gets a TriggerDeploy. A deploy after a meta update no longer waits on it succeeding (after_any), so a down builder does not hold back hives that render locally. --- AGENTS.md | 3 +- docs/swarm/builder.md | 36 +- hive-meta-flake/src/lib.rs | 132 ++++++ nix/host-modules/hive-c0re/options.nix | 7 +- nix/host-modules/swarm-controller.nix | 3 + swarm-controller/src/forge/meta_repo.rs | 90 +++- swarm-controller/src/jobq.rs | 364 +++++++++++++--- swarm-controller/src/meta_flake.rs | 530 ++++++++++++++++++++---- swarm-controller/src/startup.rs | 5 +- swarm-controller/src/webhook.rs | 5 - 10 files changed, 1027 insertions(+), 148 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index a80537af..62b2c7b2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -95,7 +95,8 @@ Per-hive (host + in-container): utility crate. - **`hive-metric/`** — one-shot CLI pushing a labeled metric over OTLP. - **`hive-meta-flake/`** — the meta flake's `flake.nix` text, a pure - render. hive-c0re and swarm-controller both call it. + render, and the reads of its pins. hive-c0re and swarm-controller both call + it. Wire types + socket plumbing: diff --git a/docs/swarm/builder.md b/docs/swarm/builder.md index f83c4613..6de55d57 100644 --- a/docs/swarm/builder.md +++ b/docs/swarm/builder.md @@ -125,14 +125,34 @@ lock doesn't wait for a build in progress and has ten minutes. The builder commits nothing. swarm-controller sends the request. It renders `internal/meta`'s `flake.nix` -with the same renderer hive-c0re uses (`hive-meta-flake`). The render pins -hyperhive to its forge `main`, and each custom agent that a hive places to its -`agent-configs/` `main`, all over `https`. The controller commits -`flake.nix` and the returned `flake.lock` to `internal/meta` in one commit, and -gives each hive's forge account read on the repo. It re-renders when it -starts, when hyperhive's `main` moves, before it deploys a merged config PR, -and before the first deploy of a new agent. That deploy waits for the commit. -A render equal to the committed `flake.nix` sends no lock. +with the same renderer hive-c0re uses (`hive-meta-flake`), all over `https`. +Each custom agent that a hive places pins to its `agent-configs/` +`main`. Hyperhive pins to one rev, a ratchet on the controller's own +running rev (`services.hyperhive.c0re.hyperhiveRev`, from `self.rev`): + +- the first commit pins hyperhive's forge `main`; +- later, the pin moves to the controller's rev only when the forge's compare + API shows that rev is a strict descendant of the committed pin: commits + ahead of it, and none the pin has that it lacks. The compare endpoint counts + histories with no common ancestor as zero commits ahead, so the check needs + both directions; +- an older controller, a rev on another line, a controller built from a dirty + tree or a forge that can't answer leaves the pin. A merge to hyperhive's + `main` moves nothing. + +The controller commits `flake.nix` and the returned `flake.lock` to +`internal/meta` in one commit, and gives each hive's forge account read on the +repo. It re-renders when it starts, before it deploys a merged config PR, and +before the first deploy of a new agent. Those deploys run whether or not the +update succeeded: a hive that builds from `internal/meta` does so only while it +pins the agent's current config (see +[Config changes](../agent-lifecycle/config-changes.md#the-swarms-meta-flake)). A render +equal to the committed `flake.nix` sends no lock. + +When a commit moves the pin from one rev to another, the controller sends a +deploy to every agent a hive places, so they rebuild on the new hyperhive. A +restart that moves nothing deploys nothing, and the first commit deploys +nothing. ## Credentials diff --git a/hive-meta-flake/src/lib.rs b/hive-meta-flake/src/lib.rs index d337d5ea..4d880481 100644 --- a/hive-meta-flake/src/lib.rs +++ b/hive-meta-flake/src/lib.rs @@ -421,3 +421,135 @@ pub fn agent_web_port(name: &str) -> u16 { // Modulo of a u32 by a u16's value is guaranteed < u16::MAX, so try_from never fails. WEB_PORT_BASE + u16::try_from(fnv1a(name) % u32::from(WEB_PORT_RANGE)).unwrap_or(0) } + +/// Whether `rev` is a full git commit id: 40 lowercase hex digits. +#[must_use] +pub fn is_rev(rev: &str) -> bool { + rev.len() == 40 && rev.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f')) +} + +/// The commit `input` is pinned to in a flake [`render_flake`] rendered: +/// the `` of its `.url = "git+?rev=";` line. `None` when +/// the flake has no such input, or its rev is not a commit id. +#[must_use] +pub fn pinned_rev<'a>(flake_nix: &'a str, input: &str) -> Option<&'a str> { + let prefix = format!(" {input}.url = \""); + let url = flake_nix + .lines() + .find_map(|l| l.strip_prefix(prefix.as_str()))? + .split_once('"')? + .0; + let (_, rev) = url.split_once("?rev=")?; + let rev = rev.split('&').next()?; + is_rev(rev).then_some(rev) +} + +/// Whether a flake [`render_flake`] rendered builds agent `name` from the +/// config commit `config_head`. `None` is an archetype agent, which has no +/// config input to compare. +#[must_use] +pub fn builds_agent_at(flake_nix: &str, name: &str, config_head: Option<&str>) -> bool { + let attr = format!("\n {name} = mkAgent {{ name = \"{name}\";"); + if !flake_nix.contains(&attr) { + return false; + } + config_head.is_none_or(|head| pinned_rev(flake_nix, &format!("agent-{name}")) == Some(head)) +} + +#[cfg(test)] +mod tests { + use swarm_wire::agent_archetype::ArchetypeRef; + + use super::*; + + const HYPERHIVE: &str = "1111111111111111111111111111111111111111"; + const ARGUS: &str = "2222222222222222222222222222222222222222"; + + fn spec(name: &str, archetype: Option, pin: Option) -> AgentSpec { + AgentSpec { + name: name.to_owned(), + port: agent_web_port(name), + archetype, + pin, + } + } + + fn rendered() -> String { + let url = |repo: &str, rev: &str| { + pinned_git_url(&format!("https://f.t/{repo}.git"), Some(rev)).unwrap() + }; + render_flake( + &url("hyperhive/hyperhive", HYPERHIVE), + &[ + spec( + "argus", + None, + Some(AgentPin { + url: url("agent-configs/argus", ARGUS), + canonical_inputs: vec![], + }), + ), + spec( + "rev", + Some(ArchetypeRef { + source: "hyperhive".to_owned(), + name: "reviewer".to_owned(), + }), + None, + ), + ], + &[], + false, + ) + .unwrap() + } + + #[test] + fn a_rendered_flake_names_the_commit_each_input_is_pinned_to() { + let flake = rendered(); + assert_eq!(pinned_rev(&flake, "hyperhive"), Some(HYPERHIVE)); + assert_eq!(pinned_rev(&flake, "agent-argus"), Some(ARGUS)); + assert_eq!( + pinned_rev(&flake, "agent-rev"), + None, + "an archetype has no input" + ); + assert_eq!(pinned_rev(&flake, "agent-ghost"), None); + } + + #[test] + fn a_pin_that_is_not_a_commit_id_is_no_pin() { + let flake = " hyperhive.url = \"git+https://f.t/h.git?rev=main\";\n"; + assert_eq!(pinned_rev(flake, "hyperhive"), None); + assert_eq!( + pinned_rev( + " hyperhive.url = \"git+https://f.t/h.git\";\n", + "hyperhive" + ), + None + ); + } + + #[test] + fn an_agent_is_built_only_at_the_config_head_the_flake_pins() { + let flake = rendered(); + assert!(builds_agent_at(&flake, "argus", Some(ARGUS))); + assert!( + !builds_agent_at(&flake, "argus", Some(HYPERHIVE)), + "a stale pin" + ); + assert!( + !builds_agent_at(&flake, "ghost", Some(ARGUS)), + "an agent the flake lacks" + ); + assert!(builds_agent_at(&flake, "rev", None)); + assert!(!builds_agent_at(&flake, "ghost", None)); + } + + #[test] + fn an_agent_whose_name_ends_another_s_is_not_that_agent() { + let flake = rendered(); + assert!(!builds_agent_at(&flake, "gus", Some(ARGUS))); + assert!(!builds_agent_at(&flake, "v", None)); + } +} diff --git a/nix/host-modules/hive-c0re/options.nix b/nix/host-modules/hive-c0re/options.nix index 17267da8..61c34b16 100644 --- a/nix/host-modules/hive-c0re/options.nix +++ b/nix/host-modules/hive-c0re/options.nix @@ -89,9 +89,12 @@ Git rev of the hyperhive checkout this host runs. hive-c0re pins the generated meta flake's `hyperhive` input to `git+/hyperhive/hyperhive.git?rev=`, and `nixpkgs` - follows that input. Wired by `nixosModules.default` from + follows that input. The swarm controller offers the same rev as the + `hyperhive` pin of `internal/meta`, which it takes when the rev is a + descendant of the committed pin. Wired by `nixosModules.default` from `self.rev`, which a dirty tree does not have; with `null`, - hive-c0re refuses to render the meta flake. + hive-c0re refuses to render the meta flake and the controller leaves + the `internal/meta` pin where it is. ''; }; agentBaseToplevel = lib.mkOption { diff --git a/nix/host-modules/swarm-controller.nix b/nix/host-modules/swarm-controller.nix index 93d0628e..c9933720 100644 --- a/nix/host-modules/swarm-controller.nix +++ b/nix/host-modules/swarm-controller.nix @@ -286,6 +286,9 @@ let # helper answers only `https` for the forge. git_base = "https://${forgeCfg.domain}"; github_disabled = !swarmCfg.github.enable; + # The rev this controller runs, which `internal/meta` pins hyperhive + # to once it descends from the committed pin; null on a dirty tree. + hyperhive_rev = config.services.hyperhive.c0re.hyperhiveRev; }; }; diff --git a/swarm-controller/src/forge/meta_repo.rs b/swarm-controller/src/forge/meta_repo.rs index 47d8adfa..8b5ef6d4 100644 --- a/swarm-controller/src/forge/meta_repo.rs +++ b/swarm-controller/src/forge/meta_repo.rs @@ -9,7 +9,7 @@ use forgejo_api::structs::{ }; use super::objects::SHARED_ORG; -use super::{Client, DefaultBranchFiles}; +use super::{Client, DefaultBranchFiles, is_repo_not_found}; use crate::meta_flake::{META_REPO, MetaFile, MetaForge}; impl MetaForge for Client { @@ -36,6 +36,27 @@ impl MetaForge for Client { .await } + async fn commits_ahead( + &self, + owner: &str, + repo: &str, + base: &str, + head: &str, + ) -> Result> { + let basehead = format!("{base}...{head}"); + match self + .api + .call(|api| api.repo_compare_diff(owner, repo, &basehead).into_future()) + .await + { + Ok(compare) => Ok(Some( + u64::try_from(compare.total_commits.unwrap_or(0)).unwrap_or(0), + )), + Err(e) if is_repo_not_found(&e) => Ok(None), + Err(e) => Err(e).with_context(|| format!("compare {basehead} in {owner}/{repo}")), + } + } + async fn commit_meta(&self, message: &str, files: Vec) -> Result<()> { let ops: Vec = files .into_iter() @@ -75,3 +96,70 @@ impl MetaForge for Client { Ok(()) } } + +#[cfg(test)] +mod tests { + use std::sync::{Arc, Mutex}; + + use super::*; + + /// A stand-in forge answering Forgejo's compare endpoint: `total_commits` + /// of 3 for `aaa...bbb`, 0 for `bbb...aaa`, 404 for anything else. + /// Records every request's path. + async fn compare_forge() -> (Client, Arc>>) { + use axum::http::{StatusCode, Uri}; + + let seen = Arc::new(Mutex::new(Vec::new())); + let recorded = Arc::clone(&seen); + let app = axum::Router::new().fallback(move |uri: Uri| { + let recorded = Arc::clone(&recorded); + async move { + recorded.lock().unwrap().push(uri.path().to_owned()); + match uri.path() { + "/api/v1/repos/o/r/compare/aaa...bbb" => ( + StatusCode::OK, + r#"{"total_commits":3,"commits":[],"files":[]}"#, + ), + "/api/v1/repos/o/r/compare/bbb...aaa" => { + (StatusCode::OK, r#"{"total_commits":0,"commits":[]}"#) + } + _ => (StatusCode::NOT_FOUND, r#"{"message":"not found"}"#), + } + } + }); + 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"); + (Client { api, url }, seen) + } + + #[tokio::test] + async fn commits_ahead_reads_the_compare_endpoint_as_base_dots_head() { + let (client, seen) = compare_forge().await; + assert_eq!( + client.commits_ahead("o", "r", "aaa", "bbb").await.unwrap(), + Some(3) + ); + assert_eq!( + client.commits_ahead("o", "r", "bbb", "aaa").await.unwrap(), + Some(0) + ); + assert_eq!( + *seen.lock().unwrap(), + [ + "/api/v1/repos/o/r/compare/aaa...bbb", + "/api/v1/repos/o/r/compare/bbb...aaa" + ] + ); + } + + #[tokio::test] + async fn a_commit_the_forge_lacks_is_no_count_not_an_error() { + let (client, _) = compare_forge().await; + assert_eq!( + client.commits_ahead("o", "r", "aaa", "zzz").await.unwrap(), + None + ); + } +} diff --git a/swarm-controller/src/jobq.rs b/swarm-controller/src/jobq.rs index 225d9195..a15c5f80 100644 --- a/swarm-controller/src/jobq.rs +++ b/swarm-controller/src/jobq.rs @@ -134,12 +134,12 @@ pub(crate) enum SwarmNodeKind { /// agent creation, where the hive deploys what it seeds. rev: Option, }, - /// Pin the swarm meta flake (`internal/meta`) to the forge's current - /// heads: render, lock on the swarm builder, commit. See `meta_flake`. + /// Pin the swarm meta flake (`internal/meta`): render, lock on the swarm + /// builder, commit. See `meta_flake`. When the commit moves the hyperhive + /// pin, a `TriggerDeploy` per placed agent follows. /// /// Carries nothing: it reads every head when it runs, so one pending node - /// covers every trigger queued behind it. Hives build from that repo, so a - /// deploy waits for this node. + /// covers every trigger queued behind it. UpdateMeta, } @@ -395,7 +395,7 @@ pub(crate) async fn run_swarm_node( ), Some(client) => publish_deploy(&client, &hive, &agent, rev).await, }, - SwarmNodeKind::UpdateMeta => update_meta(&deps, forge).await, + SwarmNodeKind::UpdateMeta => finish_update_meta(update_meta(&deps, forge).await, &builder), }; (builder, outcome) } @@ -565,64 +565,104 @@ pub(crate) async fn build_agent( .await } -/// The `UpdateMeta` arm, lifted out for the same reason. +/// The `UpdateMeta` arm, lifted out for the same reason: what the update +/// reported, and the placements it rendered for. `Err` is the node's failure. async fn update_meta( deps: &WorkerDeps, forge: Option>, -) -> hive_jobq::scheduler::Outcome { - use hive_jobq::scheduler::Outcome; - +) -> Result< + ( + meta_flake::Report, + std::collections::BTreeSet<(String, String)>, + ), + String, +> { let Some(client) = forge else { - return Outcome::Failed( + return Err( "no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token \ not in the swarm secret store), so internal/meta cannot be written" .to_owned(), ); }; let Some(meta) = meta_deps() else { - return Outcome::Failed(format!( + return Err(format!( "{} is unset, so there are no swarm values to render internal/meta from", meta_flake::SETTINGS_ENV )); }; let (Some(queue), Some(wanted)) = (&deps.queue, &deps.wanted) else { - return Outcome::Failed( - "no swarm queue is configured on this host, so neither the placed agents nor a \ - lock can be had" + return Err( + "no swarm queue is configured on this host, so neither the placed agents \ + nor a lock can be had" .to_owned(), ); }; - let agents = match placed_agents(queue, wanted, &meta.hives).await { - Ok(agents) => agents, - Err(e) => return Outcome::Failed(format!("{e:#}")), - }; - match forge_node( + let placements = placements(wanted, &meta.hives) + .await + .map_err(|e| format!("{e:#}"))?; + let agents = placed_agents(queue, &placements) + .await + .map_err(|e| format!("{e:#}"))?; + let report = forge_node( "update_meta", - meta_flake::update( - client.as_ref(), - queue, - &meta.settings, - &agents, - &meta.hives, - ), + meta_flake::update(client.as_ref(), queue, &meta.settings, &agents, &meta.hives), ) .await - { - Ok(updated) => { - tracing::info!(?updated, agents = agents.len(), "update_meta: done"); + .map_err(|e| format!("{e:#}"))?; + tracing::info!(?report, agents = agents.len(), "update_meta: done"); + Ok((report, placements)) +} + +/// The `UpdateMeta` node's outcome from what [`update_meta`] reported, with the +/// deploys a moved pin calls for declared into `builder` (the scheduler inserts +/// them under the node when it completes). Synchronous: the builder is not +/// `Sync`, so it cannot be held across an await. +fn finish_update_meta( + result: Result< + ( + meta_flake::Report, + std::collections::BTreeSet<(String, String)>, + ), + String, + >, + builder: &hive_jobq::builder::JobBuilder, +) -> hive_jobq::scheduler::Outcome { + use hive_jobq::scheduler::Outcome; + + match result { + Ok((report, placements)) => { + fan_out_if_moved(builder, &report, &placements); Outcome::Done } - Err(e) => Outcome::Failed(format!("{e:#}")), + Err(e) => Outcome::Failed(e), } } -/// Every agent some hive in `hives` places, with its archetype. -async fn placed_agents( - queue: &async_nats::Client, +/// Declare one rebuild of each `(hive, agent)` placement when `report` says +/// the agents' hyperhive moved: nothing else tells them to build again. +fn fan_out_if_moved( + grown: &hive_jobq::builder::JobBuilder, + report: &meta_flake::Report, + placements: &std::collections::BTreeSet<(String, String)>, +) { + if !report.hyperhive_moved { + return; + } + for (hive, agent) in placements { + let _ = grown.node(SwarmNodeKind::TriggerDeploy { + hive: hive.clone(), + agent: agent.clone(), + rev: None, + }); + } +} + +/// Every `(hive, agent)` a hive in `hives` places. +async fn placements( wanted: &wanted::WantedWriter, hives: &[String], -) -> Result> { - let mut names = std::collections::BTreeSet::new(); +) -> Result> { + let mut placed = std::collections::BTreeSet::new(); for hive in hives { let Some(declared) = wanted .view(hive) @@ -631,14 +671,24 @@ async fn placed_agents( else { continue; }; - names.extend( + placed.extend( declared .agents .into_iter() .filter(|(_, w)| crate::routes::agents::places_agent(w.state)) - .map(|(name, _)| name), + .map(|(name, _)| (hive.clone(), name)), ); } + Ok(placed) +} + +/// Every agent in `placements`, with its archetype. +async fn placed_agents( + queue: &async_nats::Client, + placements: &std::collections::BTreeSet<(String, String)>, +) -> Result> { + let names: std::collections::BTreeSet<&String> = + placements.iter().map(|(_, agent)| agent).collect(); let archetypes = swarm_queue_client::agent_archetype::open_or_create(queue) .await .map_err(|e| anyhow::anyhow!("open the agent-archetype bucket: {e}"))?; @@ -646,10 +696,13 @@ async fn placed_agents( for name in names { let (archetype, _, _) = crate::schedules::read_or_default::< Option, - >(&archetypes, &name) + >(&archetypes, name) .await .map_err(|e| e.context(format!("read the archetype of {name}")))?; - placed.push(meta_flake::Placed { name, archetype }); + placed.push(meta_flake::Placed { + name: name.clone(), + archetype, + }); } Ok(placed) } @@ -983,22 +1036,7 @@ pub(crate) async fn queue_merged_config_deploy( .jobq .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) - .insert_job(None, |b| { - // The hive builds the merged commit from `internal/meta`, so the - // flake must pin it before the hive is told. - let update_meta = b - .node(SwarmNodeKind::UpdateMeta) - .needs(SwarmResourceKind::Meta); - let root = update_meta.guid(); - let _deploy = b - .node(SwarmNodeKind::TriggerDeploy { - hive: hive.clone(), - agent: agent.clone(), - rev: Some(rev.clone()), - }) - .after_ok(update_meta); - vec![root] - }); + .insert_job(None, |b| merged_deploy_job(b, &hive, &agent, &rev)); match inserted { Ok(_) => tracing::info!(%agent, %rev, %hive, "config-pr merge: deploy queued"), Err(e) => { @@ -1007,6 +1045,33 @@ pub(crate) async fn queue_merged_config_deploy( } } +/// The sub-DAG a merged config PR is: the meta update, then the deploy of +/// `rev` on `hive`. +/// +/// `after_any`: a hive builds from `internal/meta` only while it pins the +/// agent's config commit, and renders locally otherwise, as a hive that +/// predates the repo always does. A failed update (no builder, an unreadable +/// config repo) must not hold those hives back. +fn merged_deploy_job( + b: &hive_jobq::builder::JobBuilder, + hive: &str, + agent: &str, + rev: &str, +) -> Vec { + let update_meta = b + .node(SwarmNodeKind::UpdateMeta) + .needs(SwarmResourceKind::Meta); + let root = update_meta.guid(); + let _deploy = b + .node(SwarmNodeKind::TriggerDeploy { + hive: hive.to_owned(), + agent: agent.to_owned(), + rev: Some(rev.to_owned()), + }) + .after_any(update_meta); + vec![root] +} + /// The sub-DAG one agent creation is: the nodes, and the edges between them. /// /// A function rather than a closure inside [`create_agent`](crate::routes::agents::create_agent) so the endpoint's @@ -1095,8 +1160,10 @@ pub(crate) fn declare_agent_job( hive: hive.to_owned(), agent: agent.to_owned(), }); - // The hive builds the agent from `internal/meta`, which can only carry it - // once the agent is declared on a hive and its config repo has a commit. + // The hive builds the agent from `internal/meta` once it pins the agent's + // config commit, and renders locally until then. `after_any` on the + // update, so one that failed (no builder, an unreadable config repo) + // still lets the hive build the agent that way. let update_meta = b .node(SwarmNodeKind::UpdateMeta) .needs(SwarmResourceKind::Meta) @@ -1131,7 +1198,7 @@ pub(crate) fn declare_agent_job( agent: agent.to_owned(), rev: None, }) - .after_ok(update_meta) + .after_any(update_meta) .after_any(mint_identity) .after_any(mint_forge_token) // `after_any` for the same reason: the container reads this token @@ -1907,6 +1974,187 @@ mod tests { ); } + /// Whether the edge from the `from` node to the `on` node accepts a + /// failed `on`, in a graph built by `declare`. + fn edge_accepts_failure( + declare: impl FnOnce(&hive_jobq::builder::JobBuilder), + from: &str, + on: &str, + ) -> bool { + use hive_jobq_wire::WireNode as _; + + let mut sched = hive_jobq::scheduler::Scheduler::new( + hive_jobq::Graph::new(), + hive_jobq::resources::ResourceTable::new(), + ); + sched + .insert_job(None, |b| { + declare(b); + Vec::new() + }) + .expect("the graph inserts"); + let graph = sched.graph(); + let id_of = |label: &str| { + graph + .nodes() + .find(|n| n.payload.label() == label) + .unwrap_or_else(|| panic!("the graph holds a {label} node")) + .id + }; + let (from, on) = (id_of(from), id_of(on)); + graph + .node(from) + .expect("just found") + .deps + .iter() + .find_map(|d| match d { + hive_jobq::Dep::Node { id, when } if *id == on => Some(*when), + _ => None, + }) + .expect("the edge exists") + .accepts(hive_jobq::TerminalState::Failed) + } + + /// A deploy that waits for the meta update must still run when the update + /// fails: a hive that renders locally never needed it. + #[test] + fn a_failed_meta_update_does_not_hold_back_the_deploys() { + let create = |b: &hive_jobq::builder::JobBuilder| { + let _ = super::declare_agent_job(b, "atlas", "pr1ma", false); + }; + let merge = |b: &hive_jobq::builder::JobBuilder| { + let _ = super::merged_deploy_job(b, "pr1ma", "atlas", "abc"); + }; + assert!(edge_accepts_failure( + create, + "trigger_deploy", + "update_meta" + )); + assert!(edge_accepts_failure(merge, "trigger_deploy", "update_meta")); + // The same lookup finds the edges that do hold a deploy back. + assert!(!edge_accepts_failure( + create, + "trigger_deploy", + "set_agent_wanted" + )); + assert!(!edge_accepts_failure( + create, + "update_meta", + "set_agent_wanted" + )); + } + + /// The scheduler's side of the same edge: with the update failed, the + /// deploy is the next node to run. + #[tokio::test] + async fn the_merged_deploy_runs_after_a_failed_meta_update() { + use hive_jobq::scheduler::{Outcome, Scheduler}; + use hive_jobq_wire::WireNode as _; + + let mut sched = Scheduler::new( + hive_jobq::Graph::new(), + hive_jobq::resources::ResourceTable::new(), + ); + sched + .insert_job(None, |b| { + super::merged_deploy_job(b, "pr1ma", "atlas", "abc") + }) + .expect("the graph inserts"); + let sched = std::sync::Arc::new(std::sync::Mutex::new(sched)); + let mut ran = Vec::new(); + while let Some(runner) = Scheduler::claim_next(&sched, |_id, kind, builder| async move { + let outcome = match kind { + SwarmNodeKind::UpdateMeta => Outcome::Failed("no builder".to_owned()), + _ => Outcome::Done, + }; + (builder, outcome) + }) { + runner.await.1.expect("no growth declared"); + ran.push( + sched + .lock() + .unwrap() + .graph() + .nodes() + .filter(|n| n.state.is_terminal()) + .map(|n| (n.payload.label(), n.state)) + .collect::>(), + ); + } + let last = ran.last().expect("both nodes ran"); + assert!( + last.contains(&("update_meta".to_owned(), hive_jobq::State::Failed)), + "{last:?}" + ); + assert!( + last.contains(&("trigger_deploy".to_owned(), hive_jobq::State::Done)), + "{last:?}" + ); + } + + /// Run an `UpdateMeta` node that reports `report`, and return the + /// `(hive, agent, rev)` of every deploy it grew. + async fn grown_deploys(report: meta_flake::Report) -> Vec<(String, String, Option)> { + use hive_jobq::scheduler::{Outcome, Scheduler}; + + let mut sched = Scheduler::new( + hive_jobq::Graph::new(), + hive_jobq::resources::ResourceTable::new(), + ); + let update = sched + .append(SwarmNodeKind::UpdateMeta, Vec::new(), None) + .expect("insert"); + let sched = std::sync::Arc::new(std::sync::Mutex::new(sched)); + let placements: std::collections::BTreeSet<(String, String)> = + [("pr1ma", "atlas"), ("pr1ma", "argus"), ("sec0nd", "iris")] + .into_iter() + .map(|(h, a)| (h.to_owned(), a.to_owned())) + .collect(); + let runner = Scheduler::claim_next(&sched, move |_id, _kind, builder| async move { + fan_out_if_moved(&builder, &report, &placements); + (builder, Outcome::Done) + }) + .expect("the node just inserted is runnable"); + runner.await.1.expect("the grown nodes insert"); + let guard = sched.lock().unwrap(); + guard + .graph() + .nodes() + .filter_map(|n| match &n.payload { + SwarmNodeKind::TriggerDeploy { hive, agent, rev } => { + assert_eq!(n.parent, Some(update), "grown under the update"); + Some((hive.clone(), agent.clone(), rev.clone())) + } + _ => None, + }) + .collect() + } + + #[tokio::test] + async fn a_moved_pin_rebuilds_every_placed_agent_and_an_unmoved_one_none() { + let moved = grown_deploys(meta_flake::Report { + updated: meta_flake::Updated::Committed, + hyperhive_moved: true, + }) + .await; + let mut moved: Vec<_> = moved.into_iter().collect(); + moved.sort(); + assert_eq!( + moved, + vec![ + ("pr1ma".to_owned(), "argus".to_owned(), None), + ("pr1ma".to_owned(), "atlas".to_owned(), None), + ("sec0nd".to_owned(), "iris".to_owned(), None), + ] + ); + let unmoved = grown_deploys(meta_flake::Report { + updated: meta_flake::Updated::Committed, + hyperhive_moved: false, + }) + .await; + assert!(unmoved.is_empty(), "{unmoved:?}"); + } + /// Node labels of the graph [`super::declare_agent_job`] builds, and the /// labels the deploy node waits on. fn creation_graph(archetype: bool) -> (Vec, Vec) { diff --git a/swarm-controller/src/meta_flake.rs b/swarm-controller/src/meta_flake.rs index 73143396..76ccb4db 100644 --- a/swarm-controller/src/meta_flake.rs +++ b/swarm-controller/src/meta_flake.rs @@ -1,11 +1,15 @@ -//! `internal/meta`: the swarm meta flake hives build their agents from. +//! `internal/meta`: the swarm meta flake hives build from. //! //! [`update`] renders `flake.nix` with `hive_meta_flake::render_flake`: hyperhive -//! pinned to its forge `main`, each custom agent's config input pinned to its +//! pinned by [`hyperhive_pin`], each custom agent's config input pinned to its //! `agent-configs/` `main`, every agent placed on some hive. The swarm //! builder locks it (`swarm_wire::builder::LockMeta`), and the flake and the //! lock are committed together in one commit. A render equal to the committed //! `flake.nix` sends no lock and commits nothing. +//! +//! The hyperhive pin is a ratchet on this controller's running rev: it moves +//! only to a rev that descends from the committed one, so deploying an older +//! controller never rolls the agents back. use std::collections::BTreeMap; use std::future::Future; @@ -39,6 +43,10 @@ pub(crate) struct Settings { git_base: String, /// The swarm turned the agents' GitHub integration off. github_disabled: bool, + /// The hyperhive commit this controller runs (`self.rev`); `None` for a + /// dirty tree, which leaves the pin where it is. + #[serde(default)] + hyperhive_rev: Option, } impl Settings { @@ -61,12 +69,23 @@ impl Settings { { bail!("{SETTINGS_ENV}: {unknown} is not a value the meta flake forwards"); } - if settings.vars.get("HIVE_FORGE_URL").is_none_or(String::is_empty) { + if settings + .vars + .get("HIVE_FORGE_URL") + .is_none_or(String::is_empty) + { bail!("{SETTINGS_ENV}: HIVE_FORGE_URL is unset"); } if !settings.git_base.starts_with("https://") { bail!("{SETTINGS_ENV}: git_base must be an https origin"); } + if let Some(rev) = settings + .hyperhive_rev + .as_deref() + .filter(|r| !hive_meta_flake::is_rev(r)) + { + bail!("{SETTINGS_ENV}: hyperhive_rev {rev:?} is not a full commit id"); + } Ok(settings) } @@ -110,6 +129,22 @@ pub(crate) trait MetaForge { repo: &str, paths: &[&str], ) -> impl Future>> + Send; + /// The number of commits `head` has that `base` lacks, as the forge's + /// compare endpoint counts them: `GET /repos/{owner}/{repo}/compare/ + /// {base}...{head}`, whose `total_commits` is the commits from the merge + /// base of the two to `head`. `None` when either commit is not on the + /// forge. + /// + /// ⚠️ Two histories with no common ancestor count 0, the same as a + /// `head` already contained in `base`. [`is_strict_descendant`] is the + /// reading that tells them apart. + fn commits_ahead( + &self, + owner: &str, + repo: &str, + base: &str, + head: &str, + ) -> impl Future>> + Send; fn ensure_meta_repo(&self) -> impl Future> + Send; fn grant_meta_read(&self, login: &str) -> impl Future> + Send; fn commit_meta( @@ -159,6 +194,16 @@ pub(crate) enum Updated { Committed, } +/// What [`update`] reports. +#[derive(Debug, PartialEq, Eq)] +pub(crate) struct Report { + pub(crate) updated: Updated, + /// The commit moved the hyperhive pin from one rev to another, which is + /// what the placed agents are rebuilt for. The first commit moves nothing: + /// no agent was ever told to build from this repo. + pub(crate) hyperhive_moved: bool, +} + /// Bring `internal/meta` up to the forge's current heads for `agents`, and /// give each of `hives`' forge accounts read on it. /// @@ -172,7 +217,7 @@ pub(crate) async fn update( settings: &Settings, agents: &[Placed], hives: &[String], -) -> Result { +) -> Result { forge.ensure_meta_repo().await?; for hive in hives { let login = swarm_secret_client::forge::hive_login(hive); @@ -180,7 +225,6 @@ pub(crate) async fn update( tracing::warn!(%hive, error = %format!("{e:#}"), "meta flake: read grant on internal/meta failed"); } } - let flake = render(forge, settings, agents).await?; let committed = forge .head_files( crate::forge::objects::SHARED_ORG, @@ -194,8 +238,20 @@ pub(crate) async fn update( .and_then(|c| c.files.iter().find(|f| f.path == path)) }; let flake_file = sha_of("flake.nix"); + let pin = hyperhive_pin( + forge, + settings, + flake_file + .map(|f| String::from_utf8_lossy(&f.bytes)) + .as_deref(), + ) + .await?; + let flake = render(forge, settings, &pin.rev, agents).await?; if flake_file.is_some_and(|f| f.bytes == flake.as_bytes()) && sha_of("flake.lock").is_some() { - return Ok(Updated::Unchanged); + return Ok(Report { + updated: Updated::Unchanged, + hyperhive_moved: false, + }); } let request = serde_json::to_vec(&LockMeta { flake_nix: flake.clone(), @@ -223,33 +279,116 @@ pub(crate) async fn update( ], ) .await?; - Ok(Updated::Committed) + Ok(Report { + updated: Updated::Committed, + hyperhive_moved: pin.moved, + }) } -/// Whether a `push` webhook body moved hyperhive's `main`, which the flake -/// pins. -pub(crate) fn moves_hyperhive(body: &[u8]) -> bool { - let Ok(v) = serde_json::from_slice::(body) else { - return false; +/// The hyperhive commit the flake pins, and whether that is a move off the +/// committed pin. +struct HyperhivePin { + rev: String, + moved: bool, +} + +/// The hyperhive pin for the next render, given the committed `flake.nix` +/// (`None` before the first commit). +/// +/// - No committed flake: the forge's `main` head, the one time the pin follows +/// `main`. +/// - Otherwise the committed pin, unless this controller's running rev is a +/// strict descendant of it ([`is_strict_descendant`]), which becomes the +/// pin. A merge to `main` moves nothing; a controller older than, or on +/// another line from, the pin leaves it. A forge that cannot answer the +/// comparison leaves it too, and the next update asks again. +async fn hyperhive_pin( + forge: &impl MetaForge, + settings: &Settings, + committed_flake: Option<&str>, +) -> Result { + let Some(flake) = committed_flake else { + let head = forge + .head_files(HYPERHIVE_OWNER, HYPERHIVE_REPO, &[]) + .await? + .with_context(|| format!("{HYPERHIVE_OWNER}/{HYPERHIVE_REPO} is empty or absent"))?; + return Ok(HyperhivePin { + rev: head.commit, + moved: false, + }); }; - let repo = v - .get("repository") - .and_then(|r| r.get("full_name")) - .and_then(serde_json::Value::as_str); - let git_ref = v.get("ref").and_then(serde_json::Value::as_str); - repo == Some(&format!("{HYPERHIVE_OWNER}/{HYPERHIVE_REPO}")) - && git_ref == Some("refs/heads/main") + let pinned = hive_meta_flake::pinned_rev(flake, "hyperhive") + .context("the committed internal/meta/flake.nix pins no hyperhive commit")?; + let keep = || HyperhivePin { + rev: pinned.to_owned(), + moved: false, + }; + let Some(running) = settings.hyperhive_rev.as_deref() else { + tracing::info!(%pinned, "meta flake: this controller has no rev; keeping the hyperhive pin"); + return Ok(keep()); + }; + match is_strict_descendant(forge, HYPERHIVE_OWNER, HYPERHIVE_REPO, pinned, running).await { + Ok(true) => { + tracing::info!(%pinned, %running, "meta flake: moving the hyperhive pin to this controller's rev"); + Ok(HyperhivePin { + rev: running.to_owned(), + moved: true, + }) + } + Ok(false) => { + tracing::info!(%pinned, %running, "meta flake: this controller's rev is not newer than the hyperhive pin; keeping it"); + Ok(keep()) + } + Err(e) => { + tracing::warn!(%pinned, %running, error = %format!("{e:#}"), "meta flake: cannot compare this controller's rev with the hyperhive pin; keeping it"); + Ok(keep()) + } + } } -/// The `flake.nix` for `agents` at the forge's current heads. -async fn render(forge: &impl MetaForge, settings: &Settings, agents: &[Placed]) -> Result { - let hyperhive = forge - .head_files(HYPERHIVE_OWNER, HYPERHIVE_REPO, &[]) +/// Whether `descendant` is a commit of `owner/repo` strictly after `ancestor`: +/// it has commits `ancestor` lacks ([`MetaForge::commits_ahead`]) and +/// `ancestor` has none it lacks. Equal commits, a `descendant` that is behind, +/// histories that diverged and histories with nothing in common are all +/// `false`; a commit the forge does not have is too. +async fn is_strict_descendant( + forge: &impl MetaForge, + owner: &str, + repo: &str, + ancestor: &str, + descendant: &str, +) -> Result { + if ancestor == descendant { + return Ok(false); + } + // Unrelated histories also count 0, so a zero here is `false` whatever + // the other side says. + let Some(ahead) = forge + .commits_ahead(owner, repo, ancestor, descendant) .await? - .with_context(|| format!("{HYPERHIVE_OWNER}/{HYPERHIVE_REPO} is empty or absent"))?; + else { + return Ok(false); + }; + if ahead == 0 { + return Ok(false); + } + let behind = forge + .commits_ahead(owner, repo, descendant, ancestor) + .await?; + Ok(behind == Some(0)) +} + +/// The `flake.nix` for `agents` at the forge's current config heads, with +/// hyperhive at `hyperhive_rev`. +async fn render( + forge: &impl MetaForge, + settings: &Settings, + hyperhive_rev: &str, + agents: &[Placed], +) -> Result { let hyperhive_url = hive_meta_flake::pinned_git_url( &settings.git_url(HYPERHIVE_OWNER, HYPERHIVE_REPO), - Some(&hyperhive.commit), + Some(hyperhive_rev), )?; let mut specs = Vec::with_capacity(agents.len()); for agent in agents { @@ -298,6 +437,7 @@ async fn render(forge: &impl MetaForge, settings: &Settings, agents: &[Placed]) #[cfg(test)] mod tests { + use std::collections::BTreeSet; use std::sync::Mutex; use super::*; @@ -305,13 +445,38 @@ mod tests { const FORGE: &str = "https://forge.t.local"; - fn settings() -> Settings { + /// A full commit id made of one hex digit. + fn rev(digit: char) -> String { + digit.to_string().repeat(40) + } + + /// The hyperhive history the fake forge knows: `1 <- 2 <- 3`, `2 <- 4` + /// (4 diverged from 3) and `9` on its own. `8` is not on the forge. + fn history() -> BTreeMap> { + [ + ('1', vec![]), + ('2', vec!['1']), + ('3', vec!['2']), + ('4', vec!['2']), + ('9', vec![]), + ] + .into_iter() + .map(|(c, parents)| (rev(c), parents.into_iter().map(rev).collect())) + .collect() + } + + fn settings_running(running: Option<&str>) -> Settings { + let running = running.map_or("null".to_owned(), |r| format!("\"{r}\"")); Settings::parse(&format!( - r#"{{"vars":{{"HIVE_FORGE_URL":"http://forge.t.local","HYPERHIVE_SWARM_NAME":"t"}},"git_base":"{FORGE}/","github_disabled":true}}"# + r#"{{"vars":{{"HIVE_FORGE_URL":"http://forge.t.local","HYPERHIVE_SWARM_NAME":"t"}},"git_base":"{FORGE}/","github_disabled":true,"hyperhive_rev":{running}}}"# )) .unwrap() } + fn settings() -> Settings { + settings_running(None) + } + fn head(commit: &str, files: &[(&str, &str)]) -> DefaultBranchFiles { DefaultBranchFiles { branch: "main".to_owned(), @@ -336,6 +501,28 @@ mod tests { repos: BTreeMap, commits: Mutex>, grants: Mutex>, + /// Hyperhive's commits and their parents. + history: BTreeMap>, + /// Every compare asked, as `(base, head)`. + compares: Mutex>, + /// Every compare fails, as a forge that is down does. + compare_fails: bool, + } + + impl FakeForge { + /// `commit` and everything it descends from; `None` for a commit not + /// on the forge. + fn ancestry(&self, commit: &str) -> Option> { + self.history.get(commit)?; + let mut seen = BTreeSet::new(); + let mut todo = vec![commit.to_owned()]; + while let Some(c) = todo.pop() { + if seen.insert(c.clone()) { + todo.extend(self.history.get(&c).cloned().unwrap_or_default()); + } + } + Some(seen) + } } impl MetaForge for FakeForge { @@ -351,6 +538,31 @@ mod tests { h })) } + /// Forgejo's `GetCompareInfo` for `base...head`: the commits `head` + /// has beyond the merge base, and none at all when the two share no + /// ancestor. + async fn commits_ahead( + &self, + _owner: &str, + _repo: &str, + base: &str, + head: &str, + ) -> Result> { + self.compares + .lock() + .unwrap() + .push((base.to_owned(), head.to_owned())); + if self.compare_fails { + bail!("the forge is down"); + } + let (Some(base), Some(head)) = (self.ancestry(base), self.ancestry(head)) else { + return Ok(None); + }; + if base.is_disjoint(&head) { + return Ok(Some(0)); + } + Ok(Some(head.difference(&base).count() as u64)) + } async fn ensure_meta_repo(&self) -> Result<()> { Ok(()) } @@ -395,9 +607,10 @@ mod tests { } } + /// A forge whose hyperhive `main` is at commit `1`, with no `internal/meta`. fn forge_with(extra: &[(&str, DefaultBranchFiles)]) -> FakeForge { let mut repos = BTreeMap::new(); - repos.insert("hyperhive/hyperhive".to_owned(), head("hh0", &[])); + repos.insert("hyperhive/hyperhive".to_owned(), head(&rev('1'), &[])); repos.insert( "agent-configs/argus".to_owned(), head( @@ -413,10 +626,23 @@ mod tests { } FakeForge { repos, + history: history(), ..FakeForge::default() } } + /// A forge with `internal/meta` committed, pinning hyperhive at `pin`. + async fn forge_pinned_at(pin: &str) -> FakeForge { + let base = forge_with(&[]); + let flake = render(&base, &settings(), pin, &argus_and_reviewer()) + .await + .unwrap(); + forge_with(&[( + "internal/meta", + head("m1", &[("flake.nix", &flake), ("flake.lock", "{}")]), + )]) + } + fn argus_and_reviewer() -> Vec { vec![ Placed { @@ -433,6 +659,23 @@ mod tests { ] } + /// The hyperhive commit the one lock sent to `locker` pins. + fn locked_pin(locker: &FakeLocker) -> String { + let sent = locker.sent.lock().unwrap(); + let [flake] = sent.as_slice() else { + panic!("one lock: {sent:?}") + }; + hive_meta_flake::pinned_rev(flake, "hyperhive") + .expect("a hyperhive pin") + .to_owned() + } + + async fn run(forge: &FakeForge, locker: &FakeLocker, settings: &Settings) -> Report { + update(forge, locker, settings, &argus_and_reviewer(), &[]) + .await + .unwrap() + } + #[tokio::test] async fn a_new_render_is_locked_and_committed_with_its_lock_in_one_commit() { let forge = forge_with(&[]); @@ -446,19 +689,29 @@ mod tests { ) .await .unwrap(); - assert_eq!(got, Updated::Committed); + assert_eq!( + got, + Report { + updated: Updated::Committed, + hyperhive_moved: false + } + ); let sent = locker.sent.lock().unwrap(); let [flake] = sent.as_slice() else { panic!("one lock: {sent:?}") }; assert!(flake.contains(&format!( - "hyperhive.url = \"git+{FORGE}/hyperhive/hyperhive.git?rev=hh0\";" + "hyperhive.url = \"git+{FORGE}/hyperhive/hyperhive.git?rev={}\";", + rev('1') ))); assert!(flake.contains(&format!( "agent-argus.url = \"git+{FORGE}/agent-configs/argus.git?rev=a1\";" ))); assert!(flake.contains("agent-argus.inputs.nixpkgs.follows = \"nixpkgs\";")); - assert!(!flake.contains("agent-rev.url"), "an archetype agent has no input"); + assert!( + !flake.contains("agent-rev.url"), + "an archetype agent has no input" + ); assert!(flake.contains("services.hyperhive.agent.github.enable = false;")); assert!(flake.contains("HYPERHIVE_SWARM_NAME = \"t\";")); let commits = forge.commits.lock().unwrap(); @@ -480,35 +733,25 @@ mod tests { #[tokio::test] async fn an_unchanged_render_sends_no_lock_and_commits_nothing() { - let first = forge_with(&[]); + let forge = forge_pinned_at(&rev('1')).await; let locker = FakeLocker::locked(); - update(&first, &locker, &settings(), &argus_and_reviewer(), &[]) - .await - .unwrap(); - let flake = locker.sent.lock().unwrap()[0].clone(); - let forge = forge_with(&[( - "internal/meta", - head("m1", &[("flake.nix", &flake), ("flake.lock", "{}")]), - )]); - let locker = FakeLocker::locked(); - let got = update(&forge, &locker, &settings(), &argus_and_reviewer(), &[]) - .await - .unwrap(); - assert_eq!(got, Updated::Unchanged); + let got = run(&forge, &locker, &settings()).await; + assert_eq!(got.updated, Updated::Unchanged); + assert!(!got.hyperhive_moved); assert!(locker.sent.lock().unwrap().is_empty()); assert!(forge.commits.lock().unwrap().is_empty()); } #[tokio::test] - async fn a_moved_head_replaces_both_committed_blobs() { + async fn a_moved_config_head_replaces_both_committed_blobs() { + let base = forge_with(&[]); + let old = render(&base, &settings(), &rev('1'), &[]).await.unwrap(); let forge = forge_with(&[( "internal/meta", - head("m1", &[("flake.nix", "old"), ("flake.lock", "{}")]), + head("m1", &[("flake.nix", &old), ("flake.lock", "{}")]), )]); let locker = FakeLocker::locked(); - update(&forge, &locker, &settings(), &argus_and_reviewer(), &[]) - .await - .unwrap(); + run(&forge, &locker, &settings()).await; let commits = forge.commits.lock().unwrap(); let shas: Vec<_> = commits[0].1.iter().map(|(_, _, sha)| sha.clone()).collect(); assert_eq!( @@ -550,34 +793,179 @@ mod tests { assert!(forge.commits.lock().unwrap().is_empty()); } - #[test] - fn only_a_push_to_hyperhive_main_moves_the_pin() { - let push = |repo: &str, git_ref: &str| { - serde_json::to_vec(&serde_json::json!({ - "ref": git_ref, - "repository": { "full_name": repo }, - "commits": [], - })) - .unwrap() + #[tokio::test] + async fn the_first_commit_pins_the_forge_main_head_whatever_the_controller_runs() { + let forge = forge_with(&[]); + let locker = FakeLocker::locked(); + let got = run(&forge, &locker, &settings_running(Some(&rev('3')))).await; + assert_eq!(locked_pin(&locker), rev('1')); + assert!( + !got.hyperhive_moved, + "no agent was told to build from it yet" + ); + assert!( + forge.compares.lock().unwrap().is_empty(), + "nothing to compare with before the first commit" + ); + } + + #[tokio::test] + async fn a_controller_on_a_descendant_moves_the_pin_and_says_so() { + let forge = forge_pinned_at(&rev('2')).await; + let locker = FakeLocker::locked(); + let got = run(&forge, &locker, &settings_running(Some(&rev('3')))).await; + assert_eq!(locked_pin(&locker), rev('3')); + assert_eq!( + got, + Report { + updated: Updated::Committed, + hyperhive_moved: true + } + ); + } + + /// Each running rev that must not move its pin; the control is the test + /// above, where a descendant does. + #[tokio::test] + async fn a_controller_that_is_not_strictly_newer_leaves_the_pin() { + for (name, pin, running) in [ + ("the same commit", rev('2'), rev('2')), + ("an ancestor", rev('2'), rev('1')), + ("a diverged commit", rev('3'), rev('4')), + ("an unrelated history", rev('2'), rev('9')), + ("a commit the forge lacks", rev('2'), rev('8')), + ] { + let forge = forge_pinned_at(&pin).await; + let locker = FakeLocker::locked(); + let got = run(&forge, &locker, &settings_running(Some(&running))).await; + assert_eq!(got.updated, Updated::Unchanged, "{name}"); + assert!(!got.hyperhive_moved, "{name}"); + assert!(locker.sent.lock().unwrap().is_empty(), "{name}"); + assert!(forge.commits.lock().unwrap().is_empty(), "{name}"); + } + } + + #[tokio::test] + async fn a_merge_to_hyperhive_main_does_not_move_the_pin() { + let mut forge = forge_pinned_at(&rev('2')).await; + forge + .repos + .insert("hyperhive/hyperhive".to_owned(), head(&rev('3'), &[])); + let locker = FakeLocker::locked(); + let got = run(&forge, &locker, &settings_running(Some(&rev('2')))).await; + assert_eq!(got.updated, Updated::Unchanged); + assert!(forge.commits.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn a_controller_without_a_rev_keeps_the_pin_and_asks_nothing() { + let forge = forge_pinned_at(&rev('2')).await; + let locker = FakeLocker::locked(); + let got = run(&forge, &locker, &settings()).await; + assert_eq!(got.updated, Updated::Unchanged); + assert!(forge.compares.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn a_forge_that_cannot_compare_keeps_the_pin_and_still_commits_the_rest() { + let mut forge = forge_pinned_at(&rev('2')).await; + forge.compare_fails = true; + // A new agent makes the render differ from the committed flake. + let mut agents = argus_and_reviewer(); + agents.push(Placed { + name: "newcomer".to_owned(), + archetype: Some(ArchetypeRef { + source: "hyperhive".to_owned(), + name: "reviewer".to_owned(), + }), + }); + let locker = FakeLocker::locked(); + let got = update( + &forge, + &locker, + &settings_running(Some(&rev('3'))), + &agents, + &[], + ) + .await + .unwrap(); + assert_eq!(locked_pin(&locker), rev('2')); + assert!(!got.hyperhive_moved); + assert_eq!(got.updated, Updated::Committed); + } + + #[tokio::test] + async fn a_committed_flake_without_a_hyperhive_pin_is_an_error_not_a_first_commit() { + let forge = forge_with(&[( + "internal/meta", + head("m1", &[("flake.nix", "{ }"), ("flake.lock", "{}")]), + )]); + let err = update(&forge, &FakeLocker::locked(), &settings(), &[], &[]) + .await + .unwrap_err(); + assert!(format!("{err:#}").contains("pins no hyperhive commit")); + } + + /// The fake is Forgejo's `GetCompareInfo`, so the descendant test rests + /// on these: a `base...head` count of 0 is both "head is contained in + /// base" and "no common ancestor", and only the reverse count tells a + /// divergence from a descendant. + #[tokio::test] + async fn descent_needs_commits_ahead_one_way_and_none_the_other() { + let forge = forge_with(&[]); + let ahead = |base: char, head: char| { + let forge = &forge; + async move { + forge + .commits_ahead("o", "r", &rev(base), &rev(head)) + .await + .unwrap() + } }; - assert!(moves_hyperhive(&push("hyperhive/hyperhive", "refs/heads/main"))); - assert!(!moves_hyperhive(&push("hyperhive/hyperhive", "refs/heads/x"))); - assert!(!moves_hyperhive(&push("agents/foo", "refs/heads/main"))); - assert!(!moves_hyperhive(b"not json")); + assert_eq!(ahead('2', '3').await, Some(1)); + assert_eq!(ahead('1', '3').await, Some(2)); + assert_eq!(ahead('3', '2').await, Some(0), "contained in base"); + assert_eq!(ahead('2', '9').await, Some(0), "no common ancestor"); + assert_eq!(ahead('3', '4').await, Some(1), "diverged: both ways"); + assert_eq!(ahead('4', '3').await, Some(1), "diverged: both ways"); + assert_eq!(ahead('2', '8').await, None); + let descends = |ancestor: char, descendant: char| { + let forge = &forge; + async move { + is_strict_descendant(forge, "o", "r", &rev(ancestor), &rev(descendant)) + .await + .unwrap() + } + }; + assert!(descends('2', '3').await); + assert!(descends('1', '3').await); + assert!(!descends('3', '2').await); + assert!(!descends('2', '2').await); + assert!(!descends('3', '4').await); + assert!(!descends('2', '9').await, "unrelated histories"); + assert!(!descends('2', '8').await); } #[test] - fn settings_refuse_an_unknown_value_a_missing_forge_and_a_plain_http_base() { - let parse = |vars: &str, git_base: &str| { + fn settings_refuse_an_unknown_value_a_missing_forge_a_plain_http_base_and_a_short_rev() { + let parse = |vars: &str, git_base: &str, rev: &str| { Settings::parse(&format!( - r#"{{"vars":{vars},"git_base":"{git_base}","github_disabled":false}}"# + r#"{{"vars":{vars},"git_base":"{git_base}","github_disabled":false,"hyperhive_rev":{rev}}}"# )) }; - assert!(parse(r#"{"HIVE_FORGE_URL":"x"}"#, "https://f").is_ok()); - let unknown = parse(r#"{"HIVE_FORGE_URL":"x","HIVE_AGENT_NATS_URL":"y"}"#, "https://f") - .unwrap_err(); + let ok = r#"{"HIVE_FORGE_URL":"x"}"#; + assert!(parse(ok, "https://f", "null").is_ok()); + let full = format!("\"{}\"", rev('a')); + assert!(parse(ok, "https://f", &full).is_ok()); + let unknown = parse( + r#"{"HIVE_FORGE_URL":"x","HIVE_AGENT_NATS_URL":"y"}"#, + "https://f", + "null", + ) + .unwrap_err(); assert!(format!("{unknown:#}").contains("HIVE_AGENT_NATS_URL")); - assert!(parse("{}", "https://f").is_err()); - assert!(parse(r#"{"HIVE_FORGE_URL":"x"}"#, "http://f").is_err()); + assert!(parse("{}", "https://f", "null").is_err()); + assert!(parse(ok, "http://f", "null").is_err()); + assert!(parse(ok, "https://f", "\"main\"").is_err()); } } diff --git a/swarm-controller/src/startup.rs b/swarm-controller/src/startup.rs index c63d255c..5bb724e4 100644 --- a/swarm-controller/src/startup.rs +++ b/swarm-controller/src/startup.rs @@ -285,8 +285,9 @@ fn spawn_forge_workers( auth: Option>, queue: Option, ) { - // `internal/meta` appears, and catches up with every head that moved while - // this daemon was down. + // `internal/meta` appears, catches up with every config head that moved + // while this daemon was down, and takes this daemon's rev as the hyperhive + // pin when that is newer. if let Err(e) = crate::jobq::queue_meta_update(jobq) { tracing::warn!(error = %format!("{e:#}"), "update_meta: queueing failed"); } diff --git a/swarm-controller/src/webhook.rs b/swarm-controller/src/webhook.rs index 5f4d71ed..6ba7f25e 100644 --- a/swarm-controller/src/webhook.rs +++ b/swarm-controller/src/webhook.rs @@ -443,11 +443,6 @@ pub(super) async fn post_webhook_forge( } } DeliveryKind::VcsActivity => { - if crate::meta_flake::moves_hyperhive(&body) - && let Err(e) = crate::jobq::queue_meta_update(&state.jobq) - { - tracing::warn!(error = %format!("{e:#}"), "webhook: queueing update_meta failed"); - } if let Some(activity) = crate::vcs_metrics::parse_push(&body) { crate::vcs_metrics::record_push(&activity); } else {