diff --git a/hive-bash-mcp/src/runner.rs b/hive-bash-mcp/src/runner.rs index 232233c5..1138eccf 100644 --- a/hive-bash-mcp/src/runner.rs +++ b/hive-bash-mcp/src/runner.rs @@ -786,17 +786,18 @@ async fn exec_cmd( let stdout = child.stdout.take().expect("stdout piped"); let stderr = child.stderr.take().expect("stderr piped"); - let out_path = out_path.to_path_buf(); - let err_path = err_path.to_path_buf(); - - let copy_out = tokio::spawn(copy_stream_to_file( - tokio::io::BufReader::new(stdout), - out_path, - )); - let copy_err = tokio::spawn(copy_stream_to_file( - tokio::io::BufReader::new(stderr), - err_path, - )); + let copy_out = { + let path = out_path.to_path_buf(); + tokio::spawn( + async move { copy_stream_to_file(tokio::io::BufReader::new(stdout), &path).await }, + ) + }; + let copy_err = { + let path = err_path.to_path_buf(); + tokio::spawn( + async move { copy_stream_to_file(tokio::io::BufReader::new(stderr), &path).await }, + ) + }; // A future that fires when the timeout elapses, or never if `None`. let timeout_fut = async { @@ -835,8 +836,8 @@ async fn exec_cmd( } }; - let _ = copy_out.await; - let _ = copy_err.await; + warn_if_copy_failed("stdout", out_path, copy_out.await); + warn_if_copy_failed("stderr", err_path, copy_err.await); let status = wait_res?; let outcome = if let Some(forced) = killed { @@ -854,19 +855,36 @@ fn u32_to_i32(v: u32) -> i32 { i32::try_from(v).unwrap_or(i32::MAX) } -async fn copy_stream_to_file(mut reader: R, path: PathBuf) +/// Copies `reader` into a new file at `path`, flushing at the end. On error +/// the bytes copied before it stay in the file. +async fn copy_stream_to_file(mut reader: R, path: &Path) -> std::io::Result<()> where R: tokio::io::AsyncRead + Unpin, { - match tokio::fs::File::create(&path).await { - Ok(mut f) => { - let _ = tokio::io::copy(&mut reader, &mut f).await; - let _ = f.flush().await; - } - Err(e) => { - tracing::warn!(path = %path.display(), error = ?e, "bash_runner: open output file failed"); - } - } + let mut f = tokio::fs::File::create(path).await?; + let copied = tokio::io::copy(&mut reader, &mut f).await; + f.flush().await?; + copied.map(|_| ()) +} + +/// Warns when a stream copy failed or its task panicked: the log file is then +/// incomplete while the task still reports its exit status. +fn warn_if_copy_failed( + stream: &str, + path: &Path, + joined: Result, tokio::task::JoinError>, +) { + let error = match joined { + Ok(Ok(())) => return, + Ok(Err(e)) => format!("{e:?}"), + Err(e) => format!("{e:?}"), + }; + tracing::warn!( + stream, + path = %path.display(), + error, + "bash_runner: capturing task output failed, the log file is incomplete" + ); } fn tail_file(path: &Path, max_bytes: usize) -> Option { @@ -1208,4 +1226,55 @@ mod tests { assert_eq!(read_task("t1").unwrap().status, TaskStatus::Pending); }); } + + /// Yields `data`, then fails with an I/O error. + struct FailAfter { + data: &'static [u8], + sent: bool, + } + + impl tokio::io::AsyncRead for FailAfter { + fn poll_read( + mut self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> std::task::Poll> { + if self.sent { + return std::task::Poll::Ready(Err(std::io::Error::other("pipe broke"))); + } + self.sent = true; + buf.put_slice(self.data); + std::task::Poll::Ready(Ok(())) + } + } + + fn block_on(f: F) -> F::Output { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(f) + } + + #[test] + fn copy_error_is_returned_and_earlier_bytes_are_kept() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t1.out"); + let reader = FailAfter { + data: b"before", + sent: false, + }; + let res = block_on(super::copy_stream_to_file(reader, &path)); + assert_eq!(res.unwrap_err().to_string(), "pipe broke"); + assert_eq!(std::fs::read(&path).unwrap(), b"before"); + } + + #[test] + fn successful_copy_writes_everything() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("t1.out"); + let res = block_on(super::copy_stream_to_file(&b"all of it"[..], &path)); + assert!(res.is_ok()); + assert_eq!(std::fs::read(&path).unwrap(), b"all of it"); + } }