From cce00a65929b6a83b2d2e1749e68aaa406b8f799 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 25 Oct 2026 18:20:41 +0100 Subject: [PATCH] swarm-builder: log each archetype eval's elapsed time, wait 5 minutes --- Cargo.lock | 1 + docs/swarm/builder.md | 5 +- swarm-builder/Cargo.toml | 1 + swarm-builder/src/archetypes.rs | 179 ++++++++++++++++++++++++++--- swarm-builder/src/nix_build.rs | 10 +- swarm-controller/src/archetypes.rs | 4 +- swarm-wire/src/builder.rs | 19 ++- 7 files changed, 195 insertions(+), 24 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7ed609cb..998f2595 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4834,6 +4834,7 @@ dependencies = [ "swarm-wire", "tokio", "tracing", + "tracing-subscriber", ] [[package]] diff --git a/docs/swarm/builder.md b/docs/swarm/builder.md index 917d29ca..fddc36c2 100644 --- a/docs/swarm/builder.md +++ b/docs/swarm/builder.md @@ -112,8 +112,9 @@ status has no link. The builder also answers `$SWARM.builder.archetypes`: it evaluates the `meta` of every agent archetype its own hyperhive source declares, with no forge token, and answers with the JSON. An evaluation doesn't wait for a -build in progress and has two minutes. What the controller does with the -answer: [`archetypes.md`](archetypes.md). +build in progress and has five minutes; the controller waits six. The builder +logs each evaluation's time as `elapsed_ms`, and a timeout's error carries the last +lines nix printed. What the controller does with the answer: [`archetypes.md`](archetypes.md). ## Meta-flake lock diff --git a/swarm-builder/Cargo.toml b/swarm-builder/Cargo.toml index bbb5d424..7e3d90fe 100644 --- a/swarm-builder/Cargo.toml +++ b/swarm-builder/Cargo.toml @@ -27,6 +27,7 @@ tracing.workspace = true # out the build slot's wait limit on paused time. [dev-dependencies] tokio = { workspace = true, features = ["test-util"] } +tracing-subscriber.workspace = true [lints] workspace = true diff --git a/swarm-builder/src/archetypes.rs b/swarm-builder/src/archetypes.rs index cf35081f..1a240bd4 100644 --- a/swarm-builder/src/archetypes.rs +++ b/swarm-builder/src/archetypes.rs @@ -7,9 +7,11 @@ use std::path::PathBuf; use std::process::Stdio; -use std::time::Duration; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; use swarm_wire::builder::ArchetypesReply; +use tokio::io::AsyncReadExt as _; /// What every evaluation is run with, fixed at start. pub struct Evaluator { @@ -60,31 +62,90 @@ impl Evaluator { } } -/// Run `cmd` to its end or `deadline`, and read the JSON object it printed or -/// the tail of what it said. -async fn run( +/// How much of nix's stderr is held while it runs, for the error of an +/// evaluation that does not finish. +const KEEP_STDERR_BYTES: usize = 64 * 1024; + +/// Run `cmd` to its end or `deadline`, log how long it took, and read the JSON +/// object it printed or the tail of what it said. +async fn run(cmd: tokio::process::Command, source: &str, deadline: Duration) -> ArchetypesReply { + let started = Instant::now(); + let reply = run_to_end(cmd, source, deadline).await; + log_elapsed(&reply, started.elapsed()); + reply +} + +/// Log how long one evaluation took, as `elapsed_ms`. +fn log_elapsed(reply: &ArchetypesReply, elapsed: Duration) { + let elapsed_ms = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX); + match reply { + ArchetypesReply::Evaluated { .. } => { + tracing::info!(elapsed_ms, "archetype evaluation finished"); + } + ArchetypesReply::Failed { .. } => { + tracing::info!(elapsed_ms, "archetype evaluation ended in failure"); + } + } +} + +async fn run_to_end( mut cmd: tokio::process::Command, source: &str, deadline: Duration, ) -> ArchetypesReply { let failed = |error_tail: String| ArchetypesReply::Failed { error_tail }; - let output = match tokio::time::timeout(deadline, cmd.output()).await { + let mut child = match cmd.spawn() { + Ok(child) => child, + Err(e) => return failed(format!("starting nix: {e}")), + }; + let mut stdout = child.stdout.take().expect("piped"); + let mut stderr = child.stderr.take().expect("piped"); + let seen = Arc::new(Mutex::new(Vec::::new())); + let mut reader = tokio::spawn({ + let seen = Arc::clone(&seen); + async move { + let mut chunk = [0u8; 8192]; + while let Ok(n) = stderr.read(&mut chunk).await { + if n == 0 { + break; + } + let mut seen = seen.lock().expect("unpoisoned"); + seen.extend_from_slice(&chunk[..n]); + if seen.len() > KEEP_STDERR_BYTES { + let cut = seen.len() - KEEP_STDERR_BYTES; + seen.drain(..cut); + } + } + } + }); + let mut out = Vec::new(); + let work = async { + let (read, status, _) = + tokio::join!(stdout.read_to_end(&mut out), child.wait(), &mut reader); + read?; + status + }; + let status = match tokio::time::timeout(deadline, work).await { Err(_) => { - return failed(format!( + reader.abort(); + let stderr = String::from_utf8_lossy(&seen.lock().expect("unpoisoned")).into_owned(); + let head = format!( "the evaluation did not finish within {} seconds", deadline.as_secs() - )); + ); + return failed(match crate::nix_build::last_lines(&stderr) { + Some(tail) => format!("{head}; nix last said:\n{tail}"), + None => head, + }); } - Ok(Err(e)) => return failed(format!("starting nix: {e}")), - Ok(Ok(output)) => output, + Ok(Err(e)) => return failed(format!("waiting for nix: {e}")), + Ok(Ok(status)) => status, }; - if !output.status.success() { - return failed(crate::nix_build::tail( - &String::from_utf8_lossy(&output.stderr), - output.status, - )); + if !status.success() { + let stderr = String::from_utf8_lossy(&seen.lock().expect("unpoisoned")).into_owned(); + return failed(crate::nix_build::tail(&stderr, status)); } - match serde_json::from_slice::(&output.stdout) { + match serde_json::from_slice::(&out) { Ok(serde_json::Value::Object(archetypes)) => ArchetypesReply::Evaluated { source: source.to_owned(), archetypes, @@ -99,6 +160,7 @@ async fn run( #[cfg(test)] mod tests { use super::*; + use tracing_subscriber::layer::SubscriberExt as _; fn evaluator() -> Evaluator { Evaluator { @@ -186,4 +248,91 @@ mod tests { "{reply:?}" ); } + + /// The `(message, elapsed_ms)` of every event logged. + #[derive(Clone, Default)] + struct Events(Arc>>); + + type Event = (String, Option); + + impl tracing_subscriber::Layer for Events { + fn on_event( + &self, + event: &tracing::Event<'_>, + _: tracing_subscriber::layer::Context<'_, S>, + ) { + struct Fields(String, Option); + impl tracing::field::Visit for Fields { + fn record_u64(&mut self, field: &tracing::field::Field, value: u64) { + if field.name() == "elapsed_ms" { + self.1 = Some(value); + } + } + fn record_debug( + &mut self, + field: &tracing::field::Field, + value: &dyn std::fmt::Debug, + ) { + if field.name() == "message" { + self.0 = format!("{value:?}"); + } + } + } + let mut fields = Fields(String::new(), None); + event.record(&mut fields); + self.0 + .lock() + .expect("unpoisoned") + .push((fields.0, fields.1)); + } + } + + #[test] + fn the_elapsed_time_is_logged_on_success_and_on_failure() { + let events = Events::default(); + tracing::subscriber::with_default( + tracing_subscriber::registry().with(events.clone()), + || { + let evaluated = ArchetypesReply::Evaluated { + source: String::new(), + archetypes: serde_json::Map::new(), + }; + let failed = ArchetypesReply::Failed { + error_tail: String::new(), + }; + log_elapsed(&evaluated, Duration::from_millis(1234)); + log_elapsed(&failed, Duration::from_millis(56)); + }, + ); + let seen = events.0.lock().expect("unpoisoned").clone(); + assert_eq!( + seen, + [ + ("archetype evaluation finished".to_owned(), Some(1234)), + ("archetype evaluation ended in failure".to_owned(), Some(56)), + ] + ); + } + + #[tokio::test] + async fn a_timeout_carries_what_nix_had_printed() { + let reply = sh( + "echo 'copying path x' >&2; sleep 30", + Duration::from_secs(1), + ) + .await; + assert!( + matches!(&reply, ArchetypesReply::Failed { error_tail } + if error_tail.contains("did not finish") && error_tail.ends_with("copying path x")), + "{reply:?}" + ); + } + + #[test] + fn the_evaluator_waits_five_minutes() { + assert_eq!( + swarm_wire::builder::ARCHETYPES_DEADLINE, + Duration::from_mins(5) + ); + } } diff --git a/swarm-builder/src/nix_build.rs b/swarm-builder/src/nix_build.rs index 6784b265..23eff5a9 100644 --- a/swarm-builder/src/nix_build.rs +++ b/swarm-builder/src/nix_build.rs @@ -346,16 +346,22 @@ async fn read_tail( /// The last [`TAIL_LINES`] lines of `stderr`, at most [`TAIL_BYTES`] bytes, /// or the exit status when it said nothing. pub(crate) fn tail(stderr: &str, status: std::process::ExitStatus) -> String { + last_lines(stderr).unwrap_or_else(|| format!("nix exited with {status} and printed nothing")) +} + +/// The last [`TAIL_LINES`] non-blank lines of `stderr`, at most [`TAIL_BYTES`] +/// bytes; `None` when it has none. +pub(crate) fn last_lines(stderr: &str) -> Option { let lines: Vec<&str> = stderr.lines().filter(|l| !l.trim().is_empty()).collect(); if lines.is_empty() { - return format!("nix exited with {status} and printed nothing"); + return None; } let joined = lines[lines.len().saturating_sub(TAIL_LINES)..].join("\n"); let mut start = joined.len().saturating_sub(TAIL_BYTES); while !joined.is_char_boundary(start) { start += 1; } - joined[start..].to_owned() + Some(joined[start..].to_owned()) } #[cfg(test)] diff --git a/swarm-controller/src/archetypes.rs b/swarm-controller/src/archetypes.rs index 59792715..227d6dc3 100644 --- a/swarm-controller/src/archetypes.rs +++ b/swarm-controller/src/archetypes.rs @@ -19,7 +19,7 @@ use swarm_controller_api::inference::InferenceSettings; use swarm_controller_api::limits::Caps; use swarm_queue_client::archetypes::{ArchetypeMeta, BUILTIN_SOURCE, SourceRecord}; use swarm_wire::builder::{ - ARCHETYPES_DEADLINE, ARCHETYPES_SUBJECT, ArchetypesReply, EvalArchetypes, + ARCHETYPES_REPLY_DEADLINE, ARCHETYPES_SUBJECT, ArchetypesReply, EvalArchetypes, }; use utoipa::ToSchema; @@ -109,7 +109,7 @@ async fn evaluate(client: &async_nats::Client) -> Result { let body = serde_json::to_vec(&EvalArchetypes {}).map_err(|e| e.to_string())?; let sent = async_nats::Request::new() .payload(body.into()) - .timeout(Some(ARCHETYPES_DEADLINE)); + .timeout(Some(ARCHETYPES_REPLY_DEADLINE)); let msg = client .send_request(ARCHETYPES_SUBJECT, sent) .await diff --git a/swarm-wire/src/builder.rs b/swarm-wire/src/builder.rs index f3def429..7d9d3814 100644 --- a/swarm-wire/src/builder.rs +++ b/swarm-wire/src/builder.rs @@ -56,9 +56,12 @@ pub enum Reply { /// hyperhive source and answers on the request's reply subject. pub const ARCHETYPES_SUBJECT: &str = "$SWARM.builder.archetypes"; -/// The longest one archetype evaluation may take, and how long the controller -/// waits for its reply. -pub const ARCHETYPES_DEADLINE: Duration = Duration::from_mins(2); +/// The longest one archetype evaluation may take. +pub const ARCHETYPES_DEADLINE: Duration = Duration::from_mins(5); + +/// How long the controller waits for an evaluation's reply: the evaluation at +/// its deadline, plus a minute for its timeout reply's trip back. +pub const ARCHETYPES_REPLY_DEADLINE: Duration = Duration::from_mins(6); /// One archetype evaluation. No field yet: the builder evaluates the hyperhive /// source it was deployed with. @@ -209,6 +212,16 @@ mod tests { assert!(REPLY_DEADLINE > BUILD_DEADLINE); } + #[test] + fn the_archetypes_deadline_is_five_minutes() { + assert_eq!(ARCHETYPES_DEADLINE, Duration::from_mins(5)); + } + + #[test] + fn the_archetypes_reply_deadline_outlasts_the_evaluation() { + assert!(ARCHETYPES_REPLY_DEADLINE > ARCHETYPES_DEADLINE); + } + #[test] fn the_lock_reply_deadline_outlasts_the_lock() { assert!(LOCK_REPLY_DEADLINE > LOCK_DEADLINE);