From bf848ba65f4f1b84fb78869aa7ffe1076070c2ec Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 17 Oct 2026 15:50:54 +0200 Subject: [PATCH] swarm-builder: refuse a build that waited too long for the slot A request that waits behind another build, e.g. one a controller restart orphaned, could start a full build after most of the controller's 61-minute reply deadline was gone. The builder now waits at most REPLY_DEADLINE - BUILD_DEADLINE - 30 s for the slot and answers Failed, without building, past that. The 30 s covers the forge token read, now bounded to 20 s, and the reply's trip back. Fixes #5437 --- docs/swarm/builder.md | 7 +- swarm-builder/Cargo.toml | 5 + swarm-builder/src/main.rs | 186 ++++++++++++++++++++++++++---- swarm-queue-client/src/builder.rs | 4 +- 4 files changed, 180 insertions(+), 22 deletions(-) diff --git a/docs/swarm/builder.md b/docs/swarm/builder.md index f918b1f2..0282af76 100644 --- a/docs/swarm/builder.md +++ b/docs/swarm/builder.md @@ -20,7 +20,11 @@ Where it runs and how it's enabled: [`services.md`](services.md#builder). finishes and writes its status on its own, older commit. 2. The node sets the status to `pending`, then sends the build as a request on the queue subject `$SWARM.builder.build` and waits up to 61 minutes for the - reply. + reply. The builder runs one build at a time and refuses a request that has + waited 30 seconds for its turn, such as one behind a build a controller + restart orphaned: a build started later could still run when the 61 + minutes end. Reading the forge token before a build has 20 seconds, and + the rest of the minute covers the reply's trip back. 3. The builder runs `nix build` of `nix/builder/check-flake`'s `check` output with every input overridden: - `hyperhive` and `nixpkgs`: the sources the builder's own host was @@ -39,6 +43,7 @@ Where it runs and how it's enabled: [`services.md`](services.md#builder). | ----------------------------------------- | --------- | --------------------------- | ------ | | built | `success` | the store path | done | | `nix` failed or timed out | `failure` | the last line `nix` printed | failed | + | the builder busy for 30 seconds | `failure` | why | failed | | no builder listening, or no reply in time | `error` | why | failed | The full tail is in the failed node's error on the Jobs page and in the diff --git a/swarm-builder/Cargo.toml b/swarm-builder/Cargo.toml index 2ba9ba11..4275fee9 100644 --- a/swarm-builder/Cargo.toml +++ b/swarm-builder/Cargo.toml @@ -22,5 +22,10 @@ swarm-secret-client.workspace = true tokio.workspace = true tracing.workspace = true +# `test-util` for `#[tokio::test(start_paused = true)]`: the slot tests wait +# out the build slot's wait limit on paused time. +[dev-dependencies] +tokio = { workspace = true, features = ["test-util"] } + [lints] workspace = true diff --git a/swarm-builder/src/main.rs b/swarm-builder/src/main.rs index b46b6fe7..8f409d12 100644 --- a/swarm-builder/src/main.rs +++ b/swarm-builder/src/main.rs @@ -12,14 +12,16 @@ //! secret is read at start and again when authelia refuses it, the forge token //! once per build, both from the store under that identity. +use std::future::Future; use std::path::PathBuf; use std::sync::Arc; +use std::time::Duration; -use anyhow::{Context, Result, bail}; +use anyhow::{Context, Result, anyhow, bail}; use futures_util::StreamExt as _; use swarm_queue_client::builder::{ - ARCHETYPES_DEADLINE, ARCHETYPES_SUBJECT, BUILD_DEADLINE, BuildAgent, QUEUE_GROUP, Reply, - SUBJECT, + ARCHETYPES_DEADLINE, ARCHETYPES_SUBJECT, BUILD_DEADLINE, BuildAgent, QUEUE_GROUP, + REPLY_DEADLINE, Reply, SUBJECT, }; use swarm_queue_client::stored_secret::StoredSecret; use swarm_queue_client::{ClientSecret, QueueConfig}; @@ -35,6 +37,23 @@ const CERT_ROLE: &str = "swarm-builder"; /// The environment prefix `swarm-builder.nix` sets. const PREFIX: &str = "SWARM_BUILDER"; +/// The longest the store login and forge token read may take before a build. +/// Each store request alone may take 30 s, so without this bound the two could +/// spend the minute `REPLY_DEADLINE` leaves after `BUILD_DEADLINE`. +const TOKEN_DEADLINE: Duration = Duration::from_secs(20); + +/// What is held back from the minute `REPLY_DEADLINE` leaves after +/// `BUILD_DEADLINE`: [`TOKEN_DEADLINE`], then 10 s to publish the reply and for +/// it to reach the controller. Both come after the slot wait. +const SLOT_MARGIN: Duration = Duration::from_secs(30); + +/// The longest a request may wait for the build slot. A build that gets the +/// slot later could still be running when the controller stops waiting for the +/// reply. +const MAX_SLOT_WAIT: Duration = REPLY_DEADLINE + .saturating_sub(BUILD_DEADLINE) + .saturating_sub(SLOT_MARGIN); + #[tokio::main] async fn main() -> Result<()> { hive_log::init(); @@ -93,10 +112,12 @@ async fn main() -> Result<()> { tokio::spawn(answer_evaluations(evaluations, queue.clone(), evaluator)); // One build at a time. swarm-controller already sends one at a time; this - // holds the bound for any request it did not serialise. + // holds the bound for any request it did not serialise, and `in_slot` + // refuses one that waited too long behind another. let slot = Arc::new(tokio::sync::Semaphore::new(1)); 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; @@ -105,10 +126,21 @@ async fn main() -> Result<()> { let slot = Arc::clone(&slot); let builder = Arc::clone(&builder); tokio::spawn(async move { - let Ok(_permit) = slot.acquire().await else { - return; + let reply = match serde_json::from_slice::(&msg.payload) { + Ok(job) => { + let Some(reply) = in_slot(&slot, received, &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 reply = handle(&msg.payload, &builder).await; match serde_json::to_vec(&reply) { Ok(body) => { if let Err(e) = queue.publish(reply_to, body.into()).await { @@ -155,27 +187,60 @@ async fn answer_evaluations( tracing::error!("the subscription to {ARCHETYPES_SUBJECT} ended"); } -/// Decode one request, build it and log the build. -async fn handle(payload: &[u8], builder: &nix_build::Builder) -> Reply { - let job: BuildAgent = match serde_json::from_slice(payload) { - Ok(job) => job, - Err(e) => { - tracing::warn!(error = %e, "an undecodable build; answering with the error"); - return Reply::Failed { - error_tail: format!("the builder could not decode the request: {e}"), - }; - } +/// `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. +async fn in_slot( + slot: &tokio::sync::Semaphore, + received: tokio::time::Instant, + job: &BuildAgent, + 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() + ), + }); }; - let reply = match forge_token().await { - Ok(token) => builder.build(&job, &token).await, + let _permit = permit.ok()?; + Some(build.await) +} + +/// 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 { + Ok(token) => builder.build(job, &token).await, Err(e) => Reply::Failed { error_tail: format!("{e:#}"), }, }; - log_build(&job, &reply); + log_build(job, &reply); reply } +/// `read`'s result, or an error once it has run for `deadline`. +async fn within(deadline: Duration, read: impl Future>) -> Result { + tokio::time::timeout(deadline, read) + .await + .unwrap_or_else(|_| { + Err(anyhow!( + "reading the forge token took more than {} s", + deadline.as_secs() + )) + }) +} + /// The one journal line a build leaves. fn log_build(job: &BuildAgent, reply: &Reply) { match reply { @@ -244,3 +309,84 @@ fn forge_url() -> Result { } Ok(url.to_owned()) } + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicBool, Ordering}; + + use tokio::sync::Semaphore; + use tokio::time::Instant; + + use super::*; + + fn job() -> BuildAgent { + BuildAgent { + agent: "atlas".to_owned(), + rev: "abc".to_owned(), + pr: 3, + } + } + + /// A request behind a build that holds the slot for `held`: its reply, + /// whether its own build ran, and how long after arriving it was answered. + async fn behind_a_build_of(held: Duration) -> (Option, bool, Duration) { + let slot = Arc::new(Semaphore::new(1)); + let running = Arc::clone(&slot).acquire_owned().await.expect("open"); + tokio::spawn(async move { + tokio::time::sleep(held).await; + drop(running); + }); + let received = Instant::now(); + let ran = AtomicBool::new(false); + let build = async { + ran.store(true, Ordering::SeqCst); + Reply::Built { + out_path: "/nix/store/x".to_owned(), + } + }; + let reply = in_slot(&slot, received, &job(), build).await; + (reply, ran.load(Ordering::SeqCst), received.elapsed()) + } + + #[tokio::test(start_paused = true)] + async fn a_request_that_waited_past_the_slot_limit_is_refused_without_building() { + let (reply, built, answered_after) = + behind_a_build_of(MAX_SLOT_WAIT + Duration::from_mins(1)).await; + assert!( + matches!(&reply, Some(Reply::Failed { error_tail }) if error_tail.contains("busy")), + "{reply:?}" + ); + assert!(!built, "a refused request must not build"); + assert_eq!(answered_after, MAX_SLOT_WAIT); + } + + #[tokio::test(start_paused = true)] + async fn a_request_that_gets_the_slot_in_time_builds() { + let (reply, built, _) = + behind_a_build_of(MAX_SLOT_WAIT.saturating_sub(Duration::from_secs(1))).await; + assert!(matches!(reply, Some(Reply::Built { .. })), "{reply:?}"); + assert!(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(); + let stuck = within(TOKEN_DEADLINE, std::future::pending::>()).await; + assert_eq!(started.elapsed(), TOKEN_DEADLINE); + assert!( + matches!(&stuck, Err(e) if e.to_string().contains("took more than 20 s")), + "{stuck:?}" + ); + let prompt = within(TOKEN_DEADLINE, async { Ok("t".to_owned()) }).await; + assert_eq!(prompt.expect("read"), "t"); + } + + #[test] + fn a_build_started_at_the_slot_limit_ends_inside_the_reply_deadline() { + assert!(MAX_SLOT_WAIT > Duration::ZERO); + assert!(MAX_SLOT_WAIT + BUILD_DEADLINE < REPLY_DEADLINE); + assert!(SLOT_MARGIN >= Duration::from_secs(30)); + assert!(SLOT_MARGIN >= TOKEN_DEADLINE + Duration::from_secs(10)); + assert!(MAX_SLOT_WAIT + SLOT_MARGIN + BUILD_DEADLINE <= REPLY_DEADLINE); + } +} diff --git a/swarm-queue-client/src/builder.rs b/swarm-queue-client/src/builder.rs index 801b4971..9ba1e6bd 100644 --- a/swarm-queue-client/src/builder.rs +++ b/swarm-queue-client/src/builder.rs @@ -22,7 +22,9 @@ pub const QUEUE_GROUP: &str = "builder"; pub const BUILD_DEADLINE: Duration = Duration::from_hours(1); /// How long the controller waits for the reply, and the queue lets the -/// builder answer: the build at its deadline, plus a minute for the rest. +/// builder answer: the build at its deadline, plus a minute the builder spends +/// on a request's wait for the build slot, the forge token read before the +/// build and the reply's trip back. pub const REPLY_DEADLINE: Duration = Duration::from_mins(61); /// One build: an agent's config repo at one commit, on the builder's own