hive-priv: stop streaming spin after stderr closes

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
+    // Drive stdout and stderr concurrently. Stderr is disabled once it
-    // them without bias — both streams drain at roughly the same rate
+    // hits EOF or an error: `next_line()` would otherwise resolve at once on
-    // as the subprocess produces output.
+    // 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]