diff --git a/swarm-builder/src/main.rs b/swarm-builder/src/main.rs index 9d1cde65..16105071 100644 --- a/swarm-builder/src/main.rs +++ b/swarm-builder/src/main.rs @@ -29,7 +29,8 @@ use std::sync::Arc; use std::time::Duration; use anyhow::{Context, Result, anyhow, bail}; -use futures_util::StreamExt as _; +use async_nats::{Message, Subject}; +use futures_util::{Stream, StreamExt as _}; use swarm_queue_client::{ClientSecret, QueueConfig}; use swarm_secret_client::forge::Credential; use swarm_secret_client::stored_secret::StoredSecret; @@ -125,76 +126,171 @@ async fn main() -> Result<()> { .map(PathBuf::from), }) .await?; - let mut requests = queue + let requests = queue .queue_subscribe(SUBJECT, QUEUE_GROUP.to_owned()) .await .with_context(|| format!("subscribing to {SUBJECT}"))?; tracing::info!(subject = SUBJECT, "builder: awaiting builds"); + let publisher = Queue(queue.clone()); let evaluations = queue .queue_subscribe(ARCHETYPES_SUBJECT, QUEUE_GROUP.to_owned()) .await .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)); + tokio::spawn(answer_evaluations( + evaluations, + publisher.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))); + tokio::spawn(answer_locks( + locks, + publisher.clone(), + Store, + 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` // 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?; + serve_meta_builds(&queue, &publisher, Store, &builder, &slot).await?; + answer_builds(requests, publisher, Store, builder, slot).await; + bail!("the subscription to {SUBJECT} ended") +} + +/// Where a reply goes. [`Queue`] in production; tests record. +trait Publisher: Clone + Send + Sync + 'static { + /// Publish `body` on `subject`; the error is the queue client's own text. + fn publish( + &self, + subject: Subject, + body: Vec, + ) -> impl Future> + Send; +} + +/// The secrets a request reads from the store, once per request. [`Store`] in +/// production; tests answer from memory. +trait Tokens: Clone + Send + Sync + 'static { + /// The forge token config repos are fetched with. + fn forge_token(&self) -> impl Future> + Send; + /// The token closures are pushed to the cache with. + fn push_token(&self) -> impl Future> + Send; +} + +/// The swarm queue connection replies are published on. +#[derive(Clone)] +struct Queue(async_nats::Client); + +impl Publisher for Queue { + async fn publish(&self, subject: Subject, body: Vec) -> Result<(), String> { + self.0 + .publish(subject, body.into()) + .await + .map_err(|e| e.to_string()) + } +} + +/// The secret store, logged in to as [`CERT_ROLE`] on every read. +#[derive(Clone, Copy)] +struct Store; + +impl Tokens for Store { + async fn forge_token(&self) -> Result { + forge_token().await + } + + async fn push_token(&self) -> Result { + attic_push_token().await + } +} + +/// Publish an encoded reply on `reply_to`. `what` names the request in the +/// log line of a failure. +async fn send_reply( + publisher: &impl Publisher, + reply_to: Subject, + encoded: serde_json::Result>, + what: &str, +) { + match encoded { + Ok(body) => { + if let Err(e) = publisher.publish(reply_to, body).await { + tracing::warn!(error = %e, "publishing {what} reply failed"); + } + } + Err(e) => tracing::warn!(error = %e, "encoding {what} reply failed"), + } +} + +/// Answer every PR check, each in its own task, with the build's result. +async fn answer_builds( + mut requests: impl Stream + Unpin, + publisher: impl Publisher, + tokens: impl Tokens, + 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 build arrived with no reply subject; dropping it"); continue; }; - let queue = queue.clone(); - let slot = Arc::clone(&slot); - let builder = Arc::clone(&builder); + let (publisher, tokens, builder, slot) = ( + publisher.clone(), + tokens.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(&job), handle(&job, &builder)).await - else { - return; - }; - reply - } - Err(e) => { - tracing::warn!(error = %e, "an undecodable build; answering with the error"); - Reply::Failed { - error_tail: format!("the builder could not decode the request: {e}"), - } - } + let Some(reply) = build_reply(&msg.payload, received, &slot, &builder, &tokens).await + else { + return; }; - 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 build's reply failed"); - } - } - Err(e) => tracing::warn!(error = %e, "encoding a build's reply failed"), - } + send_reply( + &publisher, + reply_to, + serde_json::to_vec(&reply), + "a build's", + ) + .await; }); } - bail!("the subscription to {SUBJECT} ended") +} + +/// The reply to one PR check's payload: the build's, the slot's refusal, or the +/// decode error. `None` only when the slot is closed. +async fn build_reply( + payload: &[u8], + received: tokio::time::Instant, + slot: &tokio::sync::Semaphore, + builder: &nix_build::Builder, + tokens: &impl Tokens, +) -> Option { + match serde_json::from_slice::(payload) { + Ok(job) => in_slot(slot, received, || busy(&job), handle(&job, builder, tokens)).await, + Err(e) => { + tracing::warn!(error = %e, "an undecodable build; answering with the error"); + Some(Reply::Failed { + error_tail: format!("the builder could not decode the request: {e}"), + }) + } + } } /// Answer every archetype evaluation request, one at a time, with the /// evaluation's result. async fn answer_evaluations( - mut requests: async_nats::Subscriber, - queue: async_nats::Client, + mut requests: impl Stream + Unpin, + publisher: impl Publisher, evaluator: archetypes::Evaluator, ) { while let Some(msg) = requests.next().await { @@ -211,14 +307,13 @@ async fn answer_evaluations( tracing::warn!(%error_tail, "archetype evaluation failed"); } } - match serde_json::to_vec(&reply) { - Ok(body) => { - if let Err(e) = queue.publish(reply_to, body.into()).await { - tracing::warn!(error = %e, "publishing an evaluation's reply failed"); - } - } - Err(e) => tracing::warn!(error = %e, "encoding an evaluation's reply failed"), - } + send_reply( + &publisher, + reply_to, + serde_json::to_vec(&reply), + "an evaluation's", + ) + .await; } tracing::error!("the subscription to {ARCHETYPES_SUBJECT} ended"); } @@ -226,8 +321,9 @@ async fn answer_evaluations( /// 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, + mut requests: impl Stream + Unpin, + publisher: impl Publisher, + tokens: impl Tokens, builder: Arc, ) { while let Some(msg) = requests.next().await { @@ -235,21 +331,18 @@ async fn answer_locks( 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"), - } + let reply = handle_lock(&msg.payload, &builder, &tokens).await; + send_reply(&publisher, reply_to, serde_json::to_vec(&reply), "a lock's").await; } 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 { +async fn handle_lock( + payload: &[u8], + builder: &nix_build::Builder, + tokens: &impl Tokens, +) -> LockReply { let job: LockMeta = match serde_json::from_slice(payload) { Ok(job) => job, Err(e) => { @@ -259,7 +352,7 @@ async fn handle_lock(payload: &[u8], builder: &nix_build::Builder) -> LockReply }; } }; - let reply = match within(TOKEN_DEADLINE, forge_token()).await { + let reply = match within(TOKEN_DEADLINE, tokens.forge_token()).await { Ok(token) => builder.lock(&job, &token).await, Err(e) => LockReply::Failed { error_tail: format!("{e:#}"), @@ -277,6 +370,8 @@ async fn handle_lock(payload: &[u8], builder: &nix_build::Builder) -> LockReply /// Subscribe to meta build requests and answer them in the background. async fn serve_meta_builds( queue: &async_nats::Client, + publisher: &impl Publisher, + tokens: impl Tokens, builder: &Arc, slot: &Arc, ) -> Result<()> { @@ -286,7 +381,8 @@ async fn serve_meta_builds( .with_context(|| format!("subscribing to {META_SUBJECT}"))?; tokio::spawn(answer_meta_builds( requests, - queue.clone(), + publisher.clone(), + tokens, Arc::clone(builder), Arc::clone(slot), )); @@ -296,8 +392,9 @@ async fn serve_meta_builds( /// 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, + mut requests: impl Stream + Unpin, + publisher: impl Publisher, + tokens: impl Tokens, builder: Arc, slot: Arc, ) { @@ -307,42 +404,62 @@ async fn answer_meta_builds( 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)); + let (publisher, tokens, builder, slot) = ( + publisher.clone(), + tokens.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, received + REPLY_DEADLINE - REPLY_MARGIN), - ) - .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}"), - } - } + let Some(reply) = meta_reply(&msg.payload, received, &slot, &builder, &tokens).await + else { + return; }; - 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"), - } + send_reply( + &publisher, + reply_to, + serde_json::to_vec(&reply), + "a meta build's", + ) + .await; }); } tracing::error!("the subscription to {META_SUBJECT} ended"); } +/// The reply to one meta build's payload: the build's, the slot's refusal, or +/// the decode error. `None` only when the slot is closed. +async fn meta_reply( + payload: &[u8], + received: tokio::time::Instant, + slot: &tokio::sync::Semaphore, + builder: &nix_build::Builder, + tokens: &impl Tokens, +) -> Option { + match serde_json::from_slice::(payload) { + Ok(job) => { + in_slot( + slot, + received, + || busy_meta(&job), + handle_meta( + &job, + builder, + tokens, + received + REPLY_DEADLINE - REPLY_MARGIN, + ), + ) + .await + } + Err(e) => { + tracing::warn!(error = %e, "an undecodable meta build; answering with the error"); + Some(FromMetaReply::Failed { + error_tail: format!("the builder could not decode the request: {e}"), + }) + } + } +} + /// `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. @@ -401,9 +518,10 @@ fn busy_meta(job: &BuildFromMeta) -> FromMetaReply { async fn handle_meta( job: &BuildFromMeta, builder: &nix_build::Builder, + tokens: &impl Tokens, reply_by: tokio::time::Instant, ) -> FromMetaReply { - let reply = match within(TOKEN_DEADLINE, forge_token()).await { + let reply = match within(TOKEN_DEADLINE, tokens.forge_token()).await { Ok(token) => builder.build_from_meta(job, &token).await, Err(e) => FromMetaReply::Unavailable { reason: format!("{e:#}"), @@ -412,7 +530,7 @@ async fn handle_meta( let reply = push::cache_built( reply, builder.cache.as_ref(), - attic_push_token(), + tokens.push_token(), push_budget(reply_by), &job.agent, ) @@ -449,8 +567,8 @@ async fn handle_meta( } /// 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 { +async fn handle(job: &BuildAgent, builder: &nix_build::Builder, tokens: &impl Tokens) -> Reply { + let reply = match within(TOKEN_DEADLINE, tokens.forge_token()).await { Ok(token) => builder.build(job, &token).await, Err(e) => Reply::Failed { error_tail: format!("{e:#}"), @@ -562,6 +680,12 @@ fn push_target() -> Result> { fn forge_url() -> Result { let name = format!("{PREFIX}_FORGE_URL"); let url = std::env::var(&name).with_context(|| format!("{name} is unset"))?; + check_forge_url(&name, &url) +} + +/// `url` without trailing slashes, if it is `https://[:port]`; `name` is +/// the variable it came from. +fn check_forge_url(name: &str, url: &str) -> Result { let url = url.trim_end_matches('/'); if !url.starts_with("https://") || url["https://".len()..].contains('/') { bail!("{name} is {url:?}, not https://[:port]"); @@ -704,4 +828,802 @@ mod tests { assert!(SLOT_MARGIN >= TOKEN_DEADLINE + Duration::from_secs(10)); assert!(MAX_SLOT_WAIT + SLOT_MARGIN + BUILD_DEADLINE <= REPLY_DEADLINE); } + + // The daemon's request path, end to end with a stand-in `nix`: a message in, + // the bytes published on its reply subject out. + + use serde_json::{Value, json}; + use tokio::sync::mpsc::UnboundedReceiver; + + const REV: &str = "0123456789abcdef0123456789abcdef01234567"; + const OUT: &str = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-atlas"; + + /// A directory holding a stand-in `nix` that logs each call's arguments to + /// `calls` in the same directory, then runs the script it was given. + struct Fake(PathBuf); + + impl Fake { + fn new(body: &str) -> Self { + use std::sync::atomic::AtomicU64; + static NEXT: AtomicU64 = AtomicU64::new(0); + let _ = tracing_subscriber::fmt() + .with_test_writer() + .with_max_level(tracing::Level::INFO) + .try_init(); + let dir = std::env::temp_dir().join(format!( + "swarm-builder-main-{}-{}", + std::process::id(), + NEXT.fetch_add(1, Ordering::Relaxed) + )); + std::fs::create_dir_all(&dir).expect("scratch dir"); + let fake = Self(dir); + fake.script( + "nix", + &format!( + "echo \"$*\" >> \"${{0%/*}}/calls\"; echo \"$SWARM_BUILDER_GIT_TOKEN\" >> \"${{0%/*}}/tokens\"\n{body}" + ), + ); + fake + } + + /// Write the executable `name` from a child process: a write fd held here + /// could be inherited by a child another test forks, and exec'ing a file + /// something still holds open for writing fails with `ETXTBSY`. + fn script(&self, name: &str, body: &str) { + let path = self.0.join(name); + let status = std::process::Command::new("/bin/sh") + .args([ + "-c", + r#"printf '%s' "$1" > "$2" && chmod 755 "$2""#, + "sh", + &format!("#!/bin/sh\n{body}\n"), + ]) + .arg(&path) + .status() + .expect("run /bin/sh"); + assert!(status.success(), "writing {} failed", path.display()); + } + + fn builder(&self) -> Arc { + self.builder_with(None) + } + + /// The builder with a cache whose `attic` logs its arguments to `pushes`. + fn caching_builder(&self) -> Arc { + self.script("attic", "echo \"$*\" >> \"${0%/*}/pushes\""); + self.builder_with(Some(push::Remote { + attic: self.0.join("attic"), + domain: "cache.example.org".to_owned(), + cache: "grid".to_owned(), + state_dir: self.0.clone(), + })) + } + + fn builder_with(&self, cache: Option) -> Arc { + Arc::new(nix_build::Builder { + nix: self.0.join("nix"), + check_flake: "/nix/store/c-check".to_owned(), + hyperhive: "/nix/store/h-source".to_owned(), + nixpkgs: "/nix/store/n-source".to_owned(), + forge_url: "https://forge.example".to_owned(), + config_org: "agent-configs".to_owned(), + deadline: Duration::from_secs(30), + lock_deadline: Duration::from_secs(30), + system: "x86_64-linux".to_owned(), + cache, + }) + } + + fn lines(&self, file: &str) -> Vec { + std::fs::read_to_string(self.0.join(file)) + .map(|s| s.lines().map(str::to_owned).collect()) + .unwrap_or_default() + } + + /// The arguments of every `nix` call. + fn calls(&self) -> Vec { + self.lines("calls") + } + + /// The arguments of every `attic` call. + fn pushes(&self) -> Vec { + self.lines("pushes") + } + } + + impl Drop for Fake { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } + } + + /// What `answer_*` published: the subject and the decoded body. + type Sent = UnboundedReceiver<(String, Value)>; + + /// What a token read answers. + #[derive(Clone, Copy)] + enum Read { + Token(&'static str), + Unreadable(&'static str), + Stuck, + } + + impl Read { + async fn get(self) -> Result { + match self { + Self::Token(t) => Ok(t.to_owned()), + Self::Unreadable(why) => Err(anyhow!(why)), + Self::Stuck => std::future::pending().await, + } + } + } + + fn token(value: &'static str) -> Read { + Read::Token(value) + } + + fn unreadable(why: &'static str) -> Read { + Read::Unreadable(why) + } + + fn stuck() -> Read { + Read::Stuck + } + + #[derive(Clone)] + struct FakeTokens { + forge: Read, + push: Read, + } + + impl Tokens for FakeTokens { + async fn forge_token(&self) -> Result { + self.forge.get().await + } + + async fn push_token(&self) -> Result { + self.push.get().await + } + } + + /// A publisher whose first `failing` publishes fail; the rest are recorded. + #[derive(Clone)] + struct Sink { + sent: tokio::sync::mpsc::UnboundedSender<(String, Value)>, + attempts: Arc, + failing: usize, + } + + impl Publisher for Sink { + async fn publish(&self, subject: Subject, body: Vec) -> Result<(), String> { + if self.attempts.fetch_add(1, Ordering::SeqCst) < self.failing { + return Err("the connection is closed".to_owned()); + } + let body = serde_json::from_slice(&body).expect("a JSON reply"); + self.sent + .send((subject.to_string(), body)) + .expect("test is listening"); + Ok(()) + } + } + + struct Rig { + sink: Sink, + tokens: FakeTokens, + } + + fn rig_with(forge: Read, push: Read, failing: usize) -> (Rig, Sent) { + let (sent, rx) = tokio::sync::mpsc::unbounded_channel(); + let sink = Sink { + sent, + attempts: Arc::default(), + failing, + }; + let tokens = FakeTokens { forge, push }; + (Rig { sink, tokens }, rx) + } + + fn env() -> (Rig, Sent) { + rig_with(token("forge-token"), token("push-token"), 0) + } + + async fn next(sent: &mut Sent) -> (String, Value) { + tokio::time::timeout(Duration::from_mins(2), sent.recv()) + .await + .expect("a reply in time") + .expect("the env is alive") + } + + fn msg(reply: Option<&str>, payload: &str) -> Message { + Message { + subject: Subject::from("$SWARM.builder.test"), + reply: reply.map(Subject::from), + payload: payload.to_owned().into(), + headers: None, + status: None, + description: None, + length: payload.len(), + } + } + + fn build_json(agent: &str, rev: &str, pr: u64) -> String { + json!({"agent": agent, "rev": rev, "pr": pr}).to_string() + } + + fn meta_json(agent: &str, rev: &str) -> String { + json!({"agent": agent, "meta_rev": rev}).to_string() + } + + async fn serve_builds(builder: Arc, env: Rig, msgs: Vec) { + answer_builds( + stream::iter(msgs), + env.sink, + env.tokens, + builder, + Arc::new(Semaphore::new(1)), + ) + .await; + } + + async fn serve_meta(builder: Arc, env: Rig, msgs: Vec) { + answer_meta_builds( + stream::iter(msgs), + env.sink, + env.tokens, + builder, + Arc::new(Semaphore::new(1)), + ) + .await; + } + + use futures_util::stream; + + #[tokio::test] + async fn a_build_is_answered_on_its_reply_subject_with_the_path_nix_printed() { + let fake = Fake::new(&format!("echo {OUT}")); + let (env, mut sent) = env(); + let request = msg(Some("_INBOX.7"), &build_json("atlas", REV, 3)); + serve_builds(fake.builder(), env, vec![request]).await; + let (subject, body) = next(&mut sent).await; + assert_eq!(subject, "_INBOX.7"); + assert_eq!(body, json!({"state": "built", "out_path": OUT})); + assert_eq!(fake.lines("tokens"), ["forge-token"]); + let calls = fake.calls(); + assert_eq!(calls.len(), 1, "{calls:?}"); + assert!( + calls[0].contains(&format!( + "agent-configs/atlas?ref=refs/pull/3/head&rev={REV}" + )), + "{calls:?}" + ); + } + + #[tokio::test] + async fn a_failed_build_is_answered_failed_with_what_nix_said_last() { + let fake = Fake::new( + "echo \"warning: unrelated\" >&2\n\ + echo \"error: unable to download 'https://forge.example/agent-configs/atlas': Could not resolve host\" >&2\n\ + exit 1", + ); + let (env, mut sent) = env(); + let request = msg(Some("_INBOX.1"), &build_json("atlas", REV, 3)); + serve_builds(fake.builder(), env, vec![request]).await; + let (subject, body) = next(&mut sent).await; + assert_eq!(subject, "_INBOX.1"); + assert_eq!(body["state"], "failed"); + assert!( + body["error_tail"] + .as_str() + .is_some_and(|t| t.contains("Could not resolve host")), + "{body}" + ); + assert!(body.get("out_path").is_none(), "{body}"); + } + + /// Each refused request is answered `failed` without `nix` running; the last + /// one is valid and builds, so the empty call log of the others means + /// something. + #[tokio::test] + async fn requests_that_would_bend_the_reference_are_refused_without_running_nix() { + let fake = Fake::new(&format!("echo {OUT}")); + let upper = REV.to_uppercase(); + let long = format!("{REV}0"); + let refused = [ + ("Atlas", REV, 3), + ("", REV, 3), + ("a/b", REV, 3), + ("a b", REV, 3), + ("atlas", "abc", 3), + ("atlas", upper.as_str(), 3), + ("atlas", long.as_str(), 3), + ("atlas", REV, 0), + ]; + for (agent, rev, pr) in refused { + let (env, mut sent) = env(); + let request = msg(Some("_INBOX.1"), &build_json(agent, rev, pr)); + serve_builds(fake.builder(), env, vec![request]).await; + let (_, body) = next(&mut sent).await; + assert_eq!(body["state"], "failed", "{agent:?} {rev:?} {pr}: {body}"); + assert_eq!(fake.calls(), Vec::::new(), "{agent:?} {rev:?} {pr}"); + } + let (env, mut sent) = env(); + let valid = msg(Some("_INBOX.1"), &build_json("atlas", REV, 3)); + serve_builds(fake.builder(), env, vec![valid]).await; + let (_, body) = next(&mut sent).await; + assert_eq!(body["state"], "built"); + assert_eq!(fake.calls().len(), 1); + } + + #[tokio::test] + async fn an_undecodable_request_is_answered_with_the_decode_error_and_builds_nothing() { + let fake = Fake::new(&format!("echo {OUT}")); + let payloads = [ + "", + "not json", + "{}", + r#"{"agent":"atlas","rev":"abc"}"#, + r#"{"agent":"atlas","rev":"abc","pr":"3"}"#, + r#"{"agent":7,"rev":"abc","pr":3}"#, + ]; + for payload in payloads { + let (env, mut sent) = env(); + serve_builds(fake.builder(), env, vec![msg(Some("_INBOX.1"), payload)]).await; + let (_, body) = next(&mut sent).await; + assert_eq!(body["state"], "failed", "{payload:?}"); + assert!( + body["error_tail"] + .as_str() + .is_some_and(|t| t.starts_with("the builder could not decode the request: ")), + "{payload:?}: {body}" + ); + } + assert_eq!(fake.calls(), Vec::::new()); + } + + #[tokio::test] + async fn a_request_with_no_reply_subject_is_dropped_and_the_next_one_still_builds() { + let fake = Fake::new(&format!("echo {OUT}")); + let (env, mut sent) = env(); + let msgs = vec![ + msg(None, &build_json("atlas", REV, 3)), + msg(Some("_INBOX.2"), &build_json("atlas", REV, 4)), + ]; + serve_builds(fake.builder(), env, msgs).await; + let (subject, _) = next(&mut sent).await; + assert_eq!(subject, "_INBOX.2"); + let calls = fake.calls(); + assert_eq!(calls.len(), 1, "{calls:?}"); + assert!(calls[0].contains("refs/pull/4/head"), "{calls:?}"); + assert!(sent.try_recv().is_err()); + } + + /// A forge token that cannot be read fails the build before `nix` is + /// asked for anything, and the reply says why. + #[tokio::test] + async fn a_build_whose_forge_token_cannot_be_read_is_failed_without_running_nix() { + let fake = Fake::new(&format!("echo {OUT}")); + let (env, mut sent) = rig_with( + unreadable("logging in to the secret store: connection refused"), + token("p"), + 0, + ); + let request = msg(Some("_INBOX.1"), &build_json("atlas", REV, 3)); + serve_builds(fake.builder(), env, vec![request]).await; + let (_, body) = next(&mut sent).await; + assert_eq!( + body, + json!({"state": "failed", "error_tail": "logging in to the secret store: connection refused"}) + ); + assert_eq!(fake.calls(), Vec::::new()); + } + + #[tokio::test(start_paused = true)] + async fn a_forge_token_read_that_never_returns_fails_the_build_after_its_deadline() { + let fake = Fake::new(&format!("echo {OUT}")); + let (env, mut sent) = rig_with(stuck(), token("p"), 0); + let started = Instant::now(); + let request = msg(Some("_INBOX.1"), &build_json("atlas", REV, 3)); + serve_builds(fake.builder(), env, vec![request]).await; + let (_, body) = next(&mut sent).await; + assert_eq!(body["state"], "failed"); + assert!( + body["error_tail"] + .as_str() + .is_some_and(|t| t.contains("took more than 20 s")), + "{body}" + ); + assert_eq!(started.elapsed(), TOKEN_DEADLINE); + assert_eq!(fake.calls(), Vec::::new()); + } + + /// A reply that cannot be published is logged; the builder goes on to the + /// next request. + #[tokio::test] + async fn a_reply_that_cannot_be_published_does_not_stop_the_next_build() { + let fake = Fake::new(&format!("echo {OUT}")); + let (env, mut sent) = rig_with(token("t"), token("p"), 1); + let msgs = vec![ + msg(Some("_INBOX.1"), &build_json("atlas", REV, 3)), + msg(Some("_INBOX.2"), &build_json("atlas", REV, 4)), + ]; + serve_builds(fake.builder(), env, msgs).await; + let (_, body) = next(&mut sent).await; + assert_eq!(body["state"], "built"); + assert_eq!(fake.calls().len(), 2, "both requests were built"); + assert!(sent.try_recv().is_err(), "only the second reply got out"); + } + + /// Two requests for the same head are both built, one after the other: the + /// slot serialises them, nothing deduplicates them. + #[tokio::test] + async fn two_requests_for_the_same_head_build_one_after_the_other() { + let fake = Fake::new(&format!( + "echo start >> \"${{0%/*}}/order\"; sleep 0.3; echo end >> \"${{0%/*}}/order\"; echo {OUT}" + )); + let (env, mut sent) = env(); + let same = build_json("atlas", REV, 3); + let msgs = vec![msg(Some("_INBOX.1"), &same), msg(Some("_INBOX.2"), &same)]; + serve_builds(fake.builder(), env, msgs).await; + let mut subjects = vec![next(&mut sent).await.0, next(&mut sent).await.0]; + subjects.sort(); + assert_eq!(subjects, ["_INBOX.1", "_INBOX.2"]); + assert_eq!(fake.lines("order"), ["start", "end", "start", "end"]); + assert_eq!(fake.calls().len(), 2); + } + + /// A request that arrives while the slot is held past the wait limit is + /// refused and never reaches `nix`. + #[tokio::test(start_paused = true)] + async fn a_build_that_waited_out_the_slot_is_answered_busy_without_running_nix() { + let fake = Fake::new(&format!("echo {OUT}")); + let (env, _sent) = env(); + let slot = Semaphore::new(1); + let _held = slot.acquire().await.expect("open"); + let received = Instant::now(); + let reply = build_reply( + build_json("atlas", REV, 3).as_bytes(), + received, + &slot, + &fake.builder(), + &env.tokens, + ) + .await; + assert!( + matches!(&reply, Some(Reply::Failed { error_tail }) if error_tail.contains("busy")), + "{reply:?}" + ); + assert_eq!(received.elapsed(), MAX_SLOT_WAIT); + assert_eq!(fake.calls(), Vec::::new()); + } + + #[tokio::test] + async fn a_closed_slot_gets_no_reply() { + let fake = Fake::new(&format!("echo {OUT}")); + let (env, _sent) = env(); + let slot = Semaphore::new(1); + slot.close(); + let reply = build_reply( + build_json("atlas", REV, 3).as_bytes(), + Instant::now(), + &slot, + &fake.builder(), + &env.tokens, + ) + .await; + assert!(reply.is_none(), "{reply:?}"); + assert_eq!(fake.calls(), Vec::::new()); + } + + /// `nix` for a meta build: `build` prints the path, `eval` of the option + /// prints `eval_says`. + fn meta_nix(eval_says: &str) -> Fake { + Fake::new(&format!( + "case \"$1\" in build) echo {OUT};; eval) {eval_says};; esac" + )) + } + + #[tokio::test] + async fn a_meta_build_is_answered_with_the_path_the_system_and_the_private_users_flag() { + let fake = meta_nix("echo true"); + let (env, mut sent) = env(); + let request = msg(Some("_INBOX.5"), &meta_json("atlas", REV)); + serve_meta(fake.builder(), env, vec![request]).await; + let (subject, body) = next(&mut sent).await; + assert_eq!(subject, "_INBOX.5"); + assert_eq!( + body, + json!({ + "state": "built", + "out_path": OUT, + "system": "x86_64-linux", + "cached": false, + "private_users": true, + }) + ); + assert_eq!(fake.lines("tokens"), ["forge-token"; 2]); + let calls = fake.calls(); + assert_eq!(calls.len(), 2, "{calls:?}"); + assert!( + calls[0].contains(&format!( + "git+https://forge.example/internal/meta.git?rev={REV}#nixosConfigurations.atlas.config.system.build.toplevel" + )), + "{calls:?}" + ); + assert!( + calls[1].contains("services.hyperhive.agent.user.privateUsers"), + "{calls:?}" + ); + } + + #[tokio::test] + async fn private_users_is_false_unless_nix_eval_printed_true() { + for eval_says in [ + "echo false", + "echo null", + "echo 'error: no such option' >&2; exit 1", + ] { + let fake = meta_nix(eval_says); + let (env, mut sent) = env(); + let request = msg(Some("_INBOX.1"), &meta_json("atlas", REV)); + serve_meta(fake.builder(), env, vec![request]).await; + let (_, body) = next(&mut sent).await; + assert_eq!(body["state"], "built", "{eval_says}"); + assert_eq!(body["private_users"], false, "{eval_says}"); + } + } + + #[tokio::test] + async fn a_built_meta_closure_is_pushed_and_the_reply_says_so() { + let fake = meta_nix("echo true"); + let (env, mut sent) = env(); + let request = msg(Some("_INBOX.1"), &meta_json("atlas", REV)); + serve_meta(fake.caching_builder(), env, vec![request]).await; + let (_, body) = next(&mut sent).await; + assert_eq!(body["cached"], true, "{body}"); + assert_eq!(body["private_users"], true, "the flag survives the push"); + assert_eq!(body["out_path"], OUT); + assert_eq!(fake.pushes(), [format!("push swarm:grid {OUT}")]); + } + + /// Nothing is pushed for a build that did not succeed; the presence control + /// is the test above, which uses the same cache. + #[tokio::test] + async fn a_failed_meta_build_is_failed_and_pushes_nothing() { + let fake = Fake::new("echo 'error: attribute missing' >&2; exit 1"); + let (env, mut sent) = env(); + let request = msg(Some("_INBOX.1"), &meta_json("atlas", REV)); + serve_meta(fake.caching_builder(), env, vec![request]).await; + let (_, body) = next(&mut sent).await; + assert_eq!(body["state"], "failed"); + assert!( + body["error_tail"] + .as_str() + .is_some_and(|t| t.ends_with("attribute missing")), + "{body}" + ); + assert_eq!(fake.pushes(), Vec::::new()); + } + + #[tokio::test] + async fn an_unreadable_push_token_leaves_the_meta_build_built_and_not_cached() { + let fake = meta_nix("echo true"); + let (env, mut sent) = rig_with(token("t"), unreadable("nothing at the push path"), 0); + let request = msg(Some("_INBOX.1"), &meta_json("atlas", REV)); + serve_meta(fake.caching_builder(), env, vec![request]).await; + let (_, body) = next(&mut sent).await; + assert_eq!(body["state"], "built"); + assert_eq!(body["cached"], false); + assert_eq!(fake.pushes(), Vec::::new()); + } + + #[tokio::test] + async fn a_meta_build_whose_forge_token_cannot_be_read_is_unavailable_without_running_nix() { + let fake = meta_nix("echo true"); + let (env, mut sent) = rig_with(unreadable("nothing at the forge path"), token("p"), 0); + let request = msg(Some("_INBOX.1"), &meta_json("atlas", REV)); + serve_meta(fake.caching_builder(), env, vec![request]).await; + let (_, body) = next(&mut sent).await; + assert_eq!( + body, + json!({"state": "unavailable", "reason": "nothing at the forge path"}) + ); + assert_eq!(fake.calls(), Vec::::new()); + assert_eq!(fake.pushes(), Vec::::new()); + } + + #[tokio::test] + async fn bad_meta_requests_are_failed_without_running_nix() { + let fake = meta_nix("echo true"); + for payload in [ + meta_json("Atlas", REV), + meta_json("atlas", "abc"), + "not json".to_owned(), + "{}".to_owned(), + ] { + let (env, mut sent) = env(); + serve_meta(fake.builder(), env, vec![msg(Some("_INBOX.1"), &payload)]).await; + let (_, body) = next(&mut sent).await; + assert_eq!(body["state"], "failed", "{payload}: {body}"); + } + assert_eq!(fake.calls(), Vec::::new()); + } + + #[tokio::test] + async fn a_meta_request_with_no_reply_subject_is_dropped() { + let fake = meta_nix("echo true"); + let (env, mut sent) = env(); + let msgs = vec![ + msg(None, &meta_json("atlas", REV)), + msg(Some("_INBOX.2"), &meta_json("atlas", REV)), + ]; + serve_meta(fake.builder(), env, msgs).await; + assert_eq!(next(&mut sent).await.0, "_INBOX.2"); + assert_eq!(fake.calls().len(), 2, "one build and its option read"); + } + + #[tokio::test(start_paused = true)] + async fn a_meta_build_that_waited_out_the_slot_is_unavailable() { + let fake = meta_nix("echo true"); + let (env, _sent) = env(); + let slot = Semaphore::new(1); + let _held = slot.acquire().await.expect("open"); + let reply = meta_reply( + meta_json("atlas", REV).as_bytes(), + Instant::now(), + &slot, + &fake.builder(), + &env.tokens, + ) + .await; + assert!( + matches!(&reply, Some(FromMetaReply::Unavailable { reason }) if reason.contains("busy")), + "{reply:?}" + ); + assert_eq!(fake.calls(), Vec::::new()); + } + + async fn serve_locks(builder: Arc, env: Rig, msgs: Vec) { + answer_locks(stream::iter(msgs), env.sink, env.tokens, builder).await; + } + + #[tokio::test] + async fn a_lock_is_answered_with_the_lock_nix_wrote_for_the_flake_it_was_sent() { + let fake = Fake::new("cp flake.nix flake.lock"); + let (env, mut sent) = env(); + let request = msg( + Some("_INBOX.3"), + &json!({"flake_nix": "{ a = 1; }"}).to_string(), + ); + serve_locks(fake.builder(), env, vec![request]).await; + let (subject, body) = next(&mut sent).await; + assert_eq!(subject, "_INBOX.3"); + assert_eq!(body, json!({"state": "locked", "flake_lock": "{ a = 1; }"})); + assert_eq!(fake.lines("tokens"), ["forge-token"]); + } + + #[tokio::test] + async fn a_lock_that_fails_or_whose_token_cannot_be_read_is_answered_failed() { + let fake = Fake::new("echo 'error: input unavailable' >&2; exit 1"); + let lock_request = json!({"flake_nix": "{ }"}).to_string(); + let (env, mut sent) = env(); + serve_locks( + fake.builder(), + env, + vec![msg(Some("_INBOX.1"), &lock_request)], + ) + .await; + let (_, body) = next(&mut sent).await; + assert_eq!(body["state"], "failed"); + assert!( + body["error_tail"] + .as_str() + .is_some_and(|t| t.ends_with("input unavailable")), + "{body}" + ); + + let (env, mut sent) = rig_with(unreadable("nothing at the forge path"), token("p"), 0); + let calls_before = fake.calls().len(); + serve_locks( + fake.builder(), + env, + vec![msg(Some("_INBOX.1"), &lock_request)], + ) + .await; + let (_, body) = next(&mut sent).await; + assert_eq!( + body, + json!({"state": "failed", "error_tail": "nothing at the forge path"}) + ); + assert_eq!( + fake.calls().len(), + calls_before, + "the token read came first" + ); + } + + #[tokio::test] + async fn an_undecodable_lock_is_answered_failed_and_a_lock_with_no_reply_subject_is_dropped() { + let fake = Fake::new("cp flake.nix flake.lock"); + let (env, mut sent) = env(); + let msgs = vec![ + msg(None, &json!({"flake_nix": "{ }"}).to_string()), + msg(Some("_INBOX.1"), "{}"), + ]; + serve_locks(fake.builder(), env, msgs).await; + let (subject, body) = next(&mut sent).await; + assert_eq!(subject, "_INBOX.1"); + assert_eq!(body["state"], "failed"); + assert!(sent.try_recv().is_err()); + assert_eq!(fake.calls(), Vec::::new()); + } + + fn evaluator(fake: &Fake) -> archetypes::Evaluator { + archetypes::Evaluator { + nix: fake.0.join("nix"), + hyperhive: "/nix/store/h-source".to_owned(), + nixpkgs: "/nix/store/n-source".to_owned(), + deadline: Duration::from_secs(30), + } + } + + #[tokio::test] + async fn an_archetype_evaluation_is_answered_with_the_meta_nix_printed() { + let fake = Fake::new(r#"echo '{"coder":{"description":"d"}}'"#); + let (env, mut sent) = env(); + let msgs = vec![msg(None, "{}"), msg(Some("_INBOX.4"), "{}")]; + answer_evaluations(stream::iter(msgs), env.sink, evaluator(&fake)).await; + let (subject, body) = next(&mut sent).await; + assert_eq!(subject, "_INBOX.4"); + assert_eq!( + body, + json!({ + "state": "evaluated", + "source": "/nix/store/h-source", + "archetypes": {"coder": {"description": "d"}}, + }) + ); + assert_eq!( + fake.calls().len(), + 1, + "the request with no reply was not run" + ); + } + + #[tokio::test] + async fn a_failed_archetype_evaluation_is_answered_failed() { + let fake = Fake::new("echo 'error: evaluation aborted' >&2; exit 1"); + let (env, mut sent) = env(); + let msgs = vec![msg(Some("_INBOX.4"), "{}")]; + answer_evaluations(stream::iter(msgs), env.sink, evaluator(&fake)).await; + let (_, body) = next(&mut sent).await; + assert_eq!(body["state"], "failed"); + assert!( + body["error_tail"] + .as_str() + .is_some_and(|t| t.contains("evaluation aborted")), + "{body}" + ); + } + + #[test] + fn the_forge_url_is_https_with_a_host_and_no_path() { + let ok = |url| check_forge_url("V", url).expect("accepted"); + assert_eq!(ok("https://forge.example"), "https://forge.example"); + assert_eq!( + ok("https://forge.example:3000//"), + "https://forge.example:3000" + ); + for bad in [ + "http://forge.example", + "forge.example", + "https://forge.example/sub", + "", + ] { + let e = check_forge_url("SWARM_BUILDER_FORGE_URL", bad).expect_err(bad); + assert!( + e.to_string().starts_with("SWARM_BUILDER_FORGE_URL is "), + "{bad}: {e}" + ); + } + } }