fafo

Author: barthalion

src/jobs/utils.rs

@@ -3,9 +3,13 @@ use diesel::prelude::*;
 use diesel::result::Error as DieselError;
 use log::{error, info};
 use std::fmt::Write as _;
+use std::io::Read;
 use std::os::unix::process::CommandExt;
 use std::process::{Command, Output, Stdio};
 use std::str;
+use std::sync::mpsc;
+use std::thread;
+use std::time::Duration;
 
 use crate::config::{Config, RepoConfig};
 use crate::errors::{JobError, JobResult};
@@ -128,10 +132,33 @@ macro_rules! job_log_and_error {
     }};
 }
 
+/// Reads from a stream with timeout to avoid hanging on orphaned pipes
+fn read_stream_with_timeout<R: Read + Send + 'static>(
+    mut stream: R,
+    timeout: Duration,
+) -> Result<Vec<u8>, std::io::Error> {
+    let (sender, receiver) = mpsc::channel();
+
+    thread::spawn(move || {
+        let mut buffer = Vec::new();
+        let result = stream.read_to_end(&mut buffer);
+        let _ = sender.send((result, buffer));
+    });
+
+    match receiver.recv_timeout(timeout) {
+        Ok((Ok(_bytes_read), buffer)) => Ok(buffer),
+        Ok((Err(e), _)) => Err(e),
+        Err(_) => {
+            // The orphaned child processes can keep running, but we won't wait
+            Ok(Vec::new())
+        }
+    }
+}
+
 /// Executes a command and returns its output. A JobError is returned if the command couldn't be executed, but not if
 /// it exits with a status code.
 pub fn do_command_with_output(cmd: &mut Command) -> JobResult<Output> {
-    let output = unsafe {
+    let mut child = unsafe {
         cmd.stdin(Stdio::null())
             .stdout(Stdio::piped())
             .stderr(Stdio::piped())
@@ -141,11 +168,33 @@ pub fn do_command_with_output(cmd: &mut Command) -> JobResult<Output> {
                 libc::setsid();
                 Ok(())
             })
-            .output()
+            .spawn()
             .map_err(|e| JobError::new(&format!("Failed to run {:?}: {}", &cmd, e)))?
     };
 
-    Ok(output)
+    let exit_status = child
+        .wait()
+        .map_err(|e| JobError::new(&format!("Failed to wait for {:?}: {}", &cmd, e)))?;
+
+    let stdout = if let Some(stdout_handle) = child.stdout.take() {
+        read_stream_with_timeout(stdout_handle, Duration::from_secs(30))
+            .map_err(|e| JobError::new(&format!("Failed to read stdout from {:?}: {}", &cmd, e)))?
+    } else {
+        Vec::new()
+    };
+
+    let stderr = if let Some(stderr_handle) = child.stderr.take() {
+        read_stream_with_timeout(stderr_handle, Duration::from_secs(30))
+            .map_err(|e| JobError::new(&format!("Failed to read stderr from {:?}: {}", &cmd, e)))?
+    } else {
+        Vec::new()
+    };
+
+    Ok(Output {
+        status: exit_status,
+        stdout,
+        stderr,
+    })
 }
 
 /// Executes a command. A JobError is returned if the command exits with an unsuccessful status code.