Watch
0
0
Fork
You've already forked hyperhive
0

bash-mcp: warn when capturing a task's stdout or stderr fails

copy_stream_to_file dropped the copy and flush errors, so an ENOSPC or
pipe error left a truncated .out/.err with no trace while the task still
reported its exit status. It now returns the error and the runner logs
it with the stream and file path.
This commit is contained in:
atlas 2026-10-16 14:52:46 +02:00 • committed by mara
commit fadcac8479

View file

@ -786,17 +786,18 @@ async fn exec_cmd(
let stdout = child.stdout.take().expect("stdout piped"); let stdout = child.stdout.take().expect("stdout piped");
let stderr = child.stderr.take().expect("stderr piped"); let stderr = child.stderr.take().expect("stderr piped");
let out_path = out_path.to_path_buf(); let copy_out = {
let err_path = err_path.to_path_buf(); let path = out_path.to_path_buf();
tokio::spawn(
let copy_out = tokio::spawn(copy_stream_to_file( async move { copy_stream_to_file(tokio::io::BufReader::new(stdout), &path).await },
tokio::io::BufReader::new(stdout), )
out_path, };
)); let copy_err = {
let copy_err = tokio::spawn(copy_stream_to_file( let path = err_path.to_path_buf();
tokio::io::BufReader::new(stderr), tokio::spawn(
err_path, 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`. // A future that fires when the timeout elapses, or never if `None`.
let timeout_fut = async { let timeout_fut = async {
@ -835,8 +836,8 @@ async fn exec_cmd(
} }
}; };
let _ = copy_out.await; warn_if_copy_failed("stdout", out_path, copy_out.await);
let _ = copy_err.await; warn_if_copy_failed("stderr", err_path, copy_err.await);
let status = wait_res?; let status = wait_res?;
let outcome = if let Some(forced) = killed { 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) i32::try_from(v).unwrap_or(i32::MAX)
} }
async fn copy_stream_to_file<R>(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<R>(mut reader: R, path: &Path) -> std::io::Result<()>
where where
R: tokio::io::AsyncRead + Unpin, R: tokio::io::AsyncRead + Unpin,
{ {
match tokio::fs::File::create(&path).await { let mut f = tokio::fs::File::create(path).await?;
Ok(mut f) => { let copied = tokio::io::copy(&mut reader, &mut f).await;
let _ = tokio::io::copy(&mut reader, &mut f).await; f.flush().await?;
let _ = f.flush().await; copied.map(|_| ())
} }
Err(e) => {
tracing::warn!(path = %path.display(), error = ?e, "bash_runner: open output file failed"); /// 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<std::io::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<String> { fn tail_file(path: &Path, max_bytes: usize) -> Option<String> {
@ -1208,4 +1226,55 @@ mod tests {
assert_eq!(read_task("t1").unwrap().status, TaskStatus::Pending); 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<std::io::Result<()>> {
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: std::future::Future>(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");
}
} }