From ef8e46cf9aeb77c6a89e231a73bbc5337dd23e3b Mon Sep 17 00:00:00 2001 From: HemSoft Date: Tue, 27 Oct 2026 12:45:00 -0400 Subject: [PATCH 1/4] fix: wake the command runner when output arrives instead of after a fixed nap (#281) On macos-latest the scoped-parent 4 MiB + 1 overflow sometimes hit its 3 s deadline as TimedOut. Each time the reader drained the pipe faster than the writer refilled it, collect slept a fixed 25 ms, and on a busy 3-vCPU runner those naps and their wake-ups added up. On Unix an idle turn now polls the open pipes for up to the same 25 ms and returns as soon as one has output or closes. Windows, whose anonymous pipes can't be polled, keeps the nap. The timeout, the one chunk per stream per turn and the output limits are unchanged. Co-Authored-By: Claude Opus 5.5 --- crates/codexbar-providers/src/command.rs | 10 ++- crates/codexbar-providers/src/command/pipe.rs | 63 +++++++++++++++++++ docs/COMMAND_RUNNER.md | 6 +- 3 files changed, 77 insertions(+), 2 deletions(-) diff --git a/crates/codexbar-providers/src/command.rs b/crates/codexbar-providers/src/command.rs index 972fdcc1..7fe3e83f 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())), + ); } } } diff --git a/crates/codexbar-providers/src/command/pipe.rs b/crates/codexbar-providers/src/command/pipe.rs index baf765f4..2a323fde 100644 --- a/crates/codexbar-providers/src/command/pipe.rs +++ b/crates/codexbar-providers/src/command/pipe.rs @@ -57,6 +57,36 @@ pub(super) fn read(pipe: &mut impl Pipe, bytes: &mut [u8]) -> io::Result, second: Option<&impl Pipe>, timeout: std::time::Duration) { + let mut fds: Vec = [first.map(|p| p.as_raw_fd()), second.map(|p| p.as_raw_fd())] + .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 +192,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 From 10db032e921da7683f62dc3b14ee63a9554bfdad Mon Sep 17 00:00:00 2001 From: HemSoft Date: Tue, 27 Oct 2026 16:26:59 -0400 Subject: [PATCH 2/4] refactor: keep pipe::wait's branches in one non-generic function llvm-cov counts a generic function's branches per instantiation, so the runner's instantiation of wait() left the no-open-pipe nap uncovered and lowered branch coverage. The branches now live in poll_or_nap, which the unit test covers fully. Co-Authored-By: Claude Opus 5.5 --- crates/codexbar-providers/src/command/pipe.rs | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/crates/codexbar-providers/src/command/pipe.rs b/crates/codexbar-providers/src/command/pipe.rs index 2a323fde..b02d5ead 100644 --- a/crates/codexbar-providers/src/command/pipe.rs +++ b/crates/codexbar-providers/src/command/pipe.rs @@ -63,7 +63,13 @@ pub(super) fn read(pipe: &mut impl Pipe, bytes: &mut [u8]) -> io::Result, second: Option<&impl Pipe>, timeout: std::time::Duration) { - let mut fds: Vec = [first.map(|p| p.as_raw_fd()), second.map(|p| p.as_raw_fd())] + 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 { From 4fed893bbec9922d97a175106f4b24f996123a33 Mon Sep 17 00:00:00 2001 From: HemSoft Date: Tue, 27 Oct 2026 16:35:12 -0400 Subject: [PATCH 3/4] test: pin every way a command runner collection turn ends complete()'s stdout-closed, stderr-open case was covered only when a real command's EOFs happened to arrive in separate turns; with the poll wait they arrive together. A unit test now covers each outcome directly, so branch coverage no longer depends on timing. Co-Authored-By: Claude Opus 5.5 --- crates/codexbar-providers/src/command.rs | 32 ++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/crates/codexbar-providers/src/command.rs b/crates/codexbar-providers/src/command.rs index 7fe3e83f..fa707e1f 100644 --- a/crates/codexbar-providers/src/command.rs +++ b/crates/codexbar-providers/src/command.rs @@ -280,6 +280,38 @@ mod tests { ); } + /// 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"; From 1e7ca4586c3323e37eb08343491504e8013dcc01 Mon Sep 17 00:00:00 2001 From: HemSoft Date: Tue, 27 Oct 2026 17:01:54 -0400 Subject: [PATCH 4/4] test: pin every Capture::drain outcome on a real pipe With the poll wait both streams close in the same turn, so the already-closed branch was no longer reached by the fixtures on Linux. A unit test now covers pending output, data, overflow, EOF and a closed stream directly. Co-Authored-By: Claude Opus 5.5 --- crates/codexbar-providers/src/command.rs | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/crates/codexbar-providers/src/command.rs b/crates/codexbar-providers/src/command.rs index fa707e1f..5883b9fd 100644 --- a/crates/codexbar-providers/src/command.rs +++ b/crates/codexbar-providers/src/command.rs @@ -280,6 +280,27 @@ 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]