diff --git a/AGENTS.md b/AGENTS.md index 68683535..6635cadb 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -162,7 +162,8 @@ Swarm-level: `containers.hive-matrix`; new in-container jobs are verbs here, not crates. - **`swarm-builder/`** — builds agent config PR heads for commit status, - and evaluates the agent archetypes the create form lists. + builds each agent from `internal/meta` before its deploy, and evaluates the + agent archetypes the create form lists. Design: [`docs/swarm/builder.md`](docs/swarm/builder.md). - **`swarm-ci-metrics/`** — polls the forge's Actions runs and pushes each finished run's duration and queue wait to the metrics store. Design: diff --git a/docs/agent-lifecycle/config-changes.md b/docs/agent-lifecycle/config-changes.md index 44488632..89d533d8 100644 --- a/docs/agent-lifecycle/config-changes.md +++ b/docs/agent-lifecycle/config-changes.md @@ -149,7 +149,9 @@ every agent any hive places (the pins: every agent from it. While the repo has no commit that pins an agent's current config, that agent's build fails and names the agent. The controller queues its update of the repo before the deploy request for a merged config commit -and for a new agent, and sends the deploy only when that update succeeded. +and for a new agent, and sends the deploy only when that update succeeded and +the swarm builder built the agent from the updated repo (the +[deploy gate](../swarm/builder.md#deploy-gate)). A red build skips the deploy. - **Choosing the commit.** The rebuild's `MetaSync` node reads the repo's default-branch head with the hive's read token (`forge::reader_token()`) diff --git a/docs/swarm/builder.md b/docs/swarm/builder.md index c014a68f..87388f79 100644 --- a/docs/swarm/builder.md +++ b/docs/swarm/builder.md @@ -6,6 +6,10 @@ push, the builder builds the PR's head on the agent base, and the result shows a `swarm-builder` commit status on the PR and as a `build_agent` node on the swarm UI's Jobs page. +The builder also builds each agent from `internal/meta` before the swarm sends +its deploy, and a red build holds the deploy back: see +[Deploy gate](#deploy-gate). + Where it runs and how it's enabled: [`services.md`](services.md#builder). ## What happens on a config PR @@ -102,7 +106,8 @@ status has no link. - The builder pushes nothing. The closure stays in the builder host's store until it's collected, so on a host that also runs the hive, the hive's own build can reuse whatever matches. -- It doesn't gate anything. A merge deploys whatever the status says. +- The PR check gates nothing. A merge deploys whatever the status says. The + [deploy gate](#deploy-gate) is a separate build, from `internal/meta`. - A lost webhook delivery isn't retried: the head isn't built until the next push. The jobs graph is in memory, so a controller restart drops a build in flight and leaves its status `pending`. @@ -148,11 +153,12 @@ repo. It re-renders when it starts, and again every ten minutes while the latest update hasn't succeeded (a tick adds nothing while one is pending or running, or after a success), before it deploys a merged config PR, and before the first deploy of a new agent. Those deploys run only after the -update succeeded: a hive builds only from `internal/meta`, so after a failed -update the scheduler skips the deploy node, and the graph shows the update as -failed and the deploy as skipped. Each ten-minute retry of a failed update -declares a fresh deploy behind it for every agent whose latest deploy the failed update -skipped, so the run that succeeds sends them (see +update succeeded and the [deploy gate](#deploy-gate) built the agent: a hive +builds only from `internal/meta`, so after a failed update the scheduler skips +the deploy node, and the graph shows the update as failed and the deploy as +skipped. Each ten-minute retry of a failed update declares a fresh build and +deploy behind it for every agent whose latest deploy the failed update skipped, +so the run that succeeds sends them (see [Config changes](../agent-lifecycle/config-changes.md#the-swarms-meta-flake)). A render equal to the committed `flake.nix` sends no lock. @@ -160,19 +166,72 @@ The controller sends a deploy to every agent a hive places whenever the pin in `internal/meta` differs from the pin it last sent, so they rebuild on the new hyperhive. It keeps that pin as the key `pin` in the `meta-fanout` KV bucket, which only the controller reads or writes, and a `RecordFanout` job writes the -key after every deploy of the fan-out succeeds. The check runs after each update -and once when the controller starts. +key once every deploy of the fan-out has finished or the deploy gate has skipped +it. The check runs after each update and once when the controller starts. - The first commit sends a deploy, because the key holds no pin yet. - A controller that starts with no key, as after the upgrade that added it, sends one round. A deploy is idempotent per pin, so the round is harmless. - A controller that dies after the commit, or a deploy that fails, leaves the key behind. The next update or start sends the deploys again. +- A deploy the gate skipped counts as sent. A red build holds only its own + agent back, and the key still moves, so one broken agent doesn't resend the + healthy ones on every retry. The timer declares a build and deploy again for + a deploy skipped because the build gave no verdict. - A restart that finds the key equal to the pin sends nothing. - When the bucket won't open, or its key won't read, the controller logs a warning and sends deploys only when its own update moved the pin. That case writes no key. +## Deploy gate + +Every deploy the swarm sends waits for a build of its agent from `internal/meta`, +the flake a hive builds from. Four paths send deploys, and each declares the same +build and deploy pair: a merged config PR, a new agent, the pin fan-out and the +ten-minute retry. The `build_meta` node comes before the `trigger_deploy` node +and needs the one `builder` slot, so it queues behind PR checks. The deploy +runs only when the build is green. + +The `build_meta` node reads `internal/meta`'s head commit when it runs, because +most paths declare it before the update that commits that head. It sends the +agent and that commit on `.builder.build-meta` and waits up to 61 +minutes. The builder takes the same one-build slot as PR checks, and +refuses a request that waited 30 seconds for it. It runs `nix build` of +`nixosConfigurations..config.system.build.toplevel` from +`git+https:///internal/meta.git?rev=`, with no input overridden: +the committed `flake.lock` decides every input. Like the hive's build, it uses +that attribute, so a green build here means the flake evaluates and builds for +the builder's platform. The hive doesn't reuse the result yet: it still builds +locally, from whatever head it reads at deploy time, which can be newer than +the one the swarm built. + +The node ends one of three ways: + +| Builder's answer | Node | Deploy | Timer | +| ----------------------------------------------------------------------------------------------------------------- | ------------------------------------ | ------- | --------------------------------- | +| built | done, closure recorded | runs | | +| `nix` failed or timed out (a red build) | failed, with `nix`'s last lines | skipped | declares nothing | +| no builder listening, no reply in time, or the builder couldn't start `nix` (busy for 30 seconds, no forge token) | failed, `not built, will be retried` | skipped | declares a build and deploy again | + +A red build is a verdict on the config: a config merge queues its own build and +deploy. A build that gave no verdict isn't, so the ten-minute timer declares +it again, without a meta update when the latest update succeeded, until the +builder answers. That covers a controller that updates before its builder +host does: the old builder has no subscriber for the new subject, so the +controller gets no responders at once, and the deploys go out after the builder +updates. Each retry adds a build and deploy node per affected agent while the +builder stays away. + +On a built reply the controller writes the agent's key in the `agent-closures` +KV bucket: the `internal/meta` commit, the closure's store path, the builder's +platform and the time. Only the controller writes the bucket, and the builder +holds no grant on it. A red build writes nothing and leaves the previous key. +Nothing reads the bucket yet. The closure stays unrooted in the builder host's +store. + +The controller gives the `builder` forge account read on `internal/meta` when it +updates the flake, since the repo is private. + ## Credentials | What | Where | Who reads it | @@ -183,7 +242,8 @@ and once when the controller starts. The token belongs to the `builder` forge account, which swarm-controller creates, puts in the `hives` team of `agent-configs` (read on every config -repo) and mints a `read:repository` token for, in a five-minute pass. `git` +repo, and on `internal/meta` through the meta update) and mints a +`read:repository` token for, in a five-minute pass. `git` gets the token through a credential helper reading the child's environment: it's in no argument, URL or lock, so no line `nix` prints can carry it into a status. The same `git` fetches a config's own inputs, so the builder registers diff --git a/docs/swarm/services.md b/docs/swarm/services.md index 706babc9..59c103d0 100644 --- a/docs/swarm/services.md +++ b/docs/swarm/services.md @@ -422,6 +422,11 @@ through the host's nix daemon, and swarm-controller writes the result as a `swar commit status on the head. The job shows on the swarm UI's Jobs page. A red status blocks nothing: the merge still deploys. +Separately, the builder builds each agent from `internal/meta` before the swarm +sends its deploy, and a red build holds that deploy back (`build_meta` job; +[deploy gate](builder.md#deploy-gate)). Those jobs share the one-build slot with +the PR checks. + It needs neither the forge nor a hive on its host. It reaches the queue, the secret store and the forge by name, under its own `builder` leaf, and fetches the private config repo with the read-only token of its own `builder` forge diff --git a/nix/host-modules/swarm-builder.nix b/nix/host-modules/swarm-builder.nix index 0a87b502..d8ddf1ac 100644 --- a/nix/host-modules/swarm-builder.nix +++ b/nix/host-modules/swarm-builder.nix @@ -246,6 +246,7 @@ in SWARM_BUILDER_OIDC_CLIENT_ID = clientId; SWARM_BUILDER_NIX = "${nix}/bin/nix"; SWARM_BUILDER_CHECK_FLAKE = "${../builder/check-flake}"; + SWARM_BUILDER_SYSTEM = pkgs.stdenv.hostPlatform.system; SWARM_BUILDER_HYPERHIVE_FLAKE = hyperhiveFlake; SWARM_BUILDER_NIXPKGS_FLAKE = nixpkgsFlake; SWARM_BUILDER_FORGE_URL = "https://${forgeCfg.domain}"; diff --git a/nix/module-eval/builder.nix b/nix/module-eval/builder.nix index b987a6db..61fc5189 100644 --- a/nix/module-eval/builder.nix +++ b/nix/module-eval/builder.nix @@ -140,6 +140,7 @@ let env.SWARM_BUILDER_HYPERHIVE_FLAKE == withBuilder.services.hyperhive.c0re.hyperhiveFlake && env.SWARM_BUILDER_NIXPKGS_FLAKE == withBuilder.services.hyperhive.c0re.nixpkgsFlake && lib.hasSuffix "-check-flake" env.SWARM_BUILDER_CHECK_FLAKE + && env.SWARM_BUILDER_SYSTEM == pkgs.stdenv.hostPlatform.system && env.SWARM_BUILDER_CONFIG_ORG == "agent-configs" && env.SWARM_BUILDER_FORGE_URL == "https://${withBuilder.services.hyperhive.swarm.forge.domain}"; } diff --git a/swarm-builder/README.md b/swarm-builder/README.md index 7b789383..97994501 100644 --- a/swarm-builder/README.md +++ b/swarm-builder/README.md @@ -21,6 +21,16 @@ the answer into a `swarm-builder` commit status on the PR's head. - Nothing is pushed and nothing is recorded: the closure stays in the host's store until it is collected. +## Meta builds + +The service also builds one agent from `internal/meta` on request: an agent and +a commit of the repo. It runs `nix build` of +`nixosConfigurations..config.system.build.toplevel` from that commit, +with every input at the flake's own lock, and answers with the store path and +the platform it builds for. The controller records the answer; the service +writes nothing. It takes the same one-build slot as the PR checks. See +[`docs/swarm/builder.md`](../docs/swarm/builder.md#deploy-gate). + ## Credentials The container holds one credential: its own store certificate, which nspawn diff --git a/swarm-builder/src/main.rs b/swarm-builder/src/main.rs index 86b4aad7..4d274b65 100644 --- a/swarm-builder/src/main.rs +++ b/swarm-builder/src/main.rs @@ -12,6 +12,11 @@ //! flake's `flake.nix`, and the answer is the `flake.lock` `nix` wrote for it in //! a scratch directory. The controller commits it. //! +//! Meta build requests on [`swarm_wire::builder::META_SUBJECT`] name an agent +//! and an `internal/meta` commit, and the answer is the store path of that +//! agent's toplevel, built at the flake's own lock. They take the build slot +//! PR checks take. The controller records the path; the builder writes nothing. +//! //! The store identity comes from the `BAO_*` environment. The queue client //! secret is read at start and again when authelia refuses it, the forge token //! once per build, both from the store under that identity. @@ -27,8 +32,9 @@ use swarm_queue_client::{ClientSecret, QueueConfig}; use swarm_secret_client::forge::Credential; use swarm_secret_client::stored_secret::StoredSecret; use swarm_wire::builder::{ - ARCHETYPES_DEADLINE, ARCHETYPES_SUBJECT, BUILD_DEADLINE, BuildAgent, LOCK_DEADLINE, - LOCK_SUBJECT, LockMeta, LockReply, QUEUE_GROUP, REPLY_DEADLINE, Reply, SUBJECT, + ARCHETYPES_DEADLINE, ARCHETYPES_SUBJECT, BUILD_DEADLINE, BuildAgent, BuildFromMeta, + FromMetaReply, LOCK_DEADLINE, LOCK_SUBJECT, LockMeta, LockReply, META_SUBJECT, QUEUE_GROUP, + REPLY_DEADLINE, Reply, SUBJECT, }; mod archetypes; @@ -76,6 +82,7 @@ async fn main() -> Result<()> { config_org: var("CONFIG_ORG")?, deadline: BUILD_DEADLINE, lock_deadline: LOCK_DEADLINE, + system: var("SYSTEM")?, }); let secret = swarm_secret_client::read_with_startup_retry("reading the queue client secret", || { @@ -124,8 +131,10 @@ async fn main() -> Result<()> { // One build at a time. swarm-controller already sends one at a time; this // holds the bound for any request it did not serialise, and `in_slot` - // refuses one that waited too long behind another. + // refuses one that waited too long behind another. Meta builds take the + // same slot as PR checks. let slot = Arc::new(tokio::sync::Semaphore::new(1)); + serve_meta_builds(&queue, &builder, &slot).await?; while let Some(msg) = requests.next().await { let received = tokio::time::Instant::now(); @@ -139,7 +148,8 @@ async fn main() -> Result<()> { tokio::spawn(async move { let reply = match serde_json::from_slice::(&msg.payload) { Ok(job) => { - let Some(reply) = in_slot(&slot, received, &job, handle(&job, &builder)).await + let Some(reply) = + in_slot(&slot, received, || busy(&job), handle(&job, &builder)).await else { return; }; @@ -249,36 +259,161 @@ async fn handle_lock(payload: &[u8], builder: &nix_build::Builder) -> LockReply reply } -/// `build`'s reply once `job` holds the build slot, or a refusal, without +/// Subscribe to meta build requests and answer them in the background. +async fn serve_meta_builds( + queue: &async_nats::Client, + builder: &Arc, + slot: &Arc, +) -> Result<()> { + let requests = queue + .queue_subscribe(META_SUBJECT, QUEUE_GROUP.to_owned()) + .await + .with_context(|| format!("subscribing to {META_SUBJECT}"))?; + tokio::spawn(answer_meta_builds( + requests, + queue.clone(), + Arc::clone(builder), + Arc::clone(slot), + )); + Ok(()) +} + +/// Answer every agent build from the meta flake, each in its own task, with the +/// build's result. +async fn answer_meta_builds( + mut requests: async_nats::Subscriber, + queue: async_nats::Client, + builder: Arc, + slot: Arc, +) { + while let Some(msg) = requests.next().await { + let received = tokio::time::Instant::now(); + let Some(reply_to) = msg.reply.clone() else { + tracing::warn!("a meta build arrived with no reply subject; dropping it"); + continue; + }; + let (queue, builder, slot) = (queue.clone(), Arc::clone(&builder), Arc::clone(&slot)); + tokio::spawn(async move { + let reply = match serde_json::from_slice::(&msg.payload) { + Ok(job) => { + let Some(reply) = in_slot( + &slot, + received, + || busy_meta(&job), + handle_meta(&job, &builder), + ) + .await + else { + return; + }; + reply + } + Err(e) => { + tracing::warn!(error = %e, "an undecodable meta build; answering with the error"); + FromMetaReply::Failed { + error_tail: format!("the builder could not decode the request: {e}"), + } + } + }; + match serde_json::to_vec(&reply) { + Ok(body) => { + if let Err(e) = queue.publish(reply_to, body.into()).await { + tracing::warn!(error = %e, "publishing a meta build's reply failed"); + } + } + Err(e) => tracing::warn!(error = %e, "encoding a meta build's reply failed"), + } + }); + } + tracing::error!("the subscription to {META_SUBJECT} ended"); +} + +/// `build`'s reply once a request holds the build slot, or `refuse`'s, without /// running `build`, once it has waited [`MAX_SLOT_WAIT`] since `received`. /// `None` only when the slot is closed. -async fn in_slot( +async fn in_slot( slot: &tokio::sync::Semaphore, received: tokio::time::Instant, - job: &BuildAgent, - build: impl Future, -) -> Option { + refuse: impl FnOnce() -> T, + build: impl Future, +) -> Option { let Ok(permit) = tokio::time::timeout_at(received + MAX_SLOT_WAIT, slot.acquire()).await else { - tracing::warn!( - agent = %job.agent, - rev = %job.rev, - pr = job.pr, - waited_secs = MAX_SLOT_WAIT.as_secs(), - "build refused: another build held the slot too long for this one to finish \ - before swarm-controller stops waiting" - ); - return Some(Reply::Failed { - error_tail: format!( - "the swarm builder was busy with another build for {} s, too long for this one \ - to finish before swarm-controller stops waiting for it", - MAX_SLOT_WAIT.as_secs() - ), - }); + return Some(refuse()); }; let _permit = permit.ok()?; Some(build.await) } +/// The refusal text of a request that waited [`MAX_SLOT_WAIT`] for the slot. +fn busy_tail() -> String { + format!( + "the swarm builder was busy with another build for {} s, too long for this one \ + to finish before swarm-controller stops waiting for it", + MAX_SLOT_WAIT.as_secs() + ) +} + +/// Log and answer a PR check that waited too long for the slot. +fn busy(job: &BuildAgent) -> Reply { + tracing::warn!( + agent = %job.agent, + rev = %job.rev, + pr = job.pr, + waited_secs = MAX_SLOT_WAIT.as_secs(), + "build refused: another build held the slot too long for this one to finish \ + before swarm-controller stops waiting" + ); + Reply::Failed { + error_tail: busy_tail(), + } +} + +/// Log and answer a meta build that waited too long for the slot. +fn busy_meta(job: &BuildFromMeta) -> FromMetaReply { + tracing::warn!( + agent = %job.agent, + meta_rev = %job.meta_rev, + waited_secs = MAX_SLOT_WAIT.as_secs(), + "meta build refused: another build held the slot too long for this one to finish \ + before swarm-controller stops waiting" + ); + FromMetaReply::Unavailable { + reason: busy_tail(), + } +} + +/// Build one meta request and log the build. +async fn handle_meta(job: &BuildFromMeta, builder: &nix_build::Builder) -> FromMetaReply { + let reply = match within(TOKEN_DEADLINE, forge_token()).await { + Ok(token) => builder.build_from_meta(job, &token).await, + Err(e) => FromMetaReply::Unavailable { + reason: format!("{e:#}"), + }, + }; + match &reply { + FromMetaReply::Built { out_path, system } => tracing::info!( + agent = %job.agent, + meta_rev = %job.meta_rev, + %out_path, + %system, + "agent meta build built" + ), + FromMetaReply::Failed { error_tail } => tracing::warn!( + agent = %job.agent, + meta_rev = %job.meta_rev, + %error_tail, + "agent meta build failed" + ), + FromMetaReply::Unavailable { reason } => tracing::warn!( + agent = %job.agent, + meta_rev = %job.meta_rev, + %reason, + "agent meta build not run" + ), + } + reply +} + /// Build one request and log the build. async fn handle(job: &BuildAgent, builder: &nix_build::Builder) -> Reply { let reply = match within(TOKEN_DEADLINE, forge_token()).await { @@ -406,7 +541,7 @@ mod tests { out_path: "/nix/store/x".to_owned(), } }; - let reply = in_slot(&slot, received, &job(), build).await; + let reply = in_slot(&slot, received, || busy(&job()), build).await; (reply, ran.load(Ordering::SeqCst), received.elapsed()) } @@ -430,6 +565,44 @@ mod tests { assert!(built); } + /// A meta build waits on the slot a PR check holds, and is refused at the + /// same limit. + #[tokio::test(start_paused = true)] + async fn a_meta_build_shares_the_slot_with_pr_checks() { + let meta = BuildFromMeta { + agent: "atlas".to_owned(), + meta_rev: "abc".to_owned(), + }; + let slot = Arc::new(Semaphore::new(1)); + let check = Arc::clone(&slot).acquire_owned().await.expect("open"); + let received = Instant::now(); + let refused = in_slot(&slot, received, || busy_meta(&meta), async { + FromMetaReply::Built { + out_path: "/nix/store/x".to_owned(), + system: "s".to_owned(), + } + }) + .await; + assert!( + matches!(&refused, Some(FromMetaReply::Unavailable { reason }) if reason.contains("busy")), + "{refused:?}" + ); + assert_eq!(received.elapsed(), MAX_SLOT_WAIT); + // Control: once the check lets go, the same request builds. + drop(check); + let built = in_slot(&slot, Instant::now(), || busy_meta(&meta), async { + FromMetaReply::Built { + out_path: "/nix/store/x".to_owned(), + system: "s".to_owned(), + } + }) + .await; + assert!( + matches!(built, Some(FromMetaReply::Built { .. })), + "{built:?}" + ); + } + #[tokio::test(start_paused = true)] async fn a_token_read_past_its_deadline_is_an_error_and_a_prompt_one_is_kept() { let started = Instant::now(); diff --git a/swarm-builder/src/nix_build.rs b/swarm-builder/src/nix_build.rs index 23eff5a9..71f52f67 100644 --- a/swarm-builder/src/nix_build.rs +++ b/swarm-builder/src/nix_build.rs @@ -1,7 +1,8 @@ //! One build: `nix build` of the check flake with its three inputs pointed at //! this host's hyperhive and nixpkgs and at the agent's config -//! commit. One lock: `nix flake lock` of the swarm meta flake in a scratch -//! directory. +//! commit. One meta build: `nix build` of one agent's toplevel in +//! `internal/meta` at a commit, every input at the flake's own lock. One lock: +//! `nix flake lock` of the swarm meta flake in a scratch directory. //! //! The config repo is private, so `git` fetches it with the builder's forge //! token. The token reaches `git` through a credential helper that reads it @@ -17,7 +18,9 @@ use std::path::PathBuf; use std::process::Stdio; use std::time::Duration; -use swarm_wire::builder::{BuildAgent, LockMeta, LockReply, Reply}; +use swarm_wire::builder::{ + BuildAgent, BuildFromMeta, FromMetaReply, LockMeta, LockReply, META_REPO_PATH, Reply, +}; use tokio::io::AsyncReadExt as _; /// The most of `nix`'s stderr a failed build's reply carries, from its end. @@ -54,6 +57,8 @@ pub struct Builder { pub deadline: Duration, /// How long one lock may run. pub lock_deadline: Duration, + /// The platform this host builds for, reported with each meta build. + pub system: String, } impl Builder { @@ -63,10 +68,27 @@ impl Builder { return Reply::Failed { error_tail: error }; } let mut cmd = self.nix_command(token); - cmd.args(self.args(job)); + cmd.args(BUILD_FLAGS).args(self.args(job)); run(cmd, self.deadline).await } + /// Build `job`'s agent from `internal/meta` with `token` as the forge + /// credential. + pub async fn build_from_meta(&self, job: &BuildFromMeta, token: &str) -> FromMetaReply { + if let Err(error_tail) = validate_meta(job) { + return FromMetaReply::Failed { error_tail }; + } + let mut cmd = self.nix_command(token); + cmd.args(BUILD_FLAGS).arg(self.meta_attr(job)); + match run(cmd, self.deadline).await { + Reply::Built { out_path } => FromMetaReply::Built { + out_path, + system: self.system.clone(), + }, + Reply::Failed { error_tail } => FromMetaReply::Failed { error_tail }, + } + } + /// Lock `job`'s flake with `token` as the forge credential, in a scratch /// directory removed afterwards, and answer with the lock `nix` wrote. pub async fn lock(&self, job: &LockMeta, token: &str) -> LockReply { @@ -119,21 +141,9 @@ impl Builder { cmd } - /// The `nix build` arguments for `job`. + /// The `nix build` arguments for `job`, after [`BUILD_FLAGS`]. fn args(&self, job: &BuildAgent) -> Vec { - [ - "build", - "--no-link", - "--print-out-paths", - "--no-write-lock-file", - "--option", - "accept-flake-config", - "false", - "--print-build-logs", - ] - .into_iter() - .map(str::to_owned) - .chain([ + vec![ format!("{}#check", self.check_flake), "--override-input".to_owned(), "hyperhive".to_owned(), @@ -144,8 +154,17 @@ impl Builder { "--override-input".to_owned(), "agent".to_owned(), self.agent_url(job), - ]) - .collect() + ] + } + + /// The agent's toplevel in `internal/meta` at the commit, the attribute + /// the hives build. No input is overridden: the committed `flake.lock` + /// decides all of them. + fn meta_attr(&self, job: &BuildFromMeta) -> String { + format!( + "git+{}/{META_REPO_PATH}.git?rev={}#nixosConfigurations.{}.config.system.build.toplevel", + self.forge_url, job.meta_rev, job.agent + ) } /// The config commit as a flake reference, fetched through the PR's head @@ -158,6 +177,19 @@ impl Builder { } } +/// What every `nix build` starts with: nothing linked, nothing written to the +/// flake's lock, the flake's own `nix.conf` settings refused. +const BUILD_FLAGS: [&str; 8] = [ + "build", + "--no-link", + "--print-out-paths", + "--no-write-lock-file", + "--option", + "accept-flake-config", + "false", + "--print-build-logs", +]; + /// `nix flake lock` of the flake in the working directory. The flake's config /// is not trusted, and nothing is built. const LOCK_ARGS: &[&str] = &["flake", "lock", "--option", "accept-flake-config", "false"]; @@ -207,24 +239,32 @@ fn credential_helper() -> String { /// Refuse a job whose fields would change the reference's meaning: an agent /// is one `[a-z0-9-]` segment, and a rev is a full commit id. fn validate(job: &BuildAgent) -> Result<(), String> { - let agent_ok = !job.agent.is_empty() - && job - .agent + check_agent_and_rev(&job.agent, &job.rev)?; + if job.pr == 0 { + return Err("PR 0 does not exist".to_owned()); + } + Ok(()) +} + +/// [`validate`] for a meta build: the same agent and commit rules. +fn validate_meta(job: &BuildFromMeta) -> Result<(), String> { + check_agent_and_rev(&job.agent, &job.meta_rev) +} + +fn check_agent_and_rev(agent: &str, rev: &str) -> Result<(), String> { + let agent_ok = !agent.is_empty() + && agent .bytes() .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-'); if !agent_ok { - return Err(format!("{:?} is not an agent name", job.agent)); + return Err(format!("{agent:?} is not an agent name")); } - let rev_ok = job.rev.len() == 40 - && job - .rev + let rev_ok = rev.len() == 40 + && rev .bytes() .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)); if !rev_ok { - return Err(format!("{:?} is not a full commit id", job.rev)); - } - if job.pr == 0 { - return Err("PR 0 does not exist".to_owned()); + return Err(format!("{rev:?} is not a full commit id")); } Ok(()) } @@ -387,6 +427,7 @@ mod tests { nix: PathBuf::from("nix"), deadline: BUILD_DEADLINE, lock_deadline: LOCK_DEADLINE, + system: "x86_64-linux".to_owned(), } } @@ -409,7 +450,88 @@ mod tests { .collect(); assert_eq!(overridden, ["hyperhive", "nixpkgs", "agent"]); assert!(args.iter().any(|a| a == "/nix/store/c-check#check")); - assert!(args.iter().any(|a| a == "--no-write-lock-file")); + assert!(BUILD_FLAGS.contains(&"--no-write-lock-file")); + } + + fn meta_job() -> BuildFromMeta { + BuildFromMeta { + agent: "atlas".to_owned(), + meta_rev: "0123456789abcdef0123456789abcdef01234567".to_owned(), + } + } + + #[test] + fn a_meta_build_is_the_agents_toplevel_in_internal_meta_at_the_commit() { + assert_eq!( + builder().meta_attr(&meta_job()), + "git+https://forge.example/internal/meta.git\ + ?rev=0123456789abcdef0123456789abcdef01234567\ + #nixosConfigurations.atlas.config.system.build.toplevel" + ); + } + + #[test] + fn a_meta_job_that_would_bend_the_reference_is_refused() { + for bad in [ + BuildFromMeta { + agent: "a#b".to_owned(), + ..meta_job() + }, + BuildFromMeta { + agent: String::new(), + ..meta_job() + }, + BuildFromMeta { + meta_rev: "main".to_owned(), + ..meta_job() + }, + ] { + assert!(validate_meta(&bad).is_err(), "{bad:?}"); + } + assert!(validate_meta(&meta_job()).is_ok()); + } + + #[tokio::test] + async fn a_meta_build_reports_the_path_and_the_hosts_system() { + // `echo` stands in for `nix`: it prints the arguments it was given, which + // are taken for the store path, and exits 0. + let b = Builder { + nix: PathBuf::from("echo"), + ..builder() + }; + let reply = b.build_from_meta(&meta_job(), "t").await; + assert_eq!( + reply, + FromMetaReply::Built { + out_path: format!("{} {}", BUILD_FLAGS.join(" "), b.meta_attr(&meta_job())), + system: "x86_64-linux".to_owned(), + } + ); + } + + #[tokio::test] + async fn a_failed_meta_build_is_a_failed_reply_and_a_refused_job_is_failed_without_nix() { + let b = Builder { + nix: PathBuf::from("false"), + ..builder() + }; + let failed = b.build_from_meta(&meta_job(), "t").await; + assert!(matches!(failed, FromMetaReply::Failed { .. }), "{failed:?}"); + // Control: `a_meta_build_reports_the_path_and_the_hosts_system` builds this + // same job, so the refusal below is the validation, not the stub. + let refused = b + .build_from_meta( + &BuildFromMeta { + meta_rev: "main".to_owned(), + ..meta_job() + }, + "t", + ) + .await; + assert!( + matches!(&refused, FromMetaReply::Failed { error_tail } if error_tail.contains("commit id")), + "{refused:?}" + ); } /// What the helper prints for `op` and a request with `protocol` and diff --git a/swarm-controller/src/builder.rs b/swarm-controller/src/builder.rs index 88323bca..468ba2ec 100644 --- a/swarm-controller/src/builder.rs +++ b/swarm-controller/src/builder.rs @@ -4,7 +4,8 @@ //! swarm UI's builds page for that build. //! //! Nothing is pushed, and the closure the builder made stays in its host's -//! store. A red status blocks nothing; the merge deploys as before. +//! store. A red status blocks nothing; the merge deploys as before. The build +//! that does gate a deploy is `meta_build`'s. //! //! Builds share one `SwarmResourceKind::Builder` slot, so one runs at a time, //! and a newer push replaces its PR's build while that build is still diff --git a/swarm-controller/src/jobq.rs b/swarm-controller/src/jobq.rs index 1ca5a21f..5c39b793 100644 --- a/swarm-controller/src/jobq.rs +++ b/swarm-controller/src/jobq.rs @@ -10,8 +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, meta_fanout, - meta_flake, wanted, + agent_identity, agent_renewal, auth, builder, config_pr, forge, matrix_account, meta_build, + meta_fanout, meta_flake, wanted, }; /// Node payload for the swarm-level job graph. Named `Swarm*` rather than @@ -110,6 +110,13 @@ pub(crate) enum SwarmNodeKind { /// Carries no hive: the check builds on the builder's own agent base, /// not on any hive. BuildAgent { agent: String, rev: String, pr: u64 }, + /// Build `agent` from `internal/meta`'s head on the swarm builder, and + /// record the closure in `agent-closures`. The `TriggerDeploy` behind it + /// runs only when the build is green. See `meta_build`. + /// + /// Carries no rev: it reads the head when it runs, which in most DAGs is + /// after the `UpdateMeta` that commits it. + BuildMeta { agent: String }, /// Re-mint `agent`'s queue secret if it is still stored and old enough. /// See `agent_renewal`. /// @@ -136,15 +143,16 @@ pub(crate) enum SwarmNodeKind { }, /// Pin the swarm meta flake (`internal/meta`): render, lock on the swarm /// builder, commit. See `meta_flake`. When the pin `internal/meta` holds is - /// not the one `meta_fanout` recorded, a `TriggerDeploy` per placed agent - /// follows, then a `RecordFanout`. + /// not the one `meta_fanout` recorded, a `BuildMeta` and a `TriggerDeploy` + /// per placed agent follow, then a `RecordFanout`. /// /// Carries nothing: it reads every head when it runs, so one pending node /// covers every trigger queued behind it. UpdateMeta, - /// Record `pin` in `meta_fanout` as sent to the hives. Runs only once every - /// `TriggerDeploy` declared beside it is done, so a failed or lost deploy - /// leaves the record stale and the next `UpdateMeta` sends them again. + /// Record `pin` in `meta_fanout` as sent to the hives. Runs once every + /// `TriggerDeploy` declared beside it is done or skipped by its build gate. + /// A deploy that failed or was lost leaves the record stale, and the next + /// `UpdateMeta` sends them again. RecordFanout { pin: String }, } @@ -166,6 +174,7 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind { SwarmNodeKind::MintBuilderForgeToken => "mint_builder_forge_token".to_owned(), SwarmNodeKind::MintCiMetricsForgeToken => "mint_ci_metrics_forge_token".to_owned(), SwarmNodeKind::BuildAgent { .. } => "build_agent".to_owned(), + SwarmNodeKind::BuildMeta { .. } => "build_meta".to_owned(), SwarmNodeKind::RenewAgentQueueCredential { .. } => { "renew_agent_queue_credential".to_owned() } @@ -193,6 +202,7 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind { | SwarmNodeKind::MintAgentIdentity { agent } | SwarmNodeKind::MintAgentForgeToken { agent } | SwarmNodeKind::MintAgentMatrixAccount { agent } + | SwarmNodeKind::BuildMeta { agent } | SwarmNodeKind::RenewAgentQueueCredential { agent } => { serde_json::json!({ "agent": agent }) } @@ -220,8 +230,9 @@ impl hive_jobq_wire::WireNode for SwarmNodeKind { /// has capacity 1: the scheduler's default, which no caller overrides. #[derive(Clone, Debug, PartialEq, Eq, Hash)] 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. + /// The swarm builder: one `BuildAgent` or `BuildMeta` node runs at a time, + /// and the rest wait pending. A newer push replaces its PR's pending + /// `BuildAgent`. Builder, /// `internal/meta`: one `UpdateMeta` commits at a time, so two never race /// on the same file's blob. @@ -403,6 +414,7 @@ async fn run_node_work( SwarmNodeKind::BuildAgent { agent, rev, pr } => { build_agent(&deps, forge, agent, rev, pr).await } + SwarmNodeKind::BuildMeta { agent } => build_meta(&deps, forge, agent).await, SwarmNodeKind::RenewAgentQueueCredential { agent } => renew_queue_credential(&agent).await, SwarmNodeKind::SetAgentWanted { hive, agent } => match deps.wanted { None => Outcome::Failed( @@ -428,7 +440,7 @@ async fn run_node_work( /// A forge-calling node's work, run under [`forge::FORGE_DEADLINE`]. Elapsing /// is the node's own error, naming `node`, so it fails like any other refused /// step instead of holding its executor slot. -async fn forge_node( +pub(crate) async fn forge_node( node: &str, work: impl std::future::Future>, ) -> Result { @@ -590,6 +602,15 @@ pub(crate) async fn build_agent( .await } +/// The `BuildMeta` arm, lifted out for `clippy::too_many_lines`. +async fn build_meta( + deps: &WorkerDeps, + forge: Option>, + agent: String, +) -> hive_jobq::scheduler::Outcome { + meta_build::run(forge.as_deref(), deps.queue.as_ref(), agent).await +} + /// What an `UpdateMeta` pass decided: the deploys to send, and the placements /// to send them to. struct MetaPass { @@ -717,9 +738,48 @@ async fn record_fanout( } } -/// Declare one rebuild of each `(hive, agent)` placement, which nothing else -/// tells to build again, and for a [`FanOut::Record`] the `RecordFanout` that -/// waits on every one of them. +/// The build and the deploy every path declares for one agent: a `BuildMeta` +/// of `agent`, then the `TriggerDeploy`, which runs only when the build is +/// green (`after_ok`), so a red build leaves it `Skipped`. +/// +/// `update` is the `UpdateMeta` the pair follows when it is not this pair's +/// parent. The build waits on it, since the build reads the head that update +/// commits. The deploy keeps its own edge to it too: `skipped_deploys` finds +/// the deploys a failed update skipped by that edge. +fn gated_deploy<'a>( + b: &'a hive_jobq::builder::JobBuilder, + hive: &str, + agent: &str, + rev: Option, + update: Option, +) -> hive_jobq::builder::NodeRef<'a, SwarmNodeKind, SwarmResourceKind> { + let mut build = b + .node(SwarmNodeKind::BuildMeta { + agent: agent.to_owned(), + }) + .needs(SwarmResourceKind::Builder); + if let Some(update) = update { + build = build.after_ok(update); + } + let mut deploy = b + .node(SwarmNodeKind::TriggerDeploy { + hive: hive.to_owned(), + agent: agent.to_owned(), + rev, + }) + .after_ok(build); + if let Some(update) = update { + deploy = deploy.after_ok(update); + } + deploy +} + +/// Declare one build and rebuild of each `(hive, agent)` placement, which +/// nothing else tells to build again, and for a [`FanOut::Record`] the +/// `RecordFanout` that waits on every one of them. A deploy the build gate +/// skipped counts as sent: the pin is recorded, so one red agent does not +/// resend every healthy one on each retry. A deploy skipped because the build +/// gave no verdict is re-declared by `queue_meta_update`. fn declare_fan_out( grown: &hive_jobq::builder::JobBuilder, fan_out: &FanOut, @@ -730,20 +790,18 @@ fn declare_fan_out( } let deploys: Vec = placements .iter() - .map(|(hive, agent)| { - grown - .node(SwarmNodeKind::TriggerDeploy { - hive: hive.clone(), - agent: agent.clone(), - rev: None, - }) - .guid() - }) + .map(|(hive, agent)| gated_deploy(grown, hive, agent, None, None).guid()) .collect(); if let FanOut::Record { pin } = fan_out { let mut record = grown.node(SwarmNodeKind::RecordFanout { pin: pin.clone() }); for deploy in deploys { - record = record.after_ok(deploy); + record = record.on_outcome( + deploy, + &[ + hive_jobq::TerminalState::Done, + hive_jobq::TerminalState::Skipped, + ], + ); } } } @@ -1136,13 +1194,13 @@ 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`. +/// The sub-DAG a merged config PR is: the meta update, the build of the agent +/// from the updated flake, then the deploy of `rev` on `hive`. /// /// `after_ok`: a hive builds only from `internal/meta`, so a deploy asked for /// after a failed update (no builder, an unreadable config repo) could only -/// fail its build. The failed update is `Failed` in the graph and the deploy -/// `Skipped`. +/// fail its build, and one asked for after a red build would fail the same +/// way. The failed node is `Failed` in the graph and the deploy `Skipped`. fn merged_deploy_job( b: &hive_jobq::builder::JobBuilder, hive: &str, @@ -1153,13 +1211,7 @@ fn merged_deploy_job( .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_ok(update_meta); + let _deploy = gated_deploy(b, hive, agent, Some(rev.to_owned()), Some(root)); vec![root] } @@ -1252,8 +1304,9 @@ pub(crate) fn declare_agent_job( agent: agent.to_owned(), }); // The hive builds the agent from `internal/meta` only, so the deploy needs - // the update to succeed (`after_ok`): after a failed one (no builder, an - // unreadable config repo) the hive's build could only fail. + // the update and the swarm's build from it to succeed (`after_ok`): after a + // failed update (no builder, an unreadable config repo) or a red build, the + // hive's build could only fail. let update_meta = b .node(SwarmNodeKind::UpdateMeta) .needs(SwarmResourceKind::Meta) @@ -1282,13 +1335,7 @@ pub(crate) fn declare_agent_job( // queue at all fails this node — but `TriggerDeploy` has nothing to // publish to on such a host either, so the strict edge cancels a node // that could not have succeeded anyway. - let trigger_deploy = b - .node(SwarmNodeKind::TriggerDeploy { - hive: hive.to_owned(), - agent: agent.to_owned(), - rev: None, - }) - .after_ok(update_meta) + let trigger_deploy = gated_deploy(b, hive, agent, None, Some(update_meta.guid())) .after_any(mint_identity) .after_any(mint_forge_token) // `after_any` for the same reason: the container reads this token @@ -1488,15 +1535,34 @@ pub(crate) fn spawn_meta_update_timer( }); } -/// The deploys a failed `UpdateMeta` skipped, as `(hive, agent, rev)`: for each -/// agent, its latest `TriggerDeploy` when it waited on an `UpdateMeta` that -/// `Failed` (so the scheduler skipped it). A deploy skipped for another reason -/// (the pause declaration or the config seeding failed) is left alone: the -/// update that would unblock it was skipped too, and deploying past those is -/// the race those edges close. +/// Why a deploy is `Skipped` and the timer should declare it again. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum SkipCause { + /// It waited on an `UpdateMeta` that `Failed`. + FailedUpdate, + /// It waited on a `BuildMeta` that gave no verdict ([`meta_build::is_retried`]). + UnfinishedBuild, +} + +/// A deploy [`skipped_deploys`] found. +struct SkippedDeploy { + hive: String, + agent: String, + rev: Option, + cause: SkipCause, +} + +/// The deploys the timer declares again: for each agent, its latest +/// `TriggerDeploy` when the scheduler skipped it because an `UpdateMeta` it +/// waited on `Failed`, or because the `BuildMeta` it waited on gave no verdict. +/// A deploy skipped for another reason is left alone. A red build is a verdict +/// on the config, which a config merge answers with a build of its own. The +/// pause declaration or the config seeding failing means the update that would +/// unblock the deploy was skipped too, and deploying past those is the race +/// those edges close. fn skipped_deploys( graph: &hive_jobq::Graph, -) -> Vec<(String, String, Option)> { +) -> Vec { let mut latest = std::collections::BTreeMap::new(); for n in graph.nodes() { if let SwarmNodeKind::TriggerDeploy { hive, agent, .. } = &n.payload { @@ -1511,30 +1577,52 @@ fn skipped_deploys( } latest .into_values() - .filter(|n| { - n.deps.iter().any(|d| match d { - hive_jobq::Dep::Node { id, .. } => graph.node(*id).is_some_and(|dep| { - matches!(dep.payload, SwarmNodeKind::UpdateMeta) - && dep.state == hive_jobq::State::Failed - }), - hive_jobq::Dep::Resource { .. } => false, + .filter_map(|n| { + let SwarmNodeKind::TriggerDeploy { hive, agent, rev } = &n.payload else { + return None; + }; + let waited_on = + |wanted: fn(&hive_jobq::Node) -> bool| { + n.deps.iter().any(|d| match d { + hive_jobq::Dep::Node { id, .. } => graph.node(*id).is_some_and(wanted), + hive_jobq::Dep::Resource { .. } => false, + }) + }; + let cause = if waited_on(|dep| { + matches!(dep.payload, SwarmNodeKind::UpdateMeta) + && dep.state == hive_jobq::State::Failed + }) { + SkipCause::FailedUpdate + } else if waited_on(|dep| { + matches!(dep.payload, SwarmNodeKind::BuildMeta { .. }) + && dep.state == hive_jobq::State::Failed + && dep.error.as_deref().is_some_and(meta_build::is_retried) + }) { + SkipCause::UnfinishedBuild + } else { + return None; + }; + Some(SkippedDeploy { + hive: hive.clone(), + agent: agent.clone(), + rev: rev.clone(), + cause, }) }) - .filter_map(|n| match &n.payload { - SwarmNodeKind::TriggerDeploy { hive, agent, rev } => { - Some((hive.clone(), agent.clone(), rev.clone())) - } - _ => None, - }) .collect() } -/// Insert one `UpdateMeta` job when none is in flight and the latest one, by -/// node id, is not `Done` (or none exists), with a fresh `TriggerDeploy` -/// behind it (`after_ok`) for every deploy the failed run skipped -/// ([`skipped_deploys`]), so an agent created or merged while the update was -/// failing is deployed by the run that succeeds. Returns whether a job was -/// inserted. +/// Insert one job when the latest `UpdateMeta`, by node id, is not `Done` (or +/// none exists) and none is in flight: an `UpdateMeta`, with a fresh build and +/// `TriggerDeploy` behind it ([`gated_deploy`]) for every deploy the failed +/// run skipped ([`skipped_deploys`]), so an agent created or merged while the +/// update was failing is deployed by the run that succeeds. +/// +/// When the latest `UpdateMeta` is `Done`, a deploy skipped because the builder +/// gave no verdict still gets a fresh build and deploy, with no update, so a +/// builder that was down or not yet updated is caught up once it answers. That +/// adds nodes on every tick while the builder stays away. Returns whether a job +/// was inserted. fn queue_meta_update( sched: &Mutex>, ) -> Result { @@ -1555,24 +1643,28 @@ fn queue_meta_update( return Ok(false); } update_metas.sort_by_key(|n| n.id); - if update_metas + let update_due = update_metas .last() - .is_some_and(|n| n.state == hive_jobq::State::Done) - { + .is_none_or(|n| n.state != hive_jobq::State::Done); + let skipped: Vec = skipped_deploys(sched.graph()) + .into_iter() + .filter(|s| update_due || s.cause == SkipCause::UnfinishedBuild) + .collect(); + if !update_due && skipped.is_empty() { return Ok(false); } - let skipped = skipped_deploys(sched.graph()); sched .insert_job(None, |b| { - let update = b - .node(SwarmNodeKind::UpdateMeta) - .needs(SwarmResourceKind::Meta); - for (hive, agent, rev) in skipped { - let _ = b - .node(SwarmNodeKind::TriggerDeploy { hive, agent, rev }) - .after_ok(update); - } - vec![update.guid()] + let update = update_due.then(|| { + b.node(SwarmNodeKind::UpdateMeta) + .needs(SwarmResourceKind::Meta) + .guid() + }); + let deploys: Vec<_> = skipped + .into_iter() + .map(|s| gated_deploy(b, &s.hive, &s.agent, s.rev, update).guid()) + .collect(); + update.map_or(deploys, |update| vec![update]) }) .map_err(|e| anyhow::anyhow!("{e}"))?; Ok(true) @@ -2228,12 +2320,152 @@ mod tests { )); } + /// A deploy needs the swarm's build of its agent to succeed: its edge + /// refuses a failed build, in both DAGs that declare an update. The mint + /// edge is the control that the lookup finds an edge that accepts one. + #[test] + fn a_deploy_needs_the_meta_build_to_succeed() { + for declare in [create_dag as fn(&TestBuilder), merged_dag] { + assert!(!edge_accepts_failure( + declare, + "trigger_deploy", + "build_meta" + )); + assert!(!edge_accepts_failure(declare, "build_meta", "update_meta")); + } + assert!(edge_accepts_failure( + create_dag, + "trigger_deploy", + "mint_agent_identity" + )); + } + + /// Every `TriggerDeploy` in `graph` waits, strictly, on a `BuildMeta` of the + /// same agent that needs the builder. Panics, naming `path`, when one does + /// not; returns how many deploys it checked. + fn assert_every_deploy_is_gated( + graph: &hive_jobq::Graph, + path: &str, + ) -> usize { + let mut checked = 0; + for deploy in graph.nodes() { + let SwarmNodeKind::TriggerDeploy { agent, .. } = &deploy.payload else { + continue; + }; + let gate = deploy.deps.iter().find_map(|d| { + match d { + hive_jobq::Dep::Node { id, when } => graph + .node(*id) + .filter(|n| { + matches!(&n.payload, SwarmNodeKind::BuildMeta { agent: a } if a == agent) + }) + .map(|n| (n, *when)), + hive_jobq::Dep::Resource { .. } => None, + } + }); + let (build, when) = + gate.unwrap_or_else(|| panic!("{path}: {agent}'s deploy has no build")); + assert!( + !when.accepts(hive_jobq::TerminalState::Failed), + "{path}: {agent}'s deploy runs after a failed build" + ); + assert!( + build.deps.iter().any(|d| matches!( + d, + hive_jobq::Dep::Resource { + name: SwarmResourceKind::Builder, + .. + } + )), + "{path}: {agent}'s build does not need the builder" + ); + checked += 1; + } + checked + } + + /// The four paths that send a deploy all declare the same build and deploy + /// pair, so a path that forgets the gate fails here. + #[tokio::test] + async fn all_four_deploy_paths_declare_the_gated_pair() { + let graph_of = |declare: fn(&TestBuilder)| { + 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"); + sched + }; + assert_eq!( + assert_every_deploy_is_gated(graph_of(merged_dag).graph(), "merge"), + 1 + ); + assert_eq!( + assert_every_deploy_is_gated(graph_of(create_dag).graph(), "creation"), + 1 + ); + + let pins = Arc::new(FakePins::default()); + let (_, sched) = pass(&pins, &report("p1", false), &PLACED, None, false).await; + assert_eq!( + assert_every_deploy_is_gated(sched.lock().unwrap().graph(), "fan-out"), + 3 + ); + + let sched = bob_deployed_and_atlas_skipped().await; + assert!(queue_meta_update(&sched).expect("tick")); + // bob's and atlas's first deploys and atlas's retry: all gated. + assert_eq!( + assert_every_deploy_is_gated(sched.lock().unwrap().graph(), "timer retry"), + 3 + ); + } + + /// With the swarm's build red, the deploy never runs, in both DAGs that + /// declare an update. The first run in each pair is the control, where the + /// build is green and the deploy does run. + #[tokio::test] + async fn a_red_build_means_no_deploy_runs() { + use hive_jobq::State; + + for (name, declare) in [ + ("merged", merged_dag as fn(&TestBuilder)), + ("create", create_dag as fn(&TestBuilder)), + ] { + let green = run_dag_failing(declare, None).await; + assert_eq!(state_of(&green, "build_meta"), State::Done, "{name}"); + assert_eq!(state_of(&green, "trigger_deploy"), State::Done, "{name}"); + + let red = run_dag_failing(declare, Some("build_meta")).await; + assert_eq!(state_of(&red, "update_meta"), State::Done, "{name}"); + assert_eq!(state_of(&red, "build_meta"), State::Failed, "{name}"); + assert_eq!( + state_of(&red, "trigger_deploy"), + State::Skipped, + "{name}: the deploy must not run after a red build" + ); + } + } + /// Every node of the DAG `declare` builds, run to the end with the meta /// update failing or succeeding and every other node succeeding: the /// final state of each, by label. async fn run_dag( declare: impl FnOnce(&TestBuilder), update_meta_fails: bool, + ) -> Vec<(String, hive_jobq::State)> { + run_dag_failing(declare, update_meta_fails.then_some("update_meta")).await + } + + /// [`run_dag`], with the nodes labelled `failing` failing. + async fn run_dag_failing( + declare: impl FnOnce(&TestBuilder), + failing: Option<&'static str>, ) -> Vec<(String, hive_jobq::State)> { use hive_jobq::scheduler::{Outcome, Scheduler}; use hive_jobq_wire::WireNode as _; @@ -2250,11 +2482,10 @@ mod tests { .expect("the graph inserts"); let sched = std::sync::Arc::new(std::sync::Mutex::new(sched)); while let Some(runner) = Scheduler::claim_next(&sched, |_id, kind, builder| async move { - let outcome = match kind { - SwarmNodeKind::UpdateMeta if update_meta_fails => { - Outcome::Failed("no builder".to_owned()) - } - _ => Outcome::Done, + let outcome = if failing == Some(kind.label().as_str()) { + Outcome::Failed("no builder".to_owned()) + } else { + Outcome::Done }; (builder, outcome) }) { @@ -2355,6 +2586,27 @@ mod tests { placed: &[(&str, &str)], failing_agent: Option<&'static str>, crash_before_record: bool, + ) -> (FanOut, Arc>) { + pass_with( + pins, + report, + placed, + failing_agent, + crash_before_record, + None, + ) + .await + } + + /// [`pass`], with the build of one agent failing with the given error when + /// `failing_build` names it. + async fn pass_with( + pins: &Arc, + report: &meta_flake::Report, + placed: &[(&str, &str)], + failing_agent: Option<&'static str>, + crash_before_record: bool, + failing_build: Option<(&'static str, String)>, ) -> (FanOut, Arc>) { use hive_jobq::scheduler::{Outcome, Scheduler}; use meta_fanout::PinStore as _; @@ -2380,7 +2632,8 @@ mod tests { .expect("the node just inserted is runnable"); update.await.1.expect("the grown nodes insert"); while let Some(runner) = Scheduler::claim_next(&sched, { - let (pins, sched) = (Arc::clone(pins), Arc::clone(&sched)); + let (pins, sched, failing_build) = + (Arc::clone(pins), Arc::clone(&sched), failing_build.clone()); move |_id, kind, builder| async move { let outcome = match kind { SwarmNodeKind::TriggerDeploy { agent, .. } @@ -2388,6 +2641,11 @@ mod tests { { Outcome::Failed("the hive did not take it".to_owned()) } + SwarmNodeKind::BuildMeta { agent } + if failing_build.as_ref().is_some_and(|(a, _)| *a == agent) => + { + Outcome::Failed(failing_build.map(|(_, error)| error).expect("matched")) + } SwarmNodeKind::RecordFanout { .. } if crash_before_record => { Outcome::Failed("the controller died".to_owned()) } @@ -2398,12 +2656,17 @@ mod tests { .graph() .nodes() .filter(|n| matches!(n.payload, SwarmNodeKind::TriggerDeploy { .. })) - .all(|n| n.state == hive_jobq::State::Done); + .all(|n| { + matches!( + n.state, + hive_jobq::State::Done | hive_jobq::State::Skipped + ) + }); if deploys_done { pins.record(&pin).await.expect("recorded"); Outcome::Done } else { - Outcome::Failed("recorded before the deploys were done".to_owned()) + Outcome::Failed("recorded before the deploys were settled".to_owned()) } } _ => Outcome::Done, @@ -2509,10 +2772,10 @@ mod tests { assert_eq!(fan_out, FanOut::None); } - /// The record waits on every deploy of its pass, for all of them to be - /// done, not merely finished. + /// The record waits on every deploy of its pass to be settled: done, or + /// skipped by its build gate. A failed deploy still rules it out. #[tokio::test] - async fn the_record_waits_on_every_deploy_being_done() { + async fn the_record_waits_on_every_deploy_being_done_or_skipped() { let pins = Arc::new(FakePins::default()); let (_, sched) = pass(&pins, &report("p1", false), &PLACED, None, false).await; let guard = sched.lock().unwrap(); @@ -2533,7 +2796,15 @@ mod tests { let mut deploys: Vec<_> = graph .nodes() .filter(|n| matches!(n.payload, SwarmNodeKind::TriggerDeploy { .. })) - .map(|n| (n.id, hive_jobq::DepWhen::AFTER_OK)) + .map(|n| { + ( + n.id, + hive_jobq::DepWhen::of(&[ + hive_jobq::TerminalState::Done, + hive_jobq::TerminalState::Skipped, + ]), + ) + }) .collect(); deploys.sort_by_key(|(id, _)| id.get()); assert_eq!(deploys.len(), 3); @@ -2566,6 +2837,83 @@ mod tests { assert_eq!(fan_out, FanOut::Record { pin: "p1".into() }); } + /// A red build for one agent skips that agent's deploy only, and the pin + /// is still recorded: the red agent is not sent again with every healthy + /// one on the next retry. The timer declares nothing for it either, since a + /// red build is a verdict on its config. + #[tokio::test] + async fn a_red_build_skips_its_deploy_and_the_pin_is_still_recorded() { + use hive_jobq::State; + + let pins = Arc::new(FakePins::holding("p0")); + let (_, sched) = pass_with( + &pins, + &report("p1", true), + &PLACED, + None, + false, + Some(("argus", "error: builder for argus failed".to_owned())), + ) + .await; + let mut builds = states_of(&sched, "build_meta"); + builds.sort_by_key(|s| format!("{s:?}")); + assert_eq!(builds, [State::Done, State::Done, State::Failed]); + let mut deploys = states_of(&sched, "trigger_deploy"); + deploys.sort_by_key(|s| format!("{s:?}")); + assert_eq!(deploys, [State::Done, State::Done, State::Skipped]); + assert_eq!(states_of(&sched, "record_fanout"), [State::Done]); + assert_eq!(pins.held().as_deref(), Some("p1")); + + // The failed build rolls the update up `Failed`, so the timer queues an + // update, and only an update: no deploy is declared again. + assert!(queue_meta_update(&sched).expect("tick")); + assert_eq!(states_of(&sched, "trigger_deploy").len(), 3); + assert_eq!(states_of(&sched, "build_meta").len(), 3); + assert_eq!(states_of(&sched, "update_meta").len(), 2); + } + + /// A build that gave no verdict skips its deploy and still records the pin, + /// and the timer declares a build and deploy for that agent alone. + #[tokio::test] + async fn an_unfinished_build_is_declared_again_for_that_agent_alone() { + use hive_jobq::State; + use hive_jobq_wire::WireNode as _; + + let pins = Arc::new(FakePins::holding("p0")); + let error = format!("{}no swarm builder is listening", meta_build::RETRY_PREFIX); + let (_, sched) = pass_with( + &pins, + &report("p1", true), + &PLACED, + None, + false, + Some(("argus", error)), + ) + .await; + assert_eq!(states_of(&sched, "record_fanout"), [State::Done]); + assert_eq!(pins.held().as_deref(), Some("p1")); + + assert!(queue_meta_update(&sched).expect("tick")); + let guard = sched.lock().unwrap(); + let again = |label: &str| { + let mut found: Vec<_> = guard + .graph() + .nodes() + .filter(|n| n.payload.label() == label) + .filter(|n| n.state == State::Pending) + .map(|n| match &n.payload { + SwarmNodeKind::BuildMeta { agent } + | SwarmNodeKind::TriggerDeploy { agent, .. } => agent.clone(), + other => panic!("not a build or a deploy: {other:?}"), + }) + .collect(); + found.sort(); + found + }; + assert_eq!(again("build_meta"), ["argus"]); + assert_eq!(again("trigger_deploy"), ["argus"]); + } + /// With nothing placed there is nothing to wait on, and the pin is /// recorded all the same. #[tokio::test] @@ -3428,6 +3776,115 @@ mod tests { drain(&sched, false, false).await; assert_eq!(deploys_of(&sched, "atlas"), [State::Skipped]); } + + /// Run every runnable node; the build of `agent` fails with `error` and + /// everything else succeeds. + async fn drain_failing_build(sched: &SharedSched, agent: &'static str, error: String) { + use hive_jobq::scheduler::{Outcome, Scheduler}; + + while let Some(runner) = Scheduler::claim_next(sched, { + let error = error.clone(); + move |_id, kind, builder| async move { + let outcome = match kind { + SwarmNodeKind::BuildMeta { agent: a } if a == agent => Outcome::Failed(error), + _ => Outcome::Done, + }; + (builder, outcome) + } + }) { + runner.await.1.expect("no growth declared"); + } + } + + fn fresh_sched() -> SharedSched { + std::sync::Arc::new(std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( + hive_jobq::Graph::new(), + hive_jobq::resources::ResourceTable::new(), + ))) + } + + /// The `rev` of `agent`'s deploys, oldest first. + fn deploy_revs(sched: &SharedSched, agent: &str) -> Vec> { + let guard = sched.lock().unwrap(); + let mut nodes: Vec<_> = guard + .graph() + .nodes() + .filter_map(|n| match &n.payload { + SwarmNodeKind::TriggerDeploy { agent: a, rev, .. } if a == agent => { + Some((n.id, rev.clone())) + } + _ => None, + }) + .collect(); + nodes.sort_by_key(|(id, _)| *id); + nodes.into_iter().map(|(_, rev)| rev).collect() + } + + /// A merged config's build that gave no verdict, with the update `Done`: + /// the deploy is skipped, and the timer declares a build and the same + /// deploy again, without an update, until the builder answers. + #[tokio::test] + async fn an_unfinished_build_is_retried_until_the_builder_answers() { + use hive_jobq::State; + + let sched = fresh_sched(); + sched + .lock() + .unwrap() + .insert_job(None, |b| merged_deploy_job(b, "pr1ma", "atlas", "abc")) + .expect("the graph inserts"); + let unavailable = format!("{}no swarm builder is listening", meta_build::RETRY_PREFIX); + drain_failing_build(&sched, "atlas", unavailable.clone()).await; + assert_eq!(deploys_of(&sched, "atlas"), [State::Skipped]); + assert_eq!(update_metas(&sched), 1); + + assert!(queue_meta_update(&sched).expect("tick")); + assert_eq!(update_metas(&sched), 1, "the retry needs no update"); + drain_failing_build(&sched, "atlas", unavailable).await; + assert_eq!( + deploys_of(&sched, "atlas"), + [State::Skipped, State::Skipped] + ); + + assert!(queue_meta_update(&sched).expect("tick")); + drain(&sched, false, false).await; + assert_eq!( + deploys_of(&sched, "atlas"), + [State::Skipped, State::Skipped, State::Done] + ); + // The merged deploy's rev survives every retry. + assert_eq!( + deploy_revs(&sched, "atlas"), + [ + Some("abc".to_owned()), + Some("abc".to_owned()), + Some("abc".to_owned()) + ] + ); + assert!( + !queue_meta_update(&sched).expect("tick"), + "the answer ends the retries" + ); + } + + /// A red build is a verdict: the timer declares nothing for it. The test + /// above is the control that the same shape with an unfinished build is + /// declared again. + #[tokio::test] + async fn a_red_build_is_not_declared_again() { + use hive_jobq::State; + + let sched = fresh_sched(); + sched + .lock() + .unwrap() + .insert_job(None, |b| merged_deploy_job(b, "pr1ma", "atlas", "abc")) + .expect("the graph inserts"); + drain_failing_build(&sched, "atlas", "error: builder for x failed".to_owned()).await; + assert_eq!(deploys_of(&sched, "atlas"), [State::Skipped]); + assert!(!queue_meta_update(&sched).expect("tick")); + assert_eq!(deploys_of(&sched, "atlas"), [State::Skipped]); + } } #[cfg(test)] diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index f3990f92..dcde78ed 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -51,6 +51,7 @@ mod jobq; mod linked_accounts; mod matrix_account; mod message_stream; +mod meta_build; mod meta_fanout; mod meta_flake; mod notify_flows; diff --git a/swarm-controller/src/meta_build.rs b/swarm-controller/src/meta_build.rs new file mode 100644 index 00000000..d41e6ccc --- /dev/null +++ b/swarm-controller/src/meta_build.rs @@ -0,0 +1,351 @@ +//! The `BuildMeta` node: build one agent from `internal/meta` on the swarm +//! builder before its deploy is sent, and record the closure in +//! [`swarm_wire::agent_closures`]. +//! +//! The node reads `internal/meta`'s head when it runs, because in most DAGs it +//! is declared before the `UpdateMeta` that commits that head. It sends the +//! builder the agent and the head's commit, and on a `Built` reply writes the +//! key itself: the builder holds no grant on the bucket. A `Built` reply and +//! nothing else writes, so a red build leaves the previous record. +//! +//! The node ends one of three ways. `Done` is a green build. A red build, the +//! builder's `Failed` reply, is `Failed` with `nix`'s tail: the deploy behind it +//! is skipped and the Jobs page shows why. Anything that stopped the build from +//! giving a verdict is `Failed` with [`RETRY_PREFIX`], which +//! `jobq::queue_meta_update` retries on the meta timer. + +use std::fmt::Display; +use std::future::Future; + +use anyhow::{Context as _, Result}; +use hive_jobq::scheduler::Outcome; +use swarm_wire::agent_closures::AgentClosure; +use swarm_wire::builder::{BuildFromMeta, FromMetaReply, META_SUBJECT, REPLY_DEADLINE}; + +use crate::agent_renewal::unix_now; +use crate::forge::{self, objects::SHARED_ORG}; +use crate::meta_flake::{META_REPO, MetaForge}; + +/// What a `Failed` outcome that is not a verdict on the config starts with. +pub(crate) const RETRY_PREFIX: &str = "not built, will be retried: "; + +/// Whether a `Failed` node's `error` says the build gave no verdict. +pub(crate) fn is_retried(error: &str) -> bool { + error.starts_with(RETRY_PREFIX) +} + +fn not_built(why: impl Display) -> Outcome { + Outcome::Failed(format!("{RETRY_PREFIX}{why}")) +} + +/// Where `internal/meta`'s head is read: the forge in production, a fake in +/// the tests. +pub(crate) trait MetaHead { + /// The head commit; `None` when the repo has none. + fn head(&self) -> impl Future>> + Send; +} + +impl MetaHead for forge::Client { + async fn head(&self) -> Result> { + let read = self.head_files(SHARED_ORG, META_REPO, None, &[]); + let head = crate::jobq::forge_node("build_meta", read).await?; + Ok(head.map(|h| h.commit)) + } +} + +/// Where a build is sent and its closure recorded: the queue in production, a +/// fake in the tests. +pub(crate) trait MetaBuilds { + /// `Err` is a request that never reached a builder or never came back. + fn build(&self, request: Vec) + -> impl Future> + Send; + fn record( + &self, + agent: &str, + closure: &AgentClosure, + ) -> impl Future> + Send; +} + +impl MetaBuilds for async_nats::Client { + async fn build(&self, request: Vec) -> Result { + use async_nats::RequestErrorKind; + + let sent = async_nats::Request::new() + .payload(request.into()) + .timeout(Some(REPLY_DEADLINE)); + match self.send_request(META_SUBJECT, sent).await { + Ok(msg) => serde_json::from_slice(&msg.payload) + .map_err(|e| format!("the builder's meta build reply did not decode: {e}")), + Err(e) => Err(match e.kind() { + RequestErrorKind::NoResponders => "no swarm builder is listening for meta \ + builds: deploy.builder.enable is set on no host, or its builder predates \ + the request" + .to_owned(), + RequestErrorKind::TimedOut => format!( + "the swarm builder sent no answer within {} minutes", + REPLY_DEADLINE.as_secs() / 60 + ), + _ => format!("sending the meta build to the swarm builder failed: {e}"), + }), + } + } + + async fn record(&self, agent: &str, closure: &AgentClosure) -> Result<()> { + let store = swarm_queue_client::agent_closures::open_or_create(self) + .await + .context("opening the agent-closures bucket")?; + let body = serde_json::to_vec(closure).context("encoding the closure record")?; + store + .put(agent, body.into()) + .await + .map(|_| ()) + .with_context(|| format!("writing agent-closures/{agent}")) + } +} + +/// Build `agent` from `internal/meta`'s head and record the closure. +pub(crate) async fn run( + head: Option<&impl MetaHead>, + queue: Option<&impl MetaBuilds>, + agent: String, +) -> Outcome { + let Some(head) = head else { + return not_built( + "no forge access here (SWARM_CONTROLLER_FORGE_URL unset, or its token not in the \ + swarm secret store), so internal/meta's head cannot be read", + ); + }; + let Some(queue) = queue else { + return not_built("no swarm queue is configured on this host, so no build can be sent"); + }; + let meta_rev = match head.head().await { + Ok(Some(rev)) => rev, + Ok(None) => return not_built("the swarm has not published internal/meta yet"), + Err(e) => return not_built(format!("{e:#}")), + }; + let request = match serde_json::to_vec(&BuildFromMeta { + agent: agent.clone(), + meta_rev: meta_rev.clone(), + }) { + Ok(request) => request, + Err(e) => return not_built(format!("encoding the build request: {e}")), + }; + match queue.build(request).await { + Err(why) => not_built(why), + Ok(FromMetaReply::Unavailable { reason }) => not_built(reason), + Ok(FromMetaReply::Failed { error_tail }) => Outcome::Failed(error_tail), + Ok(FromMetaReply::Built { out_path, system }) => { + let closure = AgentClosure { + meta_rev, + store_path: out_path.clone(), + system, + built_unix: unix_now(), + }; + match queue.record(&agent, &closure).await { + Ok(()) => Outcome::Done, + Err(e) => not_built(format!("built {out_path}, but recording it failed: {e:#}")), + } + } + } +} + +#[cfg(test)] +mod tests { + use std::sync::Mutex; + + use super::*; + + const REV: &str = "0123456789abcdef0123456789abcdef01234567"; + + struct FakeHead(Result, &'static str>); + + impl MetaHead for FakeHead { + async fn head(&self) -> Result> { + match self.0 { + Ok(rev) => Ok(rev.map(str::to_owned)), + Err(e) => Err(anyhow::anyhow!(e)), + } + } + } + + /// A queue that answers every build with `reply`, and keeps what it was + /// sent and what was recorded. + struct FakeQueue { + reply: Result, + record_fails: bool, + sent: Mutex>, + recorded: Mutex>, + } + + impl FakeQueue { + fn answering(reply: Result) -> Self { + Self { + reply, + record_fails: false, + sent: Mutex::new(Vec::new()), + recorded: Mutex::new(Vec::new()), + } + } + + fn built() -> Self { + Self::answering(Ok(FromMetaReply::Built { + out_path: "/nix/store/x-toplevel".to_owned(), + system: "x86_64-linux".to_owned(), + })) + } + } + + impl MetaBuilds for FakeQueue { + async fn build(&self, request: Vec) -> Result { + self.sent + .lock() + .unwrap() + .push(serde_json::from_slice(&request).expect("a build request")); + self.reply.clone() + } + + async fn record(&self, agent: &str, closure: &AgentClosure) -> Result<()> { + anyhow::ensure!(!self.record_fails, "the bucket refused the write"); + self.recorded + .lock() + .unwrap() + .push((agent.to_owned(), closure.clone())); + Ok(()) + } + } + + async fn run_with(head: &FakeHead, queue: &FakeQueue) -> Outcome { + run(Some(head), Some(queue), "atlas".to_owned()).await + } + + #[tokio::test] + async fn a_green_build_of_the_head_is_recorded_for_the_agent() { + let queue = FakeQueue::built(); + let outcome = run_with(&FakeHead(Ok(Some(REV))), &queue).await; + assert_eq!(outcome, Outcome::Done); + assert_eq!( + *queue.sent.lock().unwrap(), + [BuildFromMeta { + agent: "atlas".to_owned(), + meta_rev: REV.to_owned(), + }] + ); + let recorded = queue.recorded.lock().unwrap(); + let [(agent, closure)] = recorded.as_slice() else { + panic!("one record: {recorded:?}") + }; + assert_eq!(agent, "atlas"); + assert_eq!(closure.meta_rev, REV); + assert_eq!(closure.store_path, "/nix/store/x-toplevel"); + assert_eq!(closure.system, "x86_64-linux"); + assert!(closure.built_unix > 0); + } + + /// The builder said the build is red: the node fails with `nix`'s tail, + /// which is not a retried failure, and nothing is recorded. The green test + /// above is the control that a record is written when the build is green. + #[tokio::test] + async fn a_red_build_fails_with_its_tail_and_records_nothing() { + let queue = FakeQueue::answering(Ok(FromMetaReply::Failed { + error_tail: "error: builder for foo failed".to_owned(), + })); + let outcome = run_with(&FakeHead(Ok(Some(REV))), &queue).await; + assert_eq!( + outcome, + Outcome::Failed("error: builder for foo failed".to_owned()) + ); + assert!(queue.recorded.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn a_builder_that_gave_no_verdict_is_a_retried_failure_and_records_nothing() { + for (name, reply) in [ + ( + "no responders", + Err("no swarm builder is listening for meta builds".to_owned()), + ), + ( + "unavailable", + Ok(FromMetaReply::Unavailable { + reason: "busy".to_owned(), + }), + ), + ] { + let queue = FakeQueue::answering(reply); + let outcome = run_with(&FakeHead(Ok(Some(REV))), &queue).await; + let Outcome::Failed(error) = outcome else { + panic!("{name}: not failed") + }; + assert!(is_retried(&error), "{name}: {error}"); + assert!(queue.recorded.lock().unwrap().is_empty(), "{name}"); + } + } + + #[tokio::test] + async fn a_closure_that_cannot_be_recorded_is_a_retried_failure() { + let queue = FakeQueue { + record_fails: true, + ..FakeQueue::built() + }; + let Outcome::Failed(error) = run_with(&FakeHead(Ok(Some(REV))), &queue).await else { + panic!("not failed") + }; + assert!(is_retried(&error), "{error}"); + assert!(error.contains("/nix/store/x-toplevel"), "{error}"); + } + + #[tokio::test] + async fn without_a_head_or_a_queue_nothing_is_sent_and_the_failure_is_retried() { + for (name, head, with_queue) in [ + ("no repo yet", Some(FakeHead(Ok(None))), true), + ("forge error", Some(FakeHead(Err("forge down"))), true), + ("no forge", None, true), + ("no queue", Some(FakeHead(Ok(Some(REV)))), false), + ] { + let queue = FakeQueue::built(); + let outcome = run( + head.as_ref(), + with_queue.then_some(&queue), + "atlas".to_owned(), + ) + .await; + let Outcome::Failed(error) = outcome else { + panic!("{name}: not failed") + }; + assert!(is_retried(&error), "{name}: {error}"); + assert!(queue.sent.lock().unwrap().is_empty(), "{name}"); + } + } + + #[test] + fn the_repo_path_the_builder_fetches_is_the_one_the_controller_writes() { + assert_eq!( + swarm_wire::builder::META_REPO_PATH, + format!("{SHARED_ORG}/{META_REPO}") + ); + } + + /// The bucket on a real server: a record written for an agent reads back + /// under its key, and a second build replaces it. + #[tokio::test] + async fn the_closure_round_trips_through_the_bucket() { + let nats = crate::test_nats::TestNats::start(); + let client = nats.client().await; + let mut closure = AgentClosure { + meta_rev: REV.to_owned(), + store_path: "/nix/store/a".to_owned(), + system: "x86_64-linux".to_owned(), + built_unix: 1, + }; + client.record("atlas", &closure).await.expect("written"); + closure.store_path = "/nix/store/b".to_owned(); + client.record("atlas", &closure).await.expect("replaced"); + let store = swarm_queue_client::agent_closures::open_or_create(&client) + .await + .expect("opens"); + let got = store.get("atlas").await.expect("reads").expect("present"); + let got: AgentClosure = serde_json::from_slice(&got).expect("decodes"); + assert_eq!(got, closure); + assert!(store.get("argus").await.expect("reads").is_none()); + } +} diff --git a/swarm-controller/src/meta_fanout.rs b/swarm-controller/src/meta_fanout.rs index c8d2aed7..4b5d55b3 100644 --- a/swarm-controller/src/meta_fanout.rs +++ b/swarm-controller/src/meta_fanout.rs @@ -3,10 +3,12 @@ //! One key ([`KEY`]) in [`BUCKET`], which only this daemon reads or writes. //! The `UpdateMeta` node fans out whenever the pin `internal/meta` holds is //! not the one recorded here, and the `RecordFanout` node writes it once every -//! `TriggerDeploy` of that fan-out is done. A pass that dies, or a deploy that -//! fails, before then leaves the record stale, so the next pass sends the -//! deploys again. A deploy is idempotent per pin, so sending twice is safe and -//! sending never is not. +//! `TriggerDeploy` of that fan-out is done or skipped by its build gate. A pass +//! that dies, or a deploy that fails, before then leaves the record stale, so +//! the next pass sends the deploys again. A deploy the gate skipped counts as +//! sent: `jobq::queue_meta_update` declares it again when its build gave no +//! verdict, and a red build waits for the next config merge. A deploy is +//! idempotent per pin, so sending twice is safe and sending never is not. //! //! With no record, as on the first pass of a controller that has never written //! one, the pin is not the recorded one and the agents are sent a deploy. diff --git a/swarm-controller/src/meta_flake.rs b/swarm-controller/src/meta_flake.rs index 42f54098..d0232c13 100644 --- a/swarm-controller/src/meta_flake.rs +++ b/swarm-controller/src/meta_flake.rs @@ -215,7 +215,7 @@ pub(crate) struct Report { } /// Bring `internal/meta` up to the forge's current heads for `agents`, and -/// give each of `hives`' forge accounts read on it. +/// give each of `hives`' forge accounts, and the swarm builder's, read on it. /// /// # Errors /// A head that cannot be read, a lock the builder refuses, or a refused commit. @@ -229,10 +229,13 @@ pub(crate) async fn update( hives: &[String], ) -> Result { forge.ensure_meta_repo().await?; - for hive in hives { - let login = swarm_secret_client::forge::hive_login(hive); + let logins = hives + .iter() + .map(|hive| swarm_secret_client::forge::hive_login(hive)) + .chain([crate::forge::builder_account::LOGIN.to_owned()]); + for login in logins { if let Err(e) = forge.grant_meta_read(&login).await { - tracing::warn!(%hive, error = %format!("{e:#}"), "meta flake: read grant on internal/meta failed"); + tracing::warn!(%login, error = %format!("{e:#}"), "meta flake: read grant on internal/meta failed"); } } let committed = forge @@ -753,7 +756,10 @@ mod tests { ); assert_eq!( *forge.grants.lock().unwrap(), - vec![swarm_secret_client::forge::hive_login("pr1ma")] + vec![ + swarm_secret_client::forge::hive_login("pr1ma"), + crate::forge::builder_account::LOGIN.to_owned(), + ] ); } diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index d71db047..5070c748 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -243,8 +243,9 @@ impl Policy { } } - /// What the builder may do: hear build, archetype evaluation and meta-flake - /// lock requests and answer each on its reply subject, for as long as a build may take. + /// What the builder may do: hear build, archetype evaluation, meta-flake + /// lock and meta build requests and answer each on its reply subject, for + /// as long as a build may take. /// Its publish list is its own inbox, for /// [`Self::review_classifier_permissions`]' reason. fn builder_permissions(client_id: &str) -> Permissions { @@ -254,6 +255,7 @@ impl Policy { swarm_wire::builder::SUBJECT.to_owned(), swarm_wire::builder::ARCHETYPES_SUBJECT.to_owned(), swarm_wire::builder::LOCK_SUBJECT.to_owned(), + swarm_wire::builder::META_SUBJECT.to_owned(), Self::inbox(client_id), ]), responses: Some(swarm_wire::builder::REPLY_DEADLINE), @@ -710,6 +712,8 @@ impl Policy { swarm_wire::builder::ARCHETYPES_SUBJECT.to_owned(), // A swarm meta-flake lock, sent as a request the same way. swarm_wire::builder::LOCK_SUBJECT.to_owned(), + // An agent build from the meta flake, sent as a request the same way. + swarm_wire::builder::META_SUBJECT.to_owned(), // An operator's claude login step, sent as a request the same way. swarm_wire::claude_login::SUBJECT.to_owned(), // The notification router's deliveries log and a dry run of a flows @@ -1637,7 +1641,9 @@ mod tests { #[test] fn the_builder_hears_its_subject_and_inbox_only_and_answers_by_reply() { use swarm_queue_client::client_inbox_prefix; - use swarm_wire::builder::{ARCHETYPES_SUBJECT, LOCK_SUBJECT, REPLY_DEADLINE, SUBJECT}; + use swarm_wire::builder::{ + ARCHETYPES_SUBJECT, LOCK_SUBJECT, META_SUBJECT, REPLY_DEADLINE, SUBJECT, + }; let p = policy() .with_builders(vec![BUILDER.to_owned()]) @@ -1647,6 +1653,7 @@ mod tests { assert!(may_subscribe(&p, SUBJECT)); assert!(may_subscribe(&p, ARCHETYPES_SUBJECT)); assert!(may_subscribe(&p, LOCK_SUBJECT)); + assert!(may_subscribe(&p, META_SUBJECT)); assert!(may_subscribe(&p, &reply(&inbox))); for refused in [ ">".to_owned(), @@ -1667,11 +1674,11 @@ mod tests { #[test] fn only_the_reader_may_send_a_build() { - use swarm_wire::builder::{ARCHETYPES_SUBJECT, LOCK_SUBJECT, SUBJECT}; + use swarm_wire::builder::{ARCHETYPES_SUBJECT, LOCK_SUBJECT, META_SUBJECT, SUBJECT}; let p = with_classifier().with_builders(vec![BUILDER.to_owned()]); let reader = p.permissions("swarm-controller").expect("the reader"); - for subject in [SUBJECT, ARCHETYPES_SUBJECT, LOCK_SUBJECT] { + for subject in [SUBJECT, ARCHETYPES_SUBJECT, LOCK_SUBJECT, META_SUBJECT] { assert!(reader.publish.iter().any(|g| covers(g, subject))); for (who, grant) in [ ("a hive", p.permissions("hive-alpha").expect("a hive")), @@ -1690,6 +1697,28 @@ mod tests { } } + /// The controller writes the closure bucket; the builder, which evaluates + /// unreviewed PR configs, cannot. + #[test] + fn only_the_controller_may_write_the_closure_bucket() { + use swarm_wire::agent_closures::key_subject; + + let p = with_classifier().with_builders(vec![BUILDER.to_owned()]); + let key = key_subject("atlas"); + let controller = p.permissions("swarm-controller").expect("the reader"); + assert!(controller.publish.iter().any(|g| covers(g, &key))); + for (who, grant) in [ + ("a hive", p.permissions("hive-alpha").expect("a hive")), + ("the builder", p.permissions(BUILDER).expect("admitted")), + ] { + assert!( + !grant.publish.iter().any(|g| covers(g, &key)), + "{who} may write {key}: {:?}", + grant.publish + ); + } + } + const CLAUDE_LOGIN: &str = "swarm-claude-login"; #[test] diff --git a/swarm-queue-client/src/agent_closures.rs b/swarm-queue-client/src/agent_closures.rs new file mode 100644 index 00000000..5c8f6810 --- /dev/null +++ b/swarm-queue-client/src/agent_closures.rs @@ -0,0 +1,19 @@ +//! The connection side of the per-agent closure KV bucket. The bucket's names +//! and stored type are `swarm_wire::agent_closures`. + +/// Open the closures bucket for writing, creating it if nothing has yet. +/// +/// **Controller-side only.** `history: 1`: a reader wants the latest build. +#[cfg(feature = "kv")] +pub async fn open_or_create( + client: &async_nats::Client, +) -> Result { + let js = async_nats::jetstream::new(client.clone()); + crate::kv::open_or_create( + &js, + swarm_wire::agent_closures::BUCKET, + "agent-closures", + "The latest system closure the swarm builder built for each agent".to_owned(), + ) + .await +} diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 5cf2c750..3cbec38b 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -250,6 +250,11 @@ pub mod operator_profiles; /// `agent-archetype`. pub mod agent_archetype; +/// Opening the bucket swarm-controller records each agent's latest built +/// system closure in. The names and the stored type are +/// `swarm_wire::agent_closures`; writing needs `kv`. +pub mod agent_closures; + /// The bucket swarm-controller keeps the evaluated agent archetypes in — one /// key per source flake. The names and the stored type are unconditional; /// opening needs `kv`. diff --git a/swarm-wire/src/agent_closures.rs b/swarm-wire/src/agent_closures.rs new file mode 100644 index 00000000..2830ff2b --- /dev/null +++ b/swarm-wire/src/agent_closures.rs @@ -0,0 +1,67 @@ +//! The per-agent closure KV bucket: one key per agent, holding the latest +//! system closure the swarm builder built for it from the swarm meta flake. +//! +//! **swarm-controller is the only writer**, and creates the bucket on its first +//! write. It writes a key from a builder's `Built` reply and from nothing else, +//! so a failed build leaves the previous record in place. The builder holds no +//! grant on this bucket. +//! +//! Keyed by the agent alone, like [`crate::agent_limits`]: an agent can move +//! between hives, and its closure goes with it. + +/// The KV bucket closures are recorded in, keyed by agent name. +pub const BUCKET: &str = "agent-closures"; + +/// The stream backing [`BUCKET`]. `create_key_value` prefixes `KV_`. +pub const STREAM: &str = "KV_agent-closures"; + +/// The subject `agent`'s entry is stored under. +#[must_use] +pub fn key_subject(agent: &str) -> String { + format!("$KV.{BUCKET}.{agent}") +} + +/// One agent's latest green build: the value under `BUCKET`/``. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct AgentClosure { + /// The `internal/meta` commit the closure was built from. + pub meta_rev: String, + /// The system closure's store path, in the builder host's store. + pub store_path: String, + /// The platform the closure was built for. + pub system: String, + /// When the controller recorded the build, in unix seconds. + pub built_unix: i64, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn names_are_pinned() { + assert_eq!(STREAM, format!("KV_{BUCKET}")); + assert_eq!(key_subject("iris"), "$KV.agent-closures.iris"); + } + + #[test] + fn the_record_serializes_as_pinned() { + let value = AgentClosure { + meta_rev: "abc".to_owned(), + store_path: "/nix/store/x".to_owned(), + system: "x86_64-linux".to_owned(), + built_unix: 7, + }; + let wire = serde_json::json!({ + "meta_rev": "abc", + "store_path": "/nix/store/x", + "system": "x86_64-linux", + "built_unix": 7, + }); + assert_eq!(serde_json::to_value(&value).expect("encodes"), wire); + assert_eq!( + serde_json::from_value::(wire).expect("decodes"), + value + ); + } +} diff --git a/swarm-wire/src/builder.rs b/swarm-wire/src/builder.rs index 7d9d3814..30692cc3 100644 --- a/swarm-wire/src/builder.rs +++ b/swarm-wire/src/builder.rs @@ -1,5 +1,6 @@ //! The swarm builder's requests: the subjects swarm-controller sends a build, -//! an archetype evaluation and a meta-flake lock on, and what travels each way. +//! an archetype evaluation, a meta-flake lock and an agent build from the meta +//! flake on, and what travels each way. //! //! Three crates name these: the controller sends the requests, the builder //! answers them, and the auth-callout responder grants the subjects to each. @@ -115,6 +116,41 @@ pub enum LockReply { Failed { error_tail: String }, } +/// The subject an agent build from the swarm meta flake goes to, with the same +/// one publisher and one subscriber as [`SUBJECT`]. A subject of its own, so a +/// builder that predates the request has no subscriber and the controller gets +/// no responders at once. +pub const META_SUBJECT: &str = "$SWARM.builder.build-meta"; + +/// The swarm meta flake's repo on the forge, as `/`. +pub const META_REPO_PATH: &str = "internal/meta"; + +/// One agent built from the swarm meta flake: the toplevel of +/// `nixosConfigurations.`, with every input at the flake's own lock. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct BuildFromMeta { + /// The agent, which is also its name in the meta flake. + pub agent: String, + /// The `internal/meta` commit to build, a full 40-hex id. + pub meta_rev: String, +} + +/// The builder's answer to one [`BuildFromMeta`]. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "state", rename_all = "kebab-case")] +pub enum FromMetaReply { + /// The toplevel built; `out_path` is its store path in the builder host's + /// store and `system` the platform the builder host builds for. + Built { out_path: String, system: String }, + /// `nix` ran and the build did not finish: the last lines it printed, or + /// the deadline it hit. The controller holds the deploy back. + Failed { error_tail: String }, + /// The builder did not get to run `nix`: the build slot stayed busy or the + /// forge token could not be read. Says nothing about the config, so the + /// controller tries again. + Unavailable { reason: String }, +} + #[cfg(test)] mod tests { use serde_json::json; @@ -199,11 +235,53 @@ mod tests { } } + #[test] + fn the_meta_build_request_and_reply_serialize_as_pinned() { + let req = BuildFromMeta { + agent: "atlas".to_owned(), + meta_rev: "abc".to_owned(), + }; + let value = serde_json::to_value(&req).expect("encodes"); + assert_eq!(value, json!({"agent": "atlas", "meta_rev": "abc"})); + assert_eq!( + serde_json::from_value::(value).expect("decodes"), + req + ); + for (reply, wire) in [ + ( + FromMetaReply::Built { + out_path: "/nix/store/x".to_owned(), + system: "x86_64-linux".to_owned(), + }, + json!({"state": "built", "out_path": "/nix/store/x", "system": "x86_64-linux"}), + ), + ( + FromMetaReply::Failed { + error_tail: "e".to_owned(), + }, + json!({"state": "failed", "error_tail": "e"}), + ), + ( + FromMetaReply::Unavailable { + reason: "r".to_owned(), + }, + json!({"state": "unavailable", "reason": "r"}), + ), + ] { + assert_eq!(serde_json::to_value(&reply).expect("encodes"), wire); + assert_eq!( + serde_json::from_value::(wire).expect("decodes"), + reply + ); + } + } + #[test] fn the_subjects_are_pinned() { assert_eq!(SUBJECT, "$SWARM.builder.build"); assert_eq!(ARCHETYPES_SUBJECT, "$SWARM.builder.archetypes"); assert_eq!(LOCK_SUBJECT, "$SWARM.builder.lock-meta"); + assert_eq!(META_SUBJECT, "$SWARM.builder.build-meta"); assert_eq!(QUEUE_GROUP, "builder"); } diff --git a/swarm-wire/src/lib.rs b/swarm-wire/src/lib.rs index 774d4c83..1ce77d04 100644 --- a/swarm-wire/src/lib.rs +++ b/swarm-wire/src/lib.rs @@ -6,6 +6,7 @@ //! grants the subject) depends on this one without linking either. pub mod agent_archetype; +pub mod agent_closures; pub mod agent_icon; pub mod agent_inference; pub mod agent_limits;