use std::{
io::{self, Read, Write},
sync::Arc,
thread,
};
#[cfg(test)]
use super::config::OutputStream;
use super::{
config::{CommandConfig, OutputMode, PipeLimit, PipeSpec},
error::CommandFailure,
result::PipeOutcome,
};
use camino::Utf8PathBuf;
use tempfile::NamedTempFile;
const PIPE_CHUNK_SIZE: usize = 8192;
pub(super) fn spawn_pipe_reader<R>(
pipe: Option<R>,
spec: PipeSpec,
config: Arc<CommandConfig>,
) -> Option<thread::JoinHandle<Result<PipeOutcome, CommandFailure>>>
where
R: Read + Send + 'static,
{
pipe.map(|reader| thread::spawn(move || read_pipe(reader, spec, config.as_ref())))
}
pub(super) fn join_reader(
reader_handle: Option<thread::JoinHandle<Result<PipeOutcome, CommandFailure>>>,
spec: PipeSpec,
config: &CommandConfig,
) -> Result<PipeOutcome, CommandFailure> {
match reader_handle {
Some(join_handle) => join_handle
.join()
.map_err(|_| CommandFailure::Io(io::Error::other("pipe reader panicked")))?,
None => empty_outcome(spec, config),
}
}
fn empty_outcome(spec: PipeSpec, config: &CommandConfig) -> Result<PipeOutcome, CommandFailure> {
if matches!(spec.mode(), OutputMode::Tempfile) {
create_empty_tempfile(config, spec.stream().empty_tempfile_label())
.map(PipeOutcome::Tempfile)
} else {
Ok(PipeOutcome::Bytes(Vec::new()))
}
}
pub(super) fn cleanup_readers(
stdout_reader: &mut Option<thread::JoinHandle<Result<PipeOutcome, CommandFailure>>>,
stderr_reader: &mut Option<thread::JoinHandle<Result<PipeOutcome, CommandFailure>>>,
stdin_handle: &mut Option<thread::JoinHandle<io::Result<()>>>,
) {
join_pipe_for_cleanup("stdout", stdout_reader);
join_pipe_for_cleanup("stderr", stderr_reader);
if let Some(handle) = stdin_handle.take()
&& let Err(join_err) = handle.join()
{
tracing::warn!("stdin writer thread panicked: {join_err:?}");
}
}
pub(super) fn handle_stdin_result(
stdin_handle: Option<thread::JoinHandle<io::Result<()>>>,
status: Option<i32>,
stderr: &[u8],
) -> Result<(), CommandFailure> {
let Some(handle) = stdin_handle else {
return Ok(());
};
match handle.join() {
Ok(Ok(())) => Ok(()),
Ok(Err(err)) => {
if err.kind() == io::ErrorKind::BrokenPipe {
if status == Some(0) {
return Ok(());
}
return Err(CommandFailure::BrokenPipe {
source: err,
status,
stderr: stderr.to_vec(),
});
}
Err(CommandFailure::Io(err))
}
Err(_) => Err(CommandFailure::Io(io::Error::other(
"stdin writer panicked",
))),
}
}
fn read_pipe<R>(
reader: R,
spec: PipeSpec,
config: &CommandConfig,
) -> Result<PipeOutcome, CommandFailure>
where
R: Read,
{
let limit = spec.into_limit();
match spec.mode() {
OutputMode::Capture => read_pipe_capture(reader, limit),
OutputMode::Tempfile => {
read_pipe_tempfile(reader, limit, spec.stream().tempfile_label(), config)
}
}
}
fn read_pipe_capture<R>(mut reader: R, mut limit: PipeLimit) -> Result<PipeOutcome, CommandFailure>
where
R: Read,
{
let mut buf = Vec::new();
let mut chunk = [0_u8; PIPE_CHUNK_SIZE];
loop {
let read = reader.read(&mut chunk).map_err(CommandFailure::Io)?;
if read == 0 {
break;
}
limit.record(read)?;
buf.extend(chunk.iter().take(read).copied());
}
Ok(PipeOutcome::Bytes(buf))
}
fn read_pipe_tempfile<R>(
mut reader: R,
mut limit: PipeLimit,
label: &str,
config: &CommandConfig,
) -> Result<PipeOutcome, CommandFailure>
where
R: Read,
{
let mut tempfile = config.create_tempfile(label).map_err(CommandFailure::Io)?;
let mut chunk = [0_u8; PIPE_CHUNK_SIZE];
loop {
let read = reader.read(&mut chunk).map_err(CommandFailure::Io)?;
if read == 0 {
break;
}
limit.record(read)?;
#[expect(
clippy::indexing_slicing,
reason = "Read::read guarantees `read` does not exceed `chunk.len()`"
)]
tempfile
.write_all(&chunk[..read])
.map_err(CommandFailure::Io)?;
}
tempfile.flush().map_err(CommandFailure::Io)?;
let path = persist_tempfile(tempfile)?;
Ok(PipeOutcome::Tempfile(path))
}
fn create_empty_tempfile(
config: &CommandConfig,
label: &str,
) -> Result<Utf8PathBuf, CommandFailure> {
let mut tempfile = config.create_tempfile(label).map_err(CommandFailure::Io)?;
tempfile.flush().map_err(CommandFailure::Io)?;
persist_tempfile(tempfile)
}
fn join_pipe_for_cleanup(
label: &str,
reader_handle: &mut Option<thread::JoinHandle<Result<PipeOutcome, CommandFailure>>>,
) {
if let Some(join_handle) = reader_handle.take() {
match join_handle.join() {
Ok(Ok(_)) => {}
Ok(Err(err)) => {
tracing::warn!(stream = label, ?err, "pipe reader failed during cleanup");
}
Err(join_err) => {
tracing::warn!(stream = label, ?join_err, "pipe reader thread panicked");
}
}
}
}
fn persist_tempfile(tempfile: NamedTempFile) -> Result<Utf8PathBuf, CommandFailure> {
let temp_path = tempfile.into_temp_path();
let path = temp_path
.keep()
.map_err(|err| CommandFailure::Io(err.error))?;
Utf8PathBuf::from_path_buf(path).map_err(|_| {
CommandFailure::Io(io::Error::new(
io::ErrorKind::InvalidData,
"command tempfile path is not valid UTF-8",
))
})
}
#[cfg(test)]
mod tests {
use super::*;
use crate::stdlib::command::tests_support::test_command_config;
use anyhow::{Context, Result, anyhow, ensure};
use std::io::Cursor;
use test_support::fs;
fn assert_output_limit_error(
outcome: Result<PipeOutcome, CommandFailure>,
expected_stream: OutputStream,
expected_mode: OutputMode,
expected_limit: u64,
) {
let err =
outcome.expect_err("expected command to exceed the configured output limit for test");
match err {
CommandFailure::OutputLimit {
stream,
mode,
limit,
} => {
assert_eq!(stream, expected_stream);
assert_eq!(mode, expected_mode);
assert_eq!(limit, expected_limit);
}
other => panic!("unexpected error variant: {other:?}"),
}
}
#[test]
fn read_pipe_capture_collects_bytes_within_limit() {
let data = b"payload".to_vec();
let outcome = read_pipe_capture(
Cursor::new(data.clone()),
PipeSpec::new(OutputStream::Stdout, OutputMode::Capture, 128).into_limit(),
)
.expect("capture should succeed within the configured limit");
match outcome {
PipeOutcome::Bytes(buf) => assert_eq!(buf, data),
PipeOutcome::Tempfile(_) => panic!("capture mode should emit bytes"),
}
}
#[test]
fn read_pipe_capture_reports_limit_exceedance() {
let outcome = read_pipe_capture(
Cursor::new(vec![0_u8; 16]),
PipeSpec::new(OutputStream::Stdout, OutputMode::Capture, 8).into_limit(),
);
assert_output_limit_error(outcome, OutputStream::Stdout, OutputMode::Capture, 8);
}
#[test]
fn read_pipe_tempfile_writes_streamed_data() -> Result<()> {
let payload = vec![b'x'; 32];
let (_temp_dir, config) = test_command_config()?;
let outcome = read_pipe_tempfile(
Cursor::new(payload.clone()),
PipeSpec::new(OutputStream::Stdout, OutputMode::Tempfile, 64).into_limit(),
"stdout",
&config,
)
.map_err(|err| anyhow!("streaming should succeed within the limit, got {err:?}"))?;
let PipeOutcome::Tempfile(path) = outcome else {
return Err(anyhow!("streaming mode should emit a tempfile path"));
};
let disk = fs::read(path.as_std_path()).context("read streamed output")?;
ensure!(
disk == payload,
"streamed file contents did not match payload"
);
fs::remove_file(path.as_std_path()).context("cleanup streamed file")?;
Ok(())
}
#[test]
fn read_pipe_tempfile_respects_stream_limit() -> Result<()> {
let (_temp_dir, config) = test_command_config()?;
let outcome = read_pipe_tempfile(
Cursor::new(vec![b'y'; 32]),
PipeSpec::new(OutputStream::Stdout, OutputMode::Tempfile, 8).into_limit(),
"stdout",
&config,
);
assert_output_limit_error(outcome, OutputStream::Stdout, OutputMode::Tempfile, 8);
Ok(())
}
}