test: refactor real backend support for reliable output capture

crates/irtt-client/tests/support/mod.rs

@@ -36,15 +36,27 @@ pub fn selected_backend() -> TestBackendKind {
         Ok("real") => {
             let irtt_bin = std::env::var("IRTT_BIN").unwrap_or_else(|_| "irtt".to_string());
             match Command::new(&irtt_bin).arg("version").output() {
-                Ok(output) if output.status.success() => TestBackendKind::Real,
+                Ok(output) if output.status.success() => {
+                    debug_backend("[backend] selected backend=real");
+                    TestBackendKind::Real
+                }
                 _ => panic!("IRTT_TEST_BACKEND=real but irtt binary not found at '{irtt_bin}'"),
             }
         }
-        Ok("fake") | Err(_) => TestBackendKind::Fake,
+        Ok("fake") | Err(_) => {
+            debug_backend("[backend] selected backend=fake");
+            TestBackendKind::Fake
+        }
         Ok(other) => panic!("unknown IRTT_TEST_BACKEND value: {other}"),
     }
 }
 
+fn debug_backend(message: &str) {
+    if std::env::var("IRTT_TEST_BACKEND_DEBUG").as_deref() == Ok("1") {
+        eprintln!("{message}");
+    }
+}
+
 pub enum BackendPeer {
     Fake(FakeServer),
     Real(RealIrtServer),

crates/irtt-client/tests/support/real_irtt.rs

@@ -1,31 +1,32 @@
 #![allow(dead_code)]
 
-use std::io::{BufRead, BufReader};
+use std::io::{BufRead, BufReader, Read};
 use std::net::{SocketAddr, UdpSocket};
 use std::process::{Child, Command, Stdio};
-use std::sync::mpsc;
+use std::sync::{mpsc, Arc, Mutex};
-use std::thread;
+use std::thread::{self, JoinHandle};
 use std::time::{Duration, Instant};
 
 pub struct RealIrtServer {
     addr: SocketAddr,
     child: Child,
     keepalive_ms: Option<u64>,
+    stdout_thread: Option<JoinHandle<()>>,
+    stderr_thread: Option<JoinHandle<()>>,
 }
 
 impl RealIrtServer {
     pub fn start(hmac_key: Option<&[u8]>) -> Result<Self, String> {
         let port = Self::find_free_port()?;
         let addr = SocketAddr::from(([127, 0, 0, 1], port));
-        let bind = format!("127.0.0.1:{}", port);
+        let bind = format!("127.0.0.1:{port}");
 
         let irtt_bin = std::env::var("IRTT_BIN").unwrap_or_else(|_| "irtt".to_string());
-
         let keepalive_ms = std::env::var("IRTT_TEST_KEEP_SERVER_MS")
             .ok()
             .and_then(|v| v.parse::<u64>().ok());
 
-        debug_eprintln!("[real_irtt] backend=real irtt_bin={irtt_bin} bind={bind}");
+        debug_eprintln!("[real_irtt] selected backend=real irtt_bin={irtt_bin} bind={bind}");
 
         let mut cmd = Command::new(&irtt_bin);
         cmd.arg("server")
@@ -37,7 +38,7 @@ impl RealIrtServer {
 
         if let Some(key) = hmac_key {
             let hex_key: String = key.iter().map(|b| format!("{:02x}", b)).collect();
-            cmd.arg(format!("--hmac=0x{}", hex_key));
+            cmd.arg(format!("--hmac=0x{hex_key}"));
         }
 
         let full_cmd = format!("{cmd:?}");
@@ -45,7 +46,7 @@ impl RealIrtServer {
 
         let mut child = cmd
             .spawn()
-            .map_err(|e| format!("failed to spawn irtt server: {}", e))?;
+            .map_err(|e| format!("failed to spawn irtt server: {e}"))?;
 
         let pid = child.id();
         debug_eprintln!("[real_irtt] child pid={pid}");
@@ -53,91 +54,49 @@ impl RealIrtServer {
         let stdout = child.stdout.take().ok_or("failed to capture stdout")?;
         let stderr = child.stderr.take().ok_or("failed to capture stderr")?;
 
-        let (out_tx, out_rx) = mpsc::channel();
+        let captured = Arc::new(Mutex::new(Vec::new()));
-        let (err_tx, err_rx) = mpsc::channel();
+        let (line_tx, line_rx) = mpsc::channel();
 
-        thread::spawn(move || {
+        let stdout_thread = {
-            let reader = BufReader::new(stdout);
+            let line_tx = line_tx.clone();
-            for line in reader.lines() {
+            let captured = Arc::clone(&captured);
-                match line {
+            thread::spawn(move || drain_stream("stdout", stdout, line_tx, captured))
-                    Ok(l) => {
-                        if out_tx.send(("stdout", l)).is_err() {
-                            break;
-                        }
-                    }
-                    Err(_) => break,
-                }
-            }
-        });
-
-        thread::spawn(move || {
-            let reader = BufReader::new(stderr);
-            for line in reader.lines() {
-                match line {
-                    Ok(l) => {
-                        if err_tx.send(("stderr", l)).is_err() {
-                            break;
-                        }
-                    }
-                    Err(_) => break,
-                }
-            }
-        });
-
-        let timeout = Duration::from_secs(5);
-        let start = Instant::now();
-        let mut captured_lines = Vec::new();
-
-        let ready = loop {
-            let remaining = timeout
-                .checked_sub(start.elapsed())
-                .unwrap_or(Duration::ZERO);
-
-            let line = select_line(&out_rx, &err_rx, remaining);
-            match line {
-                Some((stream, text)) => {
-                    debug_eprintln!("[real_irtt] {stream}: {text}");
-                    captured_lines.push(format!("{stream}: {text}"));
-                    if text.contains("[ListenerStart]") {
-                        break true;
-                    }
-                    if (text.contains("[ListenerStop]") || text.contains("[ServerStop]"))
-                        && (text.contains("error") || text.contains("Error"))
-                    {
-                        break false;
-                    }
-                }
-                None => {
-                    break false;
-                }
-            }
         };
 
+        let stderr_thread = {
+            let captured = Arc::clone(&captured);
+            thread::spawn(move || drain_stream("stderr", stderr, line_tx, captured))
+        };
+
+        let ready = wait_for_ready(&line_rx, Duration::from_secs(5));
+
         if !ready {
             let exit_status = child.try_wait().ok().flatten();
-            debug_eprintln!("[real_irtt] startup failed, exit_status={exit_status:?}");
+            debug_eprintln!("[real_irtt] readiness result=false exit_status={exit_status:?}");
             child.kill().ok();
             child.wait().ok();
+            stdout_thread.join().ok();
+            stderr_thread.join().ok();
 
-            let output = captured_lines.join("\n");
+            let output = captured.lock().unwrap().join("\n");
             let extra = match exit_status {
                 Some(status) => format!("\nexit status: {status}"),
                 None => String::new(),
             };
+            debug_eprintln!("[real_irtt] failure diagnostics:\n{output}{extra}");
             return Err(format!(
                 "irtt server failed to start\noutput:\n{output}{extra}"
             ));
         }
 
-        debug_eprintln!("[real_irtt] ready on {addr}");
+        debug_eprintln!("[real_irtt] readiness result=true addr={addr}");
-
-        drop(out_rx);
-        drop(err_rx);
 
         Ok(Self {
             addr,
             child,
             keepalive_ms,
+            stdout_thread: Some(stdout_thread),
+            stderr_thread: Some(stderr_thread),
         })
     }
 
@@ -146,11 +105,11 @@ impl RealIrtServer {
     }
 
     fn find_free_port() -> Result<u16, String> {
-        let socket = UdpSocket::bind("127.0.0.1:0")
+        let socket =
-            .map_err(|e| format!("failed to find free port: {}", e))?;
+            UdpSocket::bind("127.0.0.1:0").map_err(|e| format!("failed to find free port: {e}"))?;
         let port = socket
             .local_addr()
-            .map_err(|e| format!("failed to get local addr: {}", e))?
+            .map_err(|e| format!("failed to get local addr: {e}"))?
             .port();
         drop(socket);
         Ok(port)
@@ -169,29 +128,55 @@ impl Drop for RealIrtServer {
         debug_eprintln!("[real_irtt] killing child pid={}", self.child.id());
         self.child.kill().ok();
         self.child.wait().ok();
+        if let Some(thread) = self.stdout_thread.take() {
+            thread.join().ok();
+        }
+        if let Some(thread) = self.stderr_thread.take() {
+            thread.join().ok();
+        }
     }
 }
 
-fn select_line(
+fn wait_for_ready(line_rx: &mpsc::Receiver<(&'static str, String)>, timeout: Duration) -> bool {
-    out_rx: &mpsc::Receiver<(&str, String)>,
+    let start = Instant::now();
-    err_rx: &mpsc::Receiver<(&str, String)>,
+    loop {
-    timeout: Duration,
+        let remaining = timeout
-) -> Option<(&'static str, String)> {
+            .checked_sub(start.elapsed())
-    if let Ok((_, line)) = out_rx.recv_timeout(timeout) {
+            .unwrap_or(Duration::ZERO);
-        return Some(("stdout", line));
+
+        match line_rx.recv_timeout(remaining) {
+            Ok((stream, text)) => {
+                debug_eprintln!("[real_irtt] {stream}: {text}");
+                if text.contains("[ListenerStart]") {
+                    return true;
                 }
-    if let Ok((_, line)) = err_rx.recv_timeout(Duration::ZERO) {
+                if (text.contains("[ListenerStop]") || text.contains("[ServerStop]"))
-        return Some(("stderr", line));
+                    && (text.contains("error") || text.contains("Error"))
+                {
+                    return false;
                 }
-    if timeout > Duration::ZERO {
-        if let Ok((_, line)) = out_rx.recv_timeout(Duration::from_millis(100)) {
-            return Some(("stdout", line));
             }
-        if let Ok((_, line)) = err_rx.recv_timeout(Duration::from_millis(100)) {
+            Err(_) => return false,
-            return Some(("stderr", line));
+        }
+    }
+}
+
+fn drain_stream(
+    stream: &'static str,
+    pipe: impl Read,
+    tx: mpsc::Sender<(&'static str, String)>,
+    captured: Arc<Mutex<Vec<String>>>,
+) {
+    let reader = BufReader::new(pipe);
+    for line in reader.lines() {
+        match line {
+            Ok(text) => {
+                captured.lock().unwrap().push(format!("{stream}: {text}"));
+                let _ = tx.send((stream, text));
+            }
+            Err(_) => break,
         }
     }
-    None
 }
 
 macro_rules! debug_eprintln {