swarm-builder: keep only the tail of nix's output while it runs

nix_build::run read nix's stdout and stderr with read_to_end, so a long
--print-build-logs run held its whole log in memory until nix exited,
though the reply uses only the last 30 lines (4000 bytes) of stderr and
the last non-empty line of stdout. Both streams now go through a Tail
that holds at most 128 KiB and keeps the last 64 KiB. Output under 64 KiB
reaches tail() whole, so those replies are unchanged.

docs/swarm/builder.md

@@ -40,6 +40,10 @@ Where it runs and how it's enabled: [`services.md`](services.md#builder).
    The full tail is in the failed node's error on the Jobs page and in the
    builder's journal line for the build.
 
+   The builder holds the last 64 KiB of each of `nix`'s output streams while the
+   build runs, and the reply is the last 30 lines of that (at most 4000
+   bytes), so a long build log never grows its memory.
+
 ## What the check is, and isn't
 
 - It composes the config the way a hive's meta flake does

swarm-builder/src/nix_build.rs

@@ -22,6 +22,11 @@ use tokio::io::AsyncReadExt as _;
 const TAIL_LINES: usize = 30;
 const TAIL_BYTES: usize = 4000;
 
+/// How much of each stream is held while `nix` runs, from its end. Output
+/// under this is read whole; past it, only the end is kept, so a build's log
+/// cannot grow the builder's memory.
+const KEEP_BYTES: usize = 64 * 1024;
+
 /// The variable the credential helper reads the token from, in the child only.
 const TOKEN_VAR: &str = "SWARM_BUILDER_GIT_TOKEN";
 
@@ -165,16 +170,15 @@ async fn run(mut cmd: tokio::process::Command, deadline: Duration) -> Reply {
             };
         }
     };
-    let mut stdout = child.stdout.take().expect("piped");
+    let stdout = child.stdout.take().expect("piped");
-    let mut stderr = child.stderr.take().expect("piped");
+    let stderr = child.stderr.take().expect("piped");
     let work = async {
-        let (mut out, mut err) = (Vec::new(), Vec::new());
         let (o, e, status) = tokio::join!(
-            stdout.read_to_end(&mut out),
+            read_tail(stdout, KEEP_BYTES),
-            stderr.read_to_end(&mut err),
+            read_tail(stderr, KEEP_BYTES),
             child.wait()
         );
-        o.and(e).map(|_| ())?;
+        let (out, err) = (o?, e?);
         status.map(|s| (s, out, err))
     };
     match tokio::time::timeout(deadline, work).await {
@@ -202,6 +206,68 @@ async fn run(mut cmd: tokio::process::Command, deadline: Duration) -> Reply {
     }
 }
 
+/// The last `keep` bytes of a stream, held while it is read: at most
+/// `2 * keep` bytes after each push.
+struct Tail {
+    buf: Vec<u8>,
+    keep: usize,
+    cut: bool,
+}
+
+impl Tail {
+    fn new(keep: usize) -> Self {
+        Self {
+            buf: Vec::new(),
+            keep,
+            cut: false,
+        }
+    }
+
+    fn push(&mut self, chunk: &[u8]) {
+        self.buf.extend_from_slice(chunk);
+        // Trim at twice the cap, so a copy happens once per `keep` bytes.
+        if self.buf.len() > self.keep * 2 {
+            self.trim();
+        }
+    }
+
+    fn trim(&mut self) {
+        self.buf.drain(..self.buf.len() - self.keep);
+        self.cut = true;
+    }
+
+    /// The kept bytes. When the stream was longer than `keep`, the cut line
+    /// at the front is dropped, so the result starts on a line boundary
+    /// (unless it holds no newline at all).
+    fn finish(mut self) -> Vec<u8> {
+        if self.buf.len() > self.keep {
+            self.trim();
+        }
+        if self.cut
+            && let Some(nl) = self.buf.iter().position(|&b| b == b'\n')
+        {
+            self.buf.drain(..=nl);
+        }
+        self.buf
+    }
+}
+
+/// Read `reader` to its end, keeping only its last `keep` bytes (see [`Tail`]).
+async fn read_tail(
+    mut reader: impl tokio::io::AsyncRead + Unpin,
+    keep: usize,
+) -> std::io::Result<Vec<u8>> {
+    let mut tail = Tail::new(keep);
+    let mut chunk = vec![0u8; 8192];
+    loop {
+        let n = reader.read(&mut chunk).await?;
+        if n == 0 {
+            return Ok(tail.finish());
+        }
+        tail.push(&chunk[..n]);
+    }
+}
+
 /// The last [`TAIL_LINES`] lines of `stderr`, at most [`TAIL_BYTES`] bytes,
 /// or the exit status when it said nothing.
 fn tail(stderr: &str, status: std::process::ExitStatus) -> String {
@@ -402,6 +468,128 @@ mod tests {
         assert!(error_tail.ends_with("error: attribute missing"));
     }
 
+    /// A `nix` stand-in: runs `body` as a shell script, ignoring its arguments.
+    fn fake_nix(tag: &str, body: &str) -> PathBuf {
+        use std::os::unix::fs::PermissionsExt as _;
+
+        let dir = std::env::temp_dir().join(format!(
+            "swarm-builder-fake-nix-{tag}-{}",
+            std::process::id()
+        ));
+        std::fs::create_dir_all(&dir).expect("scratch dir");
+        let path = dir.join("nix");
+        std::fs::write(&path, format!("#!/bin/sh\n{body}\n")).expect("write fake nix");
+        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).expect("chmod");
+        path
+    }
+
+    /// Builds with `nix`, retrying while the kernel refuses to exec the script
+    /// just written (`ETXTBSY`, when another test's fork briefly inherits its
+    /// open write handle).
+    async fn build_with(nix: PathBuf) -> Reply {
+        let builder = Builder { nix, ..builder() };
+        for _ in 0..20 {
+            let reply = builder.build(&job(), "token").await;
+            match &reply {
+                Reply::Failed { error_tail } if error_tail.contains("Text file busy") => {
+                    tokio::time::sleep(Duration::from_millis(50)).await;
+                }
+                _ => return reply,
+            }
+        }
+        panic!("the fake nix stayed busy")
+    }
+
+    #[tokio::test]
+    async fn eight_megabytes_of_log_leave_only_the_tail_in_the_reply() {
+        let nix = fake_nix(
+            "big",
+            "i=0; while [ $i -lt 8 ]; do \
+               head -c 1048576 /dev/zero | tr '\\0' 'x' >&2; echo >&2; i=$((i+1)); done\n\
+             printf 'noise\\n' >&2\n\
+             for i in 1 2 3; do echo \"final line $i\" >&2; done\n\
+             exit 1",
+        );
+        let Reply::Failed { error_tail } = build_with(nix).await else {
+            panic!("expected a failure")
+        };
+        assert_eq!(
+            error_tail,
+            "noise\nfinal line 1\nfinal line 2\nfinal line 3"
+        );
+    }
+
+    #[tokio::test]
+    async fn eight_megabytes_on_stdout_keep_the_last_path() {
+        let nix = fake_nix(
+            "bigout",
+            "head -c 8388608 /dev/zero | tr '\\0' 'y'; echo\n\
+             echo /nix/store/abc-nixos-system",
+        );
+        assert_eq!(
+            build_with(nix).await,
+            Reply::Built {
+                out_path: "/nix/store/abc-nixos-system".to_owned()
+            }
+        );
+    }
+
+    #[tokio::test]
+    async fn output_under_the_cap_reaches_the_reply_whole() {
+        let nix = fake_nix(
+            "small",
+            "echo 'first line' >&2; echo >&2; echo '  second line' >&2; \
+             echo 'a long stdout' ; echo /nix/store/abc-nixos-system; echo",
+        );
+        assert_eq!(
+            build_with(nix).await,
+            Reply::Built {
+                out_path: "/nix/store/abc-nixos-system".to_owned()
+            }
+        );
+        let failing = fake_nix(
+            "smallfail",
+            "echo 'first line' >&2; echo >&2; echo '  second line' >&2; exit 1",
+        );
+        let Reply::Failed { error_tail } = build_with(failing).await else {
+            panic!("expected a failure")
+        };
+        assert_eq!(error_tail, "first line\n  second line");
+    }
+
+    #[tokio::test]
+    async fn read_tail_keeps_the_end_and_starts_on_a_line() {
+        let data = "aaaa\nbbbb\ncccc\ndddd\n";
+        let whole = read_tail(data.as_bytes(), 100).await.expect("read");
+        assert_eq!(whole, data.as_bytes());
+        let cut = read_tail(data.as_bytes(), 12).await.expect("read");
+        assert_eq!(String::from_utf8(cut).expect("utf-8"), "cccc\ndddd\n");
+        let one_long_line = read_tail(&b"x".repeat(50)[..], 12).await.expect("read");
+        assert_eq!(one_long_line, b"x".repeat(12));
+    }
+
+    #[test]
+    fn the_tail_never_holds_more_than_twice_the_cap_while_reading() {
+        let mut tail = Tail::new(KEEP_BYTES);
+        let mut peak = 0;
+        for _ in 0..1024 {
+            tail.push(&[b'x'; 8192]);
+            peak = peak.max(tail.buf.len());
+        }
+        assert!(peak > KEEP_BYTES, "control: the buffer does grow: {peak}");
+        assert!(peak <= 2 * KEEP_BYTES, "{peak} bytes held");
+    }
+
+    #[tokio::test]
+    async fn read_tail_holds_at_most_the_cap_of_an_eight_megabyte_stream() {
+        let mut stream = b"line\n".repeat(8 * 1024 * 1024 / 5);
+        stream.extend_from_slice(b"the end\n");
+        let kept = read_tail(&stream[..], KEEP_BYTES).await.expect("read");
+        assert!(kept.len() <= KEEP_BYTES, "{} bytes kept", kept.len());
+        assert!(kept.ends_with(b"line\nthe end\n"));
+        assert!(kept.starts_with(b"line\n"));
+    }
+
     #[tokio::test]
     async fn a_build_past_its_deadline_is_stopped() {
         let reply = sh("sleep 30", Duration::from_millis(200)).await;