diff --git a/docs/swarm/builder.md b/docs/swarm/builder.md index 6f6273f7..1f37ba11 100644 --- a/docs/swarm/builder.md +++ b/docs/swarm/builder.md @@ -40,10 +40,6 @@ 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 6a5721cb..17d47de9 100644 --- a/swarm-builder/src/nix_build.rs +++ b/swarm-builder/src/nix_build.rs @@ -22,11 +22,6 @@ 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"; @@ -170,15 +165,16 @@ async fn run(mut cmd: tokio::process::Command, deadline: Duration) -> Reply { }; } }; - let stdout = child.stdout.take().expect("piped"); - let stderr = child.stderr.take().expect("piped"); + let mut stdout = child.stdout.take().expect("piped"); + let mut stderr = child.stderr.take().expect("piped"); let work = async { + let (mut out, mut err) = (Vec::new(), Vec::new()); let (o, e, status) = tokio::join!( - read_tail(stdout, KEEP_BYTES), - read_tail(stderr, KEEP_BYTES), + stdout.read_to_end(&mut out), + stderr.read_to_end(&mut err), child.wait() ); - let (out, err) = (o?, e?); + o.and(e).map(|_| ())?; status.map(|s| (s, out, err)) }; match tokio::time::timeout(deadline, work).await { @@ -206,69 +202,6 @@ 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 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); - } - 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 { @@ -469,132 +402,6 @@ 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)); - 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] - 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;