fix: stop a running command at once when the instance is dropped

README.md

@@ -134,7 +134,7 @@ And I stop the "prober" process
 ## Semantics worth knowing
 
 - **Processes live for one scenario.** The scenario boundary kills every process the scenario started, clears the context (environment, working directory, stdin, timeout) and forgets the last command; the end of the feature file does the same. Variables follow the host's own per-file rule.
-- **…as long as `bddkit` itself gets to finish.** The cleanup runs through the host's end-of-scenario and end-of-file calls. If `bddkit` is killed by a signal mid-run — Ctrl-C, a CI job timeout — those calls never happen, and because every command runs in its own process group, the signal that reached `bddkit` does not reach them either. Find the leftovers with `ps -o pid,pgid,command` and end each group with `kill -- -<pgid>`. On Windows this does not happen: the Job Object is closed with the `bddkit` process, and closing it kills the tree. Cleaning up on an interrupted run is tracked in [bddkit/bddkit#62](https://github.com/bddkit/bddkit/issues/62) (the host side) and [#1](https://github.com/bddkit/bddkit-exec/issues/1) (stopping a command that is still running when the interruption comes).
+- **…as long as `bddkit` itself gets to finish.** The cleanup runs through the host's end-of-scenario and end-of-file calls. If `bddkit` is killed by a signal mid-run — Ctrl-C, a CI job timeout — those calls never happen, and because every command runs in its own process group, the signal that reached `bddkit` does not reach them either. Find the leftovers with `ps -o pid,pgid,command` and end each group with `kill -- -<pgid>`. On Windows this does not happen: the Job Object is closed with the `bddkit` process, and closing it kills the tree. Cleaning up on an interrupted run is tracked in [bddkit/bddkit#62](https://github.com/bddkit/bddkit/issues/62) (the host side). Once the host drops the instance on an interruption, the plugin acts on it at once, even in the middle of `I run the command`: every process of the file is asked to stop, gets what is left of `stop_grace_secs`, and the interrupted step fails with "the run was interrupted".
 - **A one-shot command owns its whole tree.** When it exits, anything it left running in its process group or job is killed: a `sleep 30 &` would otherwise hold the output pipes open and outlive the run. Use `I start` for anything that should keep running.
 - **A non-zero exit is not a failure by itself**, exactly as an HTTP 500 is not — it is asserted. **A timeout is a failure**, exactly as a transport error is, and kills the whole tree.
 - **A mismatch waits only while the stream can still change.** On a running process it answers `not_yet`, so an armed eventual assertion looks again; on the last command, or on a process that has ended with both pipes closed, it fails at once instead of spending the whole timeout.

src/exec/mod.rs

@@ -178,12 +178,67 @@ fn reader(
     (capture, handle)
 }
 
