fix(supervisor): serialize child registration with reaping

crates/openshell-supervisor-process/src/managed_children.rs

@@ -11,25 +11,42 @@
 #![cfg(target_os = "linux")]
 
 use std::collections::HashSet;
-use std::sync::{LazyLock, Mutex};
+use std::sync::{LazyLock, Mutex, MutexGuard, PoisonError};
 
 static MANAGED_CHILDREN: LazyLock<Mutex<HashSet<i32>>> =
     LazyLock::new(|| Mutex::new(HashSet::new()));
 
-/// Add `pid` to the supervised-child set. Non-positive or out-of-range values
-/// are silently ignored.
-pub fn register(pid: u32) {
-    let Ok(pid) = i32::try_from(pid) else {
+fn lock_children() -> MutexGuard<'static, HashSet<i32>> {
+    MANAGED_CHILDREN
+        .lock()
+        .unwrap_or_else(PoisonError::into_inner)
+}
+
+fn insert(children: &mut HashSet<i32>, pid: Option<u32>) {
+    let Some(pid) = pid.and_then(|pid| i32::try_from(pid).ok()) else {
         return;
     };
-    if pid <= 0 {
-        return;
-    }
-    if let Ok(mut children) = MANAGED_CHILDREN.lock() {
+    if pid > 0 {
         children.insert(pid);
     }
 }
 
+/// Spawn a supervised child and register its PID before the orphan reaper can
+/// inspect it.
+///
+/// The registry lock intentionally spans `spawn`. Otherwise a fast child can
+/// exit and be reaped after `spawn` returns but before its PID is registered,
+/// causing the child's explicit waiter to fail with `ECHILD`.
+pub fn spawn_registered<T, E>(
+    spawn: impl FnOnce() -> Result<T, E>,
+    pid: impl FnOnce(&T) -> Option<u32>,
+) -> Result<T, E> {
+    let mut children = lock_children();
+    let child = spawn()?;
+    insert(&mut children, pid(&child));
+    Ok(child)
+}
+
 /// Remove `pid` from the supervised-child set. Non-positive or out-of-range
 /// values are silently ignored.
 pub fn unregister(pid: u32) {
@@ -39,15 +56,132 @@ pub fn unregister(pid: u32) {
     if pid <= 0 {
         return;
     }
-    if let Ok(mut children) = MANAGED_CHILDREN.lock() {
-        children.remove(&pid);
+    lock_children().remove(&pid);
+}
+
+/// Run `reap` only when `pid` is not a supervised child.
+///
+/// The registry lock remains held while `reap` runs so child creation cannot
+/// open a spawn-to-registration window between the membership check and reap.
+pub fn reap_if_unmanaged<T>(pid: i32, reap: impl FnOnce() -> T) -> Option<T> {
+    let children = lock_children();
+    if children.contains(&pid) {
+        None
+    } else {
+        Some(reap())
     }
 }
 
-/// Return `true` if `pid` is currently in the supervised-child set.
-#[must_use]
-pub fn is_managed(pid: i32) -> bool {
-    MANAGED_CHILDREN
-        .lock()
-        .is_ok_and(|children| children.contains(&pid))
+#[cfg(test)]
+mod tests {
+    use super::{reap_if_unmanaged, spawn_registered, unregister};
+    use nix::sys::wait::{Id, WaitPidFlag, WaitStatus, waitid, waitpid};
+    use nix::unistd::Pid;
+    use std::process::Command;
+    use std::sync::mpsc;
+    use std::time::Duration;
+
+    #[test]
+    fn spawned_child_is_registered_before_returning() {
+        let pid = 1_000_000_u32;
+        let child = spawn_registered(|| Ok::<_, ()>(pid), |pid| Some(*pid)).unwrap();
+        let mut reaped = false;
+
+        let result = reap_if_unmanaged(i32::try_from(child).unwrap(), || {
+            reaped = true;
+        });
+
+        assert!(result.is_none());
+        assert!(!reaped);
+        unregister(pid);
+    }
+
+    #[test]
+    fn reaper_cannot_steal_child_while_registration_is_in_progress() {
+        // Keep `spawn` paused after its logical child exists but before it
+        // returns the PID. This is the exact window that previously let the
+        // orphan reaper consume a fast child's status.
+        let pid = 1_000_002_u32;
+        let (spawn_entered_tx, spawn_entered_rx) = mpsc::channel();
+        let (complete_spawn_tx, complete_spawn_rx) = mpsc::channel();
+        let (reap_attempted_tx, reap_attempted_rx) = mpsc::channel();
+        let (reap_result_tx, reap_result_rx) = mpsc::channel();
+
+        let spawn = std::thread::spawn(move || {
+            spawn_registered(
+                || {
+                    spawn_entered_tx.send(()).unwrap();
+                    complete_spawn_rx.recv().unwrap();
+                    Ok::<_, ()>(pid)
+                },
+                |child| Some(*child),
+            )
+            .unwrap();
+        });
+        spawn_entered_rx.recv().unwrap();
+
+        let reaper = std::thread::spawn(move || {
+            reap_attempted_tx.send(()).unwrap();
+            let result = reap_if_unmanaged(i32::try_from(pid).unwrap(), || "reaped");
+            reap_result_tx.send(result).unwrap();
+        });
+        reap_attempted_rx.recv().unwrap();
+
+        assert!(
+            reap_result_rx
+                .recv_timeout(Duration::from_millis(50))
+                .is_err(),
+            "the reaper must wait until the child is registered"
+        );
+
+        complete_spawn_tx.send(()).unwrap();
+        spawn.join().unwrap();
+        assert_eq!(reap_result_rx.recv().unwrap(), None);
+        reaper.join().unwrap();
+        unregister(pid);
+    }
+
+    #[test]
+    fn unmanaged_child_can_be_reaped() {
+        let result = reap_if_unmanaged(1_000_001, || "reaped");
+
+        assert_eq!(result, Some("reaped"));
+    }
+
+    #[test]
+    fn fast_child_remains_waitable_after_orphan_reap_attempt() {
+        let mut child = spawn_registered(
+            || Command::new("sh").args(["-c", "exit 11"]).spawn(),
+            |child| Some(child.id()),
+        )
+        .expect("spawn and register fast child");
+        let child_pid = child.id();
+        let pid = Pid::from_raw(i32::try_from(child_pid).unwrap());
+
+        // Observe the completed child without consuming its status, exactly
+        // as the orphan reaper does before its managed-PID check.
+        loop {
+            match waitid(
+                Id::Pid(pid),
+                WaitPidFlag::WEXITED | WaitPidFlag::WNOHANG | WaitPidFlag::WNOWAIT,
+            ) {
+                Ok(WaitStatus::StillAlive) => std::thread::yield_now(),
+                Ok(_) => break,
+                Err(error) => panic!("observe fast child: {error}"),
+            }
+        }
+
+        let reaped = reap_if_unmanaged(pid.as_raw(), || waitpid(pid, Some(WaitPidFlag::WNOHANG)));
+        assert!(
+            reaped.is_none(),
+            "the orphan reaper must leave a registered child to its explicit waiter"
+        );
+
+        let wait = child.wait();
+        unregister(child_pid);
+        assert!(
+            wait.is_ok(),
+            "the explicit child waiter must retain the exit status"
+        );
+    }
 }

crates/openshell-supervisor-process/src/process.rs

@@ -901,17 +901,18 @@ impl ProcessHandle {
         }
 
         #[cfg(target_os = "linux")]
-        let mut child = spawn_command_with_supervisor_identity_namespace(cmd)
-            .into_diagnostic()
-            .wrap_err("failed to spawn sandbox entrypoint process")?;
+        let mut child = managed_children::spawn_registered(
+            || spawn_command_with_supervisor_identity_namespace(cmd),
+            Child::id,
+        )
+        .into_diagnostic()
+        .wrap_err("failed to spawn sandbox entrypoint process")?;
         #[cfg(not(target_os = "linux"))]
         let mut child = cmd
             .spawn()
             .into_diagnostic()
             .wrap_err("failed to spawn sandbox entrypoint process")?;
         let pid = child.id().unwrap_or(0);
-        managed_children::register(pid);
-
         let io = if let Some(master) = pty_master {
             ProcessIo::Pty(master)
         } else {
@@ -1055,8 +1056,6 @@ impl ProcessHandle {
 
         let mut child = cmd.spawn().into_diagnostic()?;
         let pid = child.id().unwrap_or(0);
-        #[cfg(target_os = "linux")]
-        managed_children::register(pid);
 
         debug!(pid, program, "Process spawned");
 

crates/openshell-supervisor-process/src/run.rs

@@ -212,12 +212,15 @@ pub async fn run_process(
                     break;
                 };
 
-                if managed_children::is_managed(pid.as_raw()) {
-                    // Let the explicit waiter own this child status.
+                let Some(reap_result) = managed_children::reap_if_unmanaged(pid.as_raw(), || {
+                    waitpid(pid, Some(WaitPidFlag::WNOHANG))
+                }) else {
+                    // Let the explicit waiter own this child status. Stop here
+                    // because WNOWAIT will keep returning this same child.
                     break;
-                }
+                };
 
-                match waitpid(pid, Some(WaitPidFlag::WNOHANG)) {
+                match reap_result {
                     Ok(WaitStatus::StillAlive)
                     | Err(nix::errno::Errno::ECHILD | nix::errno::Errno::EINTR) => {}
                     Ok(reaped) => {

crates/openshell-supervisor-process/src/ssh.rs

@@ -1342,13 +1342,14 @@ fn spawn_pty_shell(
     }
 
     #[cfg(target_os = "linux")]
-    let mut child = crate::process::spawn_std_command_with_supervisor_identity_namespace(cmd)?;
+    let mut child = managed_children::spawn_registered(
+        || crate::process::spawn_std_command_with_supervisor_identity_namespace(cmd),
+        |child| Some(child.id()),
+    )?;
     #[cfg(not(target_os = "linux"))]
     let mut child = cmd.spawn()?;
     #[cfg(target_os = "linux")]
     let child_pid = child.id();
-    #[cfg(target_os = "linux")]
-    managed_children::register(child_pid);
     let master_file = master;
 
     let (sender, receiver) = mpsc::channel::<Vec<u8>>();
@@ -1501,14 +1502,14 @@ fn spawn_pipe_exec(
     }
 
     #[cfg(target_os = "linux")]
-    let mut child = crate::process::spawn_std_command_with_supervisor_identity_namespace(cmd)?;
+    let mut child = managed_children::spawn_registered(
+        || crate::process::spawn_std_command_with_supervisor_identity_namespace(cmd),
+        |child| Some(child.id()),
+    )?;
     #[cfg(not(target_os = "linux"))]
     let mut child = cmd.spawn()?;
     #[cfg(target_os = "linux")]
     let child_pid = child.id();
-    #[cfg(target_os = "linux")]
-    managed_children::register(child_pid);
-
     let child_stdin = child.stdin.take();
     let child_stdout = child.stdout.take().expect("stdout must be piped");
     let child_stderr = child.stderr.take().expect("stderr must be piped");

e2e/rust/tests/sandbox_lifecycle.rs

@@ -470,6 +470,35 @@ async fn detached_canonical_main_nonzero_exit_reaches_error() {
     sandbox.cleanup().await;
 }
 
+#[tokio::test]
+async fn detached_main_exit_during_provisioning_is_classified_as_workload_result() {
+    let mut sandbox = SandboxGuard::create_detached_main(&["sh", "-c", "exit 11"])
+        .await
+        .expect("fast detached main exit should not be reported as a provisioning failure");
+
+    let mut get_cmd = openshell_cmd();
+    get_cmd
+        .args(["sandbox", "get", &sandbox.name])
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped());
+    let get_output = get_cmd.output().await.expect("spawn openshell sandbox get");
+    let details = normalize_output(&format!(
+        "{}{}",
+        String::from_utf8_lossy(&get_output.stdout),
+        String::from_utf8_lossy(&get_output.stderr),
+    ));
+    assert!(
+        get_output.status.success(),
+        "sandbox get failed:\n{details}"
+    );
+    assert!(
+        details.contains("Phase: Error") && details.contains("Exit Code: 11"),
+        "fast detached main should retain its workload result:\n{details}"
+    );
+
+    sandbox.cleanup().await;
+}
+
 #[tokio::test]
 async fn canonical_tty_main_uses_sandbox_environment() {
     let script = r#"printf 'canonical_env home=%s user=%s term=%s\n' "$HOME" "$USER" "$TERM"; while true; do sleep 1; done"#;