From 26d26e15c989d9e0a2aad5332b0a5f8ccf2a8013 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Thu, 27 Aug 2026 10:45:17 -0400 Subject: [PATCH] =?UTF-8?q?fix(exec):=20bounded=20post-kill=20reaps=20?= =?UTF-8?q?=E2=80=94=20D-state=20children=20can't=20wedge=20a=20turn?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Port of upstream grok-build 1.0.1 (landed upstream 2026-08-06). A child wedged in an uninterruptible kernel syscall (D-state) never observes SIGKILL, and a background descendant can hold the output pipes open past the kill; both previously stalled tool futures without bound. - kigi-tty-utils: KILL_REAP_TIMEOUT (5s) + reap_killed_bounded — wait for an already-killed child at most the bound, then abandon the corpse to tokio's orphan reaper. - local_terminal: pipe readers share their buffer with the spawner and join under the reap bound on both exit arms (aborted readers are awaited so no tail bytes race the take); the whole tree is enrolled in a ProcessGroup and group-killed on timeout; the timeout arm returns the synthetic result instead of waiting on the corpse. Grandchild kill + prompt-return pinned by a ported unix regression test. - kigi-tools: reap_killed_search_child wrapper; grep's four kill sites reap bounded; glob's stderr goes to null (an unread pipe blocks rg on warning backpressure) and its truncation kill reaps bounded. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01KSy9S31pZNWXtRZNVJ1sDf --- .../kigi-shell/src/terminal/local_terminal.rs | 174 ++++++++++++++++-- .../src/implementations/kigi/grep/mod.rs | 21 ++- .../src/implementations/opencode/glob/mod.rs | 16 +- crates/codegen/kigi-tools/src/util/mod.rs | 1 + crates/codegen/kigi-tools/src/util/spawn.rs | 16 ++ crates/codegen/kigi-tty-utils/Cargo.toml | 2 +- crates/codegen/kigi-tty-utils/src/lib.rs | 30 +++ 7 files changed, 228 insertions(+), 32 deletions(-) diff --git a/crates/codegen/kigi-shell/src/terminal/local_terminal.rs b/crates/codegen/kigi-shell/src/terminal/local_terminal.rs index 268096d..7e3ba0f 100644 --- a/crates/codegen/kigi-shell/src/terminal/local_terminal.rs +++ b/crates/codegen/kigi-shell/src/terminal/local_terminal.rs @@ -8,13 +8,66 @@ use tokio::time; use crate::terminal::runner::{ AsyncTerminalRunner, TerminalError, TerminalRunRequest, TerminalRunResult, }; +use kigi_tty_utils::KILL_REAP_TIMEOUT; pub struct LocalTerminalRunner; -async fn read_stream(mut stream: impl AsyncReadExt + Unpin) -> Vec { - let mut buffer = Vec::new(); - let _ = stream.read_to_end(&mut buffer).await; - buffer +/// A pipe reader whose buffer is shared with the spawner, so bytes read +/// before an abort are not lost with the task. +struct PipeReader { + task: tokio::task::JoinHandle<()>, + buffer: std::sync::Arc>>, +} + +impl PipeReader { + fn spawn(mut stream: impl AsyncReadExt + Unpin + Send + 'static) -> Self { + let buffer = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let task_buffer = std::sync::Arc::clone(&buffer); + let task = tokio::spawn(async move { + let mut chunk = [0u8; 8192]; + loop { + match stream.read(&mut chunk).await { + Ok(0) | Err(_) => break, + Ok(n) => Self::lock_buffer(&task_buffer).extend_from_slice(&chunk[..n]), + } + } + }); + Self { task, buffer } + } + + /// Never panics: a poisoned lock just means a holder panicked mid-append — + /// the buffer is plain bytes and stays consistent, so recover it. + fn lock_buffer(buffer: &std::sync::Mutex>) -> std::sync::MutexGuard<'_, Vec> { + buffer + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + + fn take_buffer(self) -> Vec { + std::mem::take(&mut Self::lock_buffer(&self.buffer)) + } + + /// Wait for EOF (pipe closed by all writers) at most [`KILL_REAP_TIMEOUT`], + /// then return everything read so far. + /// + /// Callers only join after the child was reaped or killed, so EOF is + /// normally immediate; the bound is a backstop for writers that outlive + /// the command and keep the pipe open — a background descendant + /// (`printf hi; sleep 30 &`) or a process wedged in uninterruptible + /// kernel I/O (D-state), which not even SIGKILL moves. + async fn join_bounded(mut self) -> Vec { + if time::timeout(KILL_REAP_TIMEOUT, &mut self.task) + .await + .is_err() + { + self.task.abort(); + // Await the cancellation: abort() is asynchronous, and a reader + // past its await point could otherwise append between the take + // below and its own completion, losing those bytes. + let _ = (&mut self.task).await; + } + self.take_buffer() + } } /// Truncate buffer to keep only the last `limit` bytes (drops oldest bytes). @@ -66,12 +119,35 @@ impl AsyncTerminalRunner for LocalTerminalRunner { // Detach from the controlling terminal so child processes // (e.g. GPG pinentry) cannot open /dev/tty and corrupt the TUI. + // (This also makes the child its own session/group leader, which the + // ProcessGroup attach below relies on.) kigi_tools::util::detach_command(&mut cmd); + // process-group-killed on the request timeout; an unreapable (D-state) + // child is abandoned to the runtime's orphan reaper let mut child = cmd .spawn() .map_err(|e| TerminalError::Other(format!("Failed to start shell: {e}")))?; + // Own the whole tree, not just the direct shell: on timeout the group + // is killed so grandchildren can't keep running (and can't hold the + // output pipes open past the kill). Group teardown is an enhancement + // over the direct-kill floor, so both enrollment failures degrade + // softly instead of dropping an already-running shell. + let process_group = match kigi_tty_utils::ProcessGroup::new() { + Ok(mut group) => { + if let Err(e) = group.attach(&child) { + // e.g. the child already exited. + tracing::debug!("ProcessGroup::attach failed: {e}"); + } + Some(group) + } + Err(e) => { + tracing::warn!("Failed to create process group: {e}"); + None + } + }; + let stdout = child .stdout .take() @@ -81,28 +157,46 @@ impl AsyncTerminalRunner for LocalTerminalRunner { .take() .ok_or_else(|| TerminalError::Other("Failed to capture stderr".into()))?; - let stdout_task = tokio::spawn(read_stream(stdout)); - let stderr_task = tokio::spawn(read_stream(stderr)); + let stdout_reader = PipeReader::spawn(stdout); + let stderr_reader = PipeReader::spawn(stderr); let mut timed_out = false; let wait_result = time::timeout(request.timeout, child.wait()).await; - let exit_status = match wait_result { + let exit_code = match wait_result { Ok(status_res) => status_res - .map_err(|e| TerminalError::Other(format!("Failed to wait for process: {e}")))?, + .map_err(|e| TerminalError::Other(format!("Failed to wait for process: {e}")))? + .code(), Err(_) => { timed_out = true; + // Kill the direct child AND the whole group (grandchildren), + // then return the synthetic timeout result without waiting + // for the corpse: `timed_out: true` already tells the caller + // everything, and a D-state child would never become reapable + // anyway. Both kills are unconditional — `kill()` on a group + // whose attach failed is a silent no-op, so the direct + // `start_kill` is the guaranteed floor (same as + // `util/subprocess.rs`). The kills close the pipes, so the + // bounded joins below return immediately in the normal case; + // the abandoned child goes to tokio's orphan reaper. if let Err(e) = child.start_kill() { tracing::warn!("Failed to kill timed-out process: {e}"); } - child.wait().await.map_err(|e| { - TerminalError::Other(format!("Failed to wait for killed process: {e}")) - })? + if let Some(group) = &process_group + && let Err(e) = group.kill() + { + tracing::warn!("Failed to kill timed-out process group: {e}"); + } + None } }; - let stdout_result = stdout_task.await.unwrap_or_else(|_| Vec::new()); - let stderr_result = stderr_task.await.unwrap_or_else(|_| Vec::new()); + // Join concurrently (worst case one reap bound, not one per pipe). + // Bounded on the natural-exit arm too: a background descendant that + // outlives the shell (`printf hi; sleep 30 &`) holds the pipes open + // past the child's exit, and an unbounded join stalls on it. + let (stdout_result, stderr_result) = + tokio::join!(stdout_reader.join_bounded(), stderr_reader.join_bounded()); let mut combined = stdout_result; combined.extend(stderr_result); @@ -112,7 +206,7 @@ impl AsyncTerminalRunner for LocalTerminalRunner { Ok(TerminalRunResult { combined_output, - exit_code: exit_status.code(), + exit_code, truncated, signal: None, timed_out, @@ -181,4 +275,56 @@ mod tests { assert_eq!(result.combined_output.trim(), "hello"); assert_eq!(result.exit_code, Some(0)); } + + /// Timing out a command whose grandchild inherited the output pipes must + /// (a) return promptly — before the group kill, the joins waited for the + /// grandchild's EOF (the full sleep here; forever for a writer wedged in + /// uninterruptible kernel I/O, the production incident this guards + /// against) — and (b) actually kill the grandchild via the group kill, + /// not merely stop waiting for it. + #[tokio::test] + #[cfg(unix)] + async fn test_timeout_kills_grandchildren_and_returns_promptly() { + let mut request = make_request("sleep 5 & echo bgpid=$!; sleep 5"); + request.timeout = std::time::Duration::from_millis(300); + + let started = std::time::Instant::now(); + let result = LocalTerminalRunner.run(request).await.unwrap(); + + assert!(result.timed_out, "run should report the timeout"); + // Prompt return: request timeout (0.3s) + pipe EOF from the group + // kill (normally instant; KILL_REAP_TIMEOUT-bounded worst case) — + // anything near the 5s sleeps means we waited for the grandchild. + assert!( + started.elapsed() < std::time::Duration::from_secs(4), + "timeout path must not wait for the killed tree (took {:?})", + started.elapsed() + ); + + // Outcome: the background grandchild must actually be dead, proving + // the group kill reached it. Poll briefly — the SIGKILL is delivered + // synchronously but init may reap the orphan a beat later. + let bg_pid = result + .combined_output + .lines() + .find_map(|l| l.trim().strip_prefix("bgpid=")) + .and_then(|p| p.trim().parse::().ok()) + .expect("background pid should have been echoed before the kill"); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + loop { + let alive = std::process::Command::new("kill") + .args(["-0", &bg_pid.to_string()]) + .status() + .map(|s| s.success()) + .unwrap_or(false); + if !alive { + break; + } + assert!( + std::time::Instant::now() < deadline, + "background grandchild {bg_pid} survived the group kill" + ); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + } } diff --git a/crates/codegen/kigi-tools/src/implementations/kigi/grep/mod.rs b/crates/codegen/kigi-tools/src/implementations/kigi/grep/mod.rs index c05aaad..6692cc5 100644 --- a/crates/codegen/kigi-tools/src/implementations/kigi/grep/mod.rs +++ b/crates/codegen/kigi-tools/src/implementations/kigi/grep/mod.rs @@ -358,18 +358,20 @@ impl kigi_tool_runtime::Tool for GrepTool { tracing::Span::current().record("wall_ms", started.elapsed().as_millis() as u64); tracing::warn!(timeout_secs = timeout.as_secs(), "grep timed out"); let _ = child.start_kill(); - let _ = child.wait().await; + crate::util::reap_killed_search_child(&mut child).await; return Ok(grep_timeout_output(timeout.as_secs())); } }; - // When `stdout_truncated`, `rg` is already killed (inside the timeout - // block, before the stderr drain); this only reaps it. - let status = child.wait().await.ok(); + // Truncated output means `rg` was already killed (inside the timeout + // block, before the stderr drain) — bounded reap, and the exit code is + // defined as 0. A natural EOF means rg is exiting, so the plain wait + // is prompt. let exit_code = if stdout_truncated { + crate::util::reap_killed_search_child(&mut child).await; 0 } else { - status.and_then(|s| s.code()).unwrap_or(-1) + child.wait().await.ok().and_then(|s| s.code()).unwrap_or(-1) }; tracing::Span::current().record("early_kill", stdout_truncated); @@ -529,7 +531,7 @@ fn grep_progress_stream( tracing::warn!(timeout_secs = secs, "grep timed out"); }); let _ = child.start_kill(); - let _ = child.wait().await; + crate::util::reap_killed_search_child(&mut child).await; // Finalize what was read (marked truncated) plus an explicit // notice, so the terminal card does not contradict what was already // streamed; with nothing streamed, the timeout-only card suffices. @@ -578,11 +580,14 @@ fn grep_progress_stream( .await; } - let status = child.wait().await.ok(); + // Truncated output means `rg` was already killed above — bounded reap, + // and the exit code is defined as 0. A natural EOF means rg is + // exiting, so the plain wait is prompt. let exit_code = if stdout_truncated { + crate::util::reap_killed_search_child(&mut child).await; 0 } else { - status.and_then(|s| s.code()).unwrap_or(-1) + child.wait().await.ok().and_then(|s| s.code()).unwrap_or(-1) }; let wall_ms = stream_started.elapsed().as_millis() as u64; diff --git a/crates/codegen/kigi-tools/src/implementations/opencode/glob/mod.rs b/crates/codegen/kigi-tools/src/implementations/opencode/glob/mod.rs index 87c4eb0..33f84b7 100644 --- a/crates/codegen/kigi-tools/src/implementations/opencode/glob/mod.rs +++ b/crates/codegen/kigi-tools/src/implementations/opencode/glob/mod.rs @@ -161,7 +161,8 @@ impl kigi_tool_runtime::Tool for GlobTool { .arg(&input.pattern) .arg(&search_dir) .stdout(Stdio::piped()) - .stderr(Stdio::piped()); + // stderr is never read; a pipe would block rg once warnings fill it. + .stderr(Stdio::null()); crate::util::detach_command(&mut cmd); cmd.stdin(Stdio::null()); @@ -206,16 +207,13 @@ impl kigi_tool_runtime::Tool for GlobTool { } } - // Consume stderr to avoid deadlocks. - if let Some(stderr_pipe) = child.stderr.take() { - let _ = stderr_pipe - .take(1_000_000) - .read_to_end(&mut Vec::new()) - .await; + if truncated_by_bytes { + // Bounded reap: a D-state rg must not stall this future forever. + crate::util::reap_killed_search_child(&mut child).await; + } else { + let _ = child.wait().await; } - let _ = child.wait().await; - let stdout = String::from_utf8_lossy(&stdout_buf); let mut truncated = truncated_by_bytes; diff --git a/crates/codegen/kigi-tools/src/util/mod.rs b/crates/codegen/kigi-tools/src/util/mod.rs index dfeed5c..c2d6065 100644 --- a/crates/codegen/kigi-tools/src/util/mod.rs +++ b/crates/codegen/kigi-tools/src/util/mod.rs @@ -29,6 +29,7 @@ pub use path_suggestions::format_not_found_error; pub use remap::{remap_json_keys, remap_schema_properties, reverse_map}; pub use spawn::{ ProcessGroup, ProcessScope, detach_command, global_process_scope, new_process_group, + reap_killed_search_child, }; pub use truncate::{ DEFAULT_SOFT_WRAP_WIDTH, ceil_char_boundary, estimate_tokens, floor_char_boundary, diff --git a/crates/codegen/kigi-tools/src/util/spawn.rs b/crates/codegen/kigi-tools/src/util/spawn.rs index e95acb8..42bfd27 100644 --- a/crates/codegen/kigi-tools/src/util/spawn.rs +++ b/crates/codegen/kigi-tools/src/util/spawn.rs @@ -8,3 +8,19 @@ pub use kigi_tty_utils::{ ProcessGroup, ProcessScope, detach_command, global_process_scope, new_process_group, }; + +/// Reap an already-killed search child, bounded by +/// [`kigi_tty_utils::KILL_REAP_TIMEOUT`]; on `None` warn and leave the corpse +/// to tokio's orphan reaper. +pub async fn reap_killed_search_child( + child: &mut tokio::process::Child, +) -> Option { + let status = kigi_tty_utils::reap_killed_bounded(child, kigi_tty_utils::KILL_REAP_TIMEOUT).await; + if status.is_none() { + tracing::warn!( + reap_timeout_secs = kigi_tty_utils::KILL_REAP_TIMEOUT.as_secs(), + "killed search child not reaped (bound expired — likely uninterruptible kernel I/O — or wait failed); abandoning to the orphan reaper" + ); + } + status +} diff --git a/crates/codegen/kigi-tty-utils/Cargo.toml b/crates/codegen/kigi-tty-utils/Cargo.toml index c754d4d..7a597f1 100644 --- a/crates/codegen/kigi-tty-utils/Cargo.toml +++ b/crates/codegen/kigi-tty-utils/Cargo.toml @@ -6,7 +6,7 @@ license = "Apache-2.0" description = "Lightweight process-spawning utilities for TTY safety — detach from controlling terminal, suppress interactive pagers, process-group lifecycle" [dependencies] -tokio = { workspace = true, features = ["process"] } +tokio = { workspace = true, features = ["process", "time"] } [target.'cfg(unix)'.dependencies] nix = { workspace = true, features = ["signal", "process"] } diff --git a/crates/codegen/kigi-tty-utils/src/lib.rs b/crates/codegen/kigi-tty-utils/src/lib.rs index 686ca26..4839af2 100644 --- a/crates/codegen/kigi-tty-utils/src/lib.rs +++ b/crates/codegen/kigi-tty-utils/src/lib.rs @@ -134,6 +134,36 @@ pub fn detach_std_command(cmd: &mut std::process::Command) { // Process group lifecycle +/// Bound on waiting for an already-killed child (or its pipe readers) before +/// abandoning it. +/// +/// A kill normally makes `child.wait()` resolve in milliseconds, but a child +/// wedged in an uninterruptible kernel syscall (D-state — e.g. a read on a +/// hard NFS mount whose server stopped responding) only observes the signal +/// when that syscall returns, which can be effectively never. Callers that +/// must not block (tool futures, turn loops) wait at most this long, then +/// abandon the corpse to the runtime's orphan reaper. +pub const KILL_REAP_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); + +/// Reap an already-killed child, waiting at most `bound` (usually +/// [`KILL_REAP_TIMEOUT`]). +/// +/// Returns the exit status when the child was reaped in time. `None` covers +/// both failure shapes — the bound expired (see [`KILL_REAP_TIMEOUT`]) and +/// `wait()` itself erred (e.g. the child was already reaped elsewhere) — the +/// caller's obligation is identical in either case: the kill signal is +/// already delivered, there is no status to report, and the corpse is left +/// to tokio's orphan reaper. Callers should log the `None` case. +pub async fn reap_killed_bounded( + child: &mut tokio::process::Child, + bound: std::time::Duration, +) -> Option { + match tokio::time::timeout(bound, child.wait()).await { + Ok(res) => res.ok(), + Err(_elapsed) => None, + } +} + /// Configure a command so the spawned child becomes the leader of a new /// process group. pub fn new_process_group(cmd: &mut tokio::process::Command) {