fix(test): remove two CI flakes — EPIPE on early-exit stdin, and an over-tight async watchdog [DEV-150, DEV-151]

libarchive_oxide-cli/tests/common/mod.rs

@@ -91,8 +91,15 @@ pub(crate) fn run(bin_name: &str, args: &[&str]) -> Output {
 }
 
 /// Runs `bin` in `cwd`, feeding `stdin_bytes` to stdin, returning the captured [`Output`].
+///
+/// A binary that rejects its arguments — e.g. `-` refused as a usage error — exits
+/// *without* reading stdin and closes the pipe, so the parent's write races the child's
+/// exit: it succeeds if the bytes fit the pipe buffer first, and fails with `BrokenPipe`
+/// (EPIPE) otherwise. That race is inherent to the behavior under test, so a broken pipe
+/// is an expected outcome here rather than a test failure; every other write error still
+/// fails loudly. Stdin is always closed before waiting so the child sees EOF.
 pub(crate) fn run_stdin(bin_name: &str, args: &[&str], cwd: &Path, stdin_bytes: &[u8]) -> Output {
-    use std::io::Write;
+    use std::io::{ErrorKind, Write};
     let mut child = Command::new(bin(bin_name))
         .args(args)
         .current_dir(cwd)
@@ -101,12 +108,14 @@ pub(crate) fn run_stdin(bin_name: &str, args: &[&str], cwd: &Path, stdin_bytes:
         .stderr(Stdio::piped())
         .spawn()
         .expect("spawn bin");
-    child
-        .stdin
-        .take()
-        .expect("stdin")
-        .write_all(stdin_bytes)
-        .expect("write stdin");
+    {
+        let mut stdin = child.stdin.take().expect("stdin");
+        match stdin.write_all(stdin_bytes) {
+            Ok(()) => {},
+            Err(error) if error.kind() == ErrorKind::BrokenPipe => {},
+            Err(error) => panic!("write stdin: {error}"),
+        }
+    }
     child.wait_with_output().expect("wait")
 }
 

libarchive_oxide/tests/async_stream_v2.rs

@@ -195,21 +195,36 @@ fn park_block_on<F: Future>(future: F) -> F::Output {
     }
 }
 
+/// Wall-clock budget for [`arm_watchdog`], deliberately generous.
+///
+/// A lost wakeup parks the [`park_block_on`] driver *immediately* — it has nothing else
+/// to poll — so a real deadlock trips this watchdog long before the budget is reached;
+/// the budget's only job is to exceed the honest runtime of the slowest runner. Emulated
+/// big-endian s390x under qemu and a loaded macOS runner are an order of magnitude
+/// slower than a native `x86_64` host, and a budget tight enough to matter there turns
+/// *slowness* into a false "deadlock" abort. It still sits well inside the job's
+/// `timeout-minutes`, so a genuine hang fails fast and attributably rather than
+/// consuming the whole job budget.
+const WATCHDOG_BUDGET: Duration = Duration::from_secs(240);
+
 /// Arms a watchdog thread that aborts the process if `label` has not signaled
-/// completion within ~30s, so a deadlocked poll fails loudly instead of hanging
-/// the test runner indefinitely.
+/// completion within [`WATCHDOG_BUDGET`], so a deadlocked poll fails loudly instead of
+/// hanging the test runner indefinitely.
 fn arm_watchdog(label: &'static str) -> Arc<AtomicBool> {
     let done = Arc::new(AtomicBool::new(false));
     let flag = Arc::clone(&done);
     thread::spawn(move || {
-        let deadline = Instant::now() + Duration::from_secs(30);
+        let deadline = Instant::now() + WATCHDOG_BUDGET;
         while Instant::now() < deadline {
             if flag.load(Ordering::SeqCst) {
                 return;
             }
             thread::sleep(Duration::from_millis(50));
         }
-        eprintln!("watchdog: `{label}` did not complete within 30s; aborting the deadlocked test");
+        eprintln!(
+            "watchdog: `{label}` did not complete within {}s; aborting the deadlocked test",
+            WATCHDOG_BUDGET.as_secs()
+        );
         std::process::abort();
     });
     done
@@ -228,7 +243,13 @@ fn async_xz_never_deadlocks_under_slow_source() {
     let expected = collect_sync(fixture());
     let archive = sync_filtered_fixture(FilterId::Xz);
     guarded("async_xz_never_deadlocks_under_slow_source", async {
-        for iteration in 0..50 {
+        // The park/unpark driver makes a lost wakeup deterministic: the executor has
+        // nothing else to poll, so a missed wake parks forever on the very first
+        // iteration. Repeats only re-exercise the NeedInput/Output interleaving, so a
+        // handful is as strong a regression guard as a large count — and a large count
+        // pushed the honest runtime past the watchdog budget on emulated (qemu s390x)
+        // and loaded runners, turning slowness into a false deadlock report.
+        for iteration in 0..8 {
             assert_eq!(
                 collect_futures(archive.clone()).await,
                 expected,