From e873195cf647e836e93508b1211b297dddcbec58 Mon Sep 17 00:00:00 2001 From: atlas Date: Fri, 16 Oct 2026 21:25:54 +0200 Subject: [PATCH 1/2] 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; From 88dd58e947d712c8a8b776836a038813fb278d28 Mon Sep 17 00:00:00 2001 From: atlas Date: Fri, 16 Oct 2026 23:19:00 +0200 Subject: [PATCH 2/2] swarm-builder: keep a cut last line that is longer than the window After a cut, Tail::finish dropped everything through the first newline. When the stream ended in one line longer than the window, that newline was the last byte, so the window came back empty and the reply fell back to the bare exit status. The cut line is now dropped only when something follows it. --- swarm-builder/src/nix_build.rs | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/swarm-builder/src/nix_build.rs b/swarm-builder/src/nix_build.rs index a0649e3b..6a5721cb 100644 --- a/swarm-builder/src/nix_build.rs +++ b/swarm-builder/src/nix_build.rs @@ -237,14 +237,15 @@ impl Tail { } /// 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). + /// at the front is dropped, so the result starts on a line boundary, + /// unless that line is all the window holds. 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') + && nl + 1 < self.buf.len() { self.buf.drain(..=nl); } @@ -566,6 +567,10 @@ mod tests { 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)); + let long_last_line = read_tail(format!("{}\n", "x".repeat(50)).as_bytes(), 12) + .await + .expect("read"); + assert_eq!(long_last_line, format!("{}\n", "x".repeat(11)).as_bytes()); } #[test]