Handle EPIPE error gracefully

Author: Notgnoshi

candumpr/src/main.rs

@@ -21,6 +21,13 @@ extern "C" fn signal_handler(_sig: libc::c_int) {
     STOP.store(true, Ordering::Relaxed);
 }
 
+/// True if any error in the chain is an EPIPE.
+fn is_broken_pipe(err: &eyre::Report) -> bool {
+    err.chain()
+        .filter_map(|e| e.downcast_ref::<std::io::Error>())
+        .any(|io| io.kind() == std::io::ErrorKind::BrokenPipe)
+}
+
 /// Log a link-state edge to stderr, ignoring repeats of the last observed state.
 fn handle_link_event(event: LinkEvent, link_up: &mut [Option<bool>], names: &[String]) {
     let (sock_id, up) = match event {
@@ -201,6 +208,10 @@ fn main() -> ExitCode {
                 Ok(mut batch) => {
                     log_error_frames(&batch, &mut bus_state, &names);
                     if let Err(e) = pipeline.write_batch(&batch) {
+                        if is_broken_pipe(&e) {
+                            tracing::debug!("output closed; shutting down");
+                            break;
+                        }
                         tracing::error!(error = ?e, "failed to write batch");
                         failed = true;
                     }
@@ -255,6 +266,9 @@ fn main() -> ExitCode {
     // Drain everything the receiver queued before it exited.
     while let Ok(mut batch) = full_rx.try_recv() {
         if let Err(e) = pipeline.write_batch(&batch) {
+            if is_broken_pipe(&e) {
+                break;
+            }
             tracing::error!(error = ?e, "failed to write batch during drain");
             failed = true;
         }

candumpr/tests/epipe.rs

@@ -0,0 +1,73 @@
+use std::io::{BufRead, BufReader};
+use std::os::unix::io::AsFd;
+use std::process::{Command, Stdio};
+use std::time::{Duration, Instant};
+
+use candumpr::can::{self, LinuxCanFrame};
+use vcan_fixture::VcanHarness;
+
+#[ctor::ctor]
+fn setup() {
+    tracing_subscriber::fmt().with_test_writer().init();
+    vcan_fixture::enter_namespace();
+}
+
+/// When the stdout consumer goes away (`candumpr can0 | head`), candumpr must notice the broken
+/// pipe on its next write and exit cleanly on its own, without being signalled.
+#[test]
+#[cfg_attr(feature = "ci", ignore = "requires vcan")]
+fn exits_cleanly_when_stdout_closes() {
+    let vcans = VcanHarness::new(1).unwrap();
+    let iface = &vcans.names()[0];
+
+    let mut child = Command::new(env!("CARGO_BIN_EXE_candumpr"))
+        .arg(iface)
+        .arg("--log-level=INFO") // TRACE level fills up the stderr buf and prevents EPIPE error
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .spawn()
+        .unwrap();
+
+    // Give the process time to set up io_uring and start receiving.
+    std::thread::sleep(Duration::from_millis(200));
+
+    let tx = can::open_can_raw_blocking(iface).unwrap();
+    let frame = LinuxCanFrame::new(0x18FECA00 | libc::CAN_EFF_FLAG, &[0xAA, 0xBB]);
+
+    // Prove the pipe is live: send one frame and read the line it produces.
+    let mut stdout = BufReader::new(child.stdout.take().unwrap());
+    can::send_frame(tx.as_fd(), &frame).unwrap();
+    let mut line = String::new();
+    stdout.read_line(&mut line).unwrap();
+    assert!(
+        line.ends_with('\n'),
+        "expected a full output line: {line:?}"
+    );
+
+    // Close the read end, like `head` exiting. The next batch write hits EPIPE.
+    drop(stdout);
+    can::send_frame(tx.as_fd(), &frame).unwrap();
+
+    let deadline = Instant::now() + Duration::from_secs(5);
+    let status = loop {
+        if let Some(status) = child.try_wait().unwrap() {
+            break status;
+        }
+        if Instant::now() >= deadline {
+            child.kill().unwrap();
+            let output = child.wait_with_output().unwrap();
+            panic!(
+                "candumpr did not exit after stdout closed. stderr:\n{}",
+                String::from_utf8_lossy(&output.stderr)
+            );
+        }
+        std::thread::sleep(Duration::from_millis(50));
+    };
+
+    let output = child.wait_with_output().unwrap();
+    eprint!("{}", String::from_utf8_lossy(&output.stderr));
+    assert!(
+        status.success(),
+        "expected clean exit after stdout closed, got {status}"
+    );
+}