swarm-builder: log each archetype eval's elapsed time, wait 5 minutes

Cargo.lock

@@ -4834,6 +4834,7 @@ dependencies = [
  "swarm-wire",
  "tokio",
  "tracing",
+ "tracing-subscriber",
 ]
 
 [[package]]

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
 

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

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::<u8>::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::<serde_json::Value>(&output.stdout) {
+    match serde_json::from_slice::<serde_json::Value>(&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<Mutex<Vec<Event>>>);
+
+    type Event = (String, Option<u64>);
+
+    impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for Events {
+        fn on_event(
+            &self,
+            event: &tracing::Event<'_>,
+            _: tracing_subscriber::layer::Context<'_, S>,
+        ) {
+            struct Fields(String, Option<u64>);
+            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)
+        );
+    }
 }

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<String> {
     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)]

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<SourceRecord, String> {
     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

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);