dicklesworthstone__coding_agent_session_search/tests/util/timeout.rs

//! Subprocess timeout-with-diagnostic wrapper (bead f2r5t / ibuuh.10.12).
//!
//! Motivation
//! ----------
//!
//! E2E test suites that spawn the `cass` binary via `assert_cmd` (or
//! `std::process::Command`) and call `.wait()` / `.assert().success()`
//! will block indefinitely when the child hangs. Under that failure
//! mode the test harness produces *no* useful output — just a silent
//! stall until the outer cargo-test timeout eventually kills the
//! runner. Operators and CI consumers are left reconstructing what
//! phase the child was in by guessing.
//!
//! This module provides [`spawn_with_timeout_or_diag`], a wrapper that
//!   1. spawns a command and polls [`std::process::Child::try_wait`],
//!   2. on success returns the normal [`std::process::Output`],
//!   3. on timeout emits a structured diagnostic dump to stderr
//!      (label, child PID, elapsed, optional `data_dir` listing, last
//!      N bytes of stdout / stderr), kills the child, and panics with
//!      a clear message.
//!
//! Tests in `tests/e2e_large_dataset.rs` and similar long-running
//! suites can swap `.assert().success()` for this wrapper to convert
//! a silent hang into a loud, diagnosable failure.

use std::io::Read;
use std::path::{Path, PathBuf};
use std::process::{Command, Output, Stdio};
use std::thread::JoinHandle;
use std::time::{Duration, Instant};

/// How often we poll the child's exit status while waiting.
const POLL_INTERVAL: Duration = Duration::from_millis(50);

/// How much of the child's streamed stdout/stderr to keep for the
/// timeout-diagnostic dump. Bounded so the dump stays readable and a
/// misbehaving child that spams output doesn't blow up test memory.
const DIAG_STREAM_TAIL_BYTES: usize = 16 * 1024;
const MAX_CAPTURE_BYTES: usize = 64 * 1024 * 1024;

struct PipeCapture {
    bytes: Vec<u8>,
    truncated: bool,
}

impl PipeCapture {
    fn empty() -> Self {
        Self {
            bytes: Vec::new(),
            truncated: false,
        }
    }
}

