diff --git a/swarm-builder/src/main.rs b/swarm-builder/src/main.rs index c68a3864..86b4aad7 100644 --- a/swarm-builder/src/main.rs +++ b/swarm-builder/src/main.rs @@ -8,6 +8,10 @@ //! [`swarm_wire::builder::ARCHETYPES_SUBJECT`] with the agent //! archetypes its hyperhive source declares, which the controller stores. //! +//! Lock requests on [`swarm_wire::builder::LOCK_SUBJECT`] carry the swarm meta +//! flake's `flake.nix`, and the answer is the `flake.lock` `nix` wrote for it in +//! a scratch directory. The controller commits it. +//! //! 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. @@ -23,8 +27,8 @@ 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, QUEUE_GROUP, - REPLY_DEADLINE, Reply, SUBJECT, + ARCHETYPES_DEADLINE, ARCHETYPES_SUBJECT, BUILD_DEADLINE, BuildAgent, LOCK_DEADLINE, + LOCK_SUBJECT, LockMeta, LockReply, QUEUE_GROUP, REPLY_DEADLINE, Reply, SUBJECT, }; mod archetypes; @@ -71,6 +75,7 @@ async fn main() -> Result<()> { forge_url: forge_url()?, config_org: var("CONFIG_ORG")?, deadline: BUILD_DEADLINE, + lock_deadline: LOCK_DEADLINE, }); let secret = swarm_secret_client::read_with_startup_retry("reading the queue client secret", || { @@ -109,6 +114,13 @@ async fn main() -> Result<()> { .with_context(|| format!("subscribing to {ARCHETYPES_SUBJECT}"))?; // Outside the build slot below: a list refresh does not wait on a build. tokio::spawn(answer_evaluations(evaluations, queue.clone(), evaluator)); + let locks = queue + .queue_subscribe(LOCK_SUBJECT, QUEUE_GROUP.to_owned()) + .await + .with_context(|| format!("subscribing to {LOCK_SUBJECT}"))?; + // Outside the build slot too: a lock builds nothing, so it does not wait + // behind an hour-long build. + tokio::spawn(answer_locks(locks, queue.clone(), Arc::clone(&builder))); // 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` @@ -186,6 +198,57 @@ async fn answer_evaluations( tracing::error!("the subscription to {ARCHETYPES_SUBJECT} ended"); } +/// Answer every meta-flake lock request, one at a time, with the lock `nix` +/// wrote or why it did not. +async fn answer_locks( + mut requests: async_nats::Subscriber, + queue: async_nats::Client, + builder: Arc, +) { + while let Some(msg) = requests.next().await { + let Some(reply_to) = msg.reply.clone() else { + tracing::warn!("a meta-flake lock arrived with no reply subject; dropping it"); + continue; + }; + let reply = handle_lock(&msg.payload, &builder).await; + 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 lock's reply failed"); + } + } + Err(e) => tracing::warn!(error = %e, "encoding a lock's reply failed"), + } + } + tracing::error!("the subscription to {LOCK_SUBJECT} ended"); +} + +/// Decode one lock request, lock it and log the lock. +async fn handle_lock(payload: &[u8], builder: &nix_build::Builder) -> LockReply { + let job: LockMeta = match serde_json::from_slice(payload) { + Ok(job) => job, + Err(e) => { + tracing::warn!(error = %e, "an undecodable lock; answering with the error"); + return LockReply::Failed { + error_tail: format!("the builder could not decode the request: {e}"), + }; + } + }; + let reply = match within(TOKEN_DEADLINE, forge_token()).await { + Ok(token) => builder.lock(&job, &token).await, + Err(e) => LockReply::Failed { + error_tail: format!("{e:#}"), + }, + }; + match &reply { + LockReply::Locked { .. } => tracing::info!("swarm meta flake locked"), + LockReply::Failed { error_tail } => { + tracing::warn!(%error_tail, "swarm meta flake lock failed"); + } + } + reply +} + /// `build`'s reply once `job` holds the build slot, or a refusal, without /// running `build`, once it has waited [`MAX_SLOT_WAIT`] since `received`. /// `None` only when the slot is closed. diff --git a/swarm-builder/src/nix_build.rs b/swarm-builder/src/nix_build.rs index 414211fe..b5c6b90d 100644 --- a/swarm-builder/src/nix_build.rs +++ b/swarm-builder/src/nix_build.rs @@ -1,6 +1,7 @@ //! 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. +//! commit. 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 @@ -16,7 +17,7 @@ use std::path::PathBuf; use std::process::Stdio; use std::time::Duration; -use swarm_wire::builder::{BuildAgent, Reply}; +use swarm_wire::builder::{BuildAgent, LockMeta, LockReply, Reply}; use tokio::io::AsyncReadExt as _; /// The most of `nix`'s stderr a failed build's reply carries, from its end. @@ -51,6 +52,8 @@ pub struct Builder { pub config_org: String, /// How long one build may run. pub deadline: Duration, + /// How long one lock may run. + pub lock_deadline: Duration, } impl Builder { @@ -59,9 +62,48 @@ impl Builder { if let Err(error) = validate(job) { return Reply::Failed { error_tail: error }; } + let mut cmd = self.nix_command(token); + cmd.args(self.args(job)); + run(cmd, self.deadline).await + } + + /// 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 { + let scratch = match Scratch::new() { + Ok(s) => s, + Err(e) => { + return LockReply::Failed { + error_tail: format!("creating the scratch directory: {e}"), + }; + } + }; + if let Err(e) = std::fs::write(scratch.0.join("flake.nix"), &job.flake_nix) { + return LockReply::Failed { + error_tail: format!("writing flake.nix: {e}"), + }; + } + let mut cmd = self.nix_command(token); + cmd.args(LOCK_ARGS).current_dir(&scratch.0); + match run_raw(cmd, self.lock_deadline).await { + Err(error_tail) => LockReply::Failed { error_tail }, + Ok((status, _, err)) if !status.success() => LockReply::Failed { + error_tail: tail(&String::from_utf8_lossy(&err), status), + }, + Ok(_) => match std::fs::read_to_string(scratch.0.join("flake.lock")) { + Ok(flake_lock) => LockReply::Locked { flake_lock }, + Err(e) => LockReply::Failed { + error_tail: format!("nix exited 0 but flake.lock is unreadable: {e}"), + }, + }, + } + } + + /// `nix` with the forge credential reachable by `git` alone, and its + /// output captured. + fn nix_command(&self, token: &str) -> tokio::process::Command { let mut cmd = tokio::process::Command::new(&self.nix); - cmd.args(self.args(job)) - .env(TOKEN_VAR, token) + cmd.env(TOKEN_VAR, token) .env("GIT_TERMINAL_PROMPT", "0") .env("GIT_CONFIG_COUNT", "1") .env(HOST_VAR, forge_host(&self.forge_url)) @@ -74,7 +116,7 @@ impl Builder { .stdout(Stdio::piped()) .stderr(Stdio::piped()) .kill_on_drop(true); - run(cmd, self.deadline).await + cmd } /// The `nix build` arguments for `job`. @@ -116,6 +158,33 @@ impl Builder { } } +/// `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"]; + +/// A fresh directory under the temp dir, removed with everything in it on drop. +struct Scratch(PathBuf); + +impl Scratch { + fn new() -> std::io::Result { + use std::sync::atomic::{AtomicU64, Ordering}; + static NEXT: AtomicU64 = AtomicU64::new(0); + let dir = std::env::temp_dir().join(format!( + "lock-meta-{}-{}", + std::process::id(), + NEXT.fetch_add(1, Ordering::Relaxed) + )); + std::fs::create_dir(&dir)?; + Ok(Self(dir)) + } +} + +impl Drop for Scratch { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + /// The host part of `forge_url` (`https://[:port]`), as `git` names it /// to a credential helper. fn forge_host(forge_url: &str) -> &str { @@ -162,14 +231,33 @@ fn validate(job: &BuildAgent) -> Result<(), String> { /// Run `cmd` to its end or `deadline`, and read the store path it printed or /// the tail of what it said. -async fn run(mut cmd: tokio::process::Command, deadline: Duration) -> Reply { +async fn run(cmd: tokio::process::Command, deadline: Duration) -> Reply { + match run_raw(cmd, deadline).await { + Err(error_tail) => Reply::Failed { error_tail }, + Ok((status, out, err)) => { + let out_path = String::from_utf8_lossy(&out) + .lines() + .rfind(|l| !l.trim().is_empty()) + .map(|l| l.trim().to_owned()); + match out_path { + Some(out_path) if status.success() => Reply::Built { out_path }, + _ => Reply::Failed { + error_tail: tail(&String::from_utf8_lossy(&err), status), + }, + } + } + } +} + +/// Run `cmd` to its end or `deadline`: its status and the kept end of its +/// stdout and stderr, or why it did not finish. +async fn run_raw( + mut cmd: tokio::process::Command, + deadline: Duration, +) -> Result<(std::process::ExitStatus, Vec, Vec), String> { let mut child = match cmd.spawn() { Ok(child) => child, - Err(e) => { - return Reply::Failed { - error_tail: format!("starting nix: {e}"), - }; - } + Err(e) => return Err(format!("starting nix: {e}")), }; let stdout = child.stdout.take().expect("piped"); let stderr = child.stderr.take().expect("piped"); @@ -183,27 +271,12 @@ async fn run(mut cmd: tokio::process::Command, deadline: Duration) -> Reply { status.map(|s| (s, out, err)) }; match tokio::time::timeout(deadline, work).await { - Err(_) => Reply::Failed { - error_tail: format!( - "the build did not finish within {} minutes", - deadline.as_secs() / 60 - ), - }, - Ok(Err(e)) => Reply::Failed { - error_tail: format!("waiting for nix: {e}"), - }, - Ok(Ok((status, out, err))) => { - let out_path = String::from_utf8_lossy(&out) - .lines() - .rfind(|l| !l.trim().is_empty()) - .map(|l| l.trim().to_owned()); - match out_path { - Some(out_path) if status.success() => Reply::Built { out_path }, - _ => Reply::Failed { - error_tail: tail(&String::from_utf8_lossy(&err), status), - }, - } - } + Err(_) => Err(format!( + "nix did not finish within {} minutes", + deadline.as_secs() / 60 + )), + Ok(Err(e)) => Err(format!("waiting for nix: {e}")), + Ok(Ok(done)) => Ok(done), } } @@ -288,7 +361,7 @@ pub(crate) fn tail(stderr: &str, status: std::process::ExitStatus) -> String { #[cfg(test)] mod tests { use super::*; - use swarm_wire::builder::BUILD_DEADLINE; + use swarm_wire::builder::{BUILD_DEADLINE, LOCK_DEADLINE}; fn job() -> BuildAgent { BuildAgent { @@ -307,6 +380,7 @@ mod tests { config_org: "agent-configs".to_owned(), nix: PathBuf::from("nix"), deadline: BUILD_DEADLINE, + lock_deadline: LOCK_DEADLINE, } } @@ -600,6 +674,72 @@ mod tests { assert!(kept.starts_with(b"line\n")); } + /// A builder whose `nix` is the shell script `body`, in a directory the + /// returned guard removes. + fn scripted(body: &str) -> (Builder, Scratch) { + use std::os::unix::fs::PermissionsExt as _; + + let dir = Scratch::new().expect("a temp dir"); + let nix = dir.0.join("nix"); + std::fs::write(&nix, format!("#!/bin/sh\n{body}\n")).expect("the script is written"); + std::fs::set_permissions(&nix, std::fs::Permissions::from_mode(0o755)) + .expect("the script is executable"); + (Builder { nix, ..builder() }, dir) + } + + fn lock_job() -> LockMeta { + LockMeta { + flake_nix: "{ outputs = _: { }; }".to_owned(), + } + } + + #[tokio::test] + async fn a_lock_answers_with_the_lock_nix_wrote_and_leaves_no_scratch() { + // The lock records where it ran, so the test can look for it after. + let (b, _guard) = scripted(r#"test "$1 $2" = "flake lock" && pwd > flake.lock"#); + let reply = b.lock(&lock_job(), "t").await; + let LockReply::Locked { flake_lock } = reply else { + panic!("{reply:?}") + }; + let ran_in = PathBuf::from(flake_lock.trim()); + assert!(ran_in.starts_with(std::env::temp_dir()), "{ran_in:?}"); + assert!( + !ran_in.exists(), + "the scratch dir {ran_in:?} was left behind" + ); + } + + #[tokio::test] + async fn a_lock_sees_the_flake_it_was_sent() { + let (b, _guard) = scripted("cp flake.nix flake.lock"); + assert_eq!( + b.lock(&lock_job(), "t").await, + LockReply::Locked { + flake_lock: lock_job().flake_nix + } + ); + } + + #[tokio::test] + async fn a_failed_lock_answers_with_what_nix_said_last() { + let (b, _guard) = scripted("echo 'error: cannot fetch' >&2; exit 1"); + let reply = b.lock(&lock_job(), "t").await; + assert!( + matches!(&reply, LockReply::Failed { error_tail } if error_tail.ends_with("error: cannot fetch")), + "{reply:?}" + ); + } + + #[tokio::test] + async fn a_lock_that_wrote_no_lock_file_failed() { + let (b, _guard) = scripted("exit 0"); + let reply = b.lock(&lock_job(), "t").await; + assert!( + matches!(&reply, LockReply::Failed { error_tail } if error_tail.contains("flake.lock")), + "{reply:?}" + ); + } + #[tokio::test] async fn a_build_past_its_deadline_is_stopped() { let reply = sh("sleep 30", Duration::from_millis(200)).await; diff --git a/swarm-nats-auth/src/policy.rs b/swarm-nats-auth/src/policy.rs index 1a824a29..c2d43a5e 100644 --- a/swarm-nats-auth/src/policy.rs +++ b/swarm-nats-auth/src/policy.rs @@ -243,8 +243,8 @@ impl Policy { } } - /// What the builder may do: hear build and archetype evaluation 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 and meta-flake + /// lock 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 { @@ -253,6 +253,7 @@ impl Policy { subscribe: Some(vec![ swarm_wire::builder::SUBJECT.to_owned(), swarm_wire::builder::ARCHETYPES_SUBJECT.to_owned(), + swarm_wire::builder::LOCK_SUBJECT.to_owned(), Self::inbox(client_id), ]), responses: Some(swarm_wire::builder::REPLY_DEADLINE), @@ -693,6 +694,8 @@ impl Policy { swarm_wire::builder::SUBJECT.to_owned(), // An archetype evaluation, sent as a request the same way. 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 operator's claude login step, sent as a request the same way. swarm_wire::claude_login::SUBJECT.to_owned(), // Verified forge deliveries, kept in the forge events stream for @@ -1614,7 +1617,7 @@ 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, REPLY_DEADLINE, SUBJECT}; + use swarm_wire::builder::{ARCHETYPES_SUBJECT, LOCK_SUBJECT, REPLY_DEADLINE, SUBJECT}; let p = policy() .with_builders(vec![BUILDER.to_owned()]) @@ -1623,6 +1626,7 @@ mod tests { let inbox = client_inbox_prefix(BUILDER); assert!(may_subscribe(&p, SUBJECT)); assert!(may_subscribe(&p, ARCHETYPES_SUBJECT)); + assert!(may_subscribe(&p, LOCK_SUBJECT)); assert!(may_subscribe(&p, &reply(&inbox))); for refused in [ ">".to_owned(), @@ -1643,11 +1647,11 @@ mod tests { #[test] fn only_the_reader_may_send_a_build() { - use swarm_wire::builder::{ARCHETYPES_SUBJECT, SUBJECT}; + use swarm_wire::builder::{ARCHETYPES_SUBJECT, LOCK_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] { + for subject in [SUBJECT, ARCHETYPES_SUBJECT, LOCK_SUBJECT] { assert!(reader.publish.iter().any(|g| covers(g, subject))); for (who, grant) in [ ("a hive", p.permissions("hive-alpha").expect("a hive")), diff --git a/swarm-wire/src/builder.rs b/swarm-wire/src/builder.rs index aeb66f54..f3def429 100644 --- a/swarm-wire/src/builder.rs +++ b/swarm-wire/src/builder.rs @@ -1,5 +1,5 @@ -//! The swarm builder's requests: the subjects swarm-controller sends a build -//! and an archetype evaluation on, and what travels each way. +//! 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. //! //! Three crates name these: the controller sends the requests, the builder //! answers them, and the auth-callout responder grants the subjects to each. @@ -80,6 +80,38 @@ pub enum ArchetypesReply { Failed { error_tail: String }, } +/// The subject a swarm meta-flake lock goes to, with the same one publisher and +/// one subscriber as [`SUBJECT`]. +pub const LOCK_SUBJECT: &str = "$SWARM.builder.lock-meta"; + +/// The longest one lock may take. A lock fetches every input's source and +/// builds nothing. +pub const LOCK_DEADLINE: Duration = Duration::from_mins(10); + +/// How long the controller waits for a lock's reply: the lock at its deadline, +/// plus a minute for the rest. +pub const LOCK_REPLY_DEADLINE: Duration = Duration::from_mins(11); + +/// One lock: the swarm meta flake's `flake.nix`, locked on its own in a scratch +/// directory. The builder writes nothing anywhere else and commits nothing; the +/// controller commits the returned lock beside the flake. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct LockMeta { + /// The rendered `flake.nix`. + pub flake_nix: String, +} + +/// The builder's answer to one [`LockMeta`]. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "state", rename_all = "kebab-case")] +pub enum LockReply { + /// The flake locked; `flake_lock` is the `flake.lock` `nix` wrote. + Locked { flake_lock: String }, + /// The lock did not finish: the last lines `nix` printed, or why the + /// builder could not start it. + Failed { error_tail: String }, +} + #[cfg(test)] mod tests { use serde_json::json; @@ -168,6 +200,7 @@ mod tests { 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!(QUEUE_GROUP, "builder"); } @@ -175,4 +208,42 @@ mod tests { fn the_reply_deadline_outlasts_the_build() { assert!(REPLY_DEADLINE > BUILD_DEADLINE); } + + #[test] + fn the_lock_reply_deadline_outlasts_the_lock() { + assert!(LOCK_REPLY_DEADLINE > LOCK_DEADLINE); + } + + #[test] + fn the_lock_request_and_reply_serialize_as_pinned() { + let req = LockMeta { + flake_nix: "{ }".to_owned(), + }; + let value = serde_json::to_value(&req).expect("encodes"); + assert_eq!(value, json!({"flake_nix": "{ }"})); + assert_eq!( + serde_json::from_value::(value).expect("decodes"), + req + ); + for (reply, wire) in [ + ( + LockReply::Locked { + flake_lock: "{}".to_owned(), + }, + json!({"state": "locked", "flake_lock": "{}"}), + ), + ( + LockReply::Failed { + error_tail: "e".to_owned(), + }, + json!({"state": "failed", "error_tail": "e"}), + ), + ] { + assert_eq!(serde_json::to_value(&reply).expect("encodes"), wire); + assert_eq!( + serde_json::from_value::(wire).expect("decodes"), + reply + ); + } + } }