fix(nesinit): a pid stops being the workload's the moment it is reaped

A pid is only a name for a process until that process is reaped; after that
the kernel may hand the same number to something else. The handle kept the
number, so a stop or a kill issued during shutdown — which every session
outcome reaches — could land on a process nobody meant, and the one aimed at
the workload would have been a SIGKILL.

The registry now clears the flag as it delivers the exit, in the same call, and
nothing signals a pid whose flag is down. Waiting for the workload on the way
out takes the same answer: already reaped is already gone.

There is a window left, between the reap and the delivery, and closing it
entirely needs a handle the kernel keeps rather than a number. Recorded rather
than papered over.

apps/nesinit/src/reap.rs

@@ -7,6 +7,7 @@
 
 use std::collections::HashMap;
 use std::io;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex};
 
 use nesprotocol::lifecycle::Exit;
@@ -83,7 +84,45 @@ pub fn refuse_oom_kill() -> io::Result<()> {
 /// component exists to report. So the reaper is the only waiter, and this is
 /// how it hands an exit to whoever asked for one.
 #[derive(Clone, Default)]
-pub struct Waiters(Arc<Mutex<HashMap<i32, oneshot::Sender<Exit>>>>);
+pub struct Waiters(Arc<Mutex<HashMap<i32, Watch>>>);
+
+/// The registry's half of one watched child.
+struct Watch {
+    exit: oneshot::Sender<Exit>,
+    running: Arc<AtomicBool>,
+}
+
+/// A child something is waiting for: its pid, its exit, and whether the pid
+/// still means what it meant.
+///
+/// The flag is the point. A pid is only a name for a process until that
+/// process is reaped, after which the kernel may hand the same number to
+/// something else — so a caller that kept a pid and signals it later can hit a
+/// process it has never heard of. Anything that signals asks here first.
+pub struct Watched {
+    pub pid: i32,
+    running: Arc<AtomicBool>,
+    exit: Option<oneshot::Receiver<Exit>>,
+}
+
+impl Watched {
+    /// The exit, which only one caller may hold — whoever takes it is the one
+    /// that reports it. What is left behind is the pid and whether it still
+    /// means this child, which is what a signaller needs.
+    pub fn take_exit(&mut self) -> Option<oneshot::Receiver<Exit>> {
+        self.exit.take()
+    }
+
+    /// Whether the pid is still this child's.
+    ///
+    /// False from the moment its exit is delivered. There is a window between
+    /// the reap and the delivery in which this still says true; closing it
+    /// entirely needs a handle the kernel keeps for us rather than a number,
+    /// and the number is what the rest of this component has.
+    pub fn running(&self) -> bool {
+        self.running.load(Ordering::SeqCst)
+    }
+}
 
 impl Waiters {
     pub fn new() -> Self {
@@ -96,29 +135,44 @@ impl Waiters {
     /// that exits immediately is reaped by a thread that has to take the same
     /// lock to deliver the exit, so it waits until the slot exists rather than
     /// finding none and dropping it.
-    pub fn watch<F>(&self, spawn: F) -> io::Result<oneshot::Receiver<Exit>>
+    pub fn watch<F>(&self, spawn: F) -> io::Result<Watched>
     where
         F: FnOnce() -> io::Result<i32>,
     {
         let mut waiting = self.lock();
         let pid = spawn()?;
         let (sender, receiver) = oneshot::channel();
-        waiting.insert(pid, sender);
-        Ok(receiver)
+        let running = Arc::new(AtomicBool::new(true));
+        waiting.insert(
+            pid,
+            Watch {
+                exit: sender,
+                running: running.clone(),
+            },
+        );
+        Ok(Watched {
+            pid,
+            running,
+            exit: Some(receiver),
+        })
     }
 
     /// Hand an exit to whoever is waiting for that pid. `false` if nobody is,
     /// which is the common case: most of what an init reaps is an orphan
     /// nothing asked about.
     pub fn deliver(&self, pid: i32, exit: Exit) -> bool {
-        let Some(sender) = self.lock().remove(&pid) else {
+        let Some(watch) = self.lock().remove(&pid) else {
             return false;
         };
+        // Before the exit goes anywhere: the pid has been reaped, so the
+        // number is free for the kernel to give to something else and nothing
+        // may signal it from here on.
+        watch.running.store(false, Ordering::SeqCst);
         // An error here means the caller stopped waiting, which is its right.
-        sender.send(exit).is_ok()
+        watch.exit.send(exit).is_ok()
     }
 
-    fn lock(&self) -> std::sync::MutexGuard<'_, HashMap<i32, oneshot::Sender<Exit>>> {
+    fn lock(&self) -> std::sync::MutexGuard<'_, HashMap<i32, Watch>> {
         self.0
             .lock()
             .unwrap_or_else(|poisoned| poisoned.into_inner())

apps/nesinit/src/workload.rs

@@ -14,7 +14,7 @@ use nesprotocol::lifecycle::{Exec, Exit, Mount};
 
 use std::os::unix::process::CommandExt;
 
-use crate::reap::Waiters;
+use crate::reap::{Waiters, Watched};
 
 /// Why something could not be done, in the words the operating system used.
 ///
@@ -55,14 +55,17 @@ pub trait Workload {
 /// The workload as a local process.
 pub struct Process {
     waiters: Waiters,
-    pid: Option<i32>,
+    running: Option<Watched>,
 }
 
 impl Process {
     /// Started through the reaper's registry, because the reaper is the only
     /// thing in this component that may call `wait`.
     pub fn new(waiters: Waiters) -> Self {
-        Self { waiters, pid: None }
+        Self {
+            waiters,
+            running: None,
+        }
     }
 
     /// Send a signal to the workload and nothing else.
@@ -70,11 +73,18 @@ impl Process {
     /// Nothing here ever signals the process group or every process: the order
     /// a shutdown promises is only true if the workload can be stopped alone.
     pub fn signal(&self, signal: libc::c_int) {
-        let Some(pid) = self.pid else { return };
+        let Some(watched) = &self.running else { return };
+        // Nothing is signalled once the exit has been delivered. The pid was
+        // freed by the reap that produced it, and the kernel is entitled to
+        // give that number to something else — a stop or a kill aimed at it
+        // then lands on a process nobody meant.
+        if !watched.running() {
+            return;
+        }
         // SAFETY: two integers, and it cannot touch this process's memory. A
         // pid that has already gone fails with ESRCH, which is exactly the
         // idempotence the callers of this rely on.
-        unsafe { libc::kill(pid, signal) };
+        unsafe { libc::kill(watched.pid, signal) };
     }
 
     /// Wait up to `grace` for the workload to leave, without a runtime.
@@ -83,7 +93,13 @@ impl Process {
     /// one pid rather than for any child, so a slow service cannot be mistaken
     /// for the workload still running.
     pub fn await_exit(&self, grace: std::time::Duration) -> bool {
-        let Some(pid) = self.pid else { return true };
+        let Some(watched) = &self.running else {
+            return true;
+        };
+        if !watched.running() {
+            return true; // already reaped, and its exit already reported
+        }
+        let pid = watched.pid;
         let deadline = std::time::Instant::now() + grace;
         loop {
             let mut status: libc::c_int = 0;
@@ -151,21 +167,18 @@ impl Workload for Process {
             });
         }
 
-        let mut pid = None;
-        let exited = self
+        let mut watched = self
             .waiters
-            .watch(|| {
-                let child = command.spawn()?;
-                let started = child.id() as i32;
-                pid = Some(started);
-                Ok(started)
-            })
+            .watch(|| Ok(command.spawn()?.id() as i32))
             .map_err(|error| Failure::new(error.to_string()))?;
-        self.pid = pid;
+
+        // The caller gets the exit and reports it; this handle keeps the pid
+        // and whether that pid is still this child's.
+        let exit = watched.take_exit().expect("a new watch has its exit");
+        self.running = Some(watched);
 
         Ok(Box::pin(async move {
-            exited
-                .await
+            exit.await
                 .map_err(|_| io::Error::other("the workload's exit was not delivered"))
         }))
     }

apps/nesinit/tests/reaping.rs

@@ -115,9 +115,10 @@ fn an_exit_reaches_whoever_asked_for_it() {
     let _alone = alone();
     let waiters = Waiters::new();
 
-    let mut exited = waiters
+    let mut watched = waiters
         .watch(|| Ok(fork_child(|| unsafe { libc::_exit(9) })))
         .expect("the child never started");
+    let mut exited = watched.take_exit().unwrap();
 
     let deadline = Instant::now() + Duration::from_secs(5);
     while Instant::now() < deadline {
@@ -154,7 +155,7 @@ fn an_exit_that_happens_before_the_caller_is_registered_is_not_lost() {
         }
     });
 
-    let mut exited = waiters
+    let mut watched = waiters
         .watch(|| {
             let pid = fork_child(|| unsafe { libc::_exit(5) });
             // Long enough for the child to exit and the reaper to reach it.
@@ -162,6 +163,7 @@ fn an_exit_that_happens_before_the_caller_is_registered_is_not_lost() {
             Ok(pid)
         })
         .expect("the child never started");
+    let mut exited = watched.take_exit().unwrap();
 
     let deadline = Instant::now() + Duration::from_secs(5);
     let exit = loop {
@@ -179,3 +181,37 @@ fn an_exit_that_happens_before_the_caller_is_registered_is_not_lost() {
     stop.store(true, Ordering::Relaxed);
     reaper.join().unwrap();
 }
+
+#[test]
+fn a_pid_stops_being_the_workload_the_moment_it_is_reaped() {
+    let _alone = alone();
+    let waiters = Waiters::new();
+
+    let mut watched = waiters
+        .watch(|| Ok(fork_child(|| unsafe { libc::_exit(0) })))
+        .expect("the child never started");
+    let mut exited = watched.take_exit().unwrap();
+    assert!(
+        watched.running(),
+        "a child that has not been reaped is still itself"
+    );
+
+    let deadline = Instant::now() + Duration::from_secs(5);
+    while Instant::now() < deadline {
+        for (pid, exit) in reap_exited() {
+            waiters.deliver(pid, exit);
+        }
+        if exited.try_recv().is_ok() {
+            break;
+        }
+        std::thread::sleep(Duration::from_millis(10));
+    }
+
+    // The pid was freed by the reap that produced that exit, and the kernel is
+    // entitled to hand the number to something else. Signalling it after this
+    // point is signalling a stranger.
+    assert!(
+        !watched.running(),
+        "a reaped pid is still being treated as the workload's",
+    );
+}