fix process lifecycle races

src/process.rs

@@ -155,10 +155,7 @@ pub(crate) fn run_captured(
     terminate_remaining_group(child.id());
     clear_active(active_process_group);
 
-    input_writer
-        .join()
-        .map_err(|_| "formatter stdin writer panicked".to_owned())?
-        .map_err(|error| format!("could not write formatter input: {error}"))?;
+    join_input_writer_after_exit(input_writer)?;
     let stdout = receive_reader(stdout, "stdout")?;
     let stderr = receive_reader(stderr, "stderr")?;
     Ok(CapturedOutput {
@@ -168,6 +165,17 @@ pub(crate) fn run_captured(
     })
 }
 
+fn join_input_writer_after_exit(
+    input_writer: thread::JoinHandle<io::Result<()>>,
+) -> Result<(), String> {
+    match input_writer.join() {
+        Err(_) => Err("formatter stdin writer panicked".to_owned()),
+        Ok(Err(error)) if error.kind() == io::ErrorKind::BrokenPipe => Ok(()),
+        Ok(Err(error)) => Err(format!("could not write formatter input: {error}")),
+        Ok(Ok(())) => Ok(()),
+    }
+}
+
 fn spawn_bounded_reader<R: Read + Send + 'static>(
     reader: R,
     limit: usize,
@@ -254,3 +262,60 @@ fn signal_group(group: u32, signal: i32) {
         }
     }
 }
+
+#[cfg(all(test, unix))]
+mod tests {
+    use std::{
+        io,
+        sync::{Mutex, atomic::AtomicBool},
+        thread,
+        time::Duration,
+    };
+
+    use super::{CommandSpec, join_input_writer_after_exit, run_captured};
+
+    fn shell_spec(script: &str) -> CommandSpec {
+        CommandSpec {
+            programs: vec!["sh".to_owned()],
+            args: vec!["-c".to_owned(), script.to_owned()],
+            working_directory: None,
+            environment: Vec::new(),
+            timeout: Duration::from_secs(1),
+            max_output_bytes: 4096,
+        }
+    }
+
+    fn run_short_lived(script: &str) -> super::CapturedOutput {
+        run_captured(
+            &shell_spec(script),
+            vec![b'x'; 8 * 1024 * 1024],
+            &AtomicBool::new(false),
+            &Mutex::new(None),
+        )
+        .expect("broken stdin pipe after child exit must not hide its output")
+    }
+
+    #[test]
+    fn post_exit_broken_pipe_preserves_success_and_nonzero_output() {
+        let success = run_short_lived("exec 0<&-; printf ready");
+        assert!(success.status.success());
+        assert_eq!(success.stdout, b"ready");
+
+        let nonzero = run_short_lived("exec 0<&-; printf bad >&2; exit 7");
+        assert_eq!(nonzero.status.code(), Some(7));
+        assert_eq!(nonzero.stderr, b"bad");
+    }
+
+    #[test]
+    fn post_exit_non_broken_pipe_writer_error_remains_a_failure() {
+        let writer = thread::spawn(|| {
+            Err(io::Error::new(
+                io::ErrorKind::PermissionDenied,
+                "injected writer failure",
+            ))
+        });
+
+        let error = join_input_writer_after_exit(writer).expect_err("writer failure must surface");
+        assert!(error.contains("injected writer failure"));
+    }
+}

src/validation.rs

@@ -145,8 +145,10 @@ impl Drop for ValidationWorker {
     fn drop(&mut self) {
         self.request_stop();
         signal_active_process_group(&self.active_process_group, libc::SIGKILL);
-        if let Some(handle) = self.thread.take() {
-            drop(handle);
+        if let Some(handle) = self.thread.take()
+            && handle.thread().id() != thread::current().id()
+        {
+            let _ = handle.join();
         }
     }
 }