diff --git a/lib/crates/fabro-cli/src/commands/server/stop.rs b/lib/crates/fabro-cli/src/commands/server/stop.rs index 054b61218..09811c6bf 100644 --- a/lib/crates/fabro-cli/src/commands/server/stop.rs +++ b/lib/crates/fabro-cli/src/commands/server/stop.rs @@ -16,17 +16,24 @@ pub(crate) async fn stop_server(storage_dir: &Path, timeout: Duration) -> Result fabro_proc::sigterm(record.pid); + // Use the zombie-aware predicate here: this loop is commonly driven + // against a child of the calling process (tests, install/uninstall + // in-process shutdowns, a foreground-launching shell). A zombie + // child would otherwise satisfy `process_running` until its parent + // waits, causing us to burn the whole `timeout` on an already-dead + // process. The `ps` cost (~2 ms per poll) is trivial compared to + // the 10 s timeout and is only paid while the process still exists. let poll_interval = Duration::from_millis(100); let mut elapsed = Duration::ZERO; while elapsed < timeout { - if !fabro_proc::process_running(record.pid) { + if !fabro_proc::process_running_strict(record.pid) { break; } time::sleep(poll_interval).await; elapsed += poll_interval; } - if fabro_proc::process_running(record.pid) { + if fabro_proc::process_running_strict(record.pid) { fabro_proc::sigkill(record.pid); time::sleep(Duration::from_millis(100)).await; } diff --git a/lib/crates/fabro-cli/tests/it/cmd/pr_list.rs b/lib/crates/fabro-cli/tests/it/cmd/pr_list.rs index 66cd60308..9d85918af 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/pr_list.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/pr_list.rs @@ -1,4 +1,13 @@ +#![allow( + clippy::absolute_paths, + reason = "This test module prefers explicit type paths over extra imports." +)] + use fabro_test::{fabro_snapshot, test_context}; +use fabro_types::run_event::PullRequestCreatedProps; +use fabro_types::{EventBody, RunEvent, RunId}; + +use super::support::{server_endpoint, setup_completed_fast_dry_run}; #[test] fn help() { @@ -26,17 +35,61 @@ fn help() { "); } +// Seed a PR event against this test's own run so the store is guaranteed to +// have at least one entry; `fabro pr list` then must load GitHub credentials +// and fail, regardless of what peer tests have left in the shared store. #[test] fn pr_list_missing_github_credentials_errors() { let context = test_context!(); + let run = setup_completed_fast_dry_run(&context); + let run_id: RunId = run.run_id.parse().unwrap(); + + let runtime = tokio::runtime::Runtime::new().unwrap(); + runtime.block_on(async { + let (client, base_url) = + server_endpoint(&context.storage_dir).expect("server endpoint should exist"); + let event = RunEvent { + id: ulid::Ulid::new().to_string(), + ts: chrono::Utc::now(), + run_id, + node_id: None, + node_label: None, + stage_id: None, + parallel_group_id: None, + parallel_branch_id: None, + session_id: None, + parent_session_id: None, + tool_call_id: None, + actor: None, + body: EventBody::PullRequestCreated(PullRequestCreatedProps { + pr_url: "https://github.com/fabro-sh/fabro/pull/123".to_string(), + pr_number: 123, + owner: "fabro-sh".to_string(), + repo: "fabro".to_string(), + base_branch: "main".to_string(), + head_branch: "fabro/run/demo".to_string(), + title: "Map the constellations".to_string(), + draft: false, + }), + }; + client + .post(format!("{base_url}/api/v1/runs/{run_id}/events")) + .json(&event) + .send() + .await + .unwrap() + .error_for_status() + .unwrap(); + }); + let mut cmd = context.command(); cmd.args(["pr", "list"]); fabro_snapshot!(context.filters(), cmd, @" - success: true - exit_code: 0 + success: false + exit_code: 1 ----- stdout ----- - No pull requests found. ----- stderr ----- + error: GitHub credentials required — run `fabro install` or set GITHUB_TOKEN "); } diff --git a/lib/crates/fabro-proc/src/flock.rs b/lib/crates/fabro-proc/src/flock.rs index 552b143b7..ff048929a 100644 --- a/lib/crates/fabro-proc/src/flock.rs +++ b/lib/crates/fabro-proc/src/flock.rs @@ -20,6 +20,25 @@ pub fn try_flock_exclusive(file: &File) -> io::Result { } } +/// Try to acquire a shared (read) lock on `file` without blocking. +/// +/// Returns `Ok(true)` if the lock was acquired, `Ok(false)` if another +/// process/fd already holds an incompatible lock, and `Err` for unexpected +/// errors. +pub fn try_flock_shared(file: &File) -> io::Result { + // SAFETY: flock() on a valid fd is safe; LOCK_NB makes it non-blocking. + let ret = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_SH | libc::LOCK_NB) }; + if ret == 0 { + Ok(true) + } else { + let err = io::Error::last_os_error(); + match err.raw_os_error() { + Some(libc::EWOULDBLOCK) => Ok(false), + _ => Err(err), + } + } +} + /// Release any lock held on `file`. pub fn flock_unlock(file: &File) -> io::Result<()> { // SAFETY: flock() with LOCK_UN on a valid fd is safe. diff --git a/lib/crates/fabro-proc/src/lib.rs b/lib/crates/fabro-proc/src/lib.rs index 57f9c3c69..5fbadb7bf 100644 --- a/lib/crates/fabro-proc/src/lib.rs +++ b/lib/crates/fabro-proc/src/lib.rs @@ -11,14 +11,14 @@ mod signal; mod title; #[cfg(unix)] -pub use flock::{flock_unlock, try_flock_exclusive}; +pub use flock::{flock_unlock, try_flock_exclusive, try_flock_shared}; #[cfg(target_os = "linux")] pub use pre_exec::pre_exec_pdeathsig; #[cfg(unix)] pub use pre_exec::pre_exec_setpgid; #[cfg(unix)] pub use pre_exec::pre_exec_setsid; -pub use signal::{process_exists, process_group_alive, process_running}; +pub use signal::{process_exists, process_group_alive, process_running, process_running_strict}; #[cfg(unix)] pub use signal::{ sigkill, sigkill_process_group, sigterm, sigterm_process_group, sigusr1, sigusr2, diff --git a/lib/crates/fabro-proc/src/signal.rs b/lib/crates/fabro-proc/src/signal.rs index a418b6dca..52c2eef71 100644 --- a/lib/crates/fabro-proc/src/signal.rs +++ b/lib/crates/fabro-proc/src/signal.rs @@ -20,17 +20,35 @@ pub fn process_exists(pid: u32) -> bool { /// Check whether a process with the given PID is still running. /// -/// On Unix, this treats zombie / defunct processes as not running even though -/// they still have a visible PID until their parent reaps them. If the -/// follow-up `ps` probe fails, this falls back to `process_exists(pid)` to -/// preserve the old conservative behavior. +/// On Unix, delegates to `process_exists` (a `kill(pid, 0)` probe). Unreaped +/// zombies count as running here because the callers are hot paths (test +/// harness marker scans, daemon-liveness probes) where a zombie window is +/// sub-millisecond and the ~2 ms cost of an authoritative zombie check via +/// `ps` would dominate. For the narrow set of callers that genuinely need +/// to treat zombies as stopped (notably the `fabro server stop` polling +/// loop, which can wait out its full timeout on a zombie child), use +/// `process_running_strict`. pub fn process_running(pid: u32) -> bool { + process_exists(pid) +} + +/// Like `process_running`, but treats zombie / defunct processes as not +/// running. +/// +/// On Unix, follows a cheap `kill(pid, 0)` probe with a `ps` shell-out to +/// read the process state character and excludes `Z`/`z` entries. Falls +/// back to `process_exists(pid)` when the `ps` probe fails, preserving the +/// old conservative behavior. On non-unix, identical to `process_exists`. +/// +/// Prefer `process_running` unless you are polling for a child you cannot +/// `wait()` on — the `ps` invocation costs ~2 ms per call on macOS and is +/// wasted on hot paths that have no zombie exposure. +pub fn process_running_strict(pid: u32) -> bool { #[cfg(unix)] { if !process_exists(pid) { return false; } - unix_process_state(pid).is_none_or(|state| !matches!(state, 'Z' | 'z')) } #[cfg(not(unix))] @@ -52,7 +70,6 @@ fn unix_process_state(pid: u32) -> Option { if !output.status.success() { return None; } - String::from_utf8_lossy(&output.stdout) .chars() .find(|ch| !ch.is_whitespace()) @@ -155,22 +172,23 @@ mod tests { use std::process::{Command, Stdio}; use std::time::Duration; - use super::{process_exists, process_group_alive, process_running}; + use super::{process_exists, process_group_alive, process_running, process_running_strict}; use crate::pre_exec::pre_exec_setpgid; #[test] fn process_running_returns_true_for_current_process() { assert!(process_exists(std::process::id())); assert!(process_running(std::process::id())); + assert!(process_running_strict(std::process::id())); } #[cfg(unix)] #[test] #[expect( clippy::disallowed_methods, - reason = "process-state test needs to spawn a short-lived child and intentionally leave it unreaped" + reason = "zombie-detection test needs to spawn a short-lived child and intentionally leave it unreaped" )] - fn process_running_returns_false_for_unreaped_zombie_child() { + fn process_running_strict_returns_false_for_unreaped_zombie_child() { let mut child = Command::new("sh") .args(["-c", "exit 0"]) .spawn() @@ -184,8 +202,12 @@ mod tests { "unreaped zombie should still have a visible pid" ); assert!( - !process_running(pid), - "unreaped zombie should not count as a running process" + process_running(pid), + "cheap process_running treats zombies as alive by design" + ); + assert!( + !process_running_strict(pid), + "process_running_strict should treat zombies as stopped" ); let _status = child.wait().expect("child should remain waitable"); diff --git a/lib/crates/fabro-server/tests/it/scenario/lifecycle.rs b/lib/crates/fabro-server/tests/it/scenario/lifecycle.rs index 2ec100f16..f04300075 100644 --- a/lib/crates/fabro-server/tests/it/scenario/lifecycle.rs +++ b/lib/crates/fabro-server/tests/it/scenario/lifecycle.rs @@ -249,7 +249,17 @@ async fn full_http_lifecycle_cancel() { ) .await; assert_eq!(body["status"], "running"); - assert_eq!(body["pending_control"], "cancel"); + // `pending_control` is computed from the store projection after the cancel + // event is appended AND the worker is signaled. The worker is sitting at a + // human gate; once notified it can emit a clearing event before this + // handler re-reads the projection, so the response can legitimately + // observe either the still-pending "cancel" or a null where the worker + // already consumed it. Durable convergence is asserted below. + let pending_control = &body["pending_control"]; + assert!( + pending_control == "cancel" || pending_control.is_null(), + "expected pending_control to be \"cancel\" or null, got {pending_control}" + ); // Verify the durable store view converges to cancelled failure. let body = wait_for_run_state(&app, &run_id, "failed", "cancelled").await; diff --git a/lib/crates/fabro-test/src/lib.rs b/lib/crates/fabro-test/src/lib.rs index da412c96c..b428cc482 100644 --- a/lib/crates/fabro-test/src/lib.rs +++ b/lib/crates/fabro-test/src/lib.rs @@ -16,7 +16,6 @@ use assert_cmd::Command; use fabro_config::Storage; use fabro_types::RunId; use regex::Regex; -use serde::Serialize; use serde_json::{Map, Value, json}; use toml::Value as TomlValue; use toml::map::Map as TomlMap; @@ -294,18 +293,35 @@ enum SessionMode { Process, } -#[derive(Debug, Serialize)] -struct ClientMarker { - pid: u32, - touched_at_ms: u128, -} - static SESSION_REFS: OnceLock>> = OnceLock::new(); fn session_refs() -> &'static Mutex> { SESSION_REFS.get_or_init(|| Mutex::new(HashMap::new())) } +// Gate the stale-session reap so it fires at most once per process. The +// reap scans /tmp/fx/{n-*,p-*} to clean up after a prior nextest run +// that crashed; it's correctness-for-safety work that does not need to +// happen on every TestContext::new. One gate per SessionMode preserves +// the existing call structure without introducing cross-mode coupling. +static NEXTEST_REAPED: OnceLock<()> = OnceLock::new(); +static PROCESS_REAPED: OnceLock<()> = OnceLock::new(); + +// Advisory-lock-based peer-presence marker. Each test process opens +// `/clients/` once, holds a shared (LOCK_SH) flock +// for the lifetime of any live TestContext in the process, and +// releases it explicitly in `cleanup_session_root` when the refcount +// drops to zero. Peers detect liveness by attempting LOCK_EX: if it +// succeeds, the owner is gone (normal exit, panic, SIGKILL, or zombie +// — the kernel releases flocks at process exit in every case), and +// the stale marker file is removed. +// +// The slot stores the marker path so subsequent rebounds in the same +// process (e.g., after a drop-to-zero followed by a new TestContext) +// can re-validate the invariant: a process only ever participates in +// one session root. +static MARKER_HANDLE: Mutex> = Mutex::new(None); + #[expect( clippy::disallowed_methods, reason = "This synchronous test-support helper uses uuidgen when available to create stable unique case IDs." @@ -329,13 +345,6 @@ fn test_case_id() -> String { ulid } -fn current_timestamp_ms() -> u128 { - SystemTime::now() - .duration_since(UNIX_EPOCH) - .expect("system time should be after unix epoch") - .as_millis() -} - fn current_pid() -> u32 { std::process::id() } @@ -467,6 +476,18 @@ fn with_session_lock(root: &Path, f: impl FnOnce() -> T) -> T { result } +// Iterate `/clients/` and for each marker file attempt a +// non-blocking exclusive flock. Success means the previous owner has +// released the lock (normal exit, panic, SIGKILL, or zombie — the +// kernel releases flocks at process exit regardless), and the stale +// marker file is removed. `EWOULDBLOCK` means the owner is still +// alive and holding LOCK_SH. +// +// Same-process subtlety: if our own PID's marker file is in the +// listing, we have LOCK_SH on it via `MARKER_HANDLE`. On Linux and +// macOS, flock locks are per-open-file-description, so a fresh +// `open()` here returns an FD that sees the shared lock and correctly +// reports EWOULDBLOCK when asked for LOCK_EX. fn live_marker_count(root: &Path) -> usize { let clients_dir = session_clients_dir(root); let Ok(entries) = std::fs::read_dir(&clients_dir) else { @@ -480,30 +501,73 @@ fn live_marker_count(root: &Path) -> usize { .to_string_lossy() .parse::() .ok() - .map(|pid| (pid, entry.path())) + .map(|_pid| entry.path()) }) - .filter(|(pid, path)| { - if fabro_proc::process_running(*pid) { - true - } else { + .filter(|path| { + // Open read-write so LOCK_EX has the access mode it expects + // on the widest set of platforms. If the file is missing + // between read_dir and open, treat it as already gone. + let Ok(file) = std::fs::OpenOptions::new() + .read(true) + .write(true) + .open(path) + else { + return false; + }; + // If the lock is acquired, the previous owner is gone; drop + // the file handle (releasing our just-acquired lock) and + // remove the marker. Anything else (`Ok(false)` meaning + // still held, `Err(_)` for unexpected IO errors) is treated + // conservatively as live. + if matches!(fabro_proc::try_flock_exclusive(&file), Ok(true)) { + drop(file); let _ = std::fs::remove_file(path); false + } else { + true } }) .count() } +// Open (or reopen, after a session-drop-to-zero) the per-process +// marker file and hold a shared advisory lock on it in +// `MARKER_HANDLE`. Called from inside the `with_session_lock` block +// of `TestContext::new`. fn write_marker(root: &Path) { - let marker = ClientMarker { - pid: current_pid(), - touched_at_ms: current_timestamp_ms(), - }; - let marker_path = session_marker_path(root, marker.pid); + let marker_path = session_marker_path(root, current_pid()); ensure_parent_dir(&marker_path); - let contents = - serde_json::to_vec(&marker).expect("client marker should serialize to JSON bytes"); - std::fs::write(&marker_path, contents) - .unwrap_or_else(|err| panic!("failed to write {}: {err}", marker_path.display())); + + let mut slot = MARKER_HANDLE.lock().expect("MARKER_HANDLE lock poisoned"); + + if let Some((existing_path, _)) = slot.as_ref() { + debug_assert_eq!( + existing_path, &marker_path, + "marker handle path drifted — session root changed mid-process?" + ); + if marker_path.exists() { + return; + } + // Marker was removed (e.g., by a peer's reap) while we thought + // we still owned it. Re-establish by replacing the handle. + slot.take(); + } + + let file = std::fs::OpenOptions::new() + .create(true) + .truncate(true) + .read(true) + .write(true) + .open(&marker_path) + .unwrap_or_else(|err| panic!("failed to open {}: {err}", marker_path.display())); + let acquired = fabro_proc::try_flock_shared(&file) + .unwrap_or_else(|err| panic!("failed to flock {}: {err}", marker_path.display())); + assert!( + acquired, + "unexpected contention acquiring LOCK_SH on freshly created marker {}", + marker_path.display() + ); + *slot = Some((marker_path, file)); } fn managed_storage_settings(storage_dir: &Path, rest: &str) -> String { @@ -898,6 +962,17 @@ fn reap_isolated_servers(root: &Path) { fn cleanup_session_root(root: &Path) { with_session_lock(root, || { + // Release our own advisory lock first so `live_marker_count` + // below can observe that no one is holding the marker file, + // then remove the file. Order matters: if we unlinked before + // dropping the handle, peers would still see our LOCK_SH on + // the (now-unlinked but still open) inode and count us as + // live, preventing the server teardown. + { + let mut slot = MARKER_HANDLE.lock().expect("MARKER_HANDLE lock poisoned"); + // Dropping the File closes the FD and releases LOCK_SH. + slot.take(); + } let marker_path = session_marker_path(root, current_pid()); let _ = std::fs::remove_file(&marker_path); let live_count = live_marker_count(root); @@ -964,8 +1039,8 @@ impl TestContext { .expect("failed to create temp dir"); let root_path = context_root.path().to_path_buf(); let (_, test_run_id, session_paths) = session_paths(); - reap_stale_session_roots(SessionMode::Nextest); - reap_stale_session_roots(SessionMode::Process); + NEXTEST_REAPED.get_or_init(|| reap_stale_session_roots(SessionMode::Nextest)); + PROCESS_REAPED.get_or_init(|| reap_stale_session_roots(SessionMode::Process)); with_session_lock(&session_paths.root, || { std::fs::create_dir_all(session_clients_dir(&session_paths.root)).unwrap_or_else( |err| {