//! `POST /api/agents/{name}/custom`: turn an archetype agent into a custom //! one, with its own `agent-configs/` repo. //! //! The repo's `flake.nix` declares the archetype's source flake as an input, at //! the URL `internal/meta` holds for that source, and imports the archetype's //! module from it. Nothing is copied: any source `internal/meta` has a pinned //! input for works. The URL's `rev` is the only pin, since the seeded repo has //! no lock. The meta flake points the agent's `hyperhive` input at its own, so //! the lock holds one hyperhive. From then on the repo's owner keeps the config //! current. The job graph is [`crate::jobq::declare_custom_job`]: repo and seed //! first, the archetype key deleted only after the seed succeeded, then the //! meta update and the deploy. A failure before the delete leaves the agent //! archetype-built and working, and the same request can be sent again. A //! failure after it leaves the agent custom, and the same request then only //! re-queues the meta update and the deploy. use axum::Json; use axum::extract::{Path, State}; use axum::http::StatusCode; use problem_details::ProblemDetails; use serde::Serialize; use swarm_queue_client::agent_archetype::ArchetypeRef; use utoipa::ToSchema; use crate::routes::agents::{claiming_hives, read_declarations}; use crate::routes::{AppState, error_problem}; use swarm_wire::hive_status::{Capabilities, NO_CONFIG_CHECKOUT}; use crate::status::{Freshness, HiveStatus}; use crate::{forge, meta_flake, schedules}; /// Answer of a queued conversion. #[derive(Serialize, ToSchema)] pub struct ConvertResponse { /// A node of the queued graph; `/api/jobq/graph` serves the rest. node_id: u64, } /// Turn an archetype agent into a custom agent. #[utoipa::path( post, path = "/api/agents/{name}/custom", params(("name" = String, Path, description = "archetype agent to turn into a custom one")), responses( (status = 200, description = "the conversion is queued", body = ConvertResponse), (status = 400, description = "the agent name is not shaped like an identifier (problem+json)", body = String), (status = 409, description = "the agent is placed on no hive or on several, or its hive has not reported the capability a converted agent's rebuild needs (problem+json)", body = String), (status = 503, description = "no swarm queue is configured here or it is not connected (problem+json)", body = String), (status = 500, description = "the queue or the job graph failed (problem+json)", body = String), ), tag = "agents" )] pub(crate) async fn post_agent_custom( State(state): State, Path(agent): Path, ) -> Result, ProblemDetails> { let agent = hive_types::Ident::parse(&agent) .map_err(|reason| error_problem(StatusCode::BAD_REQUEST, reason))? .into_string(); let client = schedules::queue(&state).map_err(schedules::refused)?; let internal = |e: anyhow::Error| error_problem(StatusCode::INTERNAL_SERVER_ERROR, &format!("{e:#}")); let store = swarm_queue_client::agent_archetype::open_or_create(&client) .await .map_err(|e| internal(e.into()))?; let (archetype, _, _) = schedules::read_or_default::>(&store, &agent) .await .map_err(internal)?; let declared = read_declarations(&state, state.hives.iter()) .await .map_err(|(hive, e)| { internal(e.context(format!("read where hive {hive} places agents"))) })?; let Some(status) = state.status.as_ref() else { return Err(error_problem( StatusCode::SERVICE_UNAVAILABLE, "no swarm queue is configured on this host", )); }; let rows = status .view(&state.hives, std::time::SystemTime::now()) .await .map_err(internal)?; let (archetype, hive) = plan(&agent, archetype, &declared, &rows) .map_err(|why| error_problem(StatusCode::CONFLICT, &why))?; let queued = { let mut sched = state .jobq .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); sched.insert_job(None, |b| { crate::jobq::declare_custom_job(b, &agent, &hive, archetype.as_ref()) }) }; let ids = queued.map_err(|e| internal(anyhow::anyhow!("{e}")))?; let [id] = ids[..] else { unreachable!("exactly one handle was asked for"); }; tracing::info!(%agent, %hive, ?archetype, "turn into custom queued"); Ok(Json(ConvertResponse { node_id: id.get() })) } /// What the request does, or why it is refused: `archetype` is the agent's /// stored one, and `None` for a custom agent, which a request only re-deploys /// (the retry of a conversion whose build failed after the key was deleted). /// The agent's hive is the one hive that places it, and must pass /// [`offers_no_config_checkout`]. fn plan( agent: &str, archetype: Option, declared: &[(String, swarm_wire::wanted::HiveWanted)], rows: &[HiveStatus], ) -> Result<(Option, String), String> { let hive = match claiming_hives(agent, declared).as_slice() { [hive] => hive.clone(), other => { return Err(format!( "agent {agent:?} is placed on {} hives, so its hive is unknown", other.len() )); } }; offers_no_config_checkout(rows, &hive)?; Ok((archetype, hive)) } /// Whether `hive`'s latest status advertises [`NO_CONFIG_CHECKOUT`]. An older /// hive keeps a checkout of a custom agent's config and would create it for a /// converted agent under the running container's mount, so a conversion waits /// for this. /// /// # Errors /// The reason, naming the hive: it has no recent status, or its status does /// not list it. An older hive lists no capabilities at all. fn offers_no_config_checkout(rows: &[HiveStatus], hive: &str) -> Result<(), String> { let capability = NO_CONFIG_CHECKOUT; let fresh = rows .iter() .find(|row| row.name == hive && row.freshness == Freshness::Fresh); let Some(snapshot) = fresh.and_then(|row| row.snapshot.as_ref()) else { return Err(format!( "hive {hive:?} has no recent status, so whether it can rebuild a converted agent is \ unknown; retry once it reports" )); }; if Capabilities::of_snapshot(snapshot).has(capability) { return Ok(()); } Err(format!( "hive {hive:?} does not advertise the {capability:?} capability, so a converted agent \ would get a config checkout on it; deploy this release to hive {hive:?} first" )) } /// What [`seed`] reads and writes on the forge: a fake in the tests. pub(crate) trait SeedForge { /// `internal/meta`'s `flake.nix` at its head. fn meta_flake(&self) -> impl Future>> + Send; /// `path` on the default branch of `agent`'s config repo; `None` when the /// repo is empty or has no such file. fn config_file( &self, agent: &str, path: &str, ) -> impl Future>> + Send; /// Commit `files` to the default branch of `agent`'s config repo. fn create_files( &self, agent: &str, files: &[(&str, String)], ) -> impl Future> + Send; } impl SeedForge for forge::Client { async fn meta_flake(&self) -> anyhow::Result> { let head = self .default_branch_files( forge::objects::SHARED_ORG, meta_flake::META_REPO, &["flake.nix"], ) .await?; Ok(head .and_then(|h| h.files.into_iter().next()) .map(|f| String::from_utf8_lossy(&f.bytes).into_owned())) } async fn config_file(&self, agent: &str, path: &str) -> anyhow::Result> { let head = self .default_branch_files(forge::CONFIG_ORG, agent, &[path]) .await?; Ok(head .and_then(|h| h.files.into_iter().next()) .map(|f| String::from_utf8_lossy(&f.bytes).into_owned())) } async fn create_files(&self, agent: &str, files: &[(&str, String)]) -> anyhow::Result<()> { self.create_config_files(agent, files, "swarm-controller: turn into custom") .await } } /// A nix identifier, so the names can be written as attribute names unquoted. fn is_nix_ident(name: &str) -> bool { let mut chars = name.chars(); chars .next() .is_some_and(|c| c.is_ascii_lowercase() || c == '_') && chars.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_' || c == '-') } /// The `flake.nix` of a converted agent: the shape every config repo has, plus /// the archetype's source as an input at `url`, and the archetype's module /// among the imports. The module is imported here, from `inputs`, because a /// module cannot take its own imports from `_module.args`. fn converted_flake_nix(archetype: &ArchetypeRef, url: &str) -> String { let ArchetypeRef { source, name } = archetype; format!( "{{\n description = \"hyperhive agent\";\n inputs = {{\n {source}.url = \"{url}\";\n }};\n outputs =\n {{ self, ... }}@inputs:\n {{\n nixosModules.default = {{\n imports = [\n ./agent.nix\n inputs.{source}.hyperhiveArchetypes.{name}.module\n ];\n _module.args.flakeInputs = builtins.removeAttrs inputs [ \"self\" ];\n }};\n }};\n}}\n" ) } /// The `agent.nix` of a converted agent: an empty module for the owner's edits. fn converted_agent_nix(agent: &str, archetype: &ArchetypeRef) -> String { let ArchetypeRef { source, name } = archetype; format!( "{{ config, pkgs, lib, ... }}:\n{{\n # Per-agent overrides for {agent}. flake.nix imports the {source} archetype\n # {name}; keeping that input current is up to this repo's owner.\n}}\n" ) } /// Seed `agent`'s config repo so it builds what the archetype built: its /// `flake.nix` declares the archetype's source as an input at the URL /// `internal/meta` holds for it today (a pinned rev), and imports the module. /// The job behind `SeedArchetypeConfig`. /// /// Safe to run again: a repo whose `flake.nix` is what this writes, whatever /// URL it holds, is left as it is, since both files go in one commit. A /// `flake.nix` that differs belongs to someone, and is an error. /// /// # Errors /// A forge failure, a name that is not a nix identifier, `internal/meta` /// holding no input for the source or one not pinned to a commit, or a different /// `flake.nix` already in the repo. pub(crate) async fn seed( forge: &impl SeedForge, agent: &str, archetype: &ArchetypeRef, ) -> anyhow::Result<()> { use anyhow::Context as _; let ArchetypeRef { source, name } = archetype; for part in [source, name] { anyhow::ensure!( is_nix_ident(part), "archetype name {part:?} is not a nix identifier" ); } let meta = forge .meta_flake() .await? .context("internal/meta has no flake.nix to read the archetype source from")?; let url = hive_meta_flake::input_url(&meta, source).with_context(|| { format!("internal/meta has no input {source:?}, so the archetype source has no URL to pin") })?; // The seeded repo has no lock, so the URL's `rev` is the only pin the input has. anyhow::ensure!( hive_meta_flake::pinned_rev(&meta, source).is_some(), "internal/meta's input {source:?} is not pinned to a commit, so the converted agent would float" ); if let Some(have) = forge.config_file(agent, "flake.nix").await? { let ours = hive_meta_flake::input_url(&have, source) .is_some_and(|held| have == converted_flake_nix(archetype, held)); anyhow::ensure!( ours, "agent-configs/{agent} already holds a different flake.nix; the conversion does not \ overwrite it" ); tracing::debug!(%agent, "turn into custom: config already seeded"); return Ok(()); } forge .create_files( agent, &[ ("agent.nix", converted_agent_nix(agent, archetype)), ("flake.nix", converted_flake_nix(archetype, url)), ], ) .await?; tracing::info!(%agent, %url, "turn into custom: seeded the config repo"); Ok(()) } /// Delete `agent`'s archetype key, so the next render and rebuild treat it as /// custom. The job behind `DropArchetype`; deleting an absent key succeeds. /// /// # Errors /// The bucket cannot be opened or written. pub(crate) async fn drop_archetype(client: &async_nats::Client, agent: &str) -> anyhow::Result<()> { swarm_queue_client::agent_archetype::open_or_create(client) .await? .delete(agent) .await?; tracing::info!(%agent, "turn into custom: archetype key deleted"); Ok(()) } #[cfg(test)] mod tests { use super::*; use std::collections::BTreeMap; use std::sync::Mutex; const URL: &str = "git+http://forge/hyperhive/hyperhive.git?rev=1111111111111111111111111111111111111111"; fn reviewer() -> ArchetypeRef { ArchetypeRef { source: "hyperhive".to_owned(), name: "reviewer".to_owned(), } } /// An in-memory forge: `internal/meta`'s flake and the files of one config /// repo. `fail_create` makes the next `create_files` fail once, without /// writing. #[derive(Default)] struct FakeForge { meta_flake: Option, repo: Mutex>, fail_create: Mutex, creates: Mutex, } impl FakeForge { fn with_inputs(inputs: &[(&str, &str)]) -> Self { use std::fmt::Write as _; let mut lines = String::new(); for (name, url) in inputs { let _ = writeln!(lines, " {name}.url = \"{url}\";"); } Self { meta_flake: Some(lines), ..Self::default() } } fn pinned() -> Self { Self::with_inputs(&[("hyperhive", URL)]) } } impl SeedForge for FakeForge { async fn meta_flake(&self) -> anyhow::Result> { Ok(self.meta_flake.clone()) } async fn config_file(&self, _agent: &str, path: &str) -> anyhow::Result> { Ok(self.repo.lock().unwrap().get(path).cloned()) } async fn create_files(&self, _agent: &str, files: &[(&str, String)]) -> anyhow::Result<()> { if std::mem::take(&mut *self.fail_create.lock().unwrap()) { anyhow::bail!("the forge refused the commit"); } *self.creates.lock().unwrap() += 1; let mut repo = self.repo.lock().unwrap(); for (path, content) in files { repo.insert((*path).to_owned(), content.clone()); } Ok(()) } } /// The seeded flake declares the source as an input at meta's URL and /// imports the module from it; nothing is copied into the repo. #[tokio::test] async fn the_seed_declares_the_source_as_an_input_and_imports_its_module() { let forge = FakeForge::pinned(); seed(&forge, "atlas", &reviewer()).await.expect("seeds"); let repo = forge.repo.lock().unwrap(); let flake = &repo["flake.nix"]; assert!( flake.contains(&format!(" hyperhive.url = \"{URL}\";")), "{flake}" ); assert!( flake.contains("inputs.hyperhive.hyperhiveArchetypes.reviewer.module"), "{flake}" ); assert!(flake.contains("_module.args.flakeInputs"), "{flake}"); assert_eq!( repo.keys().map(String::as_str).collect::>(), ["agent.nix", "flake.nix"] ); } /// A source other than hyperhive takes the URL `internal/meta` holds for /// that name, and one meta has no input for is an error. #[tokio::test] async fn another_source_takes_its_url_from_meta_and_an_unknown_one_fails() { let other = ArchetypeRef { source: "extra".to_owned(), name: "builder".to_owned(), }; let forge = FakeForge::with_inputs(&[ ("hyperhive", URL), ( "extra", "git+http://x/y.git?rev=2222222222222222222222222222222222222222", ), ]); seed(&forge, "atlas", &other).await.expect("seeds"); let flake = forge.repo.lock().unwrap()["flake.nix"].clone(); assert!( flake.contains(" extra.url = \"git+http://x/y.git?rev=2222"), "{flake}" ); assert!( flake.contains("inputs.extra.hyperhiveArchetypes.builder.module"), "{flake}" ); let why = format!( "{:#}", seed(&FakeForge::pinned(), "atlas", &other) .await .unwrap_err() ); assert!(why.contains("no input \"extra\""), "{why}"); } /// A seed that failed wrote nothing, and sending it again completes; a /// third run, after the repo is seeded, commits nothing more — even when /// meta's pin has moved since. The presence control: the second run did /// write. #[tokio::test] async fn a_seed_that_failed_is_retried_and_a_seeded_repo_is_left_alone() { let forge = FakeForge::pinned(); *forge.fail_create.lock().unwrap() = true; seed(&forge, "atlas", &reviewer()) .await .expect_err("refused"); assert!(forge.repo.lock().unwrap().is_empty(), "a failed seed wrote"); seed(&forge, "atlas", &reviewer()).await.expect("retry"); assert_eq!(*forge.creates.lock().unwrap(), 1); let moved = FakeForge { repo: Mutex::new(forge.repo.lock().unwrap().clone()), ..FakeForge::with_inputs(&[( "hyperhive", "git+http://forge/h.git?rev=3333333333333333333333333333333333333333", )]) }; seed(&moved, "atlas", &reviewer()).await.expect("again"); assert_eq!( *moved.creates.lock().unwrap(), 0, "a seeded repo was committed to again" ); } /// Someone else's `flake.nix` is never overwritten. #[tokio::test] async fn a_different_flake_nix_is_an_error() { let forge = FakeForge::pinned(); forge .repo .lock() .unwrap() .insert("flake.nix".to_owned(), "{ }".to_owned()); let why = format!( "{:#}", seed(&forge, "atlas", &reviewer()).await.unwrap_err() ); assert!(why.contains("different flake.nix"), "{why}"); assert_eq!(*forge.creates.lock().unwrap(), 0); } #[tokio::test] async fn a_missing_meta_flake_or_a_bad_name_is_an_error() { let forge = FakeForge::default(); seed(&forge, "atlas", &reviewer()) .await .expect_err("no meta flake"); let bad = ArchetypeRef { source: "hyperhive".to_owned(), name: "9; evil".to_owned(), }; seed(&FakeForge::pinned(), "atlas", &bad) .await .expect_err("bad name"); } fn row(name: &str, freshness: Freshness, snapshot: Option) -> HiveStatus { HiveStatus { name: name.to_owned(), domain: None, dashboard: None, freshness, last_seen_unix: None, age_seconds: None, snapshot, } } /// The rollout gate: a hive listing the capability is allowed; one that /// lists others, lists none (an older hive's payload), has gone quiet or /// never reported is refused, naming the hive. #[test] fn only_a_fresh_hive_listing_the_capability_may_convert() { let with = serde_json::json!({"status": "ok", "warnings": [], "capabilities": ["no-config-checkout"]}); let old = serde_json::json!({"status": "ok", "warnings": []}); let other = serde_json::json!({"status": "ok", "warnings": [], "capabilities": ["x"]}); let rows = |hive: Freshness, snapshot| { vec![ row("alpha", hive, Some(snapshot)), row("beta", Freshness::Fresh, Some(with.clone())), ] }; assert_eq!( offers_no_config_checkout(&rows(Freshness::Fresh, with.clone()), "alpha"), Ok(()) ); for refused in [old.clone(), other] { let why = offers_no_config_checkout(&rows(Freshness::Fresh, refused), "alpha").unwrap_err(); assert!( why.contains("\"alpha\"") && why.contains("no-config-checkout"), "{why}" ); } let stale = offers_no_config_checkout(&rows(Freshness::Stale, with.clone()), "alpha").unwrap_err(); assert!(stale.contains("no recent status"), "{stale}"); let never = vec![row("alpha", Freshness::NeverReported, None)]; offers_no_config_checkout(&never, "alpha").unwrap_err(); offers_no_config_checkout(&[], "alpha").unwrap_err(); } /// An old-shaped payload, without `capabilities`, still parses as the /// opaque snapshot the status reader serves. #[test] fn an_old_shaped_payload_still_parses_and_lists_nothing() { let old: serde_json::Value = serde_json::from_str(r#"{"status":"ok","warnings":[]}"#).unwrap(); assert!(old["capabilities"].as_array().is_none()); let why = offers_no_config_checkout(&[row("alpha", Freshness::Fresh, Some(old))], "alpha") .unwrap_err(); assert!(why.contains("alpha"), "{why}"); } fn placing(hive: &str, agent: &str) -> Vec<(String, swarm_wire::wanted::HiveWanted)> { let wanted = swarm_wire::wanted::HiveWanted { agents: BTreeMap::from([( agent.to_owned(), swarm_wire::wanted::AgentWanted { state: swarm_wire::wanted::AgentState::Up, }, )]), }; vec![(hive.to_owned(), wanted)] } fn reporting(hive: &str, capabilities: &[&str]) -> Vec { let snapshot = serde_json::json!({"status": "ok", "warnings": [], "capabilities": capabilities}); vec![row(hive, Freshness::Fresh, Some(snapshot))] } /// The refusals and the outcomes of a request, decided before anything is /// queued: an agent no hive places and a hive that doesn't list the /// capability are refused; an archetype agent converts, and a custom agent /// (its key already gone) is only re-deployed. The conversion is the /// control that the refusals' inputs get through otherwise. #[test] fn a_request_is_refused_for_an_unplaced_agent_or_an_old_hive() { let placed = placing("alpha", "atlas"); let ready = reporting("alpha", &["no-config-checkout"]); assert_eq!( plan("atlas", Some(reviewer()), &placed, &ready), Ok((Some(reviewer()), "alpha".to_owned())) ); assert_eq!( plan("atlas", None, &placed, &ready), Ok((None, "alpha".to_owned())) ); let unplaced = plan("atlas", Some(reviewer()), &[], &ready).unwrap_err(); assert!(unplaced.contains("0 hives"), "{unplaced}"); let old = plan("atlas", Some(reviewer()), &placed, &reporting("alpha", &[])).unwrap_err(); assert!(old.contains("\"alpha\""), "{old}"); // The retry of a custom agent waits for the capability too. plan("atlas", None, &placed, &reporting("alpha", &[])).unwrap_err(); } /// A source `internal/meta` does not pin to a commit would float in a repo /// with no lock, so the seed refuses it and writes nothing. #[tokio::test] async fn a_source_that_meta_does_not_pin_is_not_seeded() { let other = ArchetypeRef { source: "extra".to_owned(), name: "builder".to_owned(), }; let forge = FakeForge::with_inputs(&[("hyperhive", URL), ("extra", "git+http://x/y.git")]); let why = format!("{:#}", seed(&forge, "atlas", &other).await.unwrap_err()); assert!(why.contains("not pinned"), "{why}"); assert!(forge.repo.lock().unwrap().is_empty()); } }