From 666f8975a28475992eb495841ca3310ff0f4e601 Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 2 Nov 2026 22:05:58 +0100 Subject: [PATCH] hive-priv: stop streaming spin after stderr closes --- hive-priv/src/container.rs | 63 +++++++++++++++++++++++++++++++++++--- 1 file changed, 58 insertions(+), 5 deletions(-) diff --git a/hive-priv/src/container.rs b/hive-priv/src/container.rs index f5050efc..40ee26e3 100644 --- a/hive-priv/src/container.rs +++ b/hive-priv/src/container.rs @@ -643,9 +643,10 @@ async fn container_run_streaming( // Collect stderr for the error message; stream both to the client. let mut stderr_buf = String::new(); - // Drive stdout and stderr concurrently. `tokio::select!` interleaves - // them without bias — both streams drain at roughly the same rate - // as the subprocess produces output. + // Drive stdout and stderr concurrently. Stderr is disabled once it + // hits EOF or an error: `next_line()` would otherwise resolve at once on + // every turn and spin the loop until stdout closes. + let mut stderr_done = false; loop { tokio::select! { line = stdout_lines.next_line() => { @@ -661,7 +662,7 @@ async fn container_run_streaming( } } } - line = stderr_lines.next_line() => { + line = stderr_lines.next_line(), if !stderr_done => { match line { Ok(Some(l)) => { tracing::warn!(target: "nixos-container", "{l}"); @@ -671,9 +672,10 @@ async fn container_run_streaming( } stderr_buf.push_str(&l); } - Ok(None) => {} + Ok(None) => stderr_done = true, Err(e) => { tracing::warn!(error = %e, "nixos-container stderr read error"); + stderr_done = true; } } } @@ -1373,6 +1375,57 @@ mod host_tests { assert!(msg.contains("wedged"), "{msg}"); } + // ---- streaming: a closed stream must not spin the reader ---- + + /// CPU time of the calling thread. The `#[tokio::test]` runtime runs on + /// this thread, so tests running in parallel do not count. + fn thread_cpu() -> Duration { + let mut ts = libc::timespec { + tv_sec: 0, + tv_nsec: 0, + }; + let rc = unsafe { libc::clock_gettime(libc::CLOCK_THREAD_CPUTIME_ID, &raw mut ts) }; + assert_eq!(rc, 0); + Duration::new( + ts.tv_sec.try_into().unwrap(), + ts.tv_nsec.try_into().unwrap(), + ) + } + + /// Runs `create` against a `nixos-container` that executes `script`, + /// returning the client's lines and the CPU the wait cost. + async fn streamed_cpu(script: &str) -> (Vec<(PrivStream, String)>, Duration) { + let stub = Stub::new(&format!("case \"$0\" in nixos-container) {script};; esac")); + let (mut writer, peer) = pipe(); + let before = thread_cpu(); + apply_toplevel(&stub.host(), "create", "bob", TOPLEVEL, true, &mut writer) + .await + .unwrap(); + let used = thread_cpu().saturating_sub(before); + (lines_sent(writer, peer).await, used) + } + + #[tokio::test] + async fn a_closed_stderr_does_not_spin_while_stdout_stays_open() { + let (lines, used) = streamed_cpu("exec 2>&-; sleep 1; echo done").await; + assert_eq!(lines, [(PrivStream::Stdout, "done".to_owned())]); + // A spinning loop burns about 1s of CPU across the 1s sleep. + assert!( + used < Duration::from_millis(500), + "burned {used:?} of cpu in a 1s wait" + ); + } + + #[tokio::test] + async fn a_closed_stdout_does_not_spin_while_stderr_stays_open() { + let (lines, used) = streamed_cpu("exec 1>&-; sleep 1; echo done >&2").await; + assert_eq!(lines, [(PrivStream::Stderr, "done".to_owned())]); + assert!( + used < Duration::from_millis(500), + "burned {used:?} of cpu in a 1s wait" + ); + } + // ---- client events ---- #[tokio::test]