Own concurrent child Runs through collective teardown

src/child_run.rs

@@ -1,6 +1,11 @@
 //! A Run in a child `thirdshift`, started from the same Launch directory by a
 //! Spec run for one of its Tickets (ADR-0006) or by a Run for its Base fix
-//! (ADR-0008).
+//! (ADR-0008). [`Runs`] owns concurrent children and their collective cleanup;
+//! standalone [`start`] and [`Handle::wait`] serve Base fix.
+
+mod runs;
+
+pub use runs::Runs;
 
 use std::ffi::OsString;
 use std::io::{BufRead, BufReader, Read};
@@ -487,7 +492,7 @@ impl Handle {
     /// Poll the process and finished readers without waiting for unfinished
     /// work. Pending and stopping return None. A failure is returned only
     /// after graceful shutdown, reaping, and joining both readers.
-    pub fn try_wait(&mut self) -> Option<Result<Ended>> {
+    fn try_wait(&mut self) -> Option<Result<Ended>> {
         if self.completed || !self.poll_completion() {
             return None;
         }

src/child_run/execution_tests.rs

@@ -8,6 +8,70 @@ use tempfile::TempDir;
 
 const PR: &str = "https://github.com/acme/widgets/pull/31";
 
+#[test]
+fn an_empty_collection_has_no_ending_and_returns_immediately() {
+    let mut runs = Runs::default();
+    let started = Instant::now();
+    assert!(runs.next_ending().is_none());
+    assert!(started.elapsed() < Duration::from_millis(50));
+}
+
+#[test]
+fn a_pending_collection_waits_for_its_child_and_retires_it_before_delivery() {
+    let fixture = Fixture::new(&format!(
+        "open(ROOT + '/ready', 'w').close()\nwhile not os.path.exists(ROOT + '/finish'): time.sleep(.01)\nopen(ROOT + '/saved', 'w').close()\nos.write(1, b'{PR}\\n')"
+    ));
+    let mut runs = Runs::default();
+    fixture.start_in(&mut runs, 21).unwrap();
+    fixture.await_file("ready");
+    let finish = fixture.0.path().join("finish");
+    let release = thread::spawn(move || {
+        thread::sleep(Duration::from_millis(150));
+        fs::write(finish, "").unwrap();
+    });
+
+    let ending = runs.next_ending();
+    release.join().unwrap();
+    let (number, ending) = ending.unwrap();
+    assert_eq!(number, 21);
+    assert!(matches!(ending.unwrap(), Ended::Reached { pr_url: Some(url), .. } if url == PR));
+    assert!(fixture.0.path().join("saved").exists());
+    fixture.assert_stopped();
+    assert!(runs.next_ending().is_none());
+}
+
+#[test]
+fn distinct_children_deliver_out_of_order_with_their_own_endings_and_accounts() {
+    let first = Fixture::new(&format!(
+        "open(ROOT + '/ready', 'w').close()\nwhile not os.path.exists(ROOT + '/finish'): time.sleep(.01)\nos.write(1, b'{PR}\\n')\nos.write(2, b'thirdshift: Base fix: https://github.com/acme/widgets/issues/8 merged\\n')"
+    ));
+    let second = Fixture::new(
+        "os.write(2, b'thirdshift: checks failed\\nthirdshift: session log: /logs/22.jsonl\\n')\nraise SystemExit(7)",
+    );
+    let mut runs = Runs::default();
+    first.start_in(&mut runs, 21).unwrap();
+    first.await_file("ready");
+    second.start_in(&mut runs, 22).unwrap();
+
+    let (number, ending) = runs.next_ending().unwrap();
+    assert_eq!(number, 22);
+    assert!(
+        matches!(ending.unwrap(), Ended::Failed { cause, log: Some(log) }
+        if cause == "checks failed" && log == "/logs/22.jsonl")
+    );
+    second.assert_stopped();
+
+    fs::write(first.0.path().join("finish"), "").unwrap();
+    let (number, ending) = runs.next_ending().unwrap();
+    assert_eq!(number, 21);
+    assert!(
+        matches!(ending.unwrap(), Ended::Reached { pr_url: Some(url), base_fix: Some(account) }
+        if url == PR && account == "https://github.com/acme/widgets/issues/8 merged")
+    );
+    first.assert_stopped();
+    assert!(runs.next_ending().is_none());
+}
+
 /// Private fault points retain a real child and the normal execution path.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub(super) enum Fault {
@@ -50,7 +114,33 @@ impl Fixture {
     }
 
     fn start(&self) -> Result<Handle> {
-        self.start_using(Handle::start_readers)
+        self.start_number_using(248, Handle::start_readers)
+    }
+
+    fn start_in(&self, runs: &mut Runs, number: u64) -> Result<()> {
+        self.start_in_using(runs, number, Handle::start_readers)
+    }
+
+    fn start_completed_in(&self, runs: &mut Runs, number: u64) -> Result<()> {
+        // Arrange a real, completed child at the private launch seam without
+        // consuming its ending. Queue tests need both process and readers
+        // finished; process exit alone cannot guarantee reader readiness.
+        self.start_in_using(runs, number, |owned| {
+            owned.start_readers()?;
+            while !owned.poll_completion() {
+                thread::sleep(Duration::from_millis(10));
+            }
+            Ok(())
+        })
+    }
+
+    fn start_in_using(
+        &self,
+        runs: &mut Runs,
+        number: u64,
+        startup: impl FnOnce(&mut Handle) -> Result<()>,
+    ) -> Result<()> {
+        runs.start_using(number, || self.start_number_using(number, startup))
     }
 
     fn start_with_fault(&self, fault: Fault) -> Result<Handle> {
@@ -62,9 +152,17 @@ impl Fixture {
     }
 
     fn start_using(&self, startup: impl FnOnce(&mut Handle) -> Result<()>) -> Result<Handle> {
+        self.start_number_using(248, startup)
+    }
+
+    fn start_number_using(
+        &self,
+        number: u64,
+        startup: impl FnOnce(&mut Handle) -> Result<()>,
+    ) -> Result<Handle> {
         super::start_using(
             &self.0.path().join("run"),
-            &IssueUrl::parse("https://github.com/acme/widgets/issues/248").unwrap(),
+            &IssueUrl::parse(&format!("https://github.com/acme/widgets/issues/{number}")).unwrap(),
             &Given {
                 kind: Kind::Ticket {
                     spec_branch: "issue-237".to_string(),
@@ -89,6 +187,12 @@ impl Fixture {
         fs::read_to_string(self.0.path().join(name)).unwrap()
     }
 
+    fn assert_saved_on_stop(&self) {
+        assert!(self.0.path().join("saved").exists());
+        assert_eq!(self.read("terms"), "TERM\n");
+        self.assert_stopped();
+    }
+
     fn assert_stopped(&self) {
         let pid = self.read("pid").parse::<libc::pid_t>().unwrap();
         // SAFETY: probe only the exact PID recorded by this finite fixture.
@@ -230,9 +334,143 @@ fn dropping_a_live_handle_gracefully_saves_work_and_reaps_the_child() {
     let handle = fixture.start().unwrap();
     fixture.await_file("ready");
     drop(handle);
-    assert!(fixture.0.path().join("saved").exists());
-    assert_eq!(fixture.read("terms"), "TERM\n");
-    fixture.assert_stopped();
+    fixture.assert_saved_on_stop();
+}
+
+#[test]
+fn duplicate_start_is_refused_before_launch_and_keeps_the_original_child_owned() {
+    let original = Fixture::new(SAVES_ON_STOP);
+    let replacement = Fixture::new("open(ROOT + '/launched', 'w').close()");
+    let mut runs = Runs::default();
+    original.start_in(&mut runs, 21).unwrap();
+    original.await_file("ready");
+
+    let error = replacement.start_in(&mut runs, 21).unwrap_err();
+    assert!(error.to_string().contains("#21"));
+    assert!(!replacement.0.path().join("pid").exists());
+    assert!(!original.0.path().join("terms").exists());
+
+    drop(runs);
+    original.assert_saved_on_stop();
+}
+
+#[test]
+fn launch_failure_leaves_existing_children_running_and_owned_for_cleanup() {
+    let original = Fixture::new(SAVES_ON_STOP);
+    let missing = Fixture::new("");
+    fs::remove_file(missing.0.path().join("run")).unwrap();
+    let mut runs = Runs::default();
+    original.start_in(&mut runs, 21).unwrap();
+    original.await_file("ready");
+
+    let cause = format!("{:#}", missing.start_in(&mut runs, 22).unwrap_err());
+    assert!(cause.contains("could not start the Run for #22"), "{cause}");
+    assert!(!missing.0.path().join("pid").exists());
+    assert!(!original.0.path().join("terms").exists());
+
+    drop(runs);
+    original.assert_saved_on_stop();
+}
+
+fn finished_fixture() -> Fixture {
+    Fixture::new(&format!("os.write(1, b'{PR}\\n')"))
+}
+
+#[test]
+fn simultaneous_endings_are_retired_once_and_queued_identity_is_reserved_until_delivery() {
+    let first = finished_fixture();
+    let second = finished_fixture();
+    let replacement = finished_fixture();
+    let mut runs = Runs::default();
+    second.start_completed_in(&mut runs, 22).unwrap();
+    first.start_completed_in(&mut runs, 21).unwrap();
+
+    let (number, ending) = runs.next_ending().unwrap();
+    assert_eq!(number, 21);
+    assert!(matches!(ending.unwrap(), Ended::Reached { pr_url: Some(url), .. } if url == PR));
+    first.assert_stopped();
+    second.assert_stopped();
+    assert!(replacement.start_in(&mut runs, 22).is_err());
+    assert!(!replacement.0.path().join("pid").exists());
+
+    // Delivered identity may start again; the older queued ending stays first.
+    replacement.start_in(&mut runs, 21).unwrap();
+    for expected in [22, 21] {
+        let (number, ending) = runs.next_ending().unwrap();
+        assert_eq!(number, expected);
+        assert!(matches!(ending.unwrap(), Ended::Reached { pr_url: Some(url), .. } if url == PR));
+    }
+    replacement.assert_stopped();
+    assert!(runs.next_ending().is_none());
+}
+
+const SIBLING_SAVES_ON_STOP: &str = r#"
+def stop(signum, frame):
+    with open(ROOT + '/terms', 'a') as f: f.write('TERM\n')
+    open(ROOT + '/stopping', 'w').close()
+    while not os.path.exists(ROOT + '/sibling-stopping'): time.sleep(.01)
+    os.write(1, b'x' * (1024 * 1024))
+    with open(ROOT + '/saved', 'w') as f: f.write('unfinished work saved')
+    raise SystemExit(1)
+signal.signal(signal.SIGTERM, stop)
+open(ROOT + '/ready', 'w').close()
+while True: time.sleep(.01)
+"#;
+
+fn sibling_cleanup(unwind: bool) {
+    interrupt::install().unwrap();
+    assert!(!interrupt::requested());
+    let first = Fixture::new(SIBLING_SAVES_ON_STOP);
+    let second = Fixture::new(SIBLING_SAVES_ON_STOP);
+    for (child, sibling) in [(&first, &second), (&second, &first)] {
+        std::os::unix::fs::symlink(
+            sibling.0.path().join("stopping"),
+            child.0.path().join("sibling-stopping"),
+        )
+        .unwrap();
+    }
+
+    let result = std::panic::catch_unwind(|| {
+        let mut runs = Runs::default();
+        first.start_in(&mut runs, 21).unwrap();
+        second.start_in(&mut runs, 22).unwrap();
+        first.await_file("ready");
+        second.await_file("ready");
+        if unwind {
+            panic!("the Spec run owner unwound");
+        }
+    });
+
+    assert_eq!(result.is_err(), unwind);
+    assert!(
+        !interrupt::requested(),
+        "cleanup changed Command interruption"
+    );
+    for child in [&first, &second] {
+        child.assert_stopped();
+        assert_eq!(child.read("terms"), "TERM\n");
+        assert_eq!(child.read("saved"), "unfinished work saved");
+    }
+}
+
+#[test]
+fn dropping_a_collection_stops_all_siblings_before_waiting_and_saves_their_work() {
+    if isolated("dropping_a_collection_stops_all_siblings_before_waiting_and_saves_their_work")
+        .is_some()
+    {
+        return;
+    }
+    sibling_cleanup(false);
+}
+
+#[test]
+fn unwinding_a_collection_stops_all_siblings_before_waiting_and_saves_their_work() {
+    if isolated("unwinding_a_collection_stops_all_siblings_before_waiting_and_saves_their_work")
+        .is_some()
+    {
+        return;
+    }
+    sibling_cleanup(true);
 }
 
 #[test]
@@ -331,6 +569,60 @@ open(ROOT + '/ready', 'w').close()
 while True: time.sleep(.01)
 "#;
 
+#[test]
+fn interruption_reaches_every_active_child_before_a_queued_ending_is_delivered() {
+    if isolated("interruption_reaches_every_active_child_before_a_queued_ending_is_delivered")
+        .is_some()
+    {
+        return;
+    }
+    interrupt::install().unwrap();
+    let first = finished_fixture();
+    let second = finished_fixture();
+    let third = Fixture::new(GATED_STOP);
+    let fourth = Fixture::new(GATED_STOP);
+    let mut runs = Runs::default();
+    first.start_completed_in(&mut runs, 21).unwrap();
+    second.start_completed_in(&mut runs, 22).unwrap();
+    third.start_in(&mut runs, 23).unwrap();
+    fourth.start_in(&mut runs, 24).unwrap();
+    third.await_file("ready");
+    fourth.await_file("ready");
+
+    let (number, ending) = runs.next_ending().unwrap();
+    assert_eq!(number, 21);
+    assert!(matches!(ending.unwrap(), Ended::Reached { .. }));
+    first.assert_stopped();
+    second.assert_stopped();
+
+    signal_hook::low_level::raise(libc::SIGINT).unwrap();
+    let (number, ending) = runs.next_ending().unwrap();
+    assert_eq!(number, 22);
+    assert!(matches!(ending.unwrap(), Ended::Reached { .. }));
+    // No further receive is needed to notify both children, and the queued
+    // ending returned while their graceful cleanup was still pending.
+    for child in [&third, &fourth] {
+        child.await_file("stopping");
+        assert!(!child.0.path().join("saved").exists());
+        fs::write(child.0.path().join("finish"), "").unwrap();
+    }
+    signal_hook::low_level::raise(libc::SIGTERM).unwrap();
+    let mut numbers = Vec::new();
+    for _ in 0..2 {
+        let (number, ending) = runs.next_ending().unwrap();
+        numbers.push(number);
+        assert!(matches!(ending.unwrap(), Ended::Interrupted));
+    }
+    numbers.sort();
+    assert_eq!(numbers, [23, 24]);
+    assert!(runs.next_ending().is_none());
+    for child in [&third, &fourth] {
+        assert_eq!(child.read("terms"), "TERM\n");
+        assert!(child.0.path().join("saved").exists());
+        child.assert_stopped();
+    }
+}
+
 fn poll_until_stopping(handle: &mut Handle, fixture: &Fixture) {
     let deadline = Instant::now() + Duration::from_secs(4);
     while !fixture.0.path().join("stopping").exists() {