cargo pgrx regress output is no longer fully buffered (#2095)

Prior to this PR, `cargo pgrx regress` would buffer the entire
`pg_regress` output, decorate each line, and then output the results to
stdout.

Now it does it line-by-line.

It has annoyed me that I couldn't watch the test suite progress as it
runs. Now I can and you can too!

Author: eeeebbbbrrrr

cargo-pgrx/src/command/regress.rs

@@ -17,7 +17,7 @@ use pgrx_pg_config::{createdb, dropdb, PgConfig};
 use std::collections::HashSet;
 use std::env::temp_dir;
 use std::fs::{DirEntry, File};
-use std::io::{IsTerminal, Write};
+use std::io::{BufRead, BufReader, IsTerminal, Write};
 use std::path::{Path, PathBuf};
 use std::process::{Command, ExitStatus, Stdio};
 
@@ -424,11 +424,8 @@ fn run_tests(
         .parent()
         .expect("test file should be in a directory named `sql/`")
         .to_path_buf();
-    let (status, output) = pg_regress(pg_config, pg_regress_bin, dbname, &input_dir, test_files)?;
-
-    println!("{output}");
-
-    Ok(status.success())
+    pg_regress(pg_config, pg_regress_bin, dbname, &input_dir, test_files)
+        .map(|status| status.success())
 }
 
 fn create_regress_output(
@@ -446,7 +443,7 @@ fn create_regress_output(
         .parent()
         .expect("test file should be in a directory named `sql/`")
         .to_path_buf();
-    let (status, output) = pg_regress(pg_config, pg_regress_bin, dbname, &input_dir, &[test_file])?;
+    let status = pg_regress(pg_config, pg_regress_bin, dbname, &input_dir, &[test_file])?;
 
     if !status.success() {
         // pg_regress returned with an error code, but that is most likely because the test's output file
@@ -457,7 +454,6 @@ fn create_regress_output(
         if out_file.exists() {
             return Ok(Some(out_file));
         } else {
-            println!("{output}");
             std::process::exit(status.code().unwrap_or(1));
         }
     }
@@ -471,7 +467,7 @@ fn pg_regress(
     dbname: &str,
     input_dir: impl AsRef<Path>,
     tests: &[&DirEntry],
-) -> eyre::Result<(ExitStatus, String)> {
+) -> eyre::Result<ExitStatus> {
     if tests.is_empty() {
         eyre::bail!("no tests to run");
     }
@@ -520,101 +516,116 @@ fn pg_regress(
 
     tracing::trace!("running {command:?}");
 
-    let output = command.output()?;
-    let stdout = decorate_output(&String::from_utf8_lossy(&output.stdout));
-    let stderr = decorate_output(&String::from_utf8_lossy(&output.stderr));
+    let mut child = command.spawn()?;
+    let (Some(stdout), Some(stderr)) = (child.stdout.take(), child.stderr.take()) else {
+        panic!("unable to take stdout or stderr from pg_regress process");
+    };
 
-    let cmd_output = if !stdout.is_empty() && !stderr.is_empty() {
-        format!("{stdout}\n{stderr}")
-    } else if !stdout.is_empty() {
-        stdout.to_string()
-    } else {
-        stderr.to_string()
-    }
-    .trim()
-    .to_string();
+    let output_monitor = std::thread::spawn(move || {
+        let mut passed_cnt = 0;
+        let mut failed_cnt = 0;
+        let stdout = BufReader::new(stdout);
+        let stderr = BufReader::new(stderr);
+        for line in stdout.lines().chain(stderr.lines()) {
+            let line = line.unwrap();
+            let Some((line, result)) = decorate_output(line) else {
+                continue;
+            };
+
+            match result {
+                Some(TestResult::Passed) => passed_cnt += 1,
+                Some(TestResult::Failed) => failed_cnt += 1,
+                None => (),
+            }
+
+            println!("{line}");
+        }
+        (passed_cnt, failed_cnt)
+    });
+    let status = child.wait()?;
+    let (passed_cnt, failed_cnt) =
+        output_monitor.join().map_err(|_| eyre::eyre!("failed to join output monitor thread"))?;
+    println!("passed={passed_cnt} failed={failed_cnt}");
 
     #[cfg(not(target_os = "windows"))]
     {
         std::fs::remove_file(launcher_script)?;
     }
 
-    Ok((output.status, cmd_output))
+    Ok(status)
+}
+
+enum TestResult {
+    Passed,
+    Failed,
 }
 
-fn decorate_output(input: &str) -> String {
-    let mut decorated = String::with_capacity(input.len());
-    let (mut total_passed, mut total_failed) = (0, 0);
-    for line in input.lines() {
-        let mut line = line.to_string();
-        let mut is_old_line = false;
-        let mut is_new_line = false;
-
-        if line.starts_with("ok") {
-            // for pg_regress from pg16 forward, rewrite the "ok" into a colored PASS"
-            is_new_line = true;
-        } else if line.starts_with("not ok") {
-            // for pg_regress from pg16 forward, rewrite the "no ok" into a colored FAIL"
-            line = line.replace("not ok", "not_ok"); // to make parsing easier down below
-            is_new_line = true;
-        } else if line.contains("... ok") || line.contains("... FAILED") {
-            is_old_line = true;
+fn decorate_output(mut line: String) -> Option<(String, Option<TestResult>)> {
+    let mut decorated = String::with_capacity(line.len());
+    let mut test_result: Option<TestResult> = None;
+    let mut is_old_line = false;
+    let mut is_new_line = false;
+
+    if line.starts_with("ok") {
+        // for pg_regress from pg16 forward, rewrite the "ok" into a colored PASS"
+        is_new_line = true;
+    } else if line.starts_with("not ok") {
+        // for pg_regress from pg16 forward, rewrite the "no ok" into a colored FAIL"
+        line = line.replace("not ok", "not_ok"); // to make parsing easier down below
+        is_new_line = true;
+    } else if line.contains("... ok") || line.contains("... FAILED") {
+        is_old_line = true;
+    }
+
+    let parsed_test_line = if is_new_line {
+        fn split_line(line: &str) -> Option<(&str, bool, &str, &str)> {
+            let mut parts = line.split_whitespace();
+
+            let passed = parts.next()? == "ok";
+            parts.next()?; // throw away the test number
+            parts.next()?; // throw away the dash (-)
+            let test_name = parts.next()?;
+            let execution_time = parts.next()?;
+            let execution_units = parts.next()?;
+            Some((test_name, passed, execution_time, execution_units))
         }
+        split_line(&line)
+    } else if is_old_line {
+        fn split_line(line: &str) -> Option<(&str, bool, &str, &str)> {
+            let mut parts = line.split_whitespace();
+
+            parts.next()?; // throw away "test"
+            let test_name = parts.next()?;
+            parts.next()?; // throw away "..."
+            let passed = parts.next()? == "ok";
+            let execution_time = parts.next()?;
+            let execution_units = parts.next()?;
+            Some((test_name, passed, execution_time, execution_units))
+        }
+        split_line(&line)
+    } else {
+        // not a line we care about
+        return None;
+    };
 
-        let parsed_test_line = if is_new_line {
-            fn split_line(line: &str) -> Option<(&str, bool, &str, &str)> {
-                let mut parts = line.split_whitespace();
-
-                let passed = parts.next()? == "ok";
-                parts.next()?; // throw away the test number
-                parts.next()?; // throw away the dash (-)
-                let test_name = parts.next()?;
-                let execution_time = parts.next()?;
-                let execution_units = parts.next()?;
-                Some((test_name, passed, execution_time, execution_units))
-            }
-            split_line(&line)
-        } else if is_old_line {
-            fn split_line(line: &str) -> Option<(&str, bool, &str, &str)> {
-                let mut parts = line.split_whitespace();
-
-                parts.next()?; // throw away "test"
-                let test_name = parts.next()?;
-                parts.next()?; // throw away "..."
-                let passed = parts.next()? == "ok";
-                let execution_time = parts.next()?;
-                let execution_units = parts.next()?;
-                Some((test_name, passed, execution_time, execution_units))
-            }
-            split_line(&line)
+    if let Some((test_name, passed, execution_time, execution_units)) = parsed_test_line {
+        if passed {
+            test_result = Some(TestResult::Passed);
         } else {
-            // not a line we care about
-            continue;
-        };
+            test_result = Some(TestResult::Failed);
+        }
 
-        if let Some((test_name, passed, execution_time, execution_units)) = parsed_test_line {
+        decorated.push_str(&format!(
+            "{} {test_name} {execution_time}{execution_units}",
             if passed {
-                total_passed += 1
+                "PASS".bold().bright_green().to_string()
             } else {
-                total_failed += 1
+                "FAIL".bold().bright_red().to_string()
             }
-
-            decorated.push_str(&format!(
-                "{} {test_name} {execution_time}{execution_units}\n",
-                if passed {
-                    "PASS".bold().bright_green().to_string()
-                } else {
-                    "FAIL".bold().bright_red().to_string()
-                }
-            ))
-        }
-    }
-
-    if total_passed + total_failed > 0 {
-        decorated.push_str(&format!("passed={total_passed}, failed={total_failed}\n"))
+        ))
     }
 
-    decorated
+    Some((decorated, test_result))
 }
 
 fn make_test_name(entry: &DirEntry) -> String {