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

@@ -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

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

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,
+    ARCHETYPES_DEADLINE, ARCHETYPES_SUBJECT, BUILD_DEADLINE, BuildAgent, QUEUE_GROUP,
-    SUBJECT,
+    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 {
+            let reply = match serde_json::from_slice::<BuildAgent>(&msg.payload) {
+                Ok(job) => {
+                    let Some(reply) = in_slot(&slot, received, &job, handle(&job, &builder)).await
+                    else {
                         return;
                     };
-            let reply = handle(&msg.payload, &builder).await;
+                    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}"),
+                    }
+                }
+            };
             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.
+/// `build`'s reply once `job` holds the build slot, or a refusal, without
-async fn handle(payload: &[u8], builder: &nix_build::Builder) -> Reply {
+/// running `build`, once it has waited [`MAX_SLOT_WAIT`] since `received`.
-    let job: BuildAgent = match serde_json::from_slice(payload) {
+/// `None` only when the slot is closed.
-        Ok(job) => job,
+async fn in_slot(
-        Err(e) => {
+    slot: &tokio::sync::Semaphore,
-            tracing::warn!(error = %e, "an undecodable build; answering with the error");
+    received: tokio::time::Instant,
-            return Reply::Failed {
+    job: &BuildAgent,
-                error_tail: format!("the builder could not decode the request: {e}"),
+    build: impl Future<Output = Reply>,
+) -> Option<Reply> {
+    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 _permit = permit.ok()?;
+    Some(build.await)
 }
-    };
+
-    let reply = match forge_token().await {
+/// Build one request and log the build.
-        Ok(token) => builder.build(&job, &token).await,
+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<T>(deadline: Duration, read: impl Future<Output = Result<T>>) -> Result<T> {
+    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<String> {
     }
     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<Reply>, 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::<Result<String>>()).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);
+    }
+}

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