diff --git a/crates/codexbar-providers/src/command.rs b/crates/codexbar-providers/src/command.rs index 972fdcc1..5883b9fd 100644 --- a/crates/codexbar-providers/src/command.rs +++ b/crates/codexbar-providers/src/command.rs @@ -179,6 +179,10 @@ impl Capture

{ }) } + fn open_pipe(&self) -> Option<&P> { + (!self.closed).then_some(&self.pipe) + } + /// At most one chunk per stream per turn, so a noisy stream cannot starve the other stream or the deadline. fn drain(&mut self) -> Result { if self.closed { @@ -255,7 +259,11 @@ fn collect( return Ok((stdout.bytes, stderr.bytes)); } if !progress { - std::thread::sleep(CHECK_INTERVAL.min(timeout.saturating_sub(start.elapsed()))); + pipe::wait( + stdout.open_pipe(), + stderr.open_pipe(), + CHECK_INTERVAL.min(timeout.saturating_sub(start.elapsed())), + ); } } } @@ -272,6 +280,59 @@ mod tests { ); } + /// Every drain outcome on a real pipe, independent of how a command's writes and EOFs happen to interleave. + #[cfg(unix)] + #[test] + fn a_drain_reports_pending_output_eof_a_closed_stream_and_overflow() { + use std::io::Write as _; + let (reader, mut writer) = std::io::pipe().unwrap(); + let mut capture = Capture::new(reader, 4).unwrap(); + assert_eq!(capture.drain(), Ok(false), "nothing written yet"); + writer.write_all(b"abc").unwrap(); + assert_eq!(capture.drain(), Ok(true)); + assert_eq!(capture.bytes, b"abc"); + writer.write_all(b"de").unwrap(); + assert_eq!(capture.drain(), Err(CommandError::Failed), "past the 4-byte limit"); + let (reader, writer) = std::io::pipe().unwrap(); + let mut capture = Capture::new(reader, 4).unwrap(); + drop(writer); + assert_eq!(capture.drain(), Ok(true), "EOF is progress"); + assert!(capture.closed); + assert_eq!(capture.drain(), Ok(false), "a closed stream is never read again"); + } + + /// Every way a collection turn ends or goes on, independent of which pipe a real command closes first. + #[cfg(unix)] + #[test] + fn collection_ends_at_the_stop_marker_or_once_both_streams_closed_and_the_command_exited() { + let mut child = std::process::Command::new("true").spawn().unwrap(); + let (reader, _writer) = std::io::pipe().unwrap(); + let mut stderr = Capture { + pipe: reader, + bytes: b"Error: authentication required.".to_vec(), + closed: false, + limit: OUTPUT_LIMIT, + }; + let stop = |marker| StopMarker { marker, searched: 0 }; + assert_eq!( + complete(&mut child, false, &stderr, &mut stop(b"authentication required")), + Ok(true) + ); + assert_eq!(complete(&mut child, false, &stderr, &mut stop(b"")), Ok(false)); + assert_eq!( + complete(&mut child, true, &stderr, &mut stop(b"")), + Ok(false), + "stderr still open" + ); + stderr.closed = true; + let deadline = Instant::now() + Duration::from_secs(5); + while complete(&mut child, true, &stderr, &mut stop(b"")) != Ok(true) { + assert!(Instant::now() < deadline, "the exited command ends collection"); + std::thread::sleep(Duration::from_millis(5)); + } + child.wait().unwrap(); + } + #[test] fn a_stop_marker_is_found_across_reads_and_case() { let text = b"Authentication required. Please visit the URL"; diff --git a/crates/codexbar-providers/src/command/pipe.rs b/crates/codexbar-providers/src/command/pipe.rs index baf765f4..b02d5ead 100644 --- a/crates/codexbar-providers/src/command/pipe.rs +++ b/crates/codexbar-providers/src/command/pipe.rs @@ -57,6 +57,42 @@ pub(super) fn read(pipe: &mut impl Pipe, bytes: &mut [u8]) -> io::Result, second: Option<&impl Pipe>, timeout: std::time::Duration) { + poll_or_nap([first.map(|p| p.as_raw_fd()), second.map(|p| p.as_raw_fd())], timeout); +} + +/// `wait` for raw descriptors: one implementation however many pipe types call it. +#[cfg(unix)] +fn poll_or_nap(fds: [Option; 2], timeout: std::time::Duration) { + let mut fds: Vec = fds + .into_iter() + .flatten() + .map(|fd| libc::pollfd { + fd, + events: libc::POLLIN, + revents: 0, + }) + .collect(); + if fds.is_empty() { + std::thread::sleep(timeout); + return; + } + // Rounded up, so a sub-millisecond remainder still waits instead of spinning. + let ms = timeout.as_micros().div_ceil(1000).min(libc::c_int::MAX as u128) as libc::c_int; + // SAFETY: `fds` is valid writable storage for its length, and each descriptor is an open pipe this thread owns. + let _ = unsafe { libc::poll(fds.as_mut_ptr(), fds.len() as libc::nfds_t, ms) }; +} + +#[cfg(windows)] +pub(super) fn wait(_: Option<&impl Pipe>, _: Option<&impl Pipe>, timeout: std::time::Duration) { + std::thread::sleep(timeout); +} + fn read_result(result: io::Result) -> io::Result> { match result { Ok(count) => Ok(Some(count)), @@ -162,6 +198,39 @@ fn group_has_live_members(id: u32, _: std::time::Instant) -> io::Result { mod tests { use super::*; + #[cfg(unix)] + #[test] + fn an_idle_wait_ends_when_output_arrives_not_after_its_timeout() { + use std::io::Write as _; + use std::time::{Duration, Instant}; + let (reader, mut writer) = io::pipe().unwrap(); + let none: Option<&io::PipeReader> = None; + let start = Instant::now(); + wait(Some(&reader), none, Duration::from_millis(30)); + assert!( + start.elapsed() >= Duration::from_millis(25), + "no output: the whole wait" + ); + let late = std::thread::spawn(move || { + std::thread::sleep(Duration::from_millis(20)); + writer.write_all(b"x").unwrap(); + writer + }); + let start = Instant::now(); + wait(Some(&reader), none, Duration::from_secs(10)); + assert!(start.elapsed() < Duration::from_secs(5), "output ends the wait"); + drop(late.join().unwrap()); + let start = Instant::now(); + wait(Some(&reader), none, Duration::from_secs(10)); + assert!( + start.elapsed() < Duration::from_secs(5), + "readable data or a closed pipe ends it at once" + ); + let start = Instant::now(); + wait(none, none, Duration::from_millis(30)); + assert!(start.elapsed() >= Duration::from_millis(25), "no open pipe: a nap"); + } + #[test] fn nonblocking_reads_distinguish_pending_eof_data_and_failure() { assert_eq!(read_result(Ok(0)).unwrap(), Some(0)); diff --git a/docs/COMMAND_RUNNER.md b/docs/COMMAND_RUNNER.md index c802df96..f6d4ffe2 100644 --- a/docs/COMMAND_RUNNER.md +++ b/docs/COMMAND_RUNNER.md @@ -49,7 +49,11 @@ uses agy's local browser flow. the previous replacement behavior for invalid UTF-8. Nonzero exit codes return `CommandOutput { success: false, .. }` with the collected output. - The supplied timeout starts after successful process creation and covers both - pipe EOF and process exit. Idle checks sleep at most 25 ms. A timeout returns + pipe EOF and process exit. An idle check waits at most 25 ms: on Unix it + `poll`s the open pipes and returns as soon as one has output or closes, so a + writer that refills a small pipe (16 KiB on macOS) isn't read one fixed nap at + a time (#281); on Windows, whose anonymous pipes can't be polled, and once + both pipes have closed, it sleeps. A timeout returns `TimedOut` after cleanup; cleanup failure returns `Failed` instead of claiming that termination completed. Process creation itself is a native system call. - Verify owned tree termination for up to one additional second. Stop and reap