diff --git a/Cargo.lock b/Cargo.lock index edb39e32..14982be4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4890,6 +4890,7 @@ dependencies = [ "hive-jobq-metrics", "hive-jobq-wire", "hive-log", + "hive-meta-flake", "hive-types", "hmac 0.13.0", "http", diff --git a/docs/swarm/builder.md b/docs/swarm/builder.md index 864afcfe..f83c4613 100644 --- a/docs/swarm/builder.md +++ b/docs/swarm/builder.md @@ -115,6 +115,25 @@ forge token, and answers with the JSON. An evaluation doesn't wait for a build in progress and has two minutes. What the controller does with the answer: [`archetypes.md`](archetypes.md). +## Meta-flake lock + +The builder also answers `$SWARM.builder.lock-meta`. The request carries a +rendered `flake.nix`. The builder writes it into a scratch directory, runs +`nix flake lock` there with the same forge credential helper its builds use, +answers with the `flake.lock` that `nix` wrote, and removes the directory. A +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. + ## Credentials | What | Where | Who reads it | diff --git a/hive-c0re/src/lifecycle/mod.rs b/hive-c0re/src/lifecycle/mod.rs index 579e29ec..aec3b62c 100644 --- a/hive-c0re/src/lifecycle/mod.rs +++ b/hive-c0re/src/lifecycle/mod.rs @@ -53,34 +53,7 @@ pub fn container_claude_mount(name: &str) -> String { /// willing to lose (other agents may delete them). pub const CONTAINER_SHARED_MOUNT: &str = "/shared"; -/// Web UI port range, shared by every agent including the manager. -/// Deterministic from the agent's name (FNV-1a hash mod range size), so the -/// dashboard can compute the same port without asking hive-c0re. -const WEB_PORT_BASE: u16 = 8100; -const WEB_PORT_RANGE: u16 = 900; - -/// FNV-1a hash of a string — used by `agent_web_port`. -fn fnv1a(s: &str) -> u32 { - let mut hash: u32 = 2_166_136_261; - for b in s.bytes() { - hash ^= u32::from(b); - hash = hash.wrapping_mul(16_777_619); - } - hash -} - -/// Per-agent web UI port — `WEB_PORT_BASE + FNV-1a(name) % -/// WEB_PORT_RANGE` for every agent including the manager. The port -/// allocation rule reads the same for every name; collisions are -/// possible (birthday paradox at ~30 agents) and the operator -/// resolves them by renaming an agent (different hash → different -/// port). Stable across hosts, restarts, and dashboard renders — -/// no state-file dance. -#[must_use] -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) -} +pub use hive_meta_flake::agent_web_port; /// Extract the bridge gateway IP from `HIVE_NETWORK_SUBNET`. /// diff --git a/hive-meta-flake/src/lib.rs b/hive-meta-flake/src/lib.rs index 86a1d53a..d337d5ea 100644 --- a/hive-meta-flake/src/lib.rs +++ b/hive-meta-flake/src/lib.rs @@ -392,3 +392,32 @@ pub fn render_flake( out.push_str(" };\n };\n}\n"); Ok(out) } + +/// Web UI port range, shared by every agent including the manager. +/// Deterministic from the agent's name (FNV-1a hash mod range size), so the +/// dashboard can compute the same port without asking hive-c0re. +const WEB_PORT_BASE: u16 = 8100; +const WEB_PORT_RANGE: u16 = 900; + +/// FNV-1a hash of a string — used by `agent_web_port`. +fn fnv1a(s: &str) -> u32 { + let mut hash: u32 = 2_166_136_261; + for b in s.bytes() { + hash ^= u32::from(b); + hash = hash.wrapping_mul(16_777_619); + } + hash +} + +/// Per-agent web UI port — `WEB_PORT_BASE + FNV-1a(name) % +/// WEB_PORT_RANGE` for every agent including the manager. The port +/// allocation rule reads the same for every name; collisions are +/// possible (birthday paradox at ~30 agents) and the operator +/// resolves them by renaming an agent (different hash → different +/// port). Stable across hosts, restarts, and dashboard renders — +/// no state-file dance. +#[must_use] +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) +} diff --git a/nix/host-modules/swarm-controller.nix b/nix/host-modules/swarm-controller.nix index 55f14983..93d0628e 100644 --- a/nix/host-modules/swarm-controller.nix +++ b/nix/host-modules/swarm-controller.nix @@ -252,6 +252,43 @@ let SWARM_CONTROLLER_MATRIX_HOMESERVER_URL = deployCfg.swarm-controller.matrixHomeserverUrl; }; + # What `internal/meta` renders into every agent's build + # (swarm-controller/src/meta_flake.rs `Settings`): the names and values + # ./hive-c0re/environment.nix forwards, minus the per-host ones. One flake + # serves every hive, so a host's own answer here would reach all of them. + # `HIVE_AGENT_NATS_URL` is per host and reaches the container through the + # hive facts instead. + # + # The store-backed values are gated on this controller's store identity, + # where a hive gates them on its own: the swarm has a store when its + # controller can reach one. + metaEnv = { + SWARM_CONTROLLER_META_FLAKE = builtins.toJSON { + vars = { + HIVE_FORGE_URL = "http://${forgeCfg.domain}"; + } + // lib.optionalAttrs (forgeCfg.publicUrl != null) { + HIVE_FORGE_PUBLIC_URL = forgeCfg.publicUrl; + } + // lib.optionalAttrs (matrixCfg.gatewayHost != null) { + HIVE_MATRIX_URL = "http://${matrixCfg.gatewayHost}"; + } + // lib.optionalAttrs (swarmCfg.name != null) { + HYPERHIVE_SWARM_NAME = swarmCfg.name; + } + // lib.optionalAttrs haveBaoIdentity { + HIVE_AGENT_LOGS_QUERY_URL = "https://${swarmCfg.victorialogs.domain}/select/logsql/query"; + HIVE_AGENT_METRICS_URL = "https://${swarmCfg.victoriametrics.domain}"; + HIVE_AGENT_BAO_ADDR = "https://${baoCfg.domain}:${toString baoCfg.port}"; + HIVE_AGENT_BAO_SERVICES_PKI_MOUNT = deployCfg.bao.servicesPkiMountPath; + }; + # Where the pinned inputs are fetched from: the builder's credential + # helper answers only `https` for the forge. + git_base = "https://${forgeCfg.domain}"; + github_disabled = !swarmCfg.github.enable; + }; + }; + # The swarm's own display name, for `GET /api/swarm` (swarm-ui's chrome). # Gated on the option resolving, not defaulted to an empty string: an # operator who never named the swarm gets `name: null` from the @@ -831,7 +868,8 @@ in // authBridgeEnv // swarmNameEnv // matrixHomeserverEnv - // baoEnv; + // baoEnv + // metaEnv; }; }; } diff --git a/swarm-builder/README.md b/swarm-builder/README.md index ae3b640b..7b789383 100644 --- a/swarm-builder/README.md +++ b/swarm-builder/README.md @@ -39,3 +39,10 @@ every `hyperhiveArchetypes` entry in this host's hyperhive source, with nixpkgs pointed at this host's. It fetches nothing and uses no token, and it doesn't wait for a running build. See [`docs/swarm/archetypes.md`](../docs/swarm/archetypes.md). + +## Meta-flake lock + +The service also locks the swarm meta flake on request: `nix flake lock` of +the `flake.nix` it was sent, in a scratch directory, with the build's forge +credential. It answers with the lock and commits nothing. See +[`docs/swarm/builder.md`](../docs/swarm/builder.md#meta-flake-lock). diff --git a/swarm-controller/Cargo.toml b/swarm-controller/Cargo.toml index 9b70dfdc..02b61c7d 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -46,6 +46,8 @@ problem_details = { version = "0.9.0", features = ["axum"] } hive-jobq.workspace = true hive-jobq-wire.workspace = true hive-log.workspace = true +# `internal/meta`'s `flake.nix`, rendered by the same code hive-c0re renders with. +hive-meta-flake.workspace = true # The jobq-rollup OTEL exporter, wired up in `main` via # `hive_jobq_metrics::spawn_exporter` — moved to its own crate (rather than # living here as `jobq_metrics.rs`) specifically so a future second caller diff --git a/swarm-controller/src/forge.rs b/swarm-controller/src/forge.rs index e8689b92..42cec22f 100644 --- a/swarm-controller/src/forge.rs +++ b/swarm-controller/src/forge.rs @@ -52,6 +52,7 @@ pub mod classifier_account; pub mod docs_grant; pub mod hive_token; pub mod legacy_tokens; +mod meta_repo; pub mod objects; pub mod operator_profiles; pub mod pull_diff; diff --git a/swarm-controller/src/forge/meta_repo.rs b/swarm-controller/src/forge/meta_repo.rs new file mode 100644 index 00000000..47d8adfa --- /dev/null +++ b/swarm-controller/src/forge/meta_repo.rs @@ -0,0 +1,77 @@ +//! `internal/meta` on the forge: the repo the swarm meta flake is committed to +//! (`crate::meta_flake`), and the read grant each hive's account holds on it. + +use anyhow::{Context, Result}; +use base64::Engine as _; +use forgejo_api::structs::{ + AddCollaboratorOptionPermission, ChangeFileOperation, ChangeFileOperationOperation, + ChangeFilesOptions, +}; + +use super::objects::SHARED_ORG; +use super::{Client, DefaultBranchFiles}; +use crate::meta_flake::{META_REPO, MetaFile, MetaForge}; + +impl MetaForge for Client { + async fn head_files( + &self, + owner: &str, + repo: &str, + paths: &[&str], + ) -> Result> { + self.default_branch_files(owner, repo, paths).await + } + + async fn ensure_meta_repo(&self) -> Result<()> { + self.create_shared_repo(SHARED_ORG, META_REPO, true).await + } + + async fn grant_meta_read(&self, login: &str) -> Result<()> { + self.add_collaborator_to( + SHARED_ORG, + META_REPO, + login, + AddCollaboratorOptionPermission::Read, + ) + .await + } + + async fn commit_meta(&self, message: &str, files: Vec) -> Result<()> { + let ops: Vec = files + .into_iter() + .map(|f| ChangeFileOperation { + content: Some(base64::engine::general_purpose::STANDARD.encode(f.bytes)), + from_path: None, + operation: if f.sha.is_some() { + ChangeFileOperationOperation::Update + } else { + ChangeFileOperationOperation::Create + }, + path: f.path.to_owned(), + sha: f.sha, + }) + .collect(); + self.api + .call(|api| { + api.repo_change_files( + SHARED_ORG, + META_REPO, + ChangeFilesOptions { + author: None, + branch: None, + committer: None, + dates: None, + files: ops.clone(), + force_overwrite_new_branch: None, + message: Some(message.to_owned()), + new_branch: None, + signoff: None, + }, + ) + .into_future() + }) + .await + .with_context(|| format!("commit to {SHARED_ORG}/{META_REPO}"))?; + Ok(()) + } +} diff --git a/swarm-controller/src/forge/objects.rs b/swarm-controller/src/forge/objects.rs index 8e8d5c7b..7d957c3a 100644 --- a/swarm-controller/src/forge/objects.rs +++ b/swarm-controller/src/forge/objects.rs @@ -48,7 +48,7 @@ const AGENTS_ORG: &str = "agents"; /// Org hosting the operator-curated shared repos. Same org as the knowledge /// repo's, named separately for the docs repo it also holds. -pub(super) const SHARED_ORG: &str = KNOWLEDGE_ORG; +pub(crate) const SHARED_ORG: &str = KNOWLEDGE_ORG; /// The shared docs repo inside [`SHARED_ORG`]. **Private**: every agent gets /// read-only collaborator access, granted per agent by [`super::docs_grant`]. @@ -56,9 +56,8 @@ pub(super) const SHARED_ORG: &str = KNOWLEDGE_ORG; /// content into the system prompt. pub(super) const SHARED_DOCS_REPO: &str = "docs"; -/// The orgs ensured on every pass. The meta repo lives at `core/meta`, the -/// `core` user's own namespace, so no org is needed for it (and it is not a -/// swarm object: it stays with `hive-c0re`). +/// The orgs ensured on every pass. `internal/meta` is in [`SHARED_ORG`] and is +/// created by `crate::meta_flake`'s update, not by this pass. const SEEDED_ORGS: [&str; 3] = [CONFIG_ORG, SHARED_ORG, AGENTS_ORG]; /// Where the [`OPERATORS_TEAM`] must exist. Gitea teams are org-scoped, so a @@ -938,7 +937,12 @@ impl Client { } /// Create `owner/name`, empty, defaulting to `main`. - async fn create_shared_repo(&self, owner: &str, name: &str, private: bool) -> Result<()> { + pub(super) async fn create_shared_repo( + &self, + owner: &str, + name: &str, + private: bool, + ) -> Result<()> { match self .api .call(|api| { diff --git a/swarm-controller/src/jobq.rs b/swarm-controller/src/jobq.rs index 98d65b54..225d9195 100644 --- a/swarm-controller/src/jobq.rs +++ b/swarm-controller/src/jobq.rs @@ -10,7 +10,8 @@ use crate::routes::AppState; use crate::routes::agents::{claiming_hives, read_declarations}; use crate::startup::PUBLIC_URL_ENV; use crate::{ - agent_identity, agent_renewal, auth, builder, config_pr, forge, matrix_account, wanted, + agent_identity, agent_renewal, auth, builder, config_pr, forge, matrix_account, meta_flake, + wanted, }; /// Node payload for the swarm-level job graph. Named `Swarm*` rather than @@ -133,6 +134,13 @@ 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`. + /// + /// 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. + UpdateMeta, } impl hive_jobq_wire::WireNode for SwarmNodeKind { @@ -158,6 +166,7 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind { } SwarmNodeKind::SetAgentWanted { .. } => "set_agent_wanted".to_owned(), SwarmNodeKind::TriggerDeploy { .. } => "trigger_deploy".to_owned(), + SwarmNodeKind::UpdateMeta => "update_meta".to_owned(), } } @@ -185,7 +194,8 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind { | SwarmNodeKind::MintHiveForgeToken { hive } => serde_json::json!({ "hive": hive }), SwarmNodeKind::MintClassifierForgeToken | SwarmNodeKind::MintBuilderForgeToken - | SwarmNodeKind::MintCiMetricsForgeToken => serde_json::json!({}), + | SwarmNodeKind::MintCiMetricsForgeToken + | SwarmNodeKind::UpdateMeta => serde_json::json!({}), SwarmNodeKind::BuildAgent { agent, rev, pr } => { serde_json::json!({ "agent": agent, "rev": rev, "pr": pr }) } @@ -206,12 +216,16 @@ pub(crate) enum SwarmResourceKind { /// The swarm builder: one `BuildAgent` node runs at a time, and the rest /// wait pending, where a newer push can replace them. Builder, + /// `internal/meta`: one `UpdateMeta` commits at a time, so two never race + /// on the same file's blob. + Meta, } impl hive_jobq_wire::WireResource for SwarmResourceKind { fn name(&self) -> String { match self { SwarmResourceKind::Builder => "builder".to_owned(), + SwarmResourceKind::Meta => "meta".to_owned(), } } } @@ -253,6 +267,14 @@ pub(crate) struct WorkerDeps { pub(crate) matrix_homeserver: Option>, } +/// What `UpdateMeta` renders from, fixed for the process. +struct MetaDeps { + settings: meta_flake::Settings, + /// Every hive's name: each places agents, and each hive's forge account + /// gets read on the repo. + hives: Vec, +} + /// What [`SwarmNodeKind::SetAgentWanted`] declares a brand-new agent to be. /// /// A named constant rather than the literal inline, because this *is* the @@ -373,6 +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, }; (builder, outcome) } @@ -542,6 +565,95 @@ pub(crate) async fn build_agent( .await } +/// The `UpdateMeta` arm, lifted out for the same reason. +async fn update_meta( + deps: &WorkerDeps, + forge: Option>, +) -> hive_jobq::scheduler::Outcome { + use hive_jobq::scheduler::Outcome; + + let Some(client) = forge else { + return Outcome::Failed( + "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!( + "{} 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" + .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( + "update_meta", + meta_flake::update( + client.as_ref(), + queue, + &meta.settings, + &agents, + &meta.hives, + ), + ) + .await + { + Ok(updated) => { + tracing::info!(?updated, agents = agents.len(), "update_meta: done"); + Outcome::Done + } + Err(e) => Outcome::Failed(format!("{e:#}")), + } +} + +/// Every agent some hive in `hives` places, with its archetype. +async fn placed_agents( + queue: &async_nats::Client, + wanted: &wanted::WantedWriter, + hives: &[String], +) -> Result> { + let mut names = std::collections::BTreeSet::new(); + for hive in hives { + let Some(declared) = wanted + .view(hive) + .await + .map_err(|e| e.context(format!("read where hive {hive} places agents")))? + else { + continue; + }; + names.extend( + declared + .agents + .into_iter() + .filter(|(_, w)| crate::routes::agents::places_agent(w.state)) + .map(|(name, _)| name), + ); + } + let archetypes = swarm_queue_client::agent_archetype::open_or_create(queue) + .await + .map_err(|e| anyhow::anyhow!("open the agent-archetype bucket: {e}"))?; + let mut placed = Vec::with_capacity(names.len()); + for name in names { + let (archetype, _, _) = crate::schedules::read_or_default::< + Option, + >(&archetypes, &name) + .await + .map_err(|e| e.context(format!("read the archetype of {name}")))?; + placed.push(meta_flake::Placed { name, archetype }); + } + Ok(placed) +} + /// The `MintBuilderForgeToken` arm, lifted out for the same reason. async fn mint_builder_forge_token( forge: Option>, @@ -872,14 +984,20 @@ pub(crate) async fn queue_merged_config_deploy( .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) .insert_job(None, |b| { - vec![ - b.node(SwarmNodeKind::TriggerDeploy { + // 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()), }) - .guid(), - ] + .after_ok(update_meta); + vec![root] }); match inserted { Ok(_) => tracing::info!(%agent, %rev, %hive, "config-pr merge: deploy queued"), @@ -977,6 +1095,12 @@ 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. + let update_meta = b + .node(SwarmNodeKind::UpdateMeta) + .needs(SwarmResourceKind::Meta) + .after_ok(set_wanted); // Last, and specifically after the config repo is seeded: the hive // deploys by reading that repo, so a deploy asked for any earlier would // find nothing to build. This is the edge that makes creating an agent at @@ -1007,6 +1131,7 @@ pub(crate) fn declare_agent_job( agent: agent.to_owned(), rev: None, }) + .after_ok(update_meta) .after_any(mint_identity) .after_any(mint_forge_token) // `after_any` for the same reason: the container reads this token @@ -1017,6 +1142,7 @@ pub(crate) fn declare_agent_job( .after_ok(set_wanted); if let Some(init_config) = init_config { let _ = trigger_deploy.after_ok(init_config); + let _ = update_meta.after_ok(init_config); } vec![create_identity.guid()] } @@ -1183,6 +1309,64 @@ pub(crate) fn queue_builder_forge_token_mint( .map_err(|e| anyhow::anyhow!("{e}")) } +/// Insert one `UpdateMeta` job unless one is already pending, which reads the +/// same heads when it runs. Returns whether a job was inserted. +pub(crate) fn queue_meta_update( + sched: &Mutex>, +) -> Result { + let mut sched = sched + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let pending = sched.graph().nodes().any(|n| { + n.state == hive_jobq::State::Pending && matches!(n.payload, SwarmNodeKind::UpdateMeta) + }); + if pending { + return Ok(false); + } + sched + .insert_job(None, |b| { + vec![ + b.node(SwarmNodeKind::UpdateMeta) + .needs(SwarmResourceKind::Meta) + .guid(), + ] + }) + .map_err(|e| anyhow::anyhow!("{e}"))?; + Ok(true) +} + +/// [`MetaDeps`] from the environment, read on first use; `None`, logged, when +/// the settings are unset or do not parse, which `UpdateMeta` then fails on +/// by name. +fn meta_deps() -> Option<&'static MetaDeps> { + static DEPS: std::sync::OnceLock> = std::sync::OnceLock::new(); + DEPS.get_or_init(read_meta_deps).as_ref() +} + +fn read_meta_deps() -> Option { + let settings = match meta_flake::Settings::from_env() { + Ok(Some(s)) => s, + Ok(None) => { + tracing::warn!( + "{} unset; internal/meta will not be written", + meta_flake::SETTINGS_ENV + ); + return None; + } + Err(e) => { + tracing::warn!(error = %format!("{e:#}"), "internal/meta will not be written"); + return None; + } + }; + Some(MetaDeps { + settings, + hives: crate::startup::load_hives() + .into_iter() + .map(|h| h.name) + .collect(), + }) +} + /// Insert one `MintCiMetricsForgeToken` job and return its node's id. /// `forge::ci_metrics_account::spawn`'s periodic pass is the one caller. pub(crate) fn queue_ci_metrics_forge_token_mint( diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index b59030eb..f76caf53 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -50,6 +50,7 @@ mod jobq; mod linked_accounts; mod matrix_account; mod message_stream; +mod meta_flake; mod notify_flows; mod otel_http_client; mod pulls; diff --git a/swarm-controller/src/meta_flake.rs b/swarm-controller/src/meta_flake.rs new file mode 100644 index 00000000..73143396 --- /dev/null +++ b/swarm-controller/src/meta_flake.rs @@ -0,0 +1,583 @@ +//! `internal/meta`: the swarm meta flake hives build their agents 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 +//! `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. + +use std::collections::BTreeMap; +use std::future::Future; + +use anyhow::{Context, Result, bail}; +use serde::Deserialize; +use swarm_wire::agent_archetype::ArchetypeRef; +use swarm_wire::builder::{LOCK_REPLY_DEADLINE, LOCK_SUBJECT, LockMeta, LockReply}; + +use crate::forge::{CONFIG_ORG, DefaultBranchFiles}; + +/// The repo in `internal` the flake is committed to. +pub(crate) const META_REPO: &str = "meta"; + +/// The env var holding [`Settings`] as JSON, set by `swarm-controller.nix`. +pub(crate) const SETTINGS_ENV: &str = "SWARM_CONTROLLER_META_FLAKE"; + +/// Where hyperhive itself lives on the forge. +const HYPERHIVE_OWNER: &str = "hyperhive"; +const HYPERHIVE_REPO: &str = "hyperhive"; + +/// What every agent of the swarm gets, the same on every hive. +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct Settings { + /// Values by `hive_meta_flake::FORWARDED_VARS` name. `HIVE_FORGE_URL` is + /// required. + vars: BTreeMap, + /// The forge's `https` origin, which every pinned input is fetched from: + /// the swarm builder's credential helper answers only `https` for it. + git_base: String, + /// The swarm turned the agents' GitHub integration off. + github_disabled: bool, +} + +impl Settings { + /// [`Settings`] from [`SETTINGS_ENV`], `None` when it is unset. + pub(crate) fn from_env() -> Result> { + match std::env::var(SETTINGS_ENV) { + Ok(raw) => Self::parse(&raw).map(Some), + Err(std::env::VarError::NotPresent) => Ok(None), + Err(e) => Err(e).with_context(|| format!("read {SETTINGS_ENV}")), + } + } + + fn parse(raw: &str) -> Result { + let settings: Self = + serde_json::from_str(raw).with_context(|| format!("parse {SETTINGS_ENV}"))?; + if let Some(unknown) = settings + .vars + .keys() + .find(|k| !hive_meta_flake::FORWARDED_VARS.contains(&k.as_str())) + { + 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) { + 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"); + } + Ok(settings) + } + + /// The forwarded pairs with a value, in `FORWARDED_VARS` order. + fn forwarded(&self) -> Vec<(&'static str, String)> { + hive_meta_flake::FORWARDED_VARS + .iter() + .filter_map(|&name| { + self.vars + .get(name) + .filter(|v| !v.is_empty()) + .map(|v| (name, v.clone())) + }) + .collect() + } + + fn git_url(&self, owner: &str, repo: &str) -> String { + format!("{}/{owner}/{repo}.git", self.git_base.trim_end_matches('/')) + } +} + +/// One agent some hive places, with the archetype it is built from. +#[derive(Debug, Clone)] +pub(crate) struct Placed { + pub(crate) name: String, + pub(crate) archetype: Option, +} + +/// One file of a meta commit: a blob sha replaces that blob, `None` creates it. +pub(crate) struct MetaFile { + pub(crate) path: &'static str, + pub(crate) bytes: Vec, + pub(crate) sha: Option, +} + +/// The forge, as [`update`] reads and writes it. +pub(crate) trait MetaForge { + fn head_files( + &self, + owner: &str, + repo: &str, + paths: &[&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( + &self, + message: &str, + files: Vec, + ) -> impl Future> + Send; +} + +/// Where a lock is sent: the queue in production, a fake in the tests. `Err` +/// is a lock that never reached a builder or never came back. +pub(crate) trait LockSink { + fn lock(&self, request: Vec) -> impl Future> + Send; +} + +impl LockSink for async_nats::Client { + async fn lock(&self, request: Vec) -> Result { + use async_nats::RequestErrorKind; + + let sent = async_nats::Request::new() + .payload(request.into()) + .timeout(Some(LOCK_REPLY_DEADLINE)); + match self.send_request(LOCK_SUBJECT, sent).await { + Ok(msg) => serde_json::from_slice(&msg.payload) + .map_err(|e| format!("the builder's lock reply did not decode: {e}")), + Err(e) => Err(match e.kind() { + RequestErrorKind::NoResponders => "no swarm builder is listening for locks: \ + deploy.builder.enable is set on no host, or its builder predates the lock \ + request" + .to_owned(), + RequestErrorKind::TimedOut => format!( + "the swarm builder sent no lock within {} minutes", + LOCK_REPLY_DEADLINE.as_secs() / 60 + ), + _ => format!("sending the lock to the swarm builder failed: {e}"), + }), + } + } +} + +/// What [`update`] did. +#[derive(Debug, PartialEq, Eq)] +pub(crate) enum Updated { + /// The committed flake already is the render. + Unchanged, + /// A new flake and lock were committed. + Committed, +} + +/// Bring `internal/meta` up to the forge's current heads for `agents`, and +/// give each of `hives`' forge accounts read on it. +/// +/// # Errors +/// A head that cannot be read, a lock the builder refuses, or a refused commit. +/// A failed read grant is logged, not returned: the commit is what hives wait +/// on. +pub(crate) async fn update( + forge: &impl MetaForge, + locker: &impl LockSink, + settings: &Settings, + agents: &[Placed], + hives: &[String], +) -> Result { + forge.ensure_meta_repo().await?; + for hive in hives { + let login = swarm_secret_client::forge::hive_login(hive); + if let Err(e) = forge.grant_meta_read(&login).await { + 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, + META_REPO, + &["flake.nix", "flake.lock"], + ) + .await?; + let sha_of = |path: &str| { + committed + .as_ref() + .and_then(|c| c.files.iter().find(|f| f.path == path)) + }; + let flake_file = sha_of("flake.nix"); + if flake_file.is_some_and(|f| f.bytes == flake.as_bytes()) && sha_of("flake.lock").is_some() { + return Ok(Updated::Unchanged); + } + let request = serde_json::to_vec(&LockMeta { + flake_nix: flake.clone(), + }) + .context("encode the lock request")?; + let flake_lock = match locker.lock(request).await { + Ok(LockReply::Locked { flake_lock }) => flake_lock, + Ok(LockReply::Failed { error_tail }) => bail!("the builder could not lock: {error_tail}"), + Err(e) => bail!("{e}"), + }; + forge + .commit_meta( + &format!("meta: pin {} agent(s)", agents.len()), + vec![ + MetaFile { + path: "flake.nix", + bytes: flake.into_bytes(), + sha: flake_file.map(|f| f.sha.clone()), + }, + MetaFile { + path: "flake.lock", + bytes: flake_lock.into_bytes(), + sha: sha_of("flake.lock").map(|f| f.sha.clone()), + }, + ], + ) + .await?; + Ok(Updated::Committed) +} + +/// 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; + }; + 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") +} + +/// 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, &[]) + .await? + .with_context(|| format!("{HYPERHIVE_OWNER}/{HYPERHIVE_REPO} is empty or absent"))?; + let hyperhive_url = hive_meta_flake::pinned_git_url( + &settings.git_url(HYPERHIVE_OWNER, HYPERHIVE_REPO), + Some(&hyperhive.commit), + )?; + let mut specs = Vec::with_capacity(agents.len()); + for agent in agents { + let pin = if agent.archetype.is_some() { + None + } else { + let head = forge + .head_files(CONFIG_ORG, &agent.name, &["flake.lock"]) + .await? + .with_context(|| { + format!( + "agent {}: {CONFIG_ORG}/{} has no commit on the forge", + agent.name, agent.name + ) + })?; + let lock = head + .files + .first() + .map(|f| String::from_utf8_lossy(&f.bytes).into_owned()); + Some(hive_meta_flake::AgentPin { + url: hive_meta_flake::pinned_git_url( + &settings.git_url(CONFIG_ORG, &agent.name), + Some(&head.commit), + )?, + canonical_inputs: lock + .as_deref() + .map(hive_meta_flake::canonical_inputs_in_lock) + .unwrap_or_default(), + }) + }; + specs.push(hive_meta_flake::AgentSpec { + name: agent.name.clone(), + port: hive_meta_flake::agent_web_port(&agent.name), + archetype: agent.archetype.clone(), + pin, + }); + } + specs.sort_by(|a, b| a.name.cmp(&b.name)); + hive_meta_flake::render_flake( + &hyperhive_url, + &specs, + &settings.forwarded(), + settings.github_disabled, + ) +} + +#[cfg(test)] +mod tests { + use std::sync::Mutex; + + use super::*; + use crate::forge::BranchFile; + + const FORGE: &str = "https://forge.t.local"; + + fn settings() -> Settings { + Settings::parse(&format!( + r#"{{"vars":{{"HIVE_FORGE_URL":"http://forge.t.local","HYPERHIVE_SWARM_NAME":"t"}},"git_base":"{FORGE}/","github_disabled":true}}"# + )) + .unwrap() + } + + fn head(commit: &str, files: &[(&str, &str)]) -> DefaultBranchFiles { + DefaultBranchFiles { + branch: "main".to_owned(), + commit: commit.to_owned(), + web_url: None, + files: files + .iter() + .map(|(path, body)| BranchFile { + path: (*path).to_owned(), + sha: format!("blob-{path}"), + bytes: body.as_bytes().to_vec(), + }) + .collect(), + } + } + + /// One recorded commit: its message, and each file's path, text and sha. + type Commit = (String, Vec<(&'static str, String, Option)>); + + #[derive(Default)] + struct FakeForge { + repos: BTreeMap, + commits: Mutex>, + grants: Mutex>, + } + + impl MetaForge for FakeForge { + async fn head_files( + &self, + owner: &str, + repo: &str, + paths: &[&str], + ) -> Result> { + Ok(self.repos.get(&format!("{owner}/{repo}")).map(|h| { + let mut h = h.clone(); + h.files.retain(|f| paths.contains(&f.path.as_str())); + h + })) + } + async fn ensure_meta_repo(&self) -> Result<()> { + Ok(()) + } + async fn grant_meta_read(&self, login: &str) -> Result<()> { + self.grants.lock().unwrap().push(login.to_owned()); + Ok(()) + } + async fn commit_meta(&self, message: &str, files: Vec) -> Result<()> { + self.commits.lock().unwrap().push(( + message.to_owned(), + files + .into_iter() + .map(|f| (f.path, String::from_utf8(f.bytes).unwrap(), f.sha)) + .collect(), + )); + Ok(()) + } + } + + /// Answers every lock with `reply`, recording each flake it was sent. + struct FakeLocker { + reply: Result, + sent: Mutex>, + } + + impl FakeLocker { + fn locked() -> Self { + Self { + reply: Ok(LockReply::Locked { + flake_lock: "{\"version\":7}".to_owned(), + }), + sent: Mutex::default(), + } + } + } + + impl LockSink for FakeLocker { + async fn lock(&self, request: Vec) -> Result { + let req: LockMeta = serde_json::from_slice(&request).unwrap(); + self.sent.lock().unwrap().push(req.flake_nix); + self.reply.clone() + } + } + + fn forge_with(extra: &[(&str, DefaultBranchFiles)]) -> FakeForge { + let mut repos = BTreeMap::new(); + repos.insert("hyperhive/hyperhive".to_owned(), head("hh0", &[])); + repos.insert( + "agent-configs/argus".to_owned(), + head( + "a1", + &[( + "flake.lock", + r#"{"root":"root","nodes":{"root":{"inputs":{"nixpkgs":"nixpkgs"}}}}"#, + )], + ), + ); + for (name, h) in extra { + repos.insert((*name).to_owned(), h.clone()); + } + FakeForge { + repos, + ..FakeForge::default() + } + } + + fn argus_and_reviewer() -> Vec { + vec![ + Placed { + name: "argus".to_owned(), + archetype: None, + }, + Placed { + name: "rev".to_owned(), + archetype: Some(ArchetypeRef { + source: "hyperhive".to_owned(), + name: "reviewer".to_owned(), + }), + }, + ] + } + + #[tokio::test] + async fn a_new_render_is_locked_and_committed_with_its_lock_in_one_commit() { + let forge = forge_with(&[]); + let locker = FakeLocker::locked(); + let got = update( + &forge, + &locker, + &settings(), + &argus_and_reviewer(), + &["pr1ma".to_owned()], + ) + .await + .unwrap(); + assert_eq!(got, Updated::Committed); + 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\";" + ))); + 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("services.hyperhive.agent.github.enable = false;")); + assert!(flake.contains("HYPERHIVE_SWARM_NAME = \"t\";")); + let commits = forge.commits.lock().unwrap(); + let [(_, files)] = commits.as_slice() else { + panic!("one commit: {commits:?}") + }; + assert_eq!( + files, + &vec![ + ("flake.nix", flake.clone(), None), + ("flake.lock", "{\"version\":7}".to_owned(), None), + ] + ); + assert_eq!( + *forge.grants.lock().unwrap(), + vec![swarm_secret_client::forge::hive_login("pr1ma")] + ); + } + + #[tokio::test] + async fn an_unchanged_render_sends_no_lock_and_commits_nothing() { + let first = forge_with(&[]); + 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); + 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() { + let forge = forge_with(&[( + "internal/meta", + head("m1", &[("flake.nix", "old"), ("flake.lock", "{}")]), + )]); + let locker = FakeLocker::locked(); + update(&forge, &locker, &settings(), &argus_and_reviewer(), &[]) + .await + .unwrap(); + let commits = forge.commits.lock().unwrap(); + let shas: Vec<_> = commits[0].1.iter().map(|(_, _, sha)| sha.clone()).collect(); + assert_eq!( + shas, + vec![ + Some("blob-flake.nix".to_owned()), + Some("blob-flake.lock".to_owned()) + ] + ); + } + + #[tokio::test] + async fn a_refused_lock_commits_nothing() { + let forge = forge_with(&[]); + let locker = FakeLocker { + reply: Ok(LockReply::Failed { + error_tail: "no such input".to_owned(), + }), + sent: Mutex::default(), + }; + let err = update(&forge, &locker, &settings(), &argus_and_reviewer(), &[]) + .await + .unwrap_err(); + assert!(format!("{err:#}").contains("no such input")); + assert!(forge.commits.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn a_custom_agent_without_a_config_commit_fails_the_render() { + let forge = forge_with(&[]); + let agents = vec![Placed { + name: "ghost".to_owned(), + archetype: None, + }]; + let err = update(&forge, &FakeLocker::locked(), &settings(), &agents, &[]) + .await + .unwrap_err(); + assert!(format!("{err:#}").contains("agent-configs/ghost")); + 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() + }; + 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")); + } + + #[test] + fn settings_refuse_an_unknown_value_a_missing_forge_and_a_plain_http_base() { + let parse = |vars: &str, git_base: &str| { + Settings::parse(&format!( + r#"{{"vars":{vars},"git_base":"{git_base}","github_disabled":false}}"# + )) + }; + 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(); + 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()); + } +} diff --git a/swarm-controller/src/routes/agents.rs b/swarm-controller/src/routes/agents.rs index 139ca5cf..a8ffba65 100644 --- a/swarm-controller/src/routes/agents.rs +++ b/swarm-controller/src/routes/agents.rs @@ -221,7 +221,7 @@ pub(crate) async fn set_agent_state( /// /// Exhaustive for the reason [`revokes_queue_credential`] is. A `Destroyed` /// agent is gone from that hive, and the hive deploys nothing for it. -fn places_agent(state: swarm_wire::wanted::AgentState) -> bool { +pub(crate) fn places_agent(state: swarm_wire::wanted::AgentState) -> bool { use swarm_wire::wanted::AgentState; match state { AgentState::Up | AgentState::Offline | AgentState::Paused => true, diff --git a/swarm-controller/src/startup.rs b/swarm-controller/src/startup.rs index f2529e1e..c63d255c 100644 --- a/swarm-controller/src/startup.rs +++ b/swarm-controller/src/startup.rs @@ -285,6 +285,11 @@ fn spawn_forge_workers( auth: Option>, queue: Option, ) { + // `internal/meta` appears, and catches up with every head that moved while + // this daemon was down. + if let Err(e) = crate::jobq::queue_meta_update(jobq) { + tracing::warn!(error = %format!("{e:#}"), "update_meta: queueing failed"); + } forge::objects::spawn(Arc::clone(&client)); if let (Some(bridge), Some(queue)) = (auth, queue) { forge::operator_profiles::spawn(Arc::clone(&client), bridge, queue); diff --git a/swarm-controller/src/webhook.rs b/swarm-controller/src/webhook.rs index 6ba7f25e..5f4d71ed 100644 --- a/swarm-controller/src/webhook.rs +++ b/swarm-controller/src/webhook.rs @@ -443,6 +443,11 @@ 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 {