diff --git a/src/child_run.rs b/src/child_run.rs index a62afcc1d..a3de55f5b 100644 --- a/src/child_run.rs +++ b/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> { + fn try_wait(&mut self) -> Option> { if self.completed || !self.poll_completion() { return None; } diff --git a/src/child_run/execution_tests.rs b/src/child_run/execution_tests.rs index 141acd6fe..3a7a06b99 100644 --- a/src/child_run/execution_tests.rs +++ b/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 { - 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 { @@ -62,9 +152,17 @@ impl Fixture { } fn start_using(&self, startup: impl FnOnce(&mut Handle) -> Result<()>) -> Result { + self.start_number_using(248, startup) + } + + fn start_number_using( + &self, + number: u64, + startup: impl FnOnce(&mut Handle) -> Result<()>, + ) -> Result { 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::().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() { diff --git a/src/child_run/runs.rs b/src/child_run/runs.rs new file mode 100644 index 000000000..da58eb5de --- /dev/null +++ b/src/child_run/runs.rs @@ -0,0 +1,93 @@ +//! Owned concurrent child Runs. + +use std::collections::{BTreeMap, VecDeque}; +use std::thread; + +use anyhow::{Result, bail}; + +use super::{Ended, Handle, Kind, POLL}; +use crate::base_fix::BaseFixAsk; +use crate::harness::Choice; +use crate::issue::IssueUrl; + +/// Own child Runs until their numbered endings are delivered or cleanup finishes. +#[derive(Default)] +pub struct Runs { + running: BTreeMap, + endings: VecDeque<(u64, Result)>, +} + +impl Runs { + /// Start and own a child with the same launch facts as the standalone Run. + /// An issue number stays outstanding until its ending is delivered; + /// refusing a duplicate or failing to launch keeps earlier Runs owned. + pub fn start( + &mut self, + issue: &IssueUrl, + kind: Kind, + base_fix: BaseFixAsk, + harness: &Choice, + ) -> Result<()> { + self.start_using(issue.number, || { + super::start(issue, kind, base_fix, harness) + }) + } + + /// Launch and retain ownership before returning to the caller. + pub(super) fn start_using( + &mut self, + number: u64, + launch: impl FnOnce() -> Result, + ) -> Result<()> { + if self.running.contains_key(&number) + || self.endings.iter().any(|(queued, _)| *queued == number) + { + bail!("the Run for #{number} is already outstanding"); + } + let child = launch()?; + self.running.insert(number, child); + Ok(()) + } + + /// Poll every active child, then deliver the oldest numbered ending. + /// Endings discovered together follow ascending issue-number order. + /// Retirement follows reaping and reader joins; each ending is delivered + /// once. Empty returns None immediately; otherwise wait at the child + /// supervision cadence, including during Command interruption. + pub fn next_ending(&mut self) -> Option<(u64, Result)> { + loop { + // Sweep all children even with endings queued, so interruption + // reaches every active child before any result is delivered. + self.running.retain(|&number, child| { + if let Some(ending) = child.try_wait() { + self.endings.push_back((number, ending)); + false + } else { + true + } + }); + if let Some(ending) = self.endings.pop_front() { + return Some(ending); + } + if self.running.is_empty() { + return None; + } + // GitHub's polling helper returns early on interruption; child + // supervision keeps its cadence while graceful cleanup runs. + thread::sleep(POLL); + } + } +} + +impl Drop for Runs { + fn drop(&mut self) { + // Sibling salvage may rendezvous: notify every child before waiting + // for any of them. Handle::stop is nonblocking and once-only. + for child in self.running.values_mut() { + child.stop(); + } + // Reap and join through normal completion, without a deadline or a + // synthetic Command interruption. Undelivered endings are discarded. + while self.next_ending().is_some() {} + } +} diff --git a/src/spec_run.rs b/src/spec_run.rs index 5b3abb77f..ecf7a3ac1 100644 --- a/src/spec_run.rs +++ b/src/spec_run.rs @@ -12,15 +12,12 @@ mod spec_pr; mod ticket_board; -use std::collections::{BTreeMap, VecDeque}; use std::num::NonZeroUsize; -use std::thread; -use std::time::Duration; use anyhow::{Result, bail}; use crate::base_fix::BaseFixAsk; -use crate::child_run::{self, Ended, Handle, Kind}; +use crate::child_run::{Ended, Kind, Runs}; use crate::delivery::{Delivery, Opening}; use crate::failed_run::{FailedRun, interrupted_or}; use crate::github::{GitHub, Ticket}; @@ -192,11 +189,14 @@ trait Outside { /// The outside world of a Spec run on `delivery`'s Spec, from its Spec /// branch checked out in `worktree`: child `thirdshift` Runs, each of which /// may start a Base fix if `base_fix` allows one, and runs its sessions on -/// `harness`, their owned handles in `running`, GitHub, the Spec PR +/// `harness`, owned together in `children`, GitHub, the Spec PR /// `spec_pr`, and `delivery`, which takes the /// worktree when it starts. struct ChildRunsAndGitHub<'a> { spec: &'a IssueUrl, + /// Fields drop in declaration order: stop and drain child salvage before + /// Worktree cleanup, including when the Spec run owner unwinds. + children: Runs, /// Held until the Delivery starts, which takes it. worktree: Option, /// Held until it starts. @@ -204,8 +204,6 @@ struct ChildRunsAndGitHub<'a> { base_fix: BaseFixAsk, harness: &'a Choice, spec_pr: SpecPr<'a>, - running: BTreeMap, - endings: VecDeque<(u64, Result)>, } impl<'a> ChildRunsAndGitHub<'a> { @@ -214,13 +212,12 @@ impl<'a> ChildRunsAndGitHub<'a> { fn new(worktree: Worktree, delivery: Delivery<'a>, spec_pr: SpecPr<'a>) -> Self { Self { spec: delivery.issue, + children: Runs::default(), worktree: Some(worktree), base_fix: delivery.base_fix.ask_of_tickets(), harness: delivery.harness, delivery: Some(delivery), spec_pr, - running: BTreeMap::new(), - endings: VecDeque::new(), } } @@ -237,43 +234,28 @@ impl Outside for ChildRunsAndGitHub<'_> { self.worktree().push() } - /// Retain the child Run's ownership until polling delivers its ending. fn start_ticket(&mut self, number: u64) -> Result<()> { let kind = Kind::Ticket { spec_branch: self.worktree().branch().to_string(), }; let ticket = self.spec.sibling(number); - let child = child_run::start(&ticket, kind, self.base_fix.clone(), self.harness)?; - self.running.insert(number, child); - Ok(()) + self.children + .start(&ticket, kind, self.base_fix.clone(), self.harness) } fn next_ending(&mut self) -> (u64, Result) { - loop { - // Scan every child before returning an ending. In particular, - // interruption reaches all active Tickets before waiting on any. - for (&number, child) in &mut self.running { - if let Some(ending) = child.try_wait() { - self.endings.push_back((number, ending)); - } - } - for (number, _) in &self.endings { - self.running.remove(number); - } - if let Some((number, ending)) = self.endings.pop_front() { - if let Ok(ended) = &ending { - progress::step(match ended { - Ended::Reached { .. } => format!("#{number} landed"), - Ended::Interrupted => format!("#{number} interrupted"), - Ended::Failed { .. } => format!("#{number} failed"), - }); - } - return (number, ending); - } - // GitHub's polling helper returns early on interruption; child - // supervision must keep its cadence while graceful cleanup runs. - thread::sleep(Duration::from_millis(100)); + let (number, ending) = self + .children + .next_ending() + .expect("the Ticket board has an outstanding Run"); + if let Ok(ended) = &ending { + progress::step(match ended { + Ended::Reached { .. } => format!("#{number} landed"), + Ended::Interrupted => format!("#{number} interrupted"), + Ended::Failed { .. } => format!("#{number} failed"), + }); } + (number, ending) } fn tickets(&mut self) -> Result> {