use std::time::{Duration, Instant};
#[cfg(windows)]
use processkit::Command;
use processkit::{Mechanism, ProcessGroup};
use crate::common::*;
#[tokio::test]
#[ignore = "creates an OS job/cgroup"]
async fn group_reports_the_platforms_mechanism() {
let group = ProcessGroup::new().expect("create group");
let mechanism = group.mechanism();
#[cfg(windows)]
assert_eq!(mechanism, Mechanism::JobObject);
#[cfg(target_os = "linux")]
assert!(
matches!(mechanism, Mechanism::CgroupV2 | Mechanism::ProcessGroup),
"linux is cgroup v2 or its pgroup fallback, got {mechanism:?}"
);
#[cfg(all(unix, not(target_os = "linux")))]
assert_eq!(mechanism, Mechanism::ProcessGroup);
}
#[tokio::test]
#[ignore = "creates an OS job/cgroup to cross-check the read-only host query"]
async fn host_containment_matches_a_real_group() {
let host = processkit::host_containment();
let group = ProcessGroup::new().expect("create group");
assert_eq!(
host.mechanism(),
group.mechanism(),
"the read-only host query must predict the mechanism a real group gets"
);
assert_eq!(
host.parent_death_cleanup(),
processkit::Command::kill_on_parent_death_scope(),
"the host report reuses Command::kill_on_parent_death_scope()"
);
assert_eq!(host.crate_version(), env!("CARGO_PKG_VERSION"));
#[cfg(feature = "process-control")]
{
use processkit::SoftStopScope;
#[cfg(unix)]
assert_eq!(
host.soft_stop_scope(),
group.soft_stop_scope(),
"on Unix the host soft-stop reach equals a live group's (WholeTree)"
);
#[cfg(unix)]
assert_eq!(host.soft_stop_scope(), SoftStopScope::WholeTree);
#[cfg(windows)]
assert_eq!(
host.soft_stop_scope(),
SoftStopScope::OptInMembers,
"on Windows the host reports the opt-in-members maximum a Job Object can reach"
);
}
}
#[tokio::test]
#[ignore = "spawns a long-lived subprocess and asserts kill-on-drop"]
async fn dropping_group_kills_children() {
let group = ProcessGroup::new().expect("create group");
let process = group.start(&sleeper()).await.expect("spawn sleeper");
let pid = process.pid();
assert!(
pid.is_some(),
"sleeper should report a pid right after spawn"
);
drop(group);
let start = Instant::now();
let _exit = tokio::time::timeout(Duration::from_secs(10), process.wait())
.await
.expect("child outlived its group — kill-on-close did not fire")
.expect("wait completed");
assert!(
start.elapsed() < Duration::from_secs(5),
"child was not reaped promptly (took {:?})",
start.elapsed()
);
}
#[cfg(windows)]
#[tokio::test]
#[ignore = "spawns a real process tree; proves a grandchild is contained (race fix)"]
async fn windows_grandchild_is_contained() {
let tmp = std::env::temp_dir();
let tag = std::process::id();
let pidfile = tmp.join(format!("processkit_gc_{tag}.pid"));
let grandchild_ps1 = tmp.join(format!("processkit_gc_{tag}.ps1"));
let parent_ps1 = tmp.join(format!("processkit_parent_{tag}.ps1"));
let _ = std::fs::remove_file(&pidfile);
std::fs::write(
&grandchild_ps1,
format!(
"$PID | Set-Content -Encoding ascii '{}'\nStart-Sleep -Seconds 30\n",
pidfile.display()
),
)
.expect("write grandchild script");
std::fs::write(
&parent_ps1,
format!(
"Start-Process -WindowStyle Hidden -FilePath powershell \
-ArgumentList '-NoProfile','-ExecutionPolicy','Bypass','-File','{}'\n",
grandchild_ps1.display()
),
)
.expect("write parent script");
let group = ProcessGroup::new().expect("create group");
group
.start(&Command::new("powershell").args([
"-NoProfile",
"-ExecutionPolicy",
"Bypass",
"-File",
&parent_ps1.to_string_lossy(),
]))
.await
.expect("spawn parent")
.wait()
.await
.expect("parent waits");
let mut grandchild_pid = None;
poll_until(
Duration::from_secs(5),
Duration::from_millis(100),
"grandchild never recorded its PID",
|| {
if let Ok(text) = std::fs::read_to_string(&pidfile)
&& let Ok(pid) = text.trim().parse::<u32>()
{
grandchild_pid = Some(pid);
true
} else {
false
}
},
)
.await;
let pid = grandchild_pid.expect("grandchild never recorded its PID");
assert!(
windows_pid_alive(pid),
"grandchild should be alive before drop"
);
drop(group);
let mut reaped = false;
for _ in 0..50 {
if !windows_pid_alive(pid) {
reaped = true;
break;
}
tokio::time::sleep(Duration::from_millis(100)).await;
}
let _ = std::fs::remove_file(&pidfile);
let _ = std::fs::remove_file(&grandchild_ps1);
let _ = std::fs::remove_file(&parent_ps1);
assert!(
reaped,
"grandchild {pid} outlived its job — containment leaked"
);
}
#[cfg(unix)]
#[tokio::test]
#[ignore = "spawns a setsid child that forks a grandchild; proves pgroup containment reaches it"]
async fn unix_setsid_child_forks_grandchild_still_contained() {
use processkit::Command;
let tmp = std::env::temp_dir();
let pidfile = tmp.join(format!("processkit_setsid_gc_{}.pid", std::process::id()));
let _ = std::fs::remove_file(&pidfile);
let group = ProcessGroup::new().expect("create group");
let child = group
.start(
&Command::new("sh")
.args(["-c", "sleep 30 & echo $! > \"$PK_PIDFILE\"; exit 0"])
.env("PK_PIDFILE", &pidfile)
.setsid(),
)
.await
.expect("setsid child spawns (EPERM would mean the pgroup coordination broke)");
completes_within(Duration::from_secs(10), "direct child exit", child.wait())
.await
.expect("direct child waits");
let mut gc_pid = None;
poll_until(
Duration::from_secs(5),
Duration::from_millis(50),
"grandchild never recorded its pid",
|| {
if let Ok(text) = std::fs::read_to_string(&pidfile)
&& let Ok(pid) = text.trim().parse::<i32>()
{
gc_pid = Some(pid);
true
} else {
false
}
},
)
.await;
let gc = gc_pid.expect("grandchild recorded its pid");
assert!(
unsafe { libc::kill(gc, 0) } == 0,
"grandchild {gc} should be alive before the group is dropped"
);
drop(group);
poll_until(
Duration::from_secs(5),
Duration::from_millis(50),
"grandchild outlived the group drop — inherited-session containment leaked",
|| unsafe { libc::kill(gc, 0) } != 0,
)
.await;
let _ = std::fs::remove_file(&pidfile);
}
#[tokio::test]
#[ignore = "spawns a real subprocess and kills it twice"]
async fn kill_all_is_idempotent() {
let group = ProcessGroup::new().expect("create group");
let child = group.start(&sleep_secs(30)).await.expect("start sleeper");
group.kill_all().expect("first kill");
group
.kill_all()
.expect("second kill must be a no-op success, not an error");
let mut again = None;
for attempt in 0..20u32 {
match group.start(&sleep_secs(1)).await {
Ok(run) => {
again = Some(run);
break;
}
Err(e) if e.is_permission_denied() && attempt < 19 => {
tokio::time::sleep(Duration::from_millis(100)).await;
}
Err(e) => panic!("group usable after terminate: {e:?}"),
}
}
let again = again.expect("group usable after terminate (after transient retries)");
drop(again);
let _ = tokio::time::timeout(Duration::from_secs(10), child.wait())
.await
.expect("child reaped");
}
#[cfg(unix)]
#[tokio::test]
#[ignore = "spawns a fast-exiting child and tears the group down while it is a zombie"]
async fn kill_all_on_a_zombie_only_group_reports_success() {
use std::process::Stdio;
use processkit::ProcessGroup;
use tokio::io::AsyncReadExt as _;
use tokio::process::Command as TokioCommand;
let group = ProcessGroup::new().expect("create group");
let mut cmd = TokioCommand::new("sh");
cmd.arg("-c").arg("exit 0");
cmd.stdout(Stdio::piped());
cmd.stderr(Stdio::null());
let mut child = group.spawn(cmd).expect("spawn fast-exiting child");
let pid = child.id().expect("child pid") as i32;
let mut out = child.stdout.take().expect("piped stdout handle");
let mut sink = Vec::new();
completes_within(
Duration::from_secs(10),
"child exit (stdout EOF)",
out.read_to_end(&mut sink),
)
.await
.expect("read child stdout to EOF");
assert!(
unsafe { libc::kill(pid, 0) } == 0,
"the exited child must still exist as an unreaped zombie"
);
group
.kill_all()
.expect("kill_all of a zombie-only group must succeed, not raise a false EPERM");
let _ = completes_within(Duration::from_secs(10), "zombie reap", child.wait()).await;
}
#[cfg(windows)]
mod nested_job {
use std::os::windows::io::AsRawHandle;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::{Duration, Instant};
use processkit::{Command, Mechanism, ProcessGroup};
use windows_sys::Win32::Foundation::{CloseHandle, HANDLE};
use windows_sys::Win32::System::JobObjects::{
AssignProcessToJobObject, CreateJobObjectW, IsProcessInJob,
JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE, JOBOBJECT_EXTENDED_LIMIT_INFORMATION,
JobObjectExtendedLimitInformation, SetInformationJobObject,
};
use windows_sys::Win32::System::Threading::GetCurrentProcess;
use crate::common::{poll_until, windows_pid_alive};
const HELPER_FLAG: &str = "PK_NESTED_JOB_HELPER";
const HELPER_DIR: &str = "PK_NESTED_JOB_DIR";
const HELPER_TAG: &str = "PK_NESTED_JOB_TAG";
const HELPER_TEST: &str = "groups::nested_job::job_helper_process";
struct Paths {
dir: PathBuf,
tag: String,
go: PathBuf,
ready: PathBuf,
error: PathBuf,
kill: PathBuf,
killed: PathBuf,
exit: PathBuf,
gc_pidfile: PathBuf,
gc_ps1: PathBuf,
parent_ps1: PathBuf,
}
impl Paths {
fn new(dir: &Path, tag: &str) -> Self {
let at = |name: &str| dir.join(format!("processkit_nested_{tag}_{name}"));
Self {
dir: dir.to_path_buf(),
tag: tag.to_string(),
go: at("go"),
ready: at("ready"),
error: at("error"),
kill: at("kill"),
killed: at("killed"),
exit: at("exit"),
gc_pidfile: at("gc.pid"),
gc_ps1: at("gc.ps1"),
parent_ps1: at("parent.ps1"),
}
}
fn cleanup(&self) {
for p in [
&self.go,
&self.ready,
&self.error,
&self.kill,
&self.killed,
&self.exit,
&self.gc_pidfile,
&self.gc_ps1,
&self.parent_ps1,
] {
let _ = std::fs::remove_file(p);
}
let _ = std::fs::remove_file(self.ready.with_extension("tmp"));
}
}
struct OuterJob {
handle: HANDLE,
}
unsafe impl Send for OuterJob {}
impl OuterJob {
fn create() -> OuterJob {
let handle = unsafe { CreateJobObjectW(std::ptr::null(), std::ptr::null()) };
assert!(
!handle.is_null(),
"CreateJobObjectW failed: {}",
std::io::Error::last_os_error()
);
let mut info: JOBOBJECT_EXTENDED_LIMIT_INFORMATION = unsafe { std::mem::zeroed() };
info.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE;
let ok = unsafe {
SetInformationJobObject(
handle,
JobObjectExtendedLimitInformation,
std::ptr::from_ref(&info).cast(),
std::mem::size_of::<JOBOBJECT_EXTENDED_LIMIT_INFORMATION>() as u32,
)
};
if ok == 0 {
let err = std::io::Error::last_os_error();
unsafe { CloseHandle(handle) };
panic!("SetInformationJobObject failed: {err}");
}
OuterJob { handle }
}
fn assign(&self, process: HANDLE) -> bool {
unsafe { AssignProcessToJobObject(self.handle, process) != 0 }
}
}
impl Drop for OuterJob {
fn drop(&mut self) {
unsafe { CloseHandle(self.handle) };
}
}
fn current_process_in_a_job() -> bool {
let mut in_job: i32 = 0;
let ok = unsafe { IsProcessInJob(GetCurrentProcess(), std::ptr::null_mut(), &mut in_job) };
ok != 0 && in_job != 0
}
struct Ready {
helper_pid: u32,
child_pid: u32,
grandchild_pid: u32,
}
impl Ready {
fn parse(text: &str) -> Option<Ready> {
let mut helper_pid = None;
let mut child_pid = None;
let mut grandchild_pid = None;
for line in text.lines() {
let Some((key, val)) = line.split_once('=') else {
continue;
};
let val = val.trim();
match key.trim() {
"helper_pid" => helper_pid = val.parse().ok(),
"child_pid" => child_pid = val.parse().ok(),
"grandchild_pid" => grandchild_pid = val.parse().ok(),
_ => {}
}
}
Some(Ready {
helper_pid: helper_pid?,
child_pid: child_pid?,
grandchild_pid: grandchild_pid?,
})
}
}
static TAG_COUNTER: AtomicU32 = AtomicU32::new(0);
fn unique_tag(kind: &str) -> String {
format!(
"{}_{kind}_{}",
std::process::id(),
TAG_COUNTER.fetch_add(1, Ordering::Relaxed)
)
}
fn spawn_helper(paths: &Paths) -> std::process::Child {
let exe = std::env::current_exe().expect("locate the integration-test binary");
std::process::Command::new(exe)
.args([HELPER_TEST, "--exact", "--ignored"])
.env(HELPER_FLAG, "1")
.env(HELPER_DIR, &paths.dir)
.env(HELPER_TAG, &paths.tag)
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.spawn()
.expect("re-exec this test binary in nested-job helper mode")
}
async fn setup_nested(kind: &str) -> (Paths, OuterJob, Ready) {
let paths = Paths::new(&std::env::temp_dir(), &unique_tag(kind));
paths.cleanup();
let outer = OuterJob::create();
let child = spawn_helper(&paths);
let raw = child.as_raw_handle() as HANDLE;
assert!(
outer.assign(raw),
"AssignProcessToJobObject(outer job, helper) failed: {} — nested Job \
Objects appear unsupported on this host (pre-Windows 8, or a \
breakaway-forbidding outer job that rejects re-nesting)",
std::io::Error::last_os_error()
);
std::fs::write(&paths.go, b"1").expect("write the go signal");
drop(child);
let ready = wait_ready(&paths, Duration::from_secs(40)).await;
(paths, outer, ready)
}
async fn wait_ready(paths: &Paths, timeout: Duration) -> Ready {
let deadline = Instant::now() + timeout;
loop {
if let Ok(err) = std::fs::read_to_string(&paths.error) {
let err = err.trim();
if !err.is_empty() {
panic!("nested-job helper reported a hard failure: {err}");
}
}
if let Ok(text) = std::fs::read_to_string(&paths.ready)
&& let Some(ready) = Ready::parse(&text)
{
return ready;
}
assert!(
Instant::now() < deadline,
"nested-job helper never became ready within {timeout:?} (no ready/error \
file — the re-exec filter may not have matched `{HELPER_TEST}`)"
);
tokio::time::sleep(Duration::from_millis(100)).await;
}
}
#[tokio::test]
#[ignore = "re-execs a helper inside an outer job; proves nested spawn/assign + kill_all"]
async fn windows_nested_job_assign_and_kill_all() {
let (paths, outer, ready) = setup_nested("kill").await;
assert!(
windows_pid_alive(ready.helper_pid),
"helper should be alive after publishing"
);
assert!(
windows_pid_alive(ready.child_pid),
"child should be alive before kill_all"
);
assert!(
windows_pid_alive(ready.grandchild_pid),
"grandchild should be alive before kill_all"
);
std::fs::write(&paths.kill, b"1").expect("write the kill signal");
poll_until(
Duration::from_secs(20),
Duration::from_millis(100),
"helper never confirmed kill_all",
|| paths.killed.exists(),
)
.await;
poll_until(
Duration::from_secs(15),
Duration::from_millis(100),
"child outlived kill_all — inner-job containment leaked",
|| !windows_pid_alive(ready.child_pid),
)
.await;
poll_until(
Duration::from_secs(15),
Duration::from_millis(100),
"grandchild outlived kill_all — nested containment did not reach it",
|| !windows_pid_alive(ready.grandchild_pid),
)
.await;
assert!(
windows_pid_alive(ready.helper_pid),
"kill_all on the inner group must not reap the helper itself (it is not a \
member of its own job)"
);
std::fs::write(&paths.exit, b"1").expect("write the exit signal");
poll_until(
Duration::from_secs(15),
Duration::from_millis(100),
"helper did not exit after the exit signal",
|| !windows_pid_alive(ready.helper_pid),
)
.await;
drop(outer); paths.cleanup();
}
#[tokio::test]
#[ignore = "re-execs a helper subtree inside an outer job; proves closing the job reaps it all"]
async fn windows_nested_job_outer_close_reaps_tree() {
let (paths, outer, ready) = setup_nested("close").await;
assert!(
windows_pid_alive(ready.helper_pid),
"helper should be alive before the close"
);
assert!(
windows_pid_alive(ready.child_pid),
"child should be alive before the close"
);
assert!(
windows_pid_alive(ready.grandchild_pid),
"grandchild should be alive before the close"
);
drop(outer);
poll_until(
Duration::from_secs(20),
Duration::from_millis(100),
"helper outlived the outer-job close — kill-on-close did not reach it through the nesting",
|| !windows_pid_alive(ready.helper_pid),
)
.await;
poll_until(
Duration::from_secs(20),
Duration::from_millis(100),
"child outlived the outer-job close",
|| !windows_pid_alive(ready.child_pid),
)
.await;
poll_until(
Duration::from_secs(20),
Duration::from_millis(100),
"grandchild outlived the outer-job close",
|| !windows_pid_alive(ready.grandchild_pid),
)
.await;
paths.cleanup();
}
#[tokio::test]
#[ignore = "re-exec target: runs its workload only when the harness sets PK_NESTED_JOB_HELPER"]
async fn job_helper_process() {
if std::env::var_os(HELPER_FLAG).is_none() {
return;
}
let dir = PathBuf::from(std::env::var(HELPER_DIR).expect("helper dir env"));
let tag = std::env::var(HELPER_TAG).expect("helper tag env");
let paths = Paths::new(&dir, &tag);
if let Err(msg) = run_helper(&paths).await {
let _ = std::fs::write(&paths.error, &msg);
panic!("nested-job helper failed: {msg}");
}
}
async fn run_helper(paths: &Paths) -> Result<(), String> {
wait_for_file(
&paths.go,
Duration::from_secs(30),
"outer-job assignment (go signal)",
)
.await?;
if !current_process_in_a_job() {
return Err(
"helper is not inside any Job Object after the harness assign — \
the nesting premise is unmet, so this would not test nesting"
.into(),
);
}
let group = ProcessGroup::new()
.map_err(|e| format!("ProcessGroup::new failed inside the outer job: {e}"))?;
let mechanism = group.mechanism();
if mechanism != Mechanism::JobObject {
return Err(format!(
"expected the JobObject mechanism inside the nesting, got {mechanism:?} — \
containment silently degraded"
));
}
std::fs::write(
&paths.gc_ps1,
format!(
"$PID | Set-Content -Encoding ascii '{}'\nStart-Sleep -Seconds 120\n",
paths.gc_pidfile.display()
),
)
.map_err(|e| format!("write grandchild script: {e}"))?;
std::fs::write(
&paths.parent_ps1,
format!(
"Start-Process -WindowStyle Hidden -FilePath powershell \
-ArgumentList '-NoProfile','-ExecutionPolicy','Bypass','-File','{}'\n\
Start-Sleep -Seconds 120\n",
paths.gc_ps1.display()
),
)
.map_err(|e| format!("write parent script: {e}"))?;
let run = group
.start(&Command::new("powershell").args([
"-NoProfile",
"-ExecutionPolicy",
"Bypass",
"-File",
&paths.parent_ps1.to_string_lossy(),
]))
.await
.map_err(|e| {
format!(
"group.start (spawn + AssignProcessToJobObject) failed inside the outer \
job — nested containment could not be established: {e}"
)
})?;
let child_pid = run
.pid()
.ok_or("the child reported no pid right after spawn")?;
let grandchild_pid =
wait_for_pid(&paths.gc_pidfile, Duration::from_secs(30), "grandchild pid").await?;
let helper_pid = std::process::id();
write_atomic(
&paths.ready,
format!(
"helper_pid={helper_pid}\nchild_pid={child_pid}\n\
grandchild_pid={grandchild_pid}\nmechanism={mechanism:?}\n"
),
)
.map_err(|e| format!("publish ready file: {e}"))?;
if wait_for_file_opt(&paths.kill, Duration::from_secs(90)).await {
group
.kill_all()
.map_err(|e| format!("group.kill_all failed: {e}"))?;
std::fs::write(&paths.killed, b"1").map_err(|e| format!("write killed marker: {e}"))?;
drop(run);
wait_for_file_opt(&paths.exit, Duration::from_secs(90)).await;
} else {
drop(run);
}
drop(group);
Ok(())
}
async fn wait_for_file(path: &Path, timeout: Duration, what: &str) -> Result<(), String> {
let deadline = Instant::now() + timeout;
loop {
if path.exists() {
return Ok(());
}
if Instant::now() >= deadline {
return Err(format!("{what} did not arrive within {timeout:?}"));
}
tokio::time::sleep(Duration::from_millis(100)).await;
}
}
async fn wait_for_file_opt(path: &Path, timeout: Duration) -> bool {
let deadline = Instant::now() + timeout;
loop {
if path.exists() {
return true;
}
if Instant::now() >= deadline {
return false;
}
tokio::time::sleep(Duration::from_millis(100)).await;
}
}
async fn wait_for_pid(path: &Path, timeout: Duration, what: &str) -> Result<u32, String> {
let deadline = Instant::now() + timeout;
loop {
if let Ok(text) = std::fs::read_to_string(path)
&& let Ok(pid) = text.trim().parse::<u32>()
{
return Ok(pid);
}
if Instant::now() >= deadline {
return Err(format!("{what} was not published within {timeout:?}"));
}
tokio::time::sleep(Duration::from_millis(100)).await;
}
}
fn write_atomic(path: &Path, content: String) -> std::io::Result<()> {
let tmp = path.with_extension("tmp");
std::fs::write(&tmp, content)?;
std::fs::rename(&tmp, path)
}
}