From e873195cf647e836e93508b1211b297dddcbec58 Mon Sep 17 00:00:00 2001 From: atlas Date: Fri, 16 Oct 2026 21:25:54 +0200 Subject: [PATCH] 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 | 4 + swarm-builder/src/nix_build.rs | 200 ++++++++++++++++++++++++++++++++- 2 files changed, 198 insertions(+), 6 deletions(-) diff --git a/docs/swarm/builder.md b/docs/swarm/builder.md index 1f37ba11..6f6273f7 100644 --- a/docs/swarm/builder.md +++ b/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 diff --git a/swarm-builder/src/nix_build.rs b/swarm-builder/src/nix_build.rs index 17d47de9..a0649e3b 100644 --- a/swarm-builder/src/nix_build.rs +++ b/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 mut stderr = child.stderr.take().expect("piped"); + let stdout = child.stdout.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), - stderr.read_to_end(&mut err), + read_tail(stdout, KEEP_BYTES), + 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, + 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 { + 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> { + 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;