/// Spawn `cmd` and wait for it to finish, up to `timeout`. On timeout,
/// emit a diagnostic dump, kill the child, and panic.
///
/// `label` identifies the phase in the diagnostic dump so future-you
/// can tell at a glance what was hung.
///
/// `data_dir`, when supplied, is recursively listed (up to a bounded
/// number of entries) in the diagnostic dump so the reader can see
/// which index / lock / checkpoint files were or were not present at
/// the moment of the hang.
///
/// Stdin is closed (`Stdio::null()`) so the child never blocks on a
/// non-existent operator. Stdout and stderr are captured and included
/// (tail-clipped) in the diagnostic dump.
#[allow(dead_code)]
pub fn spawn_with_timeout_or_diag(
    mut cmd: Command,
    label: &str,
    data_dir: Option<&Path>,
    timeout: Duration,
) -> Output {
    cmd.stdin(Stdio::null());
    cmd.stdout(Stdio::piped());
    cmd.stderr(Stdio::piped());
    configure_child_process_group(&mut cmd);

    let start = Instant::now();
    let mut child = cmd
        .spawn()
        .unwrap_or_else(|err| panic!("spawn_with_timeout_or_diag({label}): spawn failed: {err}"));
    // 1j6x8: if the harness itself is killed (`timeout N cargo test`,
    // SIGKILL), nothing below runs and the child's process group would
    // outlive us. The reaper is an independent process that notices our
    // death and kills the group on our behalf.
    let mut reaper = spawn_harness_death_reaper(child.id());
    let stdout_reader = spawn_pipe_reader(child.stdout.take());
    let stderr_reader = spawn_pipe_reader(child.stderr.take());

    let deadline = start + timeout;
    loop {
        match child.try_wait() {
            Ok(Some(status)) => {
                kill_child_process_group(child.id());
                retire_harness_death_reaper(&mut reaper);
                let stdout = join_pipe_reader(stdout_reader, label, "stdout");
                let stderr = join_pipe_reader(stderr_reader, label, "stderr");
                let stdout = full_output_or_panic(stdout, label, "stdout");
                let stderr = full_output_or_panic(stderr, label, "stderr");
                return Output {
                    status,
                    stdout,
                    stderr,
                };
            }
            Ok(None) => {
                if Instant::now() >= deadline {
                    let pid = child.id();
                    // Kill FIRST so the stdout/stderr pipe FDs close
                    // on the child's side and the reader-thread joins
                    // below return. Waiting on a pipe whose writer is
                    // still alive but idle would otherwise block the
                    // diagnostic dump forever and defeat this helper.
                    kill_child_process_group(pid);
                    let _ = child.kill();
                    let _ = child.wait();
                    retire_harness_death_reaper(&mut reaper);
                    let stdout = join_pipe_reader(stdout_reader, label, "stdout");
                    let stderr = join_pipe_reader(stderr_reader, label, "stderr");
                    let stdout_tail = stream_tail(&stdout.bytes);
                    let stderr_tail = stream_tail(&stderr.bytes);

                    eprintln!();
                    eprintln!("================================================================");
                    eprintln!(
                        "TIMEOUT DIAGNOSTIC: phase={label:?} pid={pid} elapsed_ms={} timeout_ms={}",
                        start.elapsed().as_millis(),
                        timeout.as_millis(),
                    );
                    eprintln!("================================================================");
                    if let Some(dir) = data_dir {
                        eprintln!("--- data_dir listing ({}):", dir.display());
                        for entry in list_dir_bounded(dir, 200) {
                            eprintln!("  {entry}");
                        }
                    }
                    if stdout.truncated {
                        eprintln!(
                            "--- child stdout exceeded capture cap ({} bytes); retained latest bytes only",
                            MAX_CAPTURE_BYTES
                        );
                    }
                    eprintln!(
                        "--- child stdout tail ({} bytes of up to {}):",
                        stdout_tail.len(),
                        DIAG_STREAM_TAIL_BYTES
                    );
                    eprintln!("{}", String::from_utf8_lossy(&stdout_tail));
                    if stderr.truncated {
                        eprintln!(
                            "--- child stderr exceeded capture cap ({} bytes); retained latest bytes only",
                            MAX_CAPTURE_BYTES
                        );
                    }
                    eprintln!(
                        "--- child stderr tail ({} bytes of up to {}):",
                        stderr_tail.len(),
                        DIAG_STREAM_TAIL_BYTES
                    );
                    eprintln!("{}", String::from_utf8_lossy(&stderr_tail));
                    eprintln!("================================================================");

                    panic!(
                        "subprocess phase {label:?} exceeded timeout of {:?} (see stderr \
                         diagnostic above)",
                        timeout
                    );
                }
                std::thread::sleep(POLL_INTERVAL);
            }
            Err(err) => {
                // try_wait errored — treat as a hard failure.
                kill_child_process_group(child.id());
                let _ = child.kill();
                let _ = child.wait();
                retire_harness_death_reaper(&mut reaper);
                let _ = join_pipe_reader(stdout_reader, label, "stdout");
                let _ = join_pipe_reader(stderr_reader, label, "stderr");
                panic!("spawn_with_timeout_or_diag({label}): try_wait errored: {err}");
            }
        }
    }
}

#[cfg(unix)]
fn configure_child_process_group(cmd: &mut Command) {
    use std::os::unix::process::CommandExt;

    cmd.process_group(0);
}

#[cfg(not(unix))]
fn configure_child_process_group(_cmd: &mut Command) {}

#[cfg(unix)]
fn kill_child_process_group(pid: u32) {
    let process_group = format!("-{pid}");
    // End option parsing explicitly: the negative operand is a process group,
    // not another signal option. The same rule applies to the detached reaper.
    let _ = Command::new("/bin/kill")
        .args(["-KILL", "--", &process_group])
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .status();
}

#[cfg(not(unix))]
fn kill_child_process_group(_pid: u32) {}

/// 1j6x8: spawn a detached reaper that kills `child_pid`'s process group
/// once the current (harness) process is gone. The reaper lives in its own
/// process group so a harness-group kill does not take it down first; it
/// exits on its own as soon as either the harness or the child is gone, so
/// a normal run leaves nothing behind. Pure `std::process` — no `unsafe`,
/// no `prctl`. Unix only; elsewhere returns `None`.
#[cfg(unix)]
fn spawn_harness_death_reaper(child_pid: u32) -> Option<std::process::Child> {
    spawn_death_reaper_for(std::process::id(), child_pid)
}

