fix(task): bound buffered command output

e2e/tasks/test_task_output_buffering

@@ -0,0 +1,46 @@
+#!/usr/bin/env bash
+
+# Regression test for #8786: parallel task output must be streamed rather than
+# retained for the lifetime of the command. Replacing output still keeps a
+# bounded tail so hidden stdout can be diagnosed when a command fails.
+
+cat <<'EOF' >mise.toml
+[tasks.noisy]
+run = '''
+i=0
+while [ "$i" -lt 2000 ]; do
+  printf 'noisy-%04d\n' "$i"
+  i=$((i + 1))
+done
+'''
+
+[tasks.other]
+run = "echo other-done"
+EOF
+
+mise run noisy ::: other >output.txt 2>&1
+assert_contains "cat output.txt" "[noisy] noisy-0000"
+assert_contains "cat output.txt" "[noisy] noisy-1999"
+assert_contains "cat output.txt" "[other] other-done"
+
+cat <<'EOF' >mise.toml
+[settings.task]
+output = "replacing"
+
+[tasks.failing]
+run = '''
+i=0
+while [ "$i" -lt 10000 ]; do
+  printf 'hidden-%05d\n' "$i"
+  i=$((i + 1))
+done
+exit 1
+'''
+EOF
+
+status=0
+MISE_FORCE_PROGRESS=1 MISE_FRIENDLY_ERROR=1 RUST_BACKTRACE=0 mise run failing >output.txt 2>&1 || status=$?
+assert "test '$status' -ne 0"
+assert_contains "cat output.txt" "[output truncated; showing last 64 KiB]"
+assert_not_contains "cat output.txt" "hidden-00000"
+assert_contains "cat output.txt" "hidden-09999"

src/cmd.rs

@@ -1,4 +1,4 @@
-use std::collections::HashSet;
+use std::collections::{HashSet, VecDeque};
 use std::ffi::{OsStr, OsString};
 use std::fmt::{Debug, Display, Formatter};
 use std::io::{BufRead, BufReader, Write};
@@ -390,13 +390,71 @@ pub(crate) fn prepare_noninteractive_child(_cmd: &mut std::process::Command) {
 /// deadline we abandon the readers — any tail output is dropped.
 const PIPE_DRAIN_TIMEOUT: Duration = Duration::from_secs(10);
 
+/// Maximum amount of stdout retained for commands whose output is hidden
+/// behind a progress indicator. The tail is replayed if the command fails.
+const FAILURE_OUTPUT_TAIL_BYTES: usize = 64 * 1024;
+const FAILURE_OUTPUT_TRUNCATED_NOTICE: &str = "[output truncated; showing last 64 KiB]";
+
+#[derive(Default)]
+struct FailureOutputTail {
+    lines: VecDeque<String>,
+    bytes: usize,
+    truncated: bool,
+}
+
+impl FailureOutputTail {
+    fn push(&mut self, mut line: String) {
+        let max_line_bytes = FAILURE_OUTPUT_TAIL_BYTES.saturating_sub(1);
+        if line.len() > max_line_bytes {
+            let mut start = line.len() - max_line_bytes;
+            while !line.is_char_boundary(start) {
+                start += 1;
+            }
+            line = line[start..].to_string();
+            self.lines.clear();
+            self.bytes = 0;
+            self.truncated = true;
+        }
+
+        self.bytes = self.bytes.saturating_add(line.len().saturating_add(1));
+        self.lines.push_back(line);
+        while self.bytes > FAILURE_OUTPUT_TAIL_BYTES {
+            if let Some(line) = self.lines.pop_front() {
+                self.bytes = self.bytes.saturating_sub(line.len().saturating_add(1));
+                self.truncated = true;
+            } else {
+                break;
+            }
+        }
+    }
+
+    fn into_output(mut self) -> Vec<(String, OutputSource)> {
+        if self.truncated {
+            self.lines
+                .push_front(FAILURE_OUTPUT_TRUNCATED_NOTICE.to_string());
+        }
+        self.lines
+            .into_iter()
+            .map(|line| (line, OutputSource::Stdout))
+            .collect()
+    }
+}
+
 enum HashedProcessOutput {
     Stdout(Vec<u8>),
     Stderr(Vec<u8>),
     ReadError(&'static str, std::io::Error),
 }
 
 impl<'a> CmdLineRunner<'a> {
+    fn failure_output_tail(&self) -> Option<FailureOutputTail> {
+        if self.on_stdout.is_none() && (self.pr.is_some() || self.pr_arc.is_some()) {
+            Some(FailureOutputTail::default())
+        } else {
+            None
+        }
+    }
+
     pub fn new<P: AsRef<OsStr>>(program: P) -> Self {
         let mut cmd = Command::new(program);
         cmd.stdin(Stdio::null());
@@ -774,7 +832,7 @@ impl<'a> CmdLineRunner<'a> {
 
         let timeout_guard = self.timeout.map(|t| TimeoutGuard::new(t, id));
 
-        let mut combined_output = vec![];
+        let mut failure_output = self.failure_output_tail();
         let mut status = None;
         // Once ExitStatus arrives we set a deadline and switch to recv_timeout
         // so a grandchild that inherited the pipes can't hang us forever
@@ -805,13 +863,16 @@ impl<'a> CmdLineRunner<'a> {
             match msg {
                 ChildProcessOutput::Stdout(line) => {
                     let line = self.redactor.redact(&line);
-                    self.on_stdout(line.clone());
-                    combined_output.push((line, OutputSource::Stdout));
+                    if let Some(output) = &mut failure_output {
+                        self.on_stdout(line.clone());
+                        output.push(line);
+                    } else {
+                        self.on_stdout(line);
+                    }
                 }
                 ChildProcessOutput::Stderr(line) => {
                     let line = self.redactor.redact(&line);
-                    self.on_stderr(line.clone());
-                    combined_output.push((line, OutputSource::Stderr));
+                    self.on_stderr(line);
                 }
                 ChildProcessOutput::ExitStatus(s) => {
                     status = Some(s);
@@ -853,7 +914,10 @@ impl<'a> CmdLineRunner<'a> {
             if let Some(duration) = timeout_guard.as_ref().and_then(|g| g.timed_out()) {
                 bail!("timed out after {duration:?}");
             }
-            self.on_error(combined_output, status)?;
+            self.on_error(
+                failure_output.map_or_else(Vec::new, FailureOutputTail::into_output),
+                status,
+            )?;
         }
 
         Ok(())
@@ -971,7 +1035,7 @@ impl<'a> CmdLineRunner<'a> {
         drop(tx);
 
         let timeout_guard = self.timeout.map(|t| TimeoutGuard::new(t, id));
-        let mut combined_output = vec![];
+        let mut failure_output = self.failure_output_tail();
         let mut status = None;
         let mut wait = Box::pin(cp.wait());
         loop {
@@ -998,13 +1062,16 @@ impl<'a> CmdLineRunner<'a> {
                     match msg {
                         ChildProcessOutput::Stdout(line) => {
                             let line = self.redactor.redact(&line);
-                            self.on_stdout(line.clone());
-                            combined_output.push((line, OutputSource::Stdout));
+                            if let Some(output) = &mut failure_output {
+                                self.on_stdout(line.clone());
+                                output.push(line);
+                            } else {
+                                self.on_stdout(line);
+                            }
                         }
                         ChildProcessOutput::Stderr(line) => {
                             let line = self.redactor.redact(&line);
-                            self.on_stderr(line.clone());
-                            combined_output.push((line, OutputSource::Stderr));
+                            self.on_stderr(line);
                         }
                         ChildProcessOutput::ExitStatus(_) => {}
                         #[cfg(not(any(test, windows)))]
@@ -1043,13 +1110,16 @@ impl<'a> CmdLineRunner<'a> {
             match msg {
                 ChildProcessOutput::Stdout(line) => {
                     let line = self.redactor.redact(&line);
-                    self.on_stdout(line.clone());
-                    combined_output.push((line, OutputSource::Stdout));
+                    if let Some(output) = &mut failure_output {
+                        self.on_stdout(line.clone());
+                        output.push(line);
+                    } else {
+                        self.on_stdout(line);
+                    }
                 }
                 ChildProcessOutput::Stderr(line) => {
                     let line = self.redactor.redact(&line);
-                    self.on_stderr(line.clone());
-                    combined_output.push((line, OutputSource::Stderr));
+                    self.on_stderr(line);
                 }
                 ChildProcessOutput::ExitStatus(_) => {}
                 #[cfg(not(any(test, windows)))]
@@ -1066,7 +1136,10 @@ impl<'a> CmdLineRunner<'a> {
             if let Some(duration) = timeout_guard.as_ref().and_then(|g| g.timed_out()) {
                 bail!("timed out after {duration:?}");
             }
-            self.on_error(combined_output, status)?;
+            self.on_error(
+                failure_output.map_or_else(Vec::new, FailureOutputTail::into_output),
+                status,
+            )?;
         }
 
         Ok(())
@@ -1790,6 +1863,121 @@ mod tests {
     use pretty_assertions::assert_eq;
 
     use crate::config::Config;
+    use crate::ui::progress_report::SingleReport;
+
+    #[derive(Debug, Default)]
+    struct RecordingReport {
+        lines: Mutex<Vec<String>>,
+    }
+
+    impl SingleReport for RecordingReport {
+        fn println(&self, message: String) {
+            self.lines.lock().unwrap().push(message);
+        }
+    }
+
+    #[test]
+    fn test_failure_output_tail_preserves_output_within_limit() {
+        let mut output = super::FailureOutputTail::default();
+        output.push("first".to_string());
+        output.push("second".to_string());
+
+        let lines = output
+            .into_output()
+            .into_iter()
+            .map(|(line, _)| line)
+            .collect::<Vec<_>>();
+        assert_eq!(lines, ["first", "second"]);
+    }
+
+    #[test]
+    fn test_failure_output_tail_discards_oldest_output() {
+        let mut output = super::FailureOutputTail::default();
+        for i in 0..=super::FAILURE_OUTPUT_TAIL_BYTES / 8 {
+            output.push(format!("{i:07}"));
+        }
+
+        assert!(output.bytes <= super::FAILURE_OUTPUT_TAIL_BYTES);
+        let lines = output
+            .into_output()
+            .into_iter()
+            .map(|(line, _)| line)
+            .collect::<Vec<_>>();
+        assert_eq!(
+            lines.first().unwrap(),
+            super::FAILURE_OUTPUT_TRUNCATED_NOTICE
+        );
+        assert!(!lines.contains(&"0000000".to_string()));
+        assert_eq!(lines.last().unwrap(), "0008192");
+    }
+
+    #[test]
+    fn test_failure_output_tail_truncates_large_unicode_line() {
+        let mut output = super::FailureOutputTail::default();
+        output.push("あ".repeat(super::FAILURE_OUTPUT_TAIL_BYTES));
+
+        assert!(output.bytes <= super::FAILURE_OUTPUT_TAIL_BYTES);
+        let lines = output
+            .into_output()
+            .into_iter()
+            .map(|(line, _)| line)
+            .collect::<Vec<_>>();
+        assert_eq!(
+            lines.first().unwrap(),
+            super::FAILURE_OUTPUT_TRUNCATED_NOTICE
+        );
+        assert!(
+            lines
+                .last()
+                .unwrap()
+                .is_char_boundary(lines.last().unwrap().len())
+        );
+        assert!(lines.last().unwrap().len() < super::FAILURE_OUTPUT_TAIL_BYTES);
+    }
+
+    #[test]
+    fn test_failure_output_tail_only_enabled_for_hidden_stdout() {
+        let report = RecordingReport::default();
+        assert!(
+            super::CmdLineRunner::new("true")
+                .with_pr(&report)
+                .failure_output_tail()
+                .is_some()
+        );
+        assert!(
+            super::CmdLineRunner::new("true")
+                .with_pr(&report)
+                .with_on_stdout(|_| {})
+                .failure_output_tail()
+                .is_none()
+        );
+        assert!(
+            super::CmdLineRunner::new("true")
+                .failure_output_tail()
+                .is_none()
+        );
+    }
+
+    #[tokio::test]
+    async fn test_failure_output_tail_replayed_on_async_failure() {
+        let report = RecordingReport::default();
+        let err = super::CmdLineRunner::new("sh")
+            .args([
+                "-c",
+                "i=0; while [ $i -lt 10000 ]; do printf '%07d\\n' $i; i=$((i + 1)); done; exit 1",
+            ])
+            .with_pr(&report)
+            .execute_async()
+            .await
+            .unwrap_err();
+
+        assert!(err.to_string().contains("exited with non-zero status"));
+        let lines = report.lines.lock().unwrap();
+        assert_eq!(lines.len(), 1);
+        assert!(lines[0].starts_with(super::FAILURE_OUTPUT_TRUNCATED_NOTICE));
+        assert!(!lines[0].contains("0000000"));
+        assert!(lines[0].ends_with("0009999"));
+    }
 
     #[tokio::test]
     async fn test_cmd() {