use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
pub const LAUNCHER_SESSION_ENV: &str = "ONEPIPELINE_LAUNCHER_SESSION";
pub const LAUNCHER_ENV: &str = "ONEPIPELINE_LAUNCHER";
pub const UNKNOWN_LAUNCHER: &str = "unknown";
pub fn now_millis() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| u64::try_from(d.as_millis()).unwrap_or(u64::MAX))
.unwrap_or(0)
}
pub fn now_rfc3339() -> String {
rfc3339_from_millis(now_millis())
}
pub fn rfc3339_from_millis(millis: u64) -> String {
let secs = millis / 1_000;
let ms = millis % 1_000;
let days = i64::try_from(secs / 86_400).unwrap_or(0);
let sod = secs % 86_400;
let (year, month, day) = civil_from_days(days);
let (hour, minute, second) = (sod / 3_600, (sod % 3_600) / 60, sod % 60);
format!("{year:04}-{month:02}-{day:02}T{hour:02}:{minute:02}:{second:02}.{ms:03}Z")
}
fn civil_from_days(days: i64) -> (i64, u64, u64) {
let z = days + 719_468;
let era = if z >= 0 { z } else { z - 146_096 } / 146_097;
let doe = z - era * 146_097;
let yoe = (doe - doe / 1_460 + doe / 36_524 - doe / 146_096) / 365;
let y = yoe + era * 400;
let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
let mp = (5 * doy + 2) / 153;
let d = doy - (153 * mp + 2) / 5 + 1;
let m = if mp < 10 { mp + 3 } else { mp - 9 };
let year = if m <= 2 { y + 1 } else { y };
(year, m as u64, d as u64)
}
pub fn pid() -> u32 {
std::process::id()
}
pub fn hostname() -> String {
for key in ["HOSTNAME", "COMPUTERNAME"] {
if let Ok(value) = std::env::var(key) {
if !value.is_empty() {
return value;
}
}
}
std::fs::read_to_string("/etc/hostname")
.ok()
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
.unwrap_or_else(|| "localhost".to_string())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Teardown {
Signalled,
NothingToStop,
NotAttempted,
PartlySignalled,
#[cfg(unix)]
Refused,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Stop {
Politely,
Now,
}
pub fn stop(pid: u32, how: Stop) -> Teardown {
platform_stop(&[pid], how).0
}
pub fn stop_and_confirm(pids: &[u32], how: Stop, patience: Duration) -> Teardown {
let (established, aimed) = platform_stop(pids, how);
if established != Teardown::Signalled {
return established;
}
if gone_within(&aimed, patience) {
Teardown::Signalled
} else {
Teardown::PartlySignalled
}
}
fn gone_within(aimed: &[u32], patience: Duration) -> bool {
let deadline = Instant::now().checked_add(patience);
loop {
if !aimed.iter().any(|pid| process_may_be_live(*pid)) {
return true;
}
if deadline.is_some_and(|deadline| Instant::now() >= deadline) {
return false;
}
std::thread::sleep(PROBE_POLL);
}
}
const PROBE_POLL: Duration = Duration::from_millis(20);
fn aimable(roots: &[u32]) -> Vec<u32> {
let mut aimed: Vec<u32> = Vec::new();
for root in roots
.iter()
.copied()
.filter(|pid| *pid != 0 && *pid != self::pid())
{
if !aimed.contains(&root) {
aimed.push(root);
}
}
aimed
}
#[cfg(unix)]
fn platform_stop(roots: &[u32], how: Stop) -> (Teardown, Vec<u32>) {
let signal = match how {
Stop::Politely => libc::SIGTERM,
Stop::Now => libc::SIGKILL,
};
let mut aimed = aimable(roots);
if aimed.is_empty() {
return (Teardown::NothingToStop, aimed);
}
let Some(table) = process_table() else {
return (Teardown::NotAttempted, Vec::new());
};
for root in aimed.clone() {
for descendant in descended_from(&table, root) {
if !aimed.contains(&descendant) {
aimed.push(descendant);
}
}
}
let answers: Vec<Reached> = aimed.iter().map(|pid| signal_one(*pid, signal)).collect();
(established(&answers), aimed)
}
#[cfg(unix)]
fn established(answers: &[Reached]) -> Teardown {
let refused = answers.contains(&Reached::Refused);
let delivered = answers.contains(&Reached::Delivered);
match (refused, delivered) {
(true, true) => Teardown::PartlySignalled,
(true, false) => Teardown::Refused,
(false, true) => Teardown::Signalled,
(false, false) => Teardown::NothingToStop,
}
}
#[cfg(unix)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Reached {
Delivered,
Absent,
Refused,
}
#[cfg(unix)]
fn signal_one(pid: u32, signal: i32) -> Reached {
let Ok(raw) = i32::try_from(pid) else {
return Reached::Refused;
};
if raw <= 0 {
return Reached::Refused;
}
if unsafe { libc::kill(raw, signal) } == 0 {
return Reached::Delivered;
}
if std::io::Error::last_os_error().raw_os_error() == Some(libc::ESRCH) {
return Reached::Absent;
}
Reached::Refused
}
#[cfg(unix)]
fn descended_from(table: &[(u32, u32)], pid: u32) -> Vec<u32> {
let mut found: Vec<u32> = Vec::new();
let mut frontier = vec![pid];
while let Some(parent) = frontier.pop() {
for (child, _) in table.iter().filter(|(_, ppid)| *ppid == parent) {
if *child != pid && !found.contains(child) {
found.push(*child);
frontier.push(*child);
}
}
}
found
}
#[cfg(unix)]
fn process_table() -> Option<Vec<(u32, u32)>> {
let listed = std::process::Command::new("ps")
.args(["-A", "-o", "pid=,ppid="])
.stderr(std::process::Stdio::null())
.output()
.ok()?;
if !listed.status.success() {
return None;
}
parse_table(&String::from_utf8(listed.stdout).ok()?)
}
#[cfg(unix)]
fn parse_table(listed: &str) -> Option<Vec<(u32, u32)>> {
listed
.lines()
.filter(|line| !line.trim().is_empty())
.map(|line| {
let mut columns = line.split_whitespace();
let pid: u32 = columns.next()?.parse().ok()?;
let parent: u32 = columns.next()?.parse().ok()?;
(columns.next().is_none() && pid != 0).then_some((pid, parent))
})
.collect()
}
#[cfg(windows)]
fn platform_stop(roots: &[u32], _how: Stop) -> (Teardown, Vec<u32>) {
let aimed: Vec<u32> = aimable(roots)
.into_iter()
.filter(|pid| platform_process_may_be_live(*pid))
.collect();
if aimed.is_empty() {
return (Teardown::NothingToStop, aimed);
}
let mut walked = true;
let mut attempted = false;
for pid in &aimed {
match taskkill_established(taskkill(*pid), || platform_process_may_be_live(*pid)) {
Teardown::Signalled => attempted = true,
Teardown::PartlySignalled => {
attempted = true;
walked = false;
}
Teardown::NotAttempted => walked = false,
Teardown::NothingToStop => {}
}
}
let established = match (walked, attempted) {
(true, _) => Teardown::Signalled,
(false, true) => Teardown::PartlySignalled,
(false, false) => Teardown::NotAttempted,
};
(established, aimed)
}
#[cfg(windows)]
fn taskkill(pid: u32) -> std::io::Result<std::process::ExitStatus> {
std::process::Command::new("taskkill")
.args(["/PID", &pid.to_string(), "/T", "/F"])
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.status()
}
#[cfg(windows)]
fn taskkill_established(
ran: std::io::Result<std::process::ExitStatus>,
still_live: impl FnOnce() -> bool,
) -> Teardown {
match ran {
Ok(status) if status.success() => Teardown::Signalled,
Err(_) => Teardown::NotAttempted,
Ok(_) if still_live() => Teardown::PartlySignalled,
Ok(_) => Teardown::Signalled,
}
}
pub fn process_may_be_live(pid: u32) -> bool {
if pid == 0 {
return false;
}
platform_process_may_be_live(pid)
}
#[cfg(unix)]
fn platform_process_may_be_live(pid: u32) -> bool {
let Ok(raw) = i32::try_from(pid) else {
return true;
};
let rc = unsafe { libc::kill(raw, 0) };
if rc == 0 {
return true;
}
std::io::Error::last_os_error().raw_os_error() != Some(libc::ESRCH)
}
#[cfg(windows)]
fn platform_process_may_be_live(pid: u32) -> bool {
use windows_sys::Win32::Foundation::{CloseHandle, ERROR_INVALID_PARAMETER, WAIT_OBJECT_0};
use windows_sys::Win32::System::Threading::{
OpenProcess, WaitForSingleObject, PROCESS_SYNCHRONIZE,
};
let handle = unsafe { OpenProcess(PROCESS_SYNCHRONIZE, 0, pid) };
if handle.is_null() {
return std::io::Error::last_os_error().raw_os_error()
!= Some(ERROR_INVALID_PARAMETER as i32);
}
let waited = unsafe { WaitForSingleObject(handle, 0) };
unsafe { CloseHandle(handle) };
waited != WAIT_OBJECT_0
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StartToken(String);
impl StartToken {
pub fn recorded(&self) -> &str {
&self.0
}
pub fn matches(&self, recorded: &str) -> bool {
!recorded.is_empty() && self.0 == recorded
}
}
pub fn process_start_token(pid: u32) -> Option<StartToken> {
if pid == 0 {
return None;
}
platform_process_start_token(pid).map(StartToken)
}
#[cfg(unix)]
fn platform_process_start_token(pid: u32) -> Option<String> {
let listed = std::process::Command::new("ps")
.args(["-p", &pid.to_string(), "-o", "lstart="])
.env("TZ", "UTC")
.env("LC_ALL", "C")
.stderr(std::process::Stdio::null())
.output()
.ok()?;
if !listed.status.success() {
return None;
}
let answer = String::from_utf8(listed.stdout).ok()?;
let mut lines = answer
.lines()
.map(str::trim)
.filter(|line| !line.is_empty());
let token = lines.next()?.to_string();
lines.next().is_none().then_some(token)
}
#[cfg(windows)]
fn platform_process_start_token(pid: u32) -> Option<String> {
use windows_sys::Win32::Foundation::{CloseHandle, FILETIME, WAIT_TIMEOUT};
use windows_sys::Win32::System::Threading::{
GetProcessTimes, OpenProcess, WaitForSingleObject, PROCESS_QUERY_LIMITED_INFORMATION,
PROCESS_SYNCHRONIZE,
};
let handle = unsafe {
OpenProcess(
PROCESS_QUERY_LIMITED_INFORMATION | PROCESS_SYNCHRONIZE,
0,
pid,
)
};
if handle.is_null() {
return None;
}
let mut created = FILETIME {
dwLowDateTime: 0,
dwHighDateTime: 0,
};
let mut exited = created;
let mut kernel = created;
let mut user = created;
let read = unsafe {
GetProcessTimes(
handle,
&raw mut created,
&raw mut exited,
&raw mut kernel,
&raw mut user,
)
};
let waited = unsafe { WaitForSingleObject(handle, 0) };
unsafe { CloseHandle(handle) };
(read != 0 && waited == WAIT_TIMEOUT)
.then(|| format!("{}:{}", created.dwHighDateTime, created.dwLowDateTime))
}
pub fn open_locked_append(path: &std::path::Path) -> std::io::Result<std::fs::File> {
platform_open_locked_append(path)
}
#[cfg(unix)]
fn platform_open_locked_append(path: &std::path::Path) -> std::io::Result<std::fs::File> {
use std::os::unix::io::AsRawFd;
let file = std::fs::OpenOptions::new()
.create(true)
.append(true)
.read(true)
.open(path)?;
loop {
if unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX) } == 0 {
return Ok(file);
}
let failed = std::io::Error::last_os_error();
if failed.kind() != std::io::ErrorKind::Interrupted {
return Err(failed);
}
}
}
#[cfg(windows)]
fn platform_open_locked_append(path: &std::path::Path) -> std::io::Result<std::fs::File> {
use std::os::windows::fs::OpenOptionsExt;
use windows_sys::Win32::Foundation::ERROR_SHARING_VIOLATION;
use windows_sys::Win32::Storage::FileSystem::FILE_SHARE_READ;
const DEADLINE: std::time::Duration = std::time::Duration::from_secs(30);
let waiting_since = std::time::Instant::now();
loop {
let opened = std::fs::OpenOptions::new()
.create(true)
.write(true)
.read(true)
.share_mode(FILE_SHARE_READ)
.open(path);
match opened {
Err(e)
if e.raw_os_error() == Some(ERROR_SHARING_VIOLATION as i32)
&& waiting_since.elapsed() < DEADLINE =>
{
std::thread::sleep(std::time::Duration::from_millis(2));
}
other => return other,
}
}
}
pub fn disown_standard_handles() {
platform_disown_standard_handles();
}
#[cfg(unix)]
fn platform_disown_standard_handles() {}
#[cfg(windows)]
fn platform_disown_standard_handles() {
use windows_sys::Win32::Foundation::{
SetHandleInformation, HANDLE_FLAG_INHERIT, INVALID_HANDLE_VALUE,
};
use windows_sys::Win32::System::Console::{
GetStdHandle, STD_ERROR_HANDLE, STD_INPUT_HANDLE, STD_OUTPUT_HANDLE,
};
for which in [STD_INPUT_HANDLE, STD_OUTPUT_HANDLE, STD_ERROR_HANDLE] {
let handle = unsafe { GetStdHandle(which) };
if handle.is_null() || handle == INVALID_HANDLE_VALUE {
continue;
}
unsafe { SetHandleInformation(handle, HANDLE_FLAG_INHERIT, 0) };
}
}
pub fn launching_session() -> String {
std::env::var(LAUNCHER_SESSION_ENV)
.ok()
.filter(|s| !s.is_empty())
.unwrap_or_else(|| UNKNOWN_LAUNCHER.to_string())
}
pub fn launcher() -> String {
std::env::var(LAUNCHER_ENV)
.ok()
.filter(|s| !s.is_empty())
.unwrap_or_else(|| UNKNOWN_LAUNCHER.to_string())
}
pub fn session_digest(session: &str) -> String {
let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
for byte in session.as_bytes() {
hash ^= u64::from(*byte);
hash = hash.wrapping_mul(0x100_0000_01b3);
}
format!("{:08x}", (hash >> 32) as u32)
}
#[cfg(test)]
pub(crate) fn reaped_pid() -> u32 {
let mut child = std::process::Command::new(
std::env::current_exe().expect("the test binary knows its own path"),
)
.args(["--list", "--format", "terse"])
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.spawn()
.expect("the test binary starts");
let pid = child.id();
child.wait().expect("it exits");
pid
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_epoch_renders_as_rfc3339_millis() {
assert_eq!(rfc3339_from_millis(0), "1970-01-01T00:00:00.000Z");
}
#[test]
fn a_known_instant_renders_with_its_milliseconds() {
assert_eq!(
rfc3339_from_millis(1_786_195_785_678),
"2026-08-08T13:29:45.678Z"
);
}
#[test]
fn a_leap_day_is_not_skipped() {
assert_eq!(
rfc3339_from_millis(1_709_164_800_000),
"2024-02-29T00:00:00.000Z"
);
}
#[test]
fn now_is_rendered_in_the_envelope_shape() {
let now = now_rfc3339();
assert_eq!(now.len(), 24, "{now} is not RFC 3339 millisecond UTC");
assert!(now.ends_with('Z'), "{now} is not UTC");
}
#[test]
fn this_process_is_live_and_pid_zero_is_not() {
assert!(process_may_be_live(pid()));
assert!(!process_may_be_live(0));
}
#[test]
fn a_reaped_process_is_proved_gone() {
let dead = reaped_pid();
assert!(!process_may_be_live(dead), "pid {dead} was reaped");
}
#[test]
fn a_start_token_is_stable_for_one_process_and_absent_for_a_pid_nothing_holds() {
let mine = process_start_token(pid()).expect("this host says when a process started");
assert!(!mine.recorded().is_empty());
assert_eq!(
process_start_token(pid()),
Some(mine.clone()),
"one process gave two different start tokens"
);
assert!(mine.matches(mine.recorded()));
assert!(!mine.matches(""));
assert!(!mine.matches("some other process's start"));
let dead = reaped_pid();
assert!(
process_start_token(dead).is_none(),
"pid {dead} was reaped and still answered with a start"
);
assert!(process_start_token(0).is_none());
}
#[test]
fn an_exited_process_gives_no_start_even_while_a_handle_to_it_is_held() {
let mut child = std::process::Command::new(
std::env::current_exe().expect("the test binary knows its own path"),
)
.args(["--list", "--format", "terse"])
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.spawn()
.expect("the test binary starts");
let dead = child.id();
child.wait().expect("it exits");
assert!(
process_start_token(dead).is_none(),
"pid {dead} has exited and still answered with a start"
);
assert!(
!process_may_be_live(dead),
"pid {dead} has exited and still read as live"
);
drop(child);
}
#[cfg(unix)]
fn a_stop_reaches_the_whole_descendant_tree_and_not_the_process_beside_it(how: Stop) {
use std::os::unix::process::CommandExt;
let mut tree = std::process::Command::new("sh")
.args([
"-c",
"exec 2>&1; echo $$; sh -c 'echo $$; sh -c \"echo \\$\\$; sleep 120\" & \
sleep 120' & sleep 120",
])
.process_group(0)
.stdout(std::process::Stdio::piped())
.spawn()
.expect("a process tree");
let group = i32::try_from(tree.id()).expect("a pid is a process group id");
let mut beside = std::process::Command::new("sh")
.args(["-c", "sleep 120"])
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.spawn()
.expect("a process beside the tree");
let mut beside_in_group = std::process::Command::new("sh")
.args(["-c", "sleep 120"])
.process_group(group)
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.spawn()
.expect("a process in the tree's process group");
let mut pids = Vec::new();
{
use std::io::BufRead;
let out = std::io::BufReader::new(tree.stdout.take().expect("the tree reports itself"));
for line in out.lines().take(3) {
let line = line.expect("a reported pid");
pids.push(
line.trim()
.parse::<u32>()
.unwrap_or_else(|_| panic!("the tree said {line:?} where a pid was due")),
);
}
}
assert_eq!(pids.len(), 3, "the tree did not report three levels");
assert!(
pids.iter().all(|pid| process_may_be_live(*pid)),
"the tree was not running before it was stopped: {pids:?}"
);
stop(pids[0], how);
let patience = std::time::Duration::from_secs(10);
let reaped = ended_within(&mut tree, patience);
let deadline = std::time::Instant::now() + patience;
while std::time::Instant::now() < deadline
&& pids.iter().any(|pid| process_may_be_live(*pid))
{
std::thread::sleep(std::time::Duration::from_millis(20));
}
assert!(
reaped,
"the stop never reached the root of the tree {pids:?}"
);
let surviving: Vec<u32> = pids
.iter()
.copied()
.filter(|pid| process_may_be_live(*pid))
.collect();
assert!(
surviving.is_empty(),
"a stop left {surviving:?} of the tree {pids:?} running — the leaf is the paid one"
);
assert!(
still_running(&mut beside),
"a stop took a process that was beside the tree rather than under it"
);
assert!(
still_running(&mut beside_in_group),
"a stop took a process that shared the tree's process group without being descended \
from it — the boundary a teardown ends is descent, not the group"
);
for bystander in [&mut beside, &mut beside_in_group] {
let _ = bystander.kill();
let _ = bystander.wait();
}
}
#[cfg(unix)]
#[test]
fn a_polite_stop_reaches_the_whole_descendant_tree() {
a_stop_reaches_the_whole_descendant_tree_and_not_the_process_beside_it(Stop::Politely);
}
#[cfg(unix)]
#[test]
fn a_forceful_stop_reaches_the_whole_descendant_tree() {
a_stop_reaches_the_whole_descendant_tree_and_not_the_process_beside_it(Stop::Now);
}
#[cfg(unix)]
fn still_running(child: &mut std::process::Child) -> bool {
matches!(child.try_wait(), Ok(None))
}
#[cfg(unix)]
fn ended_within(child: &mut std::process::Child, patience: std::time::Duration) -> bool {
let deadline = std::time::Instant::now() + patience;
while std::time::Instant::now() < deadline {
if matches!(child.try_wait(), Ok(Some(_))) {
return true;
}
std::thread::sleep(std::time::Duration::from_millis(20));
}
false
}
#[cfg(unix)]
#[test]
fn a_signal_separates_a_process_it_reached_from_one_already_gone_and_from_a_broadcast() {
assert_eq!(
signal_one(pid(), 0),
Reached::Delivered,
"a live process was not reported as reached"
);
assert_eq!(
signal_one(reaped_pid(), libc::SIGTERM),
Reached::Absent,
"a process that had already exited was reported as one this teardown ended"
);
assert_eq!(
signal_one(0, libc::SIGTERM),
Reached::Refused,
"pid 0 was signalled, and to `kill` it is a whole process group"
);
}
#[test]
fn a_teardown_aimed_at_a_tree_that_has_already_gone_says_there_was_nothing_to_stop() {
let dead = reaped_pid();
assert!(
!process_may_be_live(dead),
"the reaped pid {dead} was still live, so this is not the case under test"
);
assert_eq!(
stop(dead, Stop::Politely),
Teardown::NothingToStop,
"a stop that found nothing to aim at reported having reached a tree"
);
assert_eq!(stop(0, Stop::Politely), Teardown::NothingToStop);
assert_eq!(stop(pid(), Stop::Politely), Teardown::NothingToStop);
}
#[cfg(unix)]
#[test]
fn a_teardown_refused_by_everything_it_aimed_at_reports_no_signal_at_all() {
assert_eq!(
established(&[Reached::Refused, Reached::Refused]),
Teardown::Refused,
"a teardown that delivered nothing reported part of the tree signalled"
);
assert_eq!(
established(&[Reached::Refused, Reached::Absent]),
Teardown::Refused,
"a process already gone was counted as a signal this teardown delivered"
);
assert_eq!(
established(&[Reached::Refused, Reached::Delivered]),
Teardown::PartlySignalled,
"a tree part of which took the signal was not reported as partly signalled"
);
assert_eq!(
established(&[Reached::Delivered, Reached::Absent]),
Teardown::Signalled,
"a tree every process of which was reached was not reported as reached"
);
assert_eq!(
established(&[Reached::Absent, Reached::Absent]),
Teardown::NothingToStop,
"a walk that met nothing but processes already gone reported a stop it made"
);
}
#[cfg(unix)]
#[test]
fn a_stop_that_could_signal_nothing_it_aimed_at_says_so() {
assert_eq!(
stop(u32::MAX, Stop::Politely),
Teardown::Refused,
"a stop that signalled nothing reported having reached part of a tree"
);
}
#[cfg(unix)]
fn orphaned(script: &str, levels: usize) -> Vec<u32> {
use std::io::BufRead;
let mut spawner = std::process::Command::new("sh")
.args(["-c", &format!("{script} &")])
.stdout(std::process::Stdio::piped())
.spawn()
.expect("a fixture tree");
let reported = spawner.stdout.take().expect("the tree reports itself");
let pids: Vec<u32> = std::io::BufReader::new(reported)
.lines()
.take(levels)
.map(|line| {
let line = line.expect("a reported pid");
line.trim()
.parse()
.unwrap_or_else(|_| panic!("the tree said {line:?} where a pid was due"))
})
.collect();
spawner.wait().expect("the shell that detached it exits");
assert_eq!(
pids.len(),
levels,
"the fixture reported {pids:?} where {levels} level(s) were due"
);
pids
}
#[cfg(unix)]
#[test]
fn a_stop_that_watches_reports_a_tree_that_took_the_ask_and_stayed() {
let deaf = orphaned("sh -c 'trap \"\" TERM; echo $$; sleep 120'", 1)[0];
assert_eq!(
stop_and_confirm(
&[deaf],
Stop::Politely,
std::time::Duration::from_millis(300)
),
Teardown::PartlySignalled,
"a stop watched pid {deaf} never go and still called it a clean stop"
);
assert!(
process_may_be_live(deaf),
"pid {deaf} ended on the polite ask, so the answer above proves nothing"
);
assert_eq!(
stop_and_confirm(&[deaf], Stop::Now, std::time::Duration::from_secs(10)),
Teardown::Signalled,
"a tree that went was not reported as reached"
);
assert!(
!process_may_be_live(deaf),
"the forceful ask left pid {deaf} running"
);
}
#[cfg(unix)]
#[test]
fn a_stop_aimed_at_several_roots_ends_every_tree_and_leaves_the_one_beside_them() {
let trees: Vec<Vec<u32>> = (0..2)
.map(|_| {
orphaned(
"sh -c 'echo $$; sh -c \"echo \\$\\$; sleep 120\" & sleep 120'",
2,
)
})
.collect();
let beside = orphaned("sh -c 'echo $$; sleep 120'", 1)[0];
let roots: Vec<u32> = trees.iter().map(|tree| tree[0]).collect();
let every: Vec<u32> = trees.concat();
assert!(
every.iter().all(|pid| process_may_be_live(*pid)),
"the trees {every:?} were not running before they were stopped"
);
assert_eq!(
stop_and_confirm(&roots, Stop::Now, std::time::Duration::from_secs(10)),
Teardown::Signalled,
"a stop that ended {every:?} did not report reaching them"
);
let surviving: Vec<u32> = every
.iter()
.copied()
.filter(|pid| process_may_be_live(*pid))
.collect();
assert!(
surviving.is_empty(),
"a stop of several trees left {surviving:?} of {every:?} running"
);
assert!(
process_may_be_live(beside),
"a stop of several trees took pid {beside}, which was under none of them"
);
stop(beside, Stop::Now);
}
#[cfg(windows)]
fn console_tree() -> (std::process::Child, u32) {
let mut root = std::process::Command::new("cmd")
.args(["/C", "ping -n 120 127.0.0.1"])
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.spawn()
.expect("a console process tree");
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
let mut leaf = None;
while leaf.is_none() && std::time::Instant::now() < deadline {
leaf = child_of(root.id());
if leaf.is_none() {
std::thread::sleep(std::time::Duration::from_millis(100));
}
}
match leaf {
Some(leaf) => (root, leaf),
None => {
let pid = root.id();
let _ = root.kill();
let _ = root.wait();
panic!("the tree under {pid} never started its leaf");
}
}
}
#[cfg(windows)]
fn child_of(parent: u32) -> Option<u32> {
let listed = std::process::Command::new("powershell")
.args([
"-NoProfile",
"-Command",
&format!(
"(Get-CimInstance Win32_Process -Filter 'ParentProcessId={parent}').ProcessId"
),
])
.output()
.expect("this host lists its processes");
String::from_utf8_lossy(&listed.stdout)
.lines()
.find_map(|line| line.trim().parse::<u32>().ok())
}
#[cfg(windows)]
fn all_ended_within(tree: &[u32], patience: std::time::Duration) -> bool {
let deadline = std::time::Instant::now() + patience;
while std::time::Instant::now() < deadline {
if tree.iter().all(|pid| !platform_process_may_be_live(*pid)) {
return true;
}
std::thread::sleep(std::time::Duration::from_millis(50));
}
false
}
#[cfg(windows)]
#[test]
fn a_polite_taskkill_cannot_end_a_console_process() {
let (mut root, leaf) = console_tree();
let pid = root.id();
let asked = std::process::Command::new("taskkill")
.args(["/PID", &pid.to_string(), "/T"])
.output()
.expect("taskkill runs");
let said = String::from_utf8_lossy(&asked.stderr).into_owned();
assert!(
!asked.status.success(),
"a polite taskkill reported that it ended a console tree: {said}"
);
assert!(
platform_process_may_be_live(pid) && platform_process_may_be_live(leaf),
"the polite ask ended part of the console tree {pid}/{leaf} after all, so the \
forceful ask below is not the only one that reaches it: {said}"
);
assert_eq!(
stop(pid, Stop::Now),
Teardown::Signalled,
"the forceful ask did not reach the tree the polite one could not"
);
assert!(all_ended_within(
&[pid, leaf],
std::time::Duration::from_secs(10)
));
let _ = root.wait();
}
#[cfg(windows)]
#[test]
fn a_taskkill_failure_does_not_say_which_failure_it_was() {
let (mut root, leaf) = console_tree();
let pid = root.id();
let refused = std::process::Command::new("taskkill")
.args(["/PID", &pid.to_string(), "/T"])
.output()
.expect("taskkill runs");
assert!(
platform_process_may_be_live(pid),
"the tree ended by itself"
);
let dead = reaped_pid();
assert!(
!platform_process_may_be_live(dead),
"the reaped pid {dead} was still live"
);
let absent = std::process::Command::new("taskkill")
.args(["/PID", &dead.to_string(), "/T", "/F"])
.output()
.expect("taskkill runs");
assert!(
!refused.status.success() && !absent.status.success(),
"a taskkill reported success for a tree it did not end"
);
assert_eq!(
refused.status.code(),
absent.status.code(),
"a taskkill that was refused a running process and one that found nothing to end \
report different statuses, so the teardown could read the difference off the \
status after all"
);
stop(pid, Stop::Now);
assert!(all_ended_within(
&[pid, leaf],
std::time::Duration::from_secs(10)
));
let _ = root.wait();
}
#[cfg(windows)]
fn a_stop_reaches_the_whole_console_tree(how: Stop) {
let (mut root, leaf) = console_tree();
let pid = root.id();
assert!(
platform_process_may_be_live(pid) && platform_process_may_be_live(leaf),
"the tree {pid}/{leaf} was not running before it was stopped"
);
assert_eq!(
stop(pid, how),
Teardown::Signalled,
"a stop that reached the tree {pid}/{leaf} did not report reaching it"
);
assert!(
all_ended_within(&[pid, leaf], std::time::Duration::from_secs(10)),
"a stop left part of the tree {pid}/{leaf} running — the leaf is the paid one"
);
let _ = root.wait();
}
#[cfg(windows)]
#[test]
fn a_polite_stop_reaches_the_whole_console_tree() {
a_stop_reaches_the_whole_console_tree(Stop::Politely);
}
#[cfg(windows)]
#[test]
fn a_forceful_stop_reaches_the_whole_console_tree() {
a_stop_reaches_the_whole_console_tree(Stop::Now);
}
#[cfg(windows)]
#[test]
fn a_stop_aimed_at_several_console_trees_ends_every_one_of_them() {
let (mut first, first_leaf) = console_tree();
let (mut second, second_leaf) = console_tree();
let roots = [first.id(), second.id()];
let every = [first.id(), first_leaf, second.id(), second_leaf];
assert!(
every.iter().all(|pid| platform_process_may_be_live(*pid)),
"the trees {every:?} were not running before they were stopped"
);
assert_eq!(
stop_and_confirm(&roots, Stop::Now, std::time::Duration::from_secs(10)),
Teardown::Signalled,
"a stop that ended the trees {every:?} did not report reaching them"
);
assert!(
all_ended_within(&every, std::time::Duration::from_secs(10)),
"a stop of several trees left part of {every:?} running"
);
let _ = first.wait();
let _ = second.wait();
}
#[cfg(windows)]
#[test]
fn a_failed_taskkill_is_read_from_what_is_still_running_not_from_its_status() {
use std::os::windows::process::ExitStatusExt;
let exited = |code: u32| Ok(std::process::ExitStatus::from_raw(code));
let never_asked = || panic!("liveness was asked about a teardown that settled without it");
assert_eq!(
taskkill_established(exited(0), never_asked),
Teardown::Signalled,
"a taskkill that walked the tree was not reported as having reached it"
);
assert_eq!(
taskkill_established(
Err(std::io::Error::from(std::io::ErrorKind::NotFound)),
never_asked
),
Teardown::NotAttempted,
"a taskkill that never ran was reported as having touched the tree"
);
assert_eq!(
taskkill_established(exited(128), || true),
Teardown::PartlySignalled,
"a teardown that left a process running was reported as a clean stop"
);
assert_eq!(
taskkill_established(exited(128), || false),
Teardown::Signalled,
"a tree that was already gone was reported as a process still to be found"
);
}
#[cfg(unix)]
#[test]
fn a_listing_with_a_row_it_cannot_read_is_no_listing_at_all() {
assert_eq!(
parse_table("11 10\n13 11\n"),
Some(vec![(11, 10), (13, 11)]),
"a listing every line of which is two ids was not read"
);
for unreadable in [
"11 10\nnot-a-pid also-not\n13 11\n",
"11 10\n14\n",
" PID PPID\n11 10\n",
"11 10 and-a-third\n",
] {
assert_eq!(
parse_table(unreadable),
None,
"a listing holding {unreadable:?} was read as a tree anyway"
);
}
}
#[cfg(unix)]
#[test]
fn a_blank_line_is_not_a_row_it_failed_to_read() {
assert_eq!(
parse_table("11 10\n\n13 11\n \n"),
Some(vec![(11, 10), (13, 11)])
);
}
#[cfg(unix)]
#[test]
fn a_listing_that_claims_pid_zero_is_not_acted_on() {
assert_eq!(
parse_table("0 7\n7 1\n"),
None,
"a listing claiming pid 0 was read as a tree"
);
}
#[test]
fn a_foreign_session_is_labelled_by_a_stable_digest() {
let first = session_digest("claude-code:3f9a1c2e");
assert_eq!(first, session_digest("claude-code:3f9a1c2e"));
assert_ne!(first, session_digest("claude-code:other"));
assert_eq!(first.len(), 8);
}
#[test]
fn the_host_always_names_itself() {
assert!(!hostname().is_empty());
}
}