#[cfg(unix)]
fn spawn_death_reaper_for(watched_pid: u32, child_pid: u32) -> Option<std::process::Child> {
    use std::os::unix::process::CommandExt;
    // Loop while both are alive; afterwards kill the group only if the CHILD
    // is the one still alive (the harness died). If the child is already
    // gone, do nothing — its pid could have been reused by an unrelated
    // process group.
    const REAPER_SCRIPT: &str = "while kill -0 \"$1\" 2>/dev/null && kill -0 \"$2\" 2>/dev/null; do sleep 1; done; if kill -0 \"$2\" 2>/dev/null; then /bin/kill -KILL -- \"-$2\" 2>/dev/null; fi";
    let mut cmd = Command::new("/bin/sh");
    cmd.arg("-c")
        .arg(REAPER_SCRIPT)
        .arg("cass-e2e-reaper")
        .arg(watched_pid.to_string())
        .arg(child_pid.to_string())
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::null());
    cmd.process_group(0);
    cmd.spawn().ok()
}

#[cfg(not(unix))]
fn spawn_harness_death_reaper(_child_pid: u32) -> Option<std::process::Child> {
    None
}

/// The child is already reaped by the harness: stop the reaper (it would
/// exit within a second on its own) and collect it.
fn retire_harness_death_reaper(reaper: &mut Option<std::process::Child>) {
    if let Some(mut reaper) = reaper.take() {
        let _ = reaper.kill();
        let _ = reaper.wait();
    }
}

fn spawn_pipe_reader<R>(handle: Option<R>) -> JoinHandle<PipeCapture>
where
    R: Read + Send + 'static,
{
    std::thread::spawn(move || {
        let Some(mut handle) = handle else {
            return PipeCapture::empty();
        };
        read_pipe_to_capture(&mut handle)
    })
}

fn join_pipe_reader(
    reader: JoinHandle<PipeCapture>,
    label: &str,
    stream_name: &str,
) -> PipeCapture {
    match reader.join() {
        Ok(capture) => capture,
        Err(_) => {
            eprintln!("{label}: {stream_name} reader thread panicked");
            PipeCapture::empty()
        }
    }
}

fn full_output_or_panic(capture: PipeCapture, label: &str, stream_name: &str) -> Vec<u8> {
    if capture.truncated {
        panic!(
            "spawn_with_timeout_or_diag({label}): {stream_name} exceeded capture cap of {} bytes",
            MAX_CAPTURE_BYTES
        );
    }
    capture.bytes
}

fn read_pipe_to_capture<R: Read>(handle: &mut R) -> PipeCapture {
    let mut bytes = Vec::new();
    let mut scratch = [0_u8; 8192];
    let mut write_pos = 0_usize;
    let mut truncated = false;

    loop {
        let read = match handle.read(&mut scratch) {
            Ok(0) => break,
            Ok(read) => read,
            Err(_) => break,
        };

        if bytes.len() < MAX_CAPTURE_BYTES {
            let remaining = MAX_CAPTURE_BYTES - bytes.len();
            let keep = read.min(remaining);
            bytes.extend_from_slice(&scratch[..keep]);
            if keep == read {
                continue;
            }
            truncated = true;
            for &byte in &scratch[keep..read] {
                bytes[write_pos] = byte;
                write_pos = (write_pos + 1) % MAX_CAPTURE_BYTES;
            }
        } else {
            truncated = true;
            for &byte in &scratch[..read] {
                bytes[write_pos] = byte;
                write_pos = (write_pos + 1) % MAX_CAPTURE_BYTES;
            }
        }
    }

    if truncated && write_pos != 0 {
        let mut ordered = Vec::with_capacity(bytes.len());
        ordered.extend_from_slice(&bytes[write_pos..]);
        ordered.extend_from_slice(&bytes[..write_pos]);
        bytes = ordered;
    }

    PipeCapture { bytes, truncated }
}

fn stream_tail(buf: &[u8]) -> Vec<u8> {
    if buf.len() > DIAG_STREAM_TAIL_BYTES {
        let tail_start = buf.len() - DIAG_STREAM_TAIL_BYTES;
        buf[tail_start..].to_vec()
    } else {
        buf.to_vec()
    }
}

/// Produce up to `limit` relative entries under `root`, formatted as
/// `path (size_bytes)` for files and `path/` for directories. Silently
/// ignores I/O errors so a diagnostic dump never turns into its own
/// second-order failure.
fn list_dir_bounded(root: &Path, limit: usize) -> Vec<String> {
    let mut out: Vec<String> = Vec::new();
    let mut stack: Vec<PathBuf> = vec![root.to_path_buf()];
    while let Some(dir) = stack.pop() {
        let Ok(read) = std::fs::read_dir(&dir) else {
            continue;
        };
        for entry in read.flatten() {
            if out.len() >= limit {
                out.push(format!("  ... (truncated at {limit} entries)"));
                return out;
            }
            let path = entry.path();
            let rel = path
                .strip_prefix(root)
                .unwrap_or(&path)
                .to_string_lossy()
                .into_owned();
            match std::fs::symlink_metadata(&path) {
                Ok(metadata) if metadata.file_type().is_symlink() => {
                    out.push(format!("{rel} (symlink)"));
                }
                Ok(metadata) if metadata.is_dir() => {
                    out.push(format!("{rel}/"));
                    stack.push(path);
                }
                Ok(metadata) => out.push(format!("{rel} ({} bytes)", metadata.len())),
                Err(_) => {
                    out.push(format!("{rel} (<stat failed>)"));
                }
            }
        }
    }
    out
}

#[cfg(test)]
mod tests {
    /// 1j6x8: with a watched pid that is not alive, the reaper must kill the
    /// child's process group promptly instead of leaving it orphaned.
    #[cfg(unix)]
    #[test]
    fn death_reaper_kills_child_group_when_watched_process_is_gone() -> Result<(), String> {
        use std::os::unix::process::CommandExt;
        let mut sleeper = std::process::Command::new("/bin/sh");
        sleeper
            .args(["-c", "sleep 30"])
            .stdin(std::process::Stdio::null())
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null());
        sleeper.process_group(0);
        let mut child = sleeper.spawn().map_err(|e| format!("spawn sleeper: {e}"))?;
        // Observe a process we actually started and reaped. A guessed PID near
        // a platform's maximum is not evidence that the process is absent.
        let mut watched = std::process::Command::new("/bin/sh")
            .args(["-c", "exit 0"])
            .spawn()
            .map_err(|e| format!("spawn watched process: {e}"))?;
        let dead_pid = watched.id();
        watched
            .wait()
            .map_err(|e| format!("wait watched process: {e}"))?;
        let mut reaper = super::spawn_death_reaper_for(dead_pid, child.id())
            .ok_or_else(|| "spawn reaper".to_string())?;
        let started = std::time::Instant::now();
        let mut exited = false;
        while started.elapsed() < std::time::Duration::from_secs(10) {
            if child
                .try_wait()
                .map_err(|e| format!("try_wait: {e}"))?
                .is_some()
            {
                exited = true;
                break;
            }
            std::thread::sleep(std::time::Duration::from_millis(100));
        }
        let _ = child.kill();
        let _ = child.wait();
        let _ = reaper.wait();
        if !exited {
            return Err("reaper did not kill the orphaned child within 10s".to_string());
        }
        Ok(())
    }

    use super::*;

    /// Proves the happy path: a fast-exiting child returns Output
    /// normally with no panic and no diagnostic noise.
    #[test]
    fn happy_path_returns_output_without_panicking() {
        let mut cmd = Command::new("/bin/sh");
        cmd.arg("-c").arg("printf 'hi' && exit 0");
        let out = spawn_with_timeout_or_diag(cmd, "happy_path", None, Duration::from_secs(5));
        assert!(out.status.success(), "shell must exit 0");
        assert_eq!(
            String::from_utf8_lossy(&out.stdout).trim(),
            "hi",
            "stdout must round-trip"
        );
    }

    /// Proves stdout is drained while the child is still running. A
    /// child that writes more than the kernel pipe buffer can otherwise
    /// block before exit and get misdiagnosed as a timeout.
    #[test]
    fn large_stdout_child_can_exit_without_pipe_deadlock() {
        let mut cmd = Command::new("/bin/sh");
        cmd.arg("-c").arg("yes x | head -c 1048576");
        let out = spawn_with_timeout_or_diag(cmd, "large_stdout", None, Duration::from_secs(10));
        assert!(out.status.success(), "large stdout child must exit 0");
        assert_eq!(out.stdout.len(), 1024 * 1024);
    }

    /// Proves the timeout path: a child that hangs past the deadline
    /// triggers the diagnostic-dump + kill + panic sequence.
    ///
    /// Use a shell wrapper on Unix so this also proves process-group
    /// cleanup: killing only the shell would leave `sleep` holding the
    /// stdout/stderr pipe FDs open, deadlocking the diagnostic join.
    #[test]
    fn hung_child_triggers_timeout_panic_with_diagnostic() {
        let started = Instant::now();
        let result = std::panic::catch_unwind(|| {
            #[cfg(unix)]
            let cmd = {
                let mut cmd = Command::new("/bin/sh");
                cmd.arg("-c").arg("sleep 30");
                cmd
            };
            #[cfg(not(unix))]
            let mut cmd = Command::new("/bin/sleep");
            #[cfg(not(unix))]
            cmd.arg("30");
            spawn_with_timeout_or_diag(cmd, "intentional_hang", None, Duration::from_millis(300))
        });
        let panic = result.expect_err("a stalled child must time out");
        let message = panic
            .downcast_ref::<String>()
            .map(String::as_str)
            .or_else(|| panic.downcast_ref::<&str>().copied())
            .unwrap_or_default();
        assert!(
            message.contains("exceeded timeout"),
            "unexpected panic: {message}"
        );
        // Previously should_panic accepted a 30-second pipe-drain stall after a
        // 300ms timeout. Allow CI scheduling slack, but reject that false pass.
        assert!(
            started.elapsed() < Duration::from_secs(10),
            "timeout cleanup outlived its deadline: {:?}",
            started.elapsed()
        );
    }

    #[cfg(unix)]
    #[test]
    fn exited_child_does_not_leave_descendant_output_pipes_open() {
        let mut cmd = Command::new("/bin/sh");
        cmd.args(["-c", "sleep 30 & printf ready; exit 0"]);
        let started = Instant::now();
        let output =
            spawn_with_timeout_or_diag(cmd, "orphaned_pipes", None, Duration::from_secs(2));
        assert!(output.status.success());
        assert_eq!(output.stdout, b"ready");
        assert!(
            started.elapsed() < Duration::from_secs(10),
            "an exited leader left a descendant holding its pipes: {:?}",
            started.elapsed()
        );
    }

    /// Proves the diagnostic dump includes the data_dir listing when
    /// one is supplied — even though we don't capture stderr of the
    /// panicking test, asserting `should_panic` with the right message
    /// plus eyeballing the dump locally is enough signal here. We also
    /// exercise list_dir_bounded directly so its happy-path shape is
    /// covered.
    #[test]
    fn list_dir_bounded_reports_files_and_dirs() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let root = tmp.path();
        std::fs::write(root.join("a.txt"), b"hello").unwrap();
        std::fs::create_dir(root.join("sub")).unwrap();
        std::fs::write(root.join("sub/b.bin"), b"0123456789").unwrap();

        let entries = list_dir_bounded(root, 200);
        assert!(
            entries.iter().any(|e| e.starts_with("a.txt (5 bytes)")),
            "expected a.txt entry with size; got: {entries:?}"
        );
        assert!(
            entries.iter().any(|e| e == "sub/"),
            "expected sub/ directory entry; got: {entries:?}"
        );
        assert!(
            entries
                .iter()
                .any(|e| e.starts_with("sub/b.bin (10 bytes)")),
            "expected nested file entry with size; got: {entries:?}"
        );
    }

    /// Proves the `limit` cap triggers the truncation marker so the
    /// dump never unbounded-grows on a pathological data_dir.
    #[test]
    fn list_dir_bounded_truncates_at_limit() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let root = tmp.path();
        for i in 0..10 {
            std::fs::write(root.join(format!("f-{i:02}.txt")), b"x").unwrap();
        }
        let entries = list_dir_bounded(root, 3);
        assert!(
            entries.iter().any(|e| e.contains("truncated at 3")),
            "must include the truncated-at marker once limit is exceeded; got: {entries:?}"
        );
    }

    #[cfg(unix)]
    #[test]
    fn list_dir_bounded_reports_symlinked_directories_without_following() {
        let tmp = tempfile::tempdir().expect("tempdir");
        let root = tmp.path().join("root");
        let outside = tmp.path().join("outside");
        std::fs::create_dir_all(&root).expect("create root");
        std::fs::create_dir_all(&outside).expect("create outside");
        std::fs::write(outside.join("outside-only.txt"), b"should not be listed")
            .expect("write outside marker");
        std::os::unix::fs::symlink(&outside, root.join("linked-outside"))
            .expect("create symlinked directory");

        let entries = list_dir_bounded(&root, 20);

        assert!(
            entries
                .iter()
                .any(|entry| entry == "linked-outside (symlink)"),
            "diagnostic listing must identify the symlink itself: {entries:?}"
        );
        assert!(
            entries
                .iter()
                .all(|entry| !entry.contains("outside-only.txt")),
            "diagnostic listing must not follow symlinked directories outside the data dir: {entries:?}"
        );
    }
}