+/// The trees of one instance's live jobs, kept apart from the instance so an
+/// interrupt can reach them while a step holds the instance lock. A tree is
+/// listed from spawn until `Job::finish`, which removes it before reaping:
+/// on Unix a group may only be signalled while its leader is unreaped.
+pub struct Registry {
+    grace: Duration,
+    live: Mutex<Live>,
+}
+
+#[derive(Default)]
+struct Live {
+    trees: Vec<Arc<os::Tree>>,
+    /// When `interrupt` asked, which is when the grace period began.
+    interrupted: Option<Instant>,
+}
+
+impl Registry {
+    pub fn new(grace: Duration) -> Arc<Self> {
+        Arc::new(Self {
+            grace,
+            live: Mutex::default(),
+        })
+    }
+
+    fn live(&self) -> std::sync::MutexGuard<'_, Live> {
+        self.live.lock().unwrap_or_else(|e| e.into_inner())
+    }
+
+    /// Asks every listed tree to stop (SIGTERM / CTRL_BREAK) and starts the
+    /// grace clock. Killing after the grace is up to whoever owns the jobs:
+    /// `run` for the one in flight, `Instance::reset` for the rest.
+    pub fn interrupt(&self) {
+        let mut live = self.live();
+        live.interrupted.get_or_insert_with(Instant::now);
+        if !self.grace.is_zero() {
+            for tree in &live.trees {
+                tree.interrupt();
+            }
+        }
+    }
+
+    pub fn interrupted(&self) -> bool {
+        self.live().interrupted.is_some()
+    }
+
+    /// What is left of `stop_grace_secs`, counted from the interrupt when
+    /// there was one: the request already went out, so none is spent twice.
+    pub fn grace_left(&self) -> Duration {
+        self.live()
+            .interrupted
+            .map_or(self.grace, |at| self.grace.saturating_sub(at.elapsed()))
+    }
+}
+
 /// A started command: a one-shot run once it has finished, or a background
 /// process for as long as the scenario keeps it.
 pub struct Job {
     pub spec: Spec,
     child: Child,
-    tree: os::Tree,
+    tree: Arc<os::Tree>,
+    registry: Arc<Registry>,
     pub stdout: Arc<Mutex<Capture>>,
     pub stderr: Arc<Mutex<Capture>>,
     readers: Vec<JoinHandle<()>>,
@@ -192,7 +247,7 @@ pub struct Job {
 }
 
 impl Job {
-    pub fn spawn(spec: Spec, cap: usize) -> Result<Self, String> {
+    pub fn spawn(spec: Spec, cap: usize, registry: &Arc<Registry>) -> Result<Self, String> {
         let mut child = spec.command().spawn().map_err(|e| {
             format!(
                 "cannot start `{}` in {}: {e}",
@@ -201,7 +256,7 @@ impl Job {
             )
         })?;
         let tree = match os::Tree::adopt(&mut child) {
-            Ok(tree) => tree,
+            Ok(tree) => Arc::new(tree),
             Err(e) => {
                 let _ = child.kill();
                 let _ = child.wait();
@@ -218,10 +273,19 @@ impl Job {
         }
         let (stdout, out_reader) = reader(child.stdout.take().expect("piped stdout"), cap);
         let (stderr, err_reader) = reader(child.stderr.take().expect("piped stderr"), cap);
+        {
+            let mut live = registry.live();
+            live.trees.push(Arc::clone(&tree));
+            // Started after the interrupt: nobody is left to wait on it.
+            if live.interrupted.is_some() {
+                tree.kill();
+            }
+        }
         Ok(Self {
             spec,
             child,
             tree,
+            registry: Arc::clone(registry),
             stdout,
             stderr,
             readers: vec![out_reader, err_reader],
@@ -246,6 +310,11 @@ impl Job {
             return;
         }
         self.tree.kill();
+        // Unlisted before the reap, under the lock `interrupt` signals with.
+        self.registry
+            .live()
+            .trees
+            .retain(|t| !Arc::ptr_eq(t, &self.tree));
         if let Ok(status) = self.child.wait() {
             self.exit.get_or_insert(os::exit_of(status));
         }
@@ -280,28 +349,47 @@ impl Drop for Job {
     }
 }
 
-/// How a one-shot run ended when it did not end on its own.
-pub struct TimedOut(pub Job);
+/// How a one-shot run ended.
+pub enum Ended {
+    Done(Job),
+    /// Out of time; its tree was killed.
+    TimedOut(Job),
+    /// The instance is being dropped; the tree got its grace period.
+    Interrupted(Job),
+}
 
 /// Runs to completion. Whatever the command left behind in its tree is
 /// killed once it exits: it would hold our pipes open, and nothing else
 /// would ever stop it.
-pub fn run(spec: Spec, cap: usize, timeout: Duration) -> Result<Result<Job, TimedOut>, String> {
+pub fn run(
+    spec: Spec,
+    cap: usize,
+    timeout: Duration,
+    registry: &Arc<Registry>,
+) -> Result<Ended, String> {
     let deadline = Instant::now()
         .checked_add(timeout)
         .ok_or_else(|| format!("a timeout of {timeout:?} is out of range"))?;
-    let mut job = Job::spawn(spec, cap)?;
+    let mut job = Job::spawn(spec, cap, registry)?;
     while job.poll().is_none() {
+        if registry.interrupted() {
+            stop_all([&mut job], registry.grace_left());
+            return Ok(Ended::Interrupted(job));
+        }
         if Instant::now() >= deadline {
             job.finish();
             job.drain(DRAIN);
-            return Ok(Err(TimedOut(job)));
+            return Ok(Ended::TimedOut(job));
         }
         thread::sleep(POLL);
     }
     job.finish();
     job.drain(DRAIN);
-    Ok(Ok(job))
+    // The interrupt's own request may be what ended it.
+    if registry.interrupted() {
+        return Ok(Ended::Interrupted(job));
+    }
+    Ok(Ended::Done(job))
 }
 
 /// A graceful request to every tree (SIGTERM on Unix, CTRL_BREAK on
@@ -356,14 +444,19 @@ pub fn probe(config: &InstanceConfig) -> Result<(), String> {
         stdin: None,
     };
     let described = spec.describe();
-    match run(spec, 64 * 1024, Duration::from_secs(10))? {
-        Ok(job) if job.exit() == Some(Exit::Code(0)) => Ok(()),
-        Ok(job) => Err(format!(
+    match run(
+        spec,
+        64 * 1024,
+        Duration::from_secs(10),
+        &Registry::new(Duration::ZERO),
+    )? {
+        Ended::Done(job) if job.exit() == Some(Exit::Code(0)) => Ok(()),
+        Ended::Done(job) => Err(format!(
             "`{described}` ended with {}: {}",
             job.exit().map_or("no exit".into(), |e| e.to_string()),
             text_of(&job.stderr).trim()
         )),
-        Err(_) => Err(format!("`{described}` did not finish within 10 seconds")),
+        _ => Err(format!("`{described}` did not finish within 10 seconds")),
     }
 }
 
@@ -426,10 +519,19 @@ mod tests {
         lf(text_of(c))
     }
 
+    fn registry() -> Arc<Registry> {
+        Registry::new(Duration::from_secs(2))
+    }
+
+    fn spawn(spec: Spec) -> Job {
+        Job::spawn(spec, 1024, &registry()).expect("spawn")
+    }
+
     fn completed(spec: Spec, cap: usize) -> Job {
-        run(spec, cap, Duration::from_secs(10))
-            .expect("spawn")
-            .unwrap_or_else(|_| panic!("timed out"))
+        match run(spec, cap, Duration::from_secs(10), &registry()).expect("spawn") {
+            Ended::Done(job) => job,
+            _ => panic!("did not finish"),
+        }
     }
 
     #[test]
@@ -455,9 +557,10 @@ mod tests {
             shell(script::HANG_WITH_CHILD),
             1024,
             Duration::from_millis(500),
+            &registry(),
         )
         .expect("spawn");
-        let Err(TimedOut(job)) = outcome else {
+        let Ended::TimedOut(job) = outcome else {
             panic!("it finished")
         };
         assert!(
@@ -487,7 +590,7 @@ mod tests {
 
     #[test]
     fn stop_all_ends_a_pipeline_within_the_grace_period() {
-        let mut job = Job::spawn(shell(script::PIPELINE), 1024).expect("spawn");
+        let mut job = spawn(shell(script::PIPELINE));
         let started = Instant::now();
         stop_all([&mut job], Duration::from_secs(2));
         assert!(started.elapsed() < Duration::from_secs(4));
@@ -500,7 +603,7 @@ mod tests {
 
     #[test]
     fn a_zero_grace_kills_without_asking() {
-        let mut job = Job::spawn(shell(script::SLEEP), 1024).expect("spawn");
+        let mut job = spawn(shell(script::SLEEP));
         let started = Instant::now();
         stop_all([&mut job], Duration::ZERO);
         assert!(started.elapsed() < Duration::from_secs(1));
@@ -516,7 +619,7 @@ mod tests {
     #[test]
     fn a_process_that_ignores_sigterm_is_killed_when_its_grace_runs_out() {
         let spec = shell("trap '' TERM; echo ready; while :; do sleep 0.05; done");
-        let mut job = Job::spawn(spec, 1024).expect("spawn");
+        let mut job = spawn(spec);
         let deadline = Instant::now() + Duration::from_secs(5);
         while !text(&job.stdout).contains("ready") && Instant::now() < deadline {
             thread::sleep(POLL);
@@ -531,7 +634,7 @@ mod tests {
 
     #[test]
     fn poll_sees_the_exit_before_the_process_is_reaped() {
-        let mut job = Job::spawn(shell(script::EXIT_7), 1024).expect("spawn");
+        let mut job = spawn(shell(script::EXIT_7));
         let deadline = Instant::now() + Duration::from_secs(5);
         while job.poll().is_none() && Instant::now() < deadline {
             thread::sleep(POLL);
@@ -546,7 +649,7 @@ mod tests {
     #[cfg(unix)]
     #[test]
     fn on_unix_a_polled_leader_stays_a_zombie_until_finish() {
-        let mut job = Job::spawn(shell(script::EXIT_7), 1024).expect("spawn");
+        let mut job = spawn(shell(script::EXIT_7));
         let pid = job.child.id();
         let deadline = Instant::now() + Duration::from_secs(5);
         while job.poll().is_none() && Instant::now() < deadline {
@@ -559,7 +662,7 @@ mod tests {
 
     #[test]
     fn dropping_a_job_kills_its_tree() {
-        let job = Job::spawn(shell(script::SLEEP), 1024).expect("spawn");
+        let job = spawn(shell(script::SLEEP));
         let pid = job.child.id();
         drop(job);
         assert!(!os::alive(pid));
@@ -571,7 +674,7 @@ mod tests {
             program: Program::Argv(vec!["no-such-binary-bddkit".into()]),
             ..shell("")
         };
-        let e = Job::spawn(spec, 1024).err().expect("refused");
+        let e = Job::spawn(spec, 1024, &registry()).err().expect("refused");
         assert!(e.contains("no-such-binary-bddkit"), "{e}");
     }
 

src/exec/windows.rs

@@ -86,8 +86,9 @@ pub struct Tree {
 }
 
 // SAFETY: a job handle is a kernel object handle, usable from any thread;
-// the plugin only ever touches it from one thread at a time anyway.
+// TerminateJobObject and GenerateConsoleCtrlEvent take no exclusive access.
 unsafe impl Send for Tree {}
+unsafe impl Sync for Tree {}
 
 impl Tree {
     /// Puts the suspended child in a fresh job, then lets it run.

src/instance.rs

@@ -5,7 +5,9 @@ use std::path::{Path, PathBuf};
 use std::time::Duration;
 
 use crate::config::InstanceConfig;
-use crate::exec::{self, Job, Program, Spec};
+use std::sync::Arc;
+
+use crate::exec::{self, Job, Program, Registry, Spec};
 
 /// Per-scenario preparation, set with `the command … is` steps.
 #[derive(Default)]
@@ -19,6 +21,8 @@ pub struct Context {
 
 pub struct Instance {
     pub config: InstanceConfig,
+    /// Reachable without the instance lock; see `Registry`.
+    pub registry: Arc<Registry>,
     pub context: Context,
     pub last: Option<Job>,
     /// In start order, names unique within the scenario.
@@ -28,6 +32,7 @@ pub struct Instance {
 impl Instance {
     pub fn new(config: InstanceConfig) -> Self {
         Self {
+            registry: Registry::new(config.stop_grace),
             config,
             context: Context::default(),
             last: None,
@@ -37,7 +42,7 @@ impl Instance {
 
     /// Kills every process the scenario started and forgets the rest.
     pub fn reset(&mut self) {
-        let grace = self.config.stop_grace;
+        let grace = self.registry.grace_left();
         exec::stop_all(self.processes.iter_mut().map(|(_, job)| job), grace);
         self.processes.clear();
         self.last = None;
@@ -133,8 +138,10 @@ mod tests {
             &std::env::temp_dir(),
             false,
         );
-        i.processes
-            .push(("sleeper".into(), Job::spawn(spec, 1024).expect("spawn")));
+        i.processes.push((
+            "sleeper".into(),
+            Job::spawn(spec, 1024, &i.registry).expect("spawn"),
+        ));
         i.reset();
         assert!(i.processes.is_empty() && i.context.env.is_empty() && i.last.is_none());
     }