From 7061d66ca640d2579d810ca27009db0cbc268370 Mon Sep 17 00:00:00 2001 From: Paul Logan Date: Tue, 1 Sep 2026 19:22:03 -0700 Subject: [PATCH 1/5] fix(supervisor): bound the worker population and let idle homes drain MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A dev box reached 2,770 live `wire daemon` children under a `max_workers = 16` cap: 6.2 GB resident, load average 994, swap exhausted, and a 112 MB daemon log written by all of them at once. Three defects compounded. Eviction was a request, not a guarantee. `retire_inactive_worker` sent one un-escalated SIGTERM per poll and discarded the result. A worker parked in a relay reconnect backoff (1s-30s sleeps against an unreachable relay) does not act on SIGTERM promptly, so the signal achieved nothing while the supervisor had already dropped the child from `children`. Add `kill_worker_verified`: SIGTERM, wait, escalate to SIGKILL, confirm. Evicted children were never reaped. Dropping a `Child` neither kills nor waits, and `process_alive` is a `kill(pid, 0)` probe that reports a zombie as alive, so a killed worker was indistinguishable from a survivor. Every fork-exec now lives in either `children` (intent) or a new `orphans` list (outstanding fact) until `terminate_and_reap` has both killed and reaped it. That, not the in-memory map, is what bounds the population. The husk reaper could never drain. It only removes homes with no identity and no sync history, but session adoption mints a home that gains a `private.key` within seconds — on the affected box 8,975 of 8,983 homes held one and *zero* matched the husk predicate, while the supervisor stat-ed all of them every 10s. Add `reap_idle_homes`, a separate path keyed on idleness rather than emptiness: unbound, no live lease, no pending outbox, no live daemon, and untouched for 14 days (`WIRE_IDLE_REAP_MAX_AGE_DAYS`, 0 disables). Named sessions and registry-bound homes are still never touched, and the husk predicate is unchanged. Verified live: supervisor holds 4 children across 12 poll cycles with 4 spawns, no orphan churn, and a 2.4 KB log. --- src/daemon_supervisor.rs | 426 ++++++++++++++++++++++++++++++++++++++- 1 file changed, 415 insertions(+), 11 deletions(-) diff --git a/src/daemon_supervisor.rs b/src/daemon_supervisor.rs index b63ad11..e6e6130 100644 --- a/src/daemon_supervisor.rs +++ b/src/daemon_supervisor.rs @@ -144,12 +144,87 @@ fn retire_inactive_worker(session: &crate::session::SessionInfo) { "supervisor: retiring unselected session worker '{}' pid={} version={version:?}", session.name, record.pid ); - if !crate::platform::kill_process(record.pid, false) { - eprintln!( - "supervisor: failed to signal unselected session worker '{}' pid={}", - session.name, record.pid - ); + let label = format!("unselected session worker '{}'", session.name); + kill_worker_verified(record.pid, &label); +} + +/// How long a worker gets to honour SIGTERM before the supervisor +/// escalates to SIGKILL. +const TERM_GRACE: Duration = Duration::from_millis(1500); +const TERM_POLL: Duration = Duration::from_millis(100); + +/// Terminate one worker and *verify* it actually died: SIGTERM, wait up +/// to [`TERM_GRACE`] for exit, then SIGKILL and re-check. Returns true +/// iff the pid is gone when we return. +/// +/// ## Why verification, not fire-and-forget +/// +/// The original code sent a single un-escalated SIGTERM per poll and +/// discarded the result. A worker parked in a relay reconnect backoff +/// (1s–30s sleeps against an unreachable relay) does not act on SIGTERM +/// promptly, so every poll re-signalled a process that never died while +/// the supervisor had already dropped it from its bookkeeping. Workers +/// then accumulated without bound: one dev box reached 2,770 live +/// children under a `max_workers=16` cap, 6.2 GB RSS and a load average +/// of 994. Escalating to SIGKILL and confirming death is what makes +/// eviction an actual guarantee instead of a request. +/// Terminate a child this supervisor owns and **reap it**, escalating +/// SIGTERM -> SIGKILL. Returns true iff the child has been reaped. +/// +/// Reaping is the half that `kill_worker_verified` cannot do: a killed +/// direct child stays in the process table as a zombie until somebody +/// `wait`s on it, and `process_alive` (a `kill(pid, 0)` probe) reports +/// a zombie as *alive*. Evicted workers used to be dropped without a +/// wait, so they lingered as unreapable entries and the supervisor +/// could never tell a survivor from a corpse. +fn terminate_and_reap(child: &mut Child, pid: u32, label: &str) -> bool { + if matches!(child.try_wait(), Ok(Some(_))) { + return true; + } + crate::platform::kill_process(pid, false); + let deadline = Instant::now() + TERM_GRACE; + while Instant::now() < deadline { + if matches!(child.try_wait(), Ok(Some(_))) { + return true; + } + std::thread::sleep(TERM_POLL); } + eprintln!("supervisor: {label} pid={pid} ignored SIGTERM after {TERM_GRACE:?}; sending SIGKILL"); + let _ = child.kill(); + let deadline = Instant::now() + TERM_GRACE; + while Instant::now() < deadline { + if matches!(child.try_wait(), Ok(Some(_))) { + return true; + } + std::thread::sleep(TERM_POLL); + } + eprintln!("supervisor: {label} pid={pid} not reaped after SIGKILL; retrying next poll"); + false +} + +fn kill_worker_verified(pid: u32, label: &str) -> bool { + if !crate::platform::process_alive(pid) { + return true; + } + crate::platform::kill_process(pid, false); + let deadline = Instant::now() + TERM_GRACE; + while Instant::now() < deadline { + if !crate::platform::process_alive(pid) { + return true; + } + std::thread::sleep(TERM_POLL); + } + eprintln!("supervisor: {label} pid={pid} ignored SIGTERM after {TERM_GRACE:?}; sending SIGKILL"); + crate::platform::kill_process(pid, true); + let deadline = Instant::now() + TERM_GRACE; + while Instant::now() < deadline { + if !crate::platform::process_alive(pid) { + return true; + } + std::thread::sleep(TERM_POLL); + } + eprintln!("supervisor: {label} pid={pid} SURVIVED SIGKILL; leaving it for the next poll"); + false } /// Newest mtime among a session home's activity files — the @@ -397,9 +472,126 @@ where reaped } +/// Default idle window before an *identity-bearing* by-key home is +/// reaped, in days. +const DEFAULT_IDLE_REAP_MAX_AGE_DAYS: u64 = 14; + +/// Parse the idle reap cutoff. `None` raw -> default; `0` -> `None` +/// (disabled); any other integer -> that many days; unparseable -> +/// default. +fn parse_idle_reap_max_age(raw: Option<&str>) -> Option { + match raw { + Some(v) => { + let days: u64 = v.trim().parse().unwrap_or(DEFAULT_IDLE_REAP_MAX_AGE_DAYS); + (days != 0).then(|| Duration::from_secs(days * 86_400)) + } + None => Some(Duration::from_secs(DEFAULT_IDLE_REAP_MAX_AGE_DAYS * 86_400)), + } +} + +/// Read the idle reap cutoff from the environment. +/// `WIRE_IDLE_REAP_MAX_AGE_DAYS=0` disables idle reaping entirely. +fn idle_reap_max_age_from_env() -> Option { + parse_idle_reap_max_age( + std::env::var("WIRE_IDLE_REAP_MAX_AGE_DAYS").ok().as_deref(), + ) +} + +/// Delete long-idle by-key session homes that DO hold an identity, and +/// return what was removed. +/// +/// ## Why this exists alongside [`reap_husks`] +/// +/// `reap_husks` only removes homes with no `private.key` and no sync +/// history. That made the by-key population **monotonic** in practice: +/// session adoption mints a home, the home gains an identity within +/// seconds, and from that moment no reaper could ever touch it. A real +/// box accumulated 8,983 homes (461 MB) of which *zero* matched the +/// husk predicate, while the supervisor stat-ed all of them on every +/// 10s registry poll. +/// +/// The husk predicate stays as-is — this is a strictly separate, much +/// more conservative path keyed on *idleness* rather than emptiness. A +/// dir is reaped only if ALL of these hold: +/// - its name has the by-key shape (16 lowercase hex chars) — named, +/// operator-created sessions are never touched; +/// - it is not registry-bound; +/// - it holds no live lease and no pending outbox (nothing would be +/// lost by removing it); +/// - no live daemon owns it; +/// - its last activity (or, for a never-synced home, its own mtime) is +/// older than `max_age`. Future timestamps count as young, so clock +/// skew never deletes. +/// +/// Failures are per-entry best-effort (warn + continue). +fn reap_idle_homes( + by_key_root: &Path, + max_age: Duration, + now: SystemTime, + bound_names: &std::collections::HashSet, + is_active: A, + daemon_live: D, +) -> Vec +where + A: Fn(&Path) -> bool, + D: Fn(&Path) -> bool, +{ + let mut reaped = Vec::new(); + let Ok(entries) = std::fs::read_dir(by_key_root) else { + return reaped; + }; + for entry in entries.flatten() { + let path = entry.path(); + if !path.is_dir() { + continue; + } + let Some(name) = path.file_name().and_then(|s| s.to_str()) else { + continue; + }; + let is_by_key_shape = + name.len() == 16 && name.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f')); + if !is_by_key_shape { + continue; + } + if bound_names.contains(name) { + continue; + } + if is_active(&path) { + continue; + } + if daemon_live(&path) { + continue; + } + // Prefer real activity; fall back to the home's own mtime so a + // home that never synced still ages out on this path. + let last = fs_last_active(&path).or_else(|| { + std::fs::metadata(&path) + .and_then(|m| m.modified()) + .ok() + }); + let idle_long_enough = last + .and_then(|t| now.duration_since(t).ok()) + .is_some_and(|age| age >= max_age); + if !idle_long_enough { + continue; + } + match std::fs::remove_dir_all(&path) { + Ok(()) => reaped.push(path), + Err(e) => eprintln!("supervisor: idle reap failed for {}: {e:#}", path.display()), + } + } + reaped +} + /// State the supervisor tracks per session it has spawned a child for. struct ChildState { child: Child, + /// Cached at spawn: `Child::id()` is not meaningful once the child + /// has been reaped, and the teardown pass needs a stable key. + pid: u32, + /// Session this child serves — carried so an orphaned child can + /// still name itself in teardown logs. + name: String, spawned_at: Instant, } @@ -454,9 +646,25 @@ pub fn run_supervisor(interval_secs: u64, max_workers: usize, as_json: bool) -> None => "disabled".to_string(), } ); + let idle_max_age = idle_reap_max_age_from_env(); + eprintln!( + "supervisor: idle reap cutoff = {}", + match idle_max_age { + Some(d) => format!("{} days", d.as_secs() / 86_400), + None => "disabled".to_string(), + } + ); let mut last_husk_reap: Option = None; let mut children: HashMap = HashMap::new(); + // Children the supervisor has stopped selecting but has not yet + // confirmed dead. `children` is *intent*; this is the outstanding + // *fact*. Every process we fork-exec lives in exactly one of the + // two until it has been killed AND reaped, which is what bounds + // the population — dropping a `Child` neither kills nor reaps it, + // so an untracked eviction used to leak a live process (2,770 of + // them under a max_workers=16 cap on one box). + let mut orphans: Vec = Vec::new(); // Per-session backoff that survives a child's reap → respawn → reap // cycle. Distinguishes "session crashes hard repeatedly" from // "child exited cleanly and we're spawning a fresh one". @@ -552,6 +760,24 @@ pub fn run_supervisor(interval_secs: u64, max_workers: usize, as_json: bool) -> .join(", ") ); } + if let Some(idle_age) = idle_max_age { + let idle_reaped = reap_idle_homes( + &root.join("by-key"), + idle_age, + SystemTime::now(), + &bound, + |home| fs_has_live_lease(home) || fs_has_pending_outbox(home), + // On a liveness-probe error assume live — never + // delete a home we couldn't safely inspect. + |home| existing_daemon_for_session(home).unwrap_or(true), + ); + if !idle_reaped.is_empty() { + eprintln!( + "supervisor: reaped {} idle session home(s)", + idle_reaped.len() + ); + } + } } } @@ -566,10 +792,9 @@ pub fn run_supervisor(interval_secs: u64, max_workers: usize, as_json: bool) -> .cloned() .collect(); for name in to_kill { - if let Some(mut state) = children.remove(&name) { + if let Some(state) = children.remove(&name) { eprintln!("supervisor: session '{name}' gone from registry; terminating its child"); - let _ = state.child.kill(); - let _ = state.child.wait(); + orphans.push(state); } } for session in &all_sessions { @@ -606,15 +831,17 @@ pub fn run_supervisor(interval_secs: u64, max_workers: usize, as_json: bool) -> } match spawn_child_for_session(&info.name, &info.home_dir, interval_secs) { Ok(child) => { + let pid = child.id(); eprintln!( - "supervisor: spawned child for session '{}' (pid {})", - info.name, - child.id() + "supervisor: spawned child for session '{}' (pid {pid})", + info.name ); children.insert( info.name.clone(), ChildState { child, + pid, + name: info.name.clone(), spawned_at: Instant::now(), }, ); @@ -633,6 +860,22 @@ pub fn run_supervisor(interval_secs: u64, max_workers: usize, as_json: bool) -> } } + // 5. Drain the orphan list: kill (escalating) and reap every + // child we no longer select. Anything still outstanding is + // retried on the next poll, so the live population stays + // bounded by `max_workers` plus whatever is mid-teardown. + if !orphans.is_empty() { + eprintln!( + "supervisor: {} orphan worker(s) pending teardown (cap {max_workers}, tracked {})", + orphans.len(), + children.len() + ); + } + orphans.retain_mut(|state| { + let label = format!("orphan worker for session '{}'", state.name); + !terminate_and_reap(&mut state.child, state.pid, &label) + }); + std::thread::sleep(Duration::from_secs(REGISTRY_POLL_SECS)); } } @@ -1422,6 +1665,167 @@ mod tests { assert!(reaped.is_empty()); } + const CUTOFF_14D: Duration = Duration::from_secs(14 * 86_400); + + fn far_future_days() -> SystemTime { + SystemTime::now() + Duration::from_secs(30 * 86_400) + } + + /// Give a by-key home an identity + sync history, i.e. exactly the + /// shape `reap_husks` refuses to touch. This is the population that + /// grew without bound on the box that motivated the idle reaper. + fn mk_identity_home(root: &Path, name: &str) -> PathBuf { + let home = mk_husk(root, name); + std::fs::create_dir_all(home.join("config").join("wire")).unwrap(); + std::fs::write(home.join("config").join("wire").join("private.key"), b"k").unwrap(); + std::fs::write(home.join("state").join("wire").join("last_sync.json"), b"{}").unwrap(); + home + } + + #[test] + fn idle_reap_removes_long_idle_identity_home_that_husk_reap_cannot() { + let tmp = tempfile::tempdir().unwrap(); + mk_identity_home(tmp.path(), "aaaaaaaaaaaaaaaa"); + let bound = std::collections::HashSet::new(); + // The husk reaper is blind to it — that is the bug. + let husks = reap_husks(tmp.path(), CUTOFF_48H, far_future_days(), &bound, |_| false); + assert!(husks.is_empty()); + // The idle reaper drains it. + let reaped = reap_idle_homes( + tmp.path(), + CUTOFF_14D, + far_future_days(), + &bound, + |_| false, + |_| false, + ); + assert_eq!(reaped.len(), 1); + assert!(!tmp.path().join("aaaaaaaaaaaaaaaa").exists()); + } + + #[test] + fn idle_reap_keeps_recently_active_home() { + let tmp = tempfile::tempdir().unwrap(); + mk_identity_home(tmp.path(), "bbbbbbbbbbbbbbbb"); + let bound = std::collections::HashSet::new(); + let reaped = reap_idle_homes( + tmp.path(), + CUTOFF_14D, + SystemTime::now(), + &bound, + |_| false, + |_| false, + ); + assert!(reaped.is_empty()); + assert!(tmp.path().join("bbbbbbbbbbbbbbbb").exists()); + } + + #[test] + fn idle_reap_keeps_home_with_live_lease_or_outbox() { + let tmp = tempfile::tempdir().unwrap(); + mk_identity_home(tmp.path(), "cccccccccccccccc"); + let bound = std::collections::HashSet::new(); + let reaped = reap_idle_homes( + tmp.path(), + CUTOFF_14D, + far_future_days(), + &bound, + |_| true, // active: live lease or pending outbox + |_| false, + ); + assert!(reaped.is_empty()); + } + + #[test] + fn idle_reap_keeps_home_with_live_daemon() { + let tmp = tempfile::tempdir().unwrap(); + mk_identity_home(tmp.path(), "dddddddddddddddd"); + let bound = std::collections::HashSet::new(); + let reaped = reap_idle_homes( + tmp.path(), + CUTOFF_14D, + far_future_days(), + &bound, + |_| false, + |_| true, + ); + assert!(reaped.is_empty()); + } + + #[test] + fn idle_reap_keeps_registry_bound_and_named_homes() { + let tmp = tempfile::tempdir().unwrap(); + mk_identity_home(tmp.path(), "eeeeeeeeeeeeeeee"); + mk_identity_home(tmp.path(), "peat-eagle"); + let mut bound = std::collections::HashSet::new(); + bound.insert("eeeeeeeeeeeeeeee".to_string()); + let reaped = reap_idle_homes( + tmp.path(), + CUTOFF_14D, + far_future_days(), + &bound, + |_| false, + |_| false, + ); + assert!(reaped.is_empty()); + assert!(tmp.path().join("peat-eagle").exists()); + } + + #[test] + fn idle_reap_max_age_parsing() { + assert_eq!( + parse_idle_reap_max_age(None), + Some(Duration::from_secs(DEFAULT_IDLE_REAP_MAX_AGE_DAYS * 86_400)) + ); + assert_eq!(parse_idle_reap_max_age(Some("0")), None); + assert_eq!( + parse_idle_reap_max_age(Some("3")), + Some(Duration::from_secs(3 * 86_400)) + ); + assert_eq!( + parse_idle_reap_max_age(Some("garbage")), + Some(Duration::from_secs(DEFAULT_IDLE_REAP_MAX_AGE_DAYS * 86_400)) + ); + } + + #[test] + fn kill_worker_verified_reports_dead_pid_as_gone() { + // A pid that is definitively not running must short-circuit to + // "already dead" without signalling anything. + assert!(kill_worker_verified(u32::MAX - 1, "test worker")); + } + + #[test] + fn terminate_and_reap_escalates_past_a_sigterm_ignoring_child() { + // The regression this guards: eviction used to send one + // un-escalated SIGTERM and discard both the result and the + // corpse, so a worker that did not act on SIGTERM survived + // while the supervisor dropped it from bookkeeping. + // `sh -c 'trap "" TERM; sleep 60'` ignores SIGTERM outright, so + // only SIGKILL escalation can end it — and only a `wait` can + // reap it afterwards. + let child = Command::new("sh") + .args(["-c", "trap '' TERM; sleep 60"]) + .spawn(); + let Ok(mut child) = child else { + return; // no shell available — nothing to assert + }; + let pid = child.id(); + assert!(crate::platform::process_alive(pid)); + assert!(terminate_and_reap(&mut child, pid, "sigterm-ignoring test worker")); + } + + #[test] + fn terminate_and_reap_is_idempotent_on_an_already_dead_child() { + let child = Command::new("sh").args(["-c", "exit 0"]).spawn(); + let Ok(mut child) = child else { + return; + }; + let pid = child.id(); + assert!(terminate_and_reap(&mut child, pid, "short-lived test worker")); + assert!(terminate_and_reap(&mut child, pid, "short-lived test worker")); + } + #[test] fn husk_reap_max_age_parsing() { // Unset → 48h default. From 750a7e699b43df476df5ee012796c8f789b89013 Mon Sep 17 00:00:00 2001 From: Paul Logan Date: Tue, 1 Sep 2026 20:40:39 -0700 Subject: [PATCH 2/5] fix(supervisor): don't retire a worker the supervisor already owns MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Step 3 moves an evicted child onto the orphan list and then, in the same pass, calls `retire_inactive_worker` for every session that is no longer selected — including the one just evicted. That call would kill our own child by pidfile. We hold its `Child` handle and have not waited on it, so the kill leaves a zombie, and `process_alive` is a `kill(pid, 0)` probe that reports a zombie as alive. The function therefore burned its full SIGTERM grace, escalated to SIGKILL, burned that grace too, and logged a false "SURVIVED SIGKILL" — three seconds of stall in the single-threaded poll loop, plus a bogus alarm, on every single eviction. Pass the set of pids we hold a `Child` for (selected children plus orphans awaiting teardown) and skip them. Their teardown belongs to the orphan drain in step 5, which is the only path that can actually reap them. The regression test asserts both halves: the owned child survives the call, and the call returns inside the kill grace rather than blocking on it. Without the guard it fails, and takes 3.18s to do so against 0.01s with it. --- src/daemon_supervisor.rs | 84 ++++++++++++++++++++++++++++++++++++++-- 1 file changed, 80 insertions(+), 4 deletions(-) diff --git a/src/daemon_supervisor.rs b/src/daemon_supervisor.rs index e6e6130..57c4552 100644 --- a/src/daemon_supervisor.rs +++ b/src/daemon_supervisor.rs @@ -120,7 +120,18 @@ fn classify_session_worker( /// Stop one supervisor-owned worker for a session that lifecycle planning did /// not select. A standalone daemon started by `wire up` may share the same /// pidfile and command line, so the pidfile's explicit owner is the boundary. -fn retire_inactive_worker(session: &crate::session::SessionInfo) { +/// +/// `owned_pids` names the workers this supervisor already holds a +/// `Child` handle for (selected children plus orphans awaiting +/// teardown). Those are skipped: killing one here would leave a zombie +/// that only the orphan drain can reap, and `process_alive` reports a +/// zombie as alive — so this function would burn its full SIGTERM + +/// SIGKILL grace and then log a false "SURVIVED SIGKILL" on a worker +/// that is already dead and queued for reaping. +fn retire_inactive_worker( + session: &crate::session::SessionInfo, + owned_pids: &std::collections::HashSet, +) { let pidfile = session .home_dir .join("state") @@ -135,6 +146,10 @@ fn retire_inactive_worker(session: &crate::session::SessionInfo) { if !record.supervisor_managed { return; } + if owned_pids.contains(&record.pid) { + // Ours already — the orphan drain owns its teardown. + return; + } let alive = crate::platform::process_alive(record.pid); let cmdline = crate::platform::pid_cmdline(record.pid); let Some(version) = classify_session_worker(&record, alive, cmdline.as_deref()) else { @@ -797,9 +812,16 @@ pub fn run_supervisor(interval_secs: u64, max_workers: usize, as_json: bool) -> orphans.push(state); } } + // Workers we hold a `Child` for. Their teardown belongs to the + // orphan drain in step 5, which can actually reap them. + let owned_pids: std::collections::HashSet = children + .values() + .chain(orphans.iter()) + .map(|state| state.pid) + .collect(); for session in &all_sessions { if !wanted_names.contains(&session.name) { - retire_inactive_worker(session); + retire_inactive_worker(session, &owned_pids); } } @@ -1426,7 +1448,7 @@ mod tests { let mut session = initialized_session("operator-started", false); session.home_dir = tmp.path().to_path_buf(); - retire_inactive_worker(&session); + retire_inactive_worker(&session, &std::collections::HashSet::new()); std::thread::sleep(Duration::from_millis(100)); let alive = crate::platform::process_alive(pid); let _ = crate::platform::kill_process(pid, false); @@ -1438,6 +1460,60 @@ mod tests { } #[cfg(unix)] + #[test] + fn supervisor_skips_workers_it_already_owns() { + // Regression: step 3 moves an evicted child onto the orphan + // list, then immediately calls `retire_inactive_worker` for the + // same (now unselected) session. Without the owned-pid guard + // that call kills our own child, which we have not reaped, so + // it becomes a zombie — and `process_alive` reports a zombie as + // alive. The function would burn its full SIGTERM + SIGKILL + // grace and log a false "SURVIVED SIGKILL" on every eviction. + let tmp = tempdir().unwrap(); + let state = tmp.path().join("state/wire"); + std::fs::create_dir_all(&state).unwrap(); + let mut child = std::process::Command::new("sh") + .args(["-c", "while :; do sleep 1; done", "wire", "daemon"]) + .spawn() + .unwrap(); + let pid = child.id(); + let record = crate::ensure_up::DaemonPid { + schema: crate::ensure_up::DAEMON_PID_SCHEMA.to_string(), + pid, + bin_path: "/opt/wire".to_string(), + version: env!("CARGO_PKG_VERSION").to_string(), + started_at: "2026-08-10T00:00:00Z".to_string(), + did: None, + relay_url: None, + supervisor_managed: true, + }; + std::fs::write( + state.join("daemon.pid"), + serde_json::to_vec(&record).unwrap(), + ) + .unwrap(); + let mut session = initialized_session("supervisor-owned", false); + session.home_dir = tmp.path().to_path_buf(); + + let owned: std::collections::HashSet = [pid].into_iter().collect(); + let started = Instant::now(); + retire_inactive_worker(&session, &owned); + let elapsed = started.elapsed(); + + // Left alone for the orphan drain, and returned immediately + // rather than burning the kill grace. + assert!( + child.try_wait().unwrap().is_none(), + "owned worker was killed by retire_inactive_worker" + ); + assert!( + elapsed < TERM_GRACE, + "retire_inactive_worker blocked on an owned pid: {elapsed:?}" + ); + let _ = child.kill(); + let _ = child.wait(); + } + #[test] fn supervisor_retires_its_own_inactive_daemon() { let tmp = tempdir().unwrap(); @@ -1465,7 +1541,7 @@ mod tests { let mut session = initialized_session("supervisor-owned", false); session.home_dir = tmp.path().to_path_buf(); - retire_inactive_worker(&session); + retire_inactive_worker(&session, &std::collections::HashSet::new()); std::thread::sleep(Duration::from_millis(100)); let status = child.try_wait().unwrap(); if status.is_none() { From fbc8897274eba1ba1f882d32bb7d2ba17ea3f924 Mon Sep 17 00:00:00 2001 From: Paul Logan Date: Tue, 1 Sep 2026 21:00:05 -0700 Subject: [PATCH 3/5] fix(supervisor): correct the idle reaper's guards and stop blocking the poll MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adversarial review of the previous two commits found the idle reaper could destroy live identities, and that verified kills had moved the cost onto the poll loop. Both are fixed here. The reaper's "skip registry-bound homes" guard could never fire. Registry values are human session names (`slancha-api`); by-key directories are `hex(sha256(key)[..8])`, so `bound_names.contains(dir)` compared two namespaces and was always false. The 16-hex shape filter was not a backstop either: a named session's home is `session_dir(name) = session_home_for_key(sanitize_name(name))`, which is also 16 lowercase hex. Both apparent protections were inoperative, and unlike the husk reaper this one deletes homes holding a `private.key` — losing a DID and orphaning every peer's trust entry. `reap_husks` shares the broken check but survives it because its other predicates already exclude every real session; removing those predicates is what made it load-bearing. Guard by resolved home PATH instead, taken from the same `list_sessions()` the supervisor plans from. Path identity has no namespace to confuse. Also refuse any home holding inbox history: received messages exist nowhere else and the outbox check does not cover them. The test that was supposed to catch this instead certified a fiction — it protected a directory literally named `peat-eagle`, a layout that cannot occur. Rewritten against the real resolver via `by_key_dir_name`, and it fails when the guard is removed. Retirement no longer blocks. Verifying a kill inline cost up to 3s and runs once per *unselected* session — 822 of them on the affected box — so a poll could stretch from 10s to tens of minutes, stalling spawns, reaping and the orphan drain. It now sends SIGTERM once, records the pid, and escalates to SIGKILL on a later poll, re-validating the target's cmdline before every signal so a recycled pid is never hit. Also: SIGKILL on a 1.5s grace made two truncating writes reachable — `last_sync.json` (ensure_up.rs) and `notify.cursor` (inbox_watch.rs), the only two state writers in the repo not already using tmp+rename. A torn cursor resets every peer cursor to zero and replays the whole inbox as duplicate toasts, which the code documents. Both now write atomically. Smaller review findings: idle reaping no longer sits inside the husk reaper's gate, so disabling one knob does not silently disable the other; the cheap idleness check runs before the read_dir-and-read-bodies guards; `days * 86_400` saturates rather than wrapping a fat-fingered value into a tiny cutoff. Two weak tests were tightened — escalation now asserts the grace was actually spent, and the idempotence test waits for the child instead of racing it. Known and not addressed: a supervisor killed by signal still leaks its children, since `Child` neither kills nor waits on drop. A restarted supervisor retires unselected leftovers and adopts selected ones, so the population stays bounded at roughly 2x max_workers across a restart rather than growing without limit. --- src/daemon_supervisor.rs | 401 +++++++++++++++++++++++++++------------ src/ensure_up.rs | 9 +- src/inbox_watch.rs | 12 +- 3 files changed, 296 insertions(+), 126 deletions(-) diff --git a/src/daemon_supervisor.rs b/src/daemon_supervisor.rs index 57c4552..8253435 100644 --- a/src/daemon_supervisor.rs +++ b/src/daemon_supervisor.rs @@ -131,6 +131,7 @@ fn classify_session_worker( fn retire_inactive_worker( session: &crate::session::SessionInfo, owned_pids: &std::collections::HashSet, + pending: &mut HashMap, ) { let pidfile = session .home_dir @@ -155,12 +156,36 @@ fn retire_inactive_worker( let Some(version) = classify_session_worker(&record, alive, cmdline.as_deref()) else { return; }; - eprintln!( - "supervisor: retiring unselected session worker '{}' pid={} version={version:?}", - session.name, record.pid - ); - let label = format!("unselected session worker '{}'", session.name); - kill_worker_verified(record.pid, &label); + // Non-blocking, spread across polls. This runs once per *unselected* + // session — 822 of them on the box that motivated this code — so it + // must never sleep: verifying a kill inline at ~3s each would turn a + // 10s poll into a 40-minute one and stall spawns, child reaping and + // the orphan drain along with it. Signal once, escalate on a later + // poll. + // + // `classify_session_worker` re-validates the pid's cmdline above on + // every pass, so a worker that exits and has its pid recycled fails + // that check before we would ever signal the stranger. + match pending.get(&record.pid) { + None => { + eprintln!( + "supervisor: retiring unselected session worker '{}' pid={} version={version:?}", + session.name, record.pid + ); + crate::platform::kill_process(record.pid, false); + pending.insert(record.pid, Instant::now()); + } + Some(sent) if sent.elapsed() >= TERM_GRACE => { + eprintln!( + "supervisor: unselected session worker '{}' pid={} ignored SIGTERM for {:?}; sending SIGKILL", + session.name, + record.pid, + sent.elapsed() + ); + crate::platform::kill_process(record.pid, true); + } + Some(_) => {} + } } /// How long a worker gets to honour SIGTERM before the supervisor @@ -168,26 +193,11 @@ fn retire_inactive_worker( const TERM_GRACE: Duration = Duration::from_millis(1500); const TERM_POLL: Duration = Duration::from_millis(100); -/// Terminate one worker and *verify* it actually died: SIGTERM, wait up -/// to [`TERM_GRACE`] for exit, then SIGKILL and re-check. Returns true -/// iff the pid is gone when we return. -/// -/// ## Why verification, not fire-and-forget -/// -/// The original code sent a single un-escalated SIGTERM per poll and -/// discarded the result. A worker parked in a relay reconnect backoff -/// (1s–30s sleeps against an unreachable relay) does not act on SIGTERM -/// promptly, so every poll re-signalled a process that never died while -/// the supervisor had already dropped it from its bookkeeping. Workers -/// then accumulated without bound: one dev box reached 2,770 live -/// children under a `max_workers=16` cap, 6.2 GB RSS and a load average -/// of 994. Escalating to SIGKILL and confirming death is what makes -/// eviction an actual guarantee instead of a request. /// Terminate a child this supervisor owns and **reap it**, escalating /// SIGTERM -> SIGKILL. Returns true iff the child has been reaped. /// -/// Reaping is the half that `kill_worker_verified` cannot do: a killed -/// direct child stays in the process table as a zombie until somebody +/// Reaping is the half a bare signal cannot do: a killed direct child +/// stays in the process table as a zombie until somebody /// `wait`s on it, and `process_alive` (a `kill(pid, 0)` probe) reports /// a zombie as *alive*. Evicted workers used to be dropped without a /// wait, so they lingered as unreapable entries and the supervisor @@ -217,31 +227,6 @@ fn terminate_and_reap(child: &mut Child, pid: u32, label: &str) -> bool { false } -fn kill_worker_verified(pid: u32, label: &str) -> bool { - if !crate::platform::process_alive(pid) { - return true; - } - crate::platform::kill_process(pid, false); - let deadline = Instant::now() + TERM_GRACE; - while Instant::now() < deadline { - if !crate::platform::process_alive(pid) { - return true; - } - std::thread::sleep(TERM_POLL); - } - eprintln!("supervisor: {label} pid={pid} ignored SIGTERM after {TERM_GRACE:?}; sending SIGKILL"); - crate::platform::kill_process(pid, true); - let deadline = Instant::now() + TERM_GRACE; - while Instant::now() < deadline { - if !crate::platform::process_alive(pid) { - return true; - } - std::thread::sleep(TERM_POLL); - } - eprintln!("supervisor: {label} pid={pid} SURVIVED SIGKILL; leaving it for the next poll"); - false -} - /// Newest mtime among a session home's activity files — the /// supervisor's "last actually *synced*" signal. These live under the /// session's `state/wire/` subtree (same root the per-session daemon @@ -296,6 +281,15 @@ fn fs_has_pending_outbox(home: &Path) -> bool { }) } +/// True iff the home holds received-message history. The outbox check +/// asks "would we lose something we owe a peer"; this asks "would we +/// lose something a peer already sent us". A home with an inbox is +/// never idle-reapable no matter how long it has sat. +fn fs_has_inbox_history(home: &Path) -> bool { + let inbox = home.join("state").join("wire").join("inbox"); + std::fs::read_dir(inbox).is_ok_and(|entries| entries.flatten().next().is_some()) +} + #[derive(Debug, Clone)] struct SupervisorPlan { selected: Vec, @@ -498,7 +492,7 @@ fn parse_idle_reap_max_age(raw: Option<&str>) -> Option { match raw { Some(v) => { let days: u64 = v.trim().parse().unwrap_or(DEFAULT_IDLE_REAP_MAX_AGE_DAYS); - (days != 0).then(|| Duration::from_secs(days * 86_400)) + (days != 0).then(|| Duration::from_secs(days.saturating_mul(86_400))) } None => Some(Duration::from_secs(DEFAULT_IDLE_REAP_MAX_AGE_DAYS * 86_400)), } @@ -525,25 +519,43 @@ fn idle_reap_max_age_from_env() -> Option { /// husk predicate, while the supervisor stat-ed all of them on every /// 10s registry poll. /// -/// The husk predicate stays as-is — this is a strictly separate, much -/// more conservative path keyed on *idleness* rather than emptiness. A -/// dir is reaped only if ALL of these hold: -/// - its name has the by-key shape (16 lowercase hex chars) — named, -/// operator-created sessions are never touched; -/// - it is not registry-bound; -/// - it holds no live lease and no pending outbox (nothing would be -/// lost by removing it); +/// The husk predicate stays as-is — this is a strictly separate path +/// keyed on *idleness* rather than emptiness. A dir is reaped only if +/// ALL of these hold: +/// - its name has the by-key shape (16 lowercase hex chars); +/// - its path is not in `protected` (see below); +/// - it holds no live lease, no pending outbox, and no inbox history — +/// nothing would be lost by removing it; /// - no live daemon owns it; /// - its last activity (or, for a never-synced home, its own mtime) is /// older than `max_age`. Future timestamps count as young, so clock /// skew never deletes. /// +/// ## `protected` is matched by PATH, never by name +/// +/// The obvious guard — "skip anything in the registry" — does not work +/// by name, and `reap_husks` gets away with it only because its other +/// predicates already exclude every real session. Registry values are +/// human session names (`slancha-api`); by-key directories are +/// `hex(sha256(key)[..8])`. A `bound_names.contains(dir_name)` test +/// compares two different namespaces and is therefore *always false*. +/// The 16-hex shape filter is not a backstop either: a named session's +/// home is `session_dir(name) = session_home_for_key(sanitize_name(name))`, +/// which is also 16 lowercase hex. So both "protections" a name-based +/// guard appears to offer are inoperative, and this reaper — unlike the +/// husk one — deletes homes that hold a `private.key`. Getting it wrong +/// destroys a DID and orphans every peer's trust entry. +/// +/// The caller therefore passes resolved home *paths* from +/// `list_sessions()`, which is the same source the supervisor plans +/// from. Path identity has no namespace to confuse. +/// /// Failures are per-entry best-effort (warn + continue). fn reap_idle_homes( by_key_root: &Path, max_age: Duration, now: SystemTime, - bound_names: &std::collections::HashSet, + protected: &std::collections::HashSet, is_active: A, daemon_live: D, ) -> Vec @@ -568,17 +580,14 @@ where if !is_by_key_shape { continue; } - if bound_names.contains(name) { - continue; - } - if is_active(&path) { - continue; - } - if daemon_live(&path) { + if protected.contains(&path) { continue; } - // Prefer real activity; fall back to the home's own mtime so a - // home that never synced still ages out on this path. + // Cheapest decisive test first: almost every home fails the + // idleness check, and the guards below cost a read_dir plus the + // body of every pending outbox file. Prefer real activity; fall + // back to the home's own mtime so a home that never synced + // still ages out on this path. let last = fs_last_active(&path).or_else(|| { std::fs::metadata(&path) .and_then(|m| m.modified()) @@ -590,6 +599,12 @@ where if !idle_long_enough { continue; } + if is_active(&path) { + continue; + } + if daemon_live(&path) { + continue; + } match std::fs::remove_dir_all(&path) { Ok(()) => reaped.push(path), Err(e) => eprintln!("supervisor: idle reap failed for {}: {e:#}", path.display()), @@ -670,6 +685,7 @@ pub fn run_supervisor(interval_secs: u64, max_workers: usize, as_json: bool) -> } ); let mut last_husk_reap: Option = None; + let mut last_idle_reap: Option = None; let mut children: HashMap = HashMap::new(); // Children the supervisor has stopped selecting but has not yet @@ -680,6 +696,10 @@ pub fn run_supervisor(interval_secs: u64, max_workers: usize, as_json: bool) -> // so an untracked eviction used to leak a live process (2,770 of // them under a max_workers=16 cap on one box). let mut orphans: Vec = Vec::new(); + // Unselected workers signalled but not yet confirmed dead: pid -> + // when SIGTERM was sent. Lets retirement escalate across polls + // instead of blocking inside one. + let mut pending_retire: HashMap = HashMap::new(); // Per-session backoff that survives a child's reap → respawn → reap // cycle. Distinguishes "session crashes hard repeatedly" from // "child exited cleanly and we're spawning a fresh one". @@ -775,24 +795,43 @@ pub fn run_supervisor(interval_secs: u64, max_workers: usize, as_json: bool) -> .join(", ") ); } - if let Some(idle_age) = idle_max_age { - let idle_reaped = reap_idle_homes( - &root.join("by-key"), - idle_age, - SystemTime::now(), - &bound, - |home| fs_has_live_lease(home) || fs_has_pending_outbox(home), - // On a liveness-probe error assume live — never - // delete a home we couldn't safely inspect. - |home| existing_daemon_for_session(home).unwrap_or(true), - ); - if !idle_reaped.is_empty() { - eprintln!( - "supervisor: reaped {} idle session home(s)", - idle_reaped.len() - ); - } - } + } + } + + // 2c. Idle sweep. Gated independently of the husk sweep: the two + // cutoffs are documented as separate knobs, so disabling one + // must not silently disable the other. + if let Some(idle_age) = idle_max_age + && last_idle_reap.is_none_or(|t| t.elapsed() >= HUSK_REAP_INTERVAL) + && let Ok(root) = crate::session::sessions_root() + { + last_idle_reap = Some(Instant::now()); + // Protect by resolved path, never by name — see + // `reap_idle_homes`. Every home the registry currently + // knows about is off limits, whether or not it is selected. + let protected: std::collections::HashSet = all_sessions + .iter() + .map(|session| session.home_dir.clone()) + .collect(); + let idle_reaped = reap_idle_homes( + &root.join("by-key"), + idle_age, + SystemTime::now(), + &protected, + |home| { + fs_has_live_lease(home) + || fs_has_pending_outbox(home) + || fs_has_inbox_history(home) + }, + // On a liveness-probe error assume live — never delete a + // home we couldn't safely inspect. + |home| existing_daemon_for_session(home).unwrap_or(true), + ); + if !idle_reaped.is_empty() { + eprintln!( + "supervisor: reaped {} idle session home(s)", + idle_reaped.len() + ); } } @@ -821,9 +860,12 @@ pub fn run_supervisor(interval_secs: u64, max_workers: usize, as_json: bool) -> .collect(); for session in &all_sessions { if !wanted_names.contains(&session.name) { - retire_inactive_worker(session, &owned_pids); + retire_inactive_worker(session, &owned_pids, &mut pending_retire); } } + // Forget workers that are gone, so a recycled pid can never + // inherit a stale escalation deadline. + pending_retire.retain(|pid, _| crate::platform::process_alive(*pid)); // 4. Spawn missing children, respecting backoff + existing // pidfiles (operator-spawned daemons coexist). @@ -1448,7 +1490,11 @@ mod tests { let mut session = initialized_session("operator-started", false); session.home_dir = tmp.path().to_path_buf(); - retire_inactive_worker(&session, &std::collections::HashSet::new()); + retire_inactive_worker( + &session, + &std::collections::HashSet::new(), + &mut HashMap::new(), + ); std::thread::sleep(Duration::from_millis(100)); let alive = crate::platform::process_alive(pid); let _ = crate::platform::kill_process(pid, false); @@ -1497,7 +1543,7 @@ mod tests { let owned: std::collections::HashSet = [pid].into_iter().collect(); let started = Instant::now(); - retire_inactive_worker(&session, &owned); + retire_inactive_worker(&session, &owned, &mut HashMap::new()); let elapsed = started.elapsed(); // Left alone for the orphan drain, and returned immediately @@ -1541,7 +1587,11 @@ mod tests { let mut session = initialized_session("supervisor-owned", false); session.home_dir = tmp.path().to_path_buf(); - retire_inactive_worker(&session, &std::collections::HashSet::new()); + retire_inactive_worker( + &session, + &std::collections::HashSet::new(), + &mut HashMap::new(), + ); std::thread::sleep(Duration::from_millis(100)); let status = child.try_wait().unwrap(); if status.is_none() { @@ -1763,6 +1813,7 @@ mod tests { let tmp = tempfile::tempdir().unwrap(); mk_identity_home(tmp.path(), "aaaaaaaaaaaaaaaa"); let bound = std::collections::HashSet::new(); + let protected = std::collections::HashSet::new(); // The husk reaper is blind to it — that is the bug. let husks = reap_husks(tmp.path(), CUTOFF_48H, far_future_days(), &bound, |_| false); assert!(husks.is_empty()); @@ -1771,7 +1822,7 @@ mod tests { tmp.path(), CUTOFF_14D, far_future_days(), - &bound, + &protected, |_| false, |_| false, ); @@ -1783,12 +1834,12 @@ mod tests { fn idle_reap_keeps_recently_active_home() { let tmp = tempfile::tempdir().unwrap(); mk_identity_home(tmp.path(), "bbbbbbbbbbbbbbbb"); - let bound = std::collections::HashSet::new(); + let protected = std::collections::HashSet::new(); let reaped = reap_idle_homes( tmp.path(), CUTOFF_14D, SystemTime::now(), - &bound, + &protected, |_| false, |_| false, ); @@ -1800,13 +1851,13 @@ mod tests { fn idle_reap_keeps_home_with_live_lease_or_outbox() { let tmp = tempfile::tempdir().unwrap(); mk_identity_home(tmp.path(), "cccccccccccccccc"); - let bound = std::collections::HashSet::new(); + let protected = std::collections::HashSet::new(); let reaped = reap_idle_homes( tmp.path(), CUTOFF_14D, far_future_days(), - &bound, - |_| true, // active: live lease or pending outbox + &protected, + |_| true, // active: live lease, pending outbox or inbox |_| false, ); assert!(reaped.is_empty()); @@ -1816,12 +1867,12 @@ mod tests { fn idle_reap_keeps_home_with_live_daemon() { let tmp = tempfile::tempdir().unwrap(); mk_identity_home(tmp.path(), "dddddddddddddddd"); - let bound = std::collections::HashSet::new(); + let protected = std::collections::HashSet::new(); let reaped = reap_idle_homes( tmp.path(), CUTOFF_14D, far_future_days(), - &bound, + &protected, |_| false, |_| true, ); @@ -1829,24 +1880,62 @@ mod tests { } #[test] - fn idle_reap_keeps_registry_bound_and_named_homes() { + fn idle_reap_keeps_homes_the_registry_knows_about() { + // The layout this must survive is the REAL one. A named + // session's home is not a directory called "peat-eagle" — it is + // `by-key/`, exactly the + // same shape as an adoption husk. Registry *values* are names + // and by-key *directories* are hashes, so the obvious + // `bound_names.contains(dir_name)` guard compares two + // namespaces and never fires. An earlier version of this test + // invented a layout in which both guards worked and therefore + // certified a reaper that would have deleted live identities. let tmp = tempfile::tempdir().unwrap(); - mk_identity_home(tmp.path(), "eeeeeeeeeeeeeeee"); - mk_identity_home(tmp.path(), "peat-eagle"); - let mut bound = std::collections::HashSet::new(); - bound.insert("eeeeeeeeeeeeeeee".to_string()); + let named_dir = crate::session::by_key_dir_name("slancha-api"); + assert_eq!(named_dir.len(), 16, "named session home is by-key shaped"); + let named_home = mk_identity_home(tmp.path(), &named_dir); + let husk_home = mk_identity_home(tmp.path(), "aaaaaaaaaaaaaaaa"); + + // Protection is by resolved path, as the supervisor passes it. + let protected: std::collections::HashSet = + [named_home.clone()].into_iter().collect(); let reaped = reap_idle_homes( tmp.path(), CUTOFF_14D, far_future_days(), - &bound, + &protected, |_| false, |_| false, ); - assert!(reaped.is_empty()); - assert!(tmp.path().join("peat-eagle").exists()); + + assert!(named_home.exists(), "deleted a registry-known session home"); + assert_eq!(reaped, vec![husk_home]); } + #[test] + fn idle_reap_keeps_home_holding_inbox_history() { + // Received messages are not recoverable from anywhere else, and + // the outbox guard does not cover them. + let tmp = tempfile::tempdir().unwrap(); + let home = mk_identity_home(tmp.path(), "ffffffffffffffff"); + std::fs::create_dir_all(home.join("state").join("wire").join("inbox")).unwrap(); + std::fs::write( + home.join("state").join("wire").join("inbox").join("m.jsonl"), + b"{}", + ) + .unwrap(); + assert!(fs_has_inbox_history(&home)); + let reaped = reap_idle_homes( + tmp.path(), + CUTOFF_14D, + far_future_days(), + &std::collections::HashSet::new(), + |h| fs_has_inbox_history(h), + |_| false, + ); + assert!(reaped.is_empty()); + assert!(home.exists()); + } #[test] fn idle_reap_max_age_parsing() { assert_eq!( @@ -1864,13 +1953,6 @@ mod tests { ); } - #[test] - fn kill_worker_verified_reports_dead_pid_as_gone() { - // A pid that is definitively not running must short-circuit to - // "already dead" without signalling anything. - assert!(kill_worker_verified(u32::MAX - 1, "test worker")); - } - #[test] fn terminate_and_reap_escalates_past_a_sigterm_ignoring_child() { // The regression this guards: eviction used to send one @@ -1880,26 +1962,99 @@ mod tests { // `sh -c 'trap "" TERM; sleep 60'` ignores SIGTERM outright, so // only SIGKILL escalation can end it — and only a `wait` can // reap it afterwards. - let child = Command::new("sh") + let mut child = Command::new("sh") .args(["-c", "trap '' TERM; sleep 60"]) - .spawn(); - let Ok(mut child) = child else { - return; // no shell available — nothing to assert - }; + .spawn() + .expect("spawning a shell"); let pid = child.id(); assert!(crate::platform::process_alive(pid)); + let started = Instant::now(); assert!(terminate_and_reap(&mut child, pid, "sigterm-ignoring test worker")); + // Must have gone the long way round: SIGTERM, full grace, then + // SIGKILL. An implementation that skipped the grace, or opened + // with SIGKILL, would return well inside TERM_GRACE and pass a + // bare `assert!(...)` while breaking clean shutdown for every + // worker that *does* honour SIGTERM. + assert!( + started.elapsed() >= TERM_GRACE, + "returned in {:?}, before the SIGTERM grace elapsed", + started.elapsed() + ); } #[test] - fn terminate_and_reap_is_idempotent_on_an_already_dead_child() { - let child = Command::new("sh").args(["-c", "exit 0"]).spawn(); - let Ok(mut child) = child else { - return; - }; + fn terminate_and_reap_returns_immediately_for_an_exited_child() { + let mut child = Command::new("sh") + .args(["-c", "exit 0"]) + .spawn() + .expect("spawning a shell"); let pid = child.id(); + // Wait for the exit first, otherwise this races: an unexited + // child sends the call down the SIGTERM path and the test can no + // longer tell which branch it exercised. + child.wait().expect("waiting for the child"); + let started = Instant::now(); assert!(terminate_and_reap(&mut child, pid, "short-lived test worker")); - assert!(terminate_and_reap(&mut child, pid, "short-lived test worker")); + assert!( + started.elapsed() < TERM_GRACE, + "burned the kill grace on an already-dead child: {:?}", + started.elapsed() + ); + } + + #[test] + fn retire_escalates_across_polls_without_ever_blocking() { + // Retirement runs once per unselected session — hundreds per + // poll — so it must signal and return, never verify inline. + let tmp = tempdir().unwrap(); + let state = tmp.path().join("state/wire"); + std::fs::create_dir_all(&state).unwrap(); + let mut child = std::process::Command::new("sh") + .args(["-c", "trap '' TERM; while :; do sleep 1; done", "wire", "daemon"]) + .spawn() + .unwrap(); + let pid = child.id(); + let record = crate::ensure_up::DaemonPid { + schema: crate::ensure_up::DAEMON_PID_SCHEMA.to_string(), + pid, + bin_path: "/opt/wire".to_string(), + version: env!("CARGO_PKG_VERSION").to_string(), + started_at: "2026-08-10T00:00:00Z".to_string(), + did: None, + relay_url: None, + supervisor_managed: true, + }; + std::fs::write( + state.join("daemon.pid"), + serde_json::to_vec(&record).unwrap(), + ) + .unwrap(); + let mut session = initialized_session("stubborn", false); + session.home_dir = tmp.path().to_path_buf(); + let owned = std::collections::HashSet::new(); + let mut pending = HashMap::new(); + + // Poll 1: SIGTERM only, and it returns immediately even though + // this worker ignores SIGTERM entirely. + let started = Instant::now(); + retire_inactive_worker(&session, &owned, &mut pending); + assert!( + started.elapsed() < TERM_GRACE, + "blocked inside a single poll: {:?}", + started.elapsed() + ); + assert!(pending.contains_key(&pid)); + assert!(child.try_wait().unwrap().is_none(), "SIGTERM should not have killed it"); + + // Poll 2, before the grace elapses: still no escalation. + retire_inactive_worker(&session, &owned, &mut pending); + assert!(child.try_wait().unwrap().is_none()); + + // Poll N, after the grace: escalate to SIGKILL. + pending.insert(pid, Instant::now() - TERM_GRACE - Duration::from_millis(50)); + retire_inactive_worker(&session, &owned, &mut pending); + let status = child.wait().expect("reaping the killed worker"); + assert!(!status.success(), "worker should have been SIGKILLed"); } #[test] diff --git a/src/ensure_up.rs b/src/ensure_up.rs index 003d2b8..16bc2f5 100644 --- a/src/ensure_up.rs +++ b/src/ensure_up.rs @@ -402,7 +402,14 @@ pub fn write_last_sync_record(push_n: usize, pull_n: usize, rejected_n: usize) { std::fs::create_dir_all(parent)?; } let body = serde_json::to_vec_pretty(&record)?; - std::fs::write(&path, body)?; + // tmp + rename: `fs::write` truncates in place, so a daemon + // killed mid-write leaves a 0-byte or half-JSON file and + // `read_last_sync_record` then reports a healthy session as + // "never synced". The supervisor SIGKILLs unresponsive workers + // on a short grace, so this window is reachable by design. + let tmp = path.with_extension("json.tmp"); + std::fs::write(&tmp, body)?; + std::fs::rename(&tmp, &path)?; Ok(()) })() .map_err(|e| eprintln!("daemon: last-sync persist error (non-fatal): {e:#}")); diff --git a/src/inbox_watch.rs b/src/inbox_watch.rs index 8cf7df2..58362d9 100644 --- a/src/inbox_watch.rs +++ b/src/inbox_watch.rs @@ -197,8 +197,16 @@ impl InboxWatcher { std::fs::create_dir_all(parent).with_context(|| format!("creating {parent:?}"))?; } let bytes = serde_json::to_vec(&self.cursors)?; - std::fs::write(cursor_path, bytes) - .with_context(|| format!("writing cursor file {cursor_path:?}"))?; + // tmp + rename. `fs::write` truncates in place, and a torn + // cursor file resets ALL peer cursors to zero — the duplicate + // toast storm documented in `load`. The supervisor SIGKILLs + // unresponsive workers on a short grace, so a truncating write + // here is a reachable path to that storm, not a theoretical one. + let tmp = cursor_path.with_extension("cursor.tmp"); + std::fs::write(&tmp, bytes) + .with_context(|| format!("writing cursor file {tmp:?}"))?; + std::fs::rename(&tmp, cursor_path) + .with_context(|| format!("renaming {tmp:?} -> {cursor_path:?}"))?; Ok(()) } From b7155db79f7198db8cf67035d957dd6977544698 Mon Sep 17 00:00:00 2001 From: Paul Logan Date: Tue, 1 Sep 2026 21:11:06 -0700 Subject: [PATCH 4/5] style: cargo fmt --- src/daemon_supervisor.rs | 50 ++++++++++++++++++++++++++++------------ src/inbox_watch.rs | 3 +-- 2 files changed, 36 insertions(+), 17 deletions(-) diff --git a/src/daemon_supervisor.rs b/src/daemon_supervisor.rs index 8253435..711505b 100644 --- a/src/daemon_supervisor.rs +++ b/src/daemon_supervisor.rs @@ -214,7 +214,9 @@ fn terminate_and_reap(child: &mut Child, pid: u32, label: &str) -> bool { } std::thread::sleep(TERM_POLL); } - eprintln!("supervisor: {label} pid={pid} ignored SIGTERM after {TERM_GRACE:?}; sending SIGKILL"); + eprintln!( + "supervisor: {label} pid={pid} ignored SIGTERM after {TERM_GRACE:?}; sending SIGKILL" + ); let _ = child.kill(); let deadline = Instant::now() + TERM_GRACE; while Instant::now() < deadline { @@ -501,9 +503,7 @@ fn parse_idle_reap_max_age(raw: Option<&str>) -> Option { /// Read the idle reap cutoff from the environment. /// `WIRE_IDLE_REAP_MAX_AGE_DAYS=0` disables idle reaping entirely. fn idle_reap_max_age_from_env() -> Option { - parse_idle_reap_max_age( - std::env::var("WIRE_IDLE_REAP_MAX_AGE_DAYS").ok().as_deref(), - ) + parse_idle_reap_max_age(std::env::var("WIRE_IDLE_REAP_MAX_AGE_DAYS").ok().as_deref()) } /// Delete long-idle by-key session homes that DO hold an identity, and @@ -588,11 +588,8 @@ where // body of every pending outbox file. Prefer real activity; fall // back to the home's own mtime so a home that never synced // still ages out on this path. - let last = fs_last_active(&path).or_else(|| { - std::fs::metadata(&path) - .and_then(|m| m.modified()) - .ok() - }); + let last = fs_last_active(&path) + .or_else(|| std::fs::metadata(&path).and_then(|m| m.modified()).ok()); let idle_long_enough = last .and_then(|t| now.duration_since(t).ok()) .is_some_and(|age| age >= max_age); @@ -1804,7 +1801,11 @@ mod tests { let home = mk_husk(root, name); std::fs::create_dir_all(home.join("config").join("wire")).unwrap(); std::fs::write(home.join("config").join("wire").join("private.key"), b"k").unwrap(); - std::fs::write(home.join("state").join("wire").join("last_sync.json"), b"{}").unwrap(); + std::fs::write( + home.join("state").join("wire").join("last_sync.json"), + b"{}", + ) + .unwrap(); home } @@ -1920,7 +1921,10 @@ mod tests { let home = mk_identity_home(tmp.path(), "ffffffffffffffff"); std::fs::create_dir_all(home.join("state").join("wire").join("inbox")).unwrap(); std::fs::write( - home.join("state").join("wire").join("inbox").join("m.jsonl"), + home.join("state") + .join("wire") + .join("inbox") + .join("m.jsonl"), b"{}", ) .unwrap(); @@ -1969,7 +1973,11 @@ mod tests { let pid = child.id(); assert!(crate::platform::process_alive(pid)); let started = Instant::now(); - assert!(terminate_and_reap(&mut child, pid, "sigterm-ignoring test worker")); + assert!(terminate_and_reap( + &mut child, + pid, + "sigterm-ignoring test worker" + )); // Must have gone the long way round: SIGTERM, full grace, then // SIGKILL. An implementation that skipped the grace, or opened // with SIGKILL, would return well inside TERM_GRACE and pass a @@ -1994,7 +2002,11 @@ mod tests { // longer tell which branch it exercised. child.wait().expect("waiting for the child"); let started = Instant::now(); - assert!(terminate_and_reap(&mut child, pid, "short-lived test worker")); + assert!(terminate_and_reap( + &mut child, + pid, + "short-lived test worker" + )); assert!( started.elapsed() < TERM_GRACE, "burned the kill grace on an already-dead child: {:?}", @@ -2010,7 +2022,12 @@ mod tests { let state = tmp.path().join("state/wire"); std::fs::create_dir_all(&state).unwrap(); let mut child = std::process::Command::new("sh") - .args(["-c", "trap '' TERM; while :; do sleep 1; done", "wire", "daemon"]) + .args([ + "-c", + "trap '' TERM; while :; do sleep 1; done", + "wire", + "daemon", + ]) .spawn() .unwrap(); let pid = child.id(); @@ -2044,7 +2061,10 @@ mod tests { started.elapsed() ); assert!(pending.contains_key(&pid)); - assert!(child.try_wait().unwrap().is_none(), "SIGTERM should not have killed it"); + assert!( + child.try_wait().unwrap().is_none(), + "SIGTERM should not have killed it" + ); // Poll 2, before the grace elapses: still no escalation. retire_inactive_worker(&session, &owned, &mut pending); diff --git a/src/inbox_watch.rs b/src/inbox_watch.rs index 58362d9..8103719 100644 --- a/src/inbox_watch.rs +++ b/src/inbox_watch.rs @@ -203,8 +203,7 @@ impl InboxWatcher { // unresponsive workers on a short grace, so a truncating write // here is a reachable path to that storm, not a theoretical one. let tmp = cursor_path.with_extension("cursor.tmp"); - std::fs::write(&tmp, bytes) - .with_context(|| format!("writing cursor file {tmp:?}"))?; + std::fs::write(&tmp, bytes).with_context(|| format!("writing cursor file {tmp:?}"))?; std::fs::rename(&tmp, cursor_path) .with_context(|| format!("renaming {tmp:?} -> {cursor_path:?}"))?; Ok(()) From 6634adc5483983b2f0a1caace3e3ad97176c3e33 Mon Sep 17 00:00:00 2001 From: Paul Logan Date: Tue, 1 Sep 2026 21:18:32 -0700 Subject: [PATCH 5/5] style: drop a redundant closure clippy flags under -D warnings --- src/daemon_supervisor.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/daemon_supervisor.rs b/src/daemon_supervisor.rs index 711505b..4a1a910 100644 --- a/src/daemon_supervisor.rs +++ b/src/daemon_supervisor.rs @@ -1934,7 +1934,7 @@ mod tests { CUTOFF_14D, far_future_days(), &std::collections::HashSet::new(), - |h| fs_has_inbox_history(h), + fs_has_inbox_history, |_| false, ); assert!(reaped.is_empty());