use std::io::Write;
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;
use crate::live_tail::LiveTail;
use crate::output_cap::{CappedOutput, MAX_CAPTURE_BYTES};
use anyhow::{Context, Result};
use futures::StreamExt;
use kanade_shared::subject;
use kanade_shared::wire::{Command, RunAs, Shell};
use rand::RngExt;
use tokio::io::AsyncReadExt;
use tokio::process::Command as ProcessCommand;
use tracing::{debug, info, warn};
use uuid::Uuid;
pub(crate) const POWERSHELL_UTF8_PRELUDE: &str = "[Console]::OutputEncoding = New-Object System.Text.UTF8Encoding $false; \
$OutputEncoding = [Console]::OutputEncoding; ";
fn staging_dir() -> Result<PathBuf> {
static SLOT: OnceLock<Mutex<Option<PathBuf>>> = OnceLock::new();
let slot = SLOT.get_or_init(|| Mutex::new(None));
let mut guard = slot.lock().expect("staging_dir mutex poisoned");
if let Some(p) = guard.as_ref() {
return Ok(p.clone());
}
let uuid = Uuid::new_v4().simple().to_string();
#[cfg(target_os = "macos")]
let dir = crate::process_as_user_macos::create_staging_dir(&uuid)?;
#[cfg(not(target_os = "macos"))]
let dir = {
let dir = if cfg!(target_os = "windows") {
let category = std::env::var_os("ProgramData")
.map(PathBuf::from)
.unwrap_or_else(|| PathBuf::from(r"C:\ProgramData"))
.join("Kanade")
.join("agent-scripts");
std::fs::create_dir_all(&category)
.with_context(|| format!("create_dir_all {}", category.display()))?;
category.join(&uuid)
} else {
std::env::temp_dir().join(format!("kanade-agent-{uuid}"))
};
std::fs::create_dir(&dir).with_context(|| format!("create_dir {}", dir.display()))?;
dir
};
*guard = Some(dir.clone());
Ok(dir)
}
pub(crate) struct TempPowerShellScript {
path: PathBuf,
}
impl TempPowerShellScript {
pub fn write(body: &str) -> Result<Self> {
let dir = staging_dir()?;
let path = dir.join(format!("kanade-{}.ps1", Uuid::new_v4().simple()));
let mut options = std::fs::OpenOptions::new();
options.write(true).create_new(true);
#[cfg(target_os = "macos")]
std::os::unix::fs::OpenOptionsExt::mode(&mut options, 0o644);
let mut f = options
.open(&path)
.with_context(|| format!("create_new {}", path.display()))?;
#[cfg(target_os = "macos")]
{
use std::os::unix::fs::PermissionsExt;
f.set_permissions(std::fs::Permissions::from_mode(0o644))
.with_context(|| format!("chmod 644 {}", path.display()))?;
}
f.write_all(&[0xEF, 0xBB, 0xBF])
.with_context(|| format!("write BOM {}", path.display()))?;
f.write_all(body.as_bytes())
.with_context(|| format!("write body {}", path.display()))?;
Ok(Self { path })
}
pub fn path(&self) -> &Path {
&self.path
}
}
impl Drop for TempPowerShellScript {
fn drop(&mut self) {
let _ = std::fs::remove_file(&self.path);
}
}
pub(crate) struct TempPowerShellLaunch {
launcher: TempPowerShellScript,
_user: TempPowerShellScript,
}
impl TempPowerShellLaunch {
pub fn stage(user_body: &str) -> Result<Self> {
let user = TempPowerShellScript::write(user_body)?;
let user_path = user.path().to_string_lossy().replace('\'', "''");
let launcher_body = format!(
"{POWERSHELL_UTF8_PRELUDE}$global:LASTEXITCODE = 0\n\
& '{user_path}' @args\n\
if (-not $?) {{ exit $LASTEXITCODE }}\n"
);
let launcher = TempPowerShellScript::write(&launcher_body)?;
Ok(Self {
launcher,
_user: user,
})
}
pub fn launcher_path(&self) -> &Path {
self.launcher.path()
}
}
pub enum ExecOutcome {
Completed {
exit_code: i32,
stdout: String,
stderr: String,
},
Killed {
stdout: String,
stderr: String,
},
Timeout {
stdout: String,
stderr: String,
},
}
pub async fn apply_jitter(cmd: &Command) {
if let Some(j) = cmd.jitter_secs.filter(|&s| s > 1) {
let secs = rand::rng().random_range(0..j);
info!(
jitter_secs = j,
sleep_secs = secs,
"applying jitter before exec"
);
tokio::time::sleep(Duration::from_secs(secs)).await;
}
}
pub async fn run_command_with_kill(
client: &async_nats::Client,
cmd: &Command,
live: Option<Arc<LiveTail>>,
) -> Result<ExecOutcome> {
#[cfg(not(target_os = "macos"))]
if !matches!(cmd.run_as, RunAs::System) {
return run_in_user_session_dispatch(client, cmd, live).await;
}
let _launch: Option<TempPowerShellLaunch>;
let launcher_path_owned: Option<String>;
let (program, args): (&str, Vec<&str>) = match cmd.shell {
Shell::Powershell => {
let launch = TempPowerShellLaunch::stage(&cmd.script)?;
launcher_path_owned = Some(launch.launcher_path().to_string_lossy().into_owned());
_launch = Some(launch);
(
"powershell",
vec![
"-NoProfile",
"-NonInteractive",
"-ExecutionPolicy",
"Bypass",
"-File",
launcher_path_owned.as_deref().unwrap(),
],
)
}
Shell::Cmd => {
_launch = None;
("cmd", vec!["/C", &cmd.script])
}
Shell::Sh => {
_launch = None;
("sh", vec!["-c", &cmd.script])
}
Shell::Pwsh => {
let launch = TempPowerShellLaunch::stage(&cmd.script)?;
launcher_path_owned = Some(launch.launcher_path().to_string_lossy().into_owned());
_launch = Some(launch);
let mut args = vec!["-NoProfile", "-NonInteractive"];
#[cfg(target_os = "windows")]
args.extend_from_slice(&["-ExecutionPolicy", "Bypass"]);
args.extend_from_slice(&["-File", launcher_path_owned.as_deref().unwrap()]);
("pwsh", args)
}
};
let mut builder = host_command(cmd, program, &args)?;
builder
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.kill_on_drop(true);
#[cfg(unix)]
spawn_in_own_session(&mut builder);
let mut child = builder
.spawn()
.with_context(|| format!("spawn {program}"))?;
let tree = KillTree::attach(&child);
let stdout_handle = child.stdout.take();
let stderr_handle = child.stderr.take();
let (capture_stop, stop) = tokio::sync::watch::channel(false);
let live_out = live.clone();
let out_stop = stop.clone();
let stdout_task = tokio::spawn(async move {
drain_to_string(stdout_handle, live_out, Stream::Stdout, out_stop).await
});
let live_err = live.clone();
let stderr_task = tokio::spawn(async move {
drain_to_string(stderr_handle, live_err, Stream::Stderr, stop).await
});
let timeout_dur = Duration::from_secs(cmd.timeout_secs.max(1));
let inner = match &cmd.exec_id {
Some(eid) => {
let kill_subject = subject::kill(eid);
let mut kill_sub = client
.subscribe(kill_subject.clone())
.await
.with_context(|| format!("subscribe {kill_subject}"))?;
client.flush().await.ok();
debug!(exec_id = %eid, subject = %kill_subject, "kill listener armed");
tokio::select! {
status = child.wait() => {
debug!(exec_id = %eid, "child exited (wait arm fired)");
let s = status?;
OutcomeInner::Completed(s.code().unwrap_or(-1))
}
msg = kill_sub.next() => {
info!(exec_id = %eid, has_msg = msg.is_some(), "kill arm fired");
tree.terminate(&mut child).await;
OutcomeInner::Killed
}
_ = tokio::time::sleep(timeout_dur) => {
info!(exec_id = %eid, "timeout arm fired");
tree.terminate(&mut child).await;
OutcomeInner::Timeout
}
}
}
None => {
tokio::select! {
status = child.wait() => {
let s = status?;
OutcomeInner::Completed(s.code().unwrap_or(-1))
}
_ = tokio::time::sleep(timeout_dur) => {
tree.terminate(&mut child).await;
OutcomeInner::Timeout
}
}
}
};
let (stdout, stderr) = finish_capture(stdout_task, stderr_task, capture_stop).await?;
Ok(match inner {
OutcomeInner::Completed(code) => ExecOutcome::Completed {
exit_code: code,
stdout,
stderr,
},
OutcomeInner::Killed => ExecOutcome::Killed { stdout, stderr },
OutcomeInner::Timeout => ExecOutcome::Timeout { stdout, stderr },
})
}
fn host_command(cmd: &Command, program: &str, args: &[&str]) -> Result<ProcessCommand> {
#[cfg(target_os = "macos")]
if !matches!(cmd.run_as, RunAs::System) {
return crate::process_as_user_macos::session_command(
cmd.run_as,
program,
args,
cmd.cwd.as_deref(),
);
}
let mut builder = ProcessCommand::new(program);
builder.args(args);
#[cfg(target_os = "macos")]
builder.env("PATH", crate::process_as_user_macos::job_path());
if let Some(dir) = cmd.cwd.as_deref().filter(|s| !s.is_empty()) {
#[cfg(target_os = "windows")]
{
match crate::cwd_expand::open_self_token()
.and_then(|tok| crate::cwd_expand::expand(dir, tok.handle()))
{
Ok(expanded) => {
builder.current_dir(expanded);
}
Err(e) => {
warn!(error = %e, raw_cwd = %dir, "cwd expansion failed; using raw value");
builder.current_dir(dir);
}
}
}
#[cfg(target_os = "macos")]
{
builder.current_dir(crate::process_as_user_macos::expand_agent_cwd(dir));
}
#[cfg(not(any(target_os = "windows", target_os = "macos")))]
{
builder.current_dir(dir);
}
}
Ok(builder)
}
#[cfg(unix)]
fn spawn_in_own_session(builder: &mut ProcessCommand) {
unsafe {
builder.pre_exec(|| {
if libc::setsid() == -1 {
Err(std::io::Error::last_os_error())
} else {
Ok(())
}
});
}
}
#[cfg(unix)]
const KILL_TERM_GRACE: Duration = Duration::from_secs(5);
struct KillTree {
#[cfg(target_os = "windows")]
job: Option<crate::job_object::JobObject>,
#[cfg(unix)]
group: Option<libc::pid_t>,
}
impl KillTree {
fn attach(child: &tokio::process::Child) -> Self {
#[cfg(target_os = "windows")]
{
let job = child.raw_handle().and_then(|h| {
crate::job_object::JobObject::assign_handle(windows::Win32::Foundation::HANDLE(h))
.map_err(|e| {
warn!(error = %e, "job object assign failed; kill falls back to single-process terminate");
})
.ok()
});
Self { job }
}
#[cfg(unix)]
{
Self {
group: child.id().and_then(|pid| libc::pid_t::try_from(pid).ok()),
}
}
}
async fn terminate(&self, child: &mut tokio::process::Child) {
#[cfg(target_os = "windows")]
if let Some(job) = &self.job {
job.terminate();
return;
}
#[cfg(unix)]
if let Some(pgid) = self.group {
signal_group(pgid, libc::SIGTERM);
if tokio::time::timeout(KILL_TERM_GRACE, child.wait())
.await
.is_err()
{
info!(
pgid,
"host still running after SIGTERM grace; sending SIGKILL"
);
}
signal_group(pgid, libc::SIGKILL);
if let Err(e) = child.wait().await {
warn!(error = %e, "wait after killing the process group failed");
}
return;
}
if let Err(e) = child.kill().await {
warn!(error = %e, "child.kill failed (process may already be dead)");
}
}
}
#[cfg(unix)]
fn signal_group(pgid: libc::pid_t, signal: libc::c_int) {
if unsafe { libc::killpg(pgid, signal) } == -1 {
let err = std::io::Error::last_os_error();
if err.raw_os_error() != Some(libc::ESRCH) {
warn!(error = %err, pgid, signal, "killpg failed");
}
}
}
pub(crate) const OUTPUT_DRAIN_GRACE: Duration = Duration::from_secs(2);
type Capture = (String, Option<anyhow::Error>);
pub(crate) async fn finish_capture(
stdout: tokio::task::JoinHandle<Capture>,
stderr: tokio::task::JoinHandle<Capture>,
stop: tokio::sync::watch::Sender<bool>,
) -> Result<(String, String)> {
let joined = async { tokio::join!(stdout, stderr) };
tokio::pin!(joined);
let (out, err) = match tokio::time::timeout(OUTPUT_DRAIN_GRACE, &mut joined).await {
Ok(pair) => pair,
Err(_) => {
let _ = stop.send(true);
joined.await
}
};
let (stdout, stdout_err) = out.context("stdout capture join")?;
let (mut stderr, stderr_err) = err.context("stderr capture join")?;
for (stream, error) in [("stdout", stdout_err), ("stderr", stderr_err)] {
if let Some(error) = error {
warn!(%stream, %error, "capture ended early (kept partial output)");
stderr.push_str(&format!(
"\n[agent: {stream} capture ended early: {error}]\n"
));
}
}
Ok((stdout, stderr))
}
enum OutcomeInner {
Completed(i32),
Killed,
Timeout,
}
#[derive(Clone, Copy, Debug)]
enum Stream {
Stdout,
Stderr,
}
async fn drain_to_string<R>(
reader: Option<R>,
live: Option<Arc<LiveTail>>,
stream: Stream,
mut stop: tokio::sync::watch::Receiver<bool>,
) -> (String, Option<anyhow::Error>)
where
R: tokio::io::AsyncRead + Unpin,
{
let mut buf = CappedOutput::new(MAX_CAPTURE_BYTES);
let mut err: Option<anyhow::Error> = None;
if let Some(mut s) = reader {
let mut chunk = [0u8; 8 * 1024];
loop {
let read = tokio::select! {
biased;
_ = stop.changed() => {
err = Some(anyhow::anyhow!("output drain deadline reached; a descendant may still hold the pipe"));
break;
}
read = s.read(&mut chunk) => read,
};
match read {
Ok(0) => break,
Ok(n) => {
let slice = &chunk[..n];
buf.push(slice);
if let Some(lt) = &live {
match stream {
Stream::Stdout => lt.push_stdout(slice),
Stream::Stderr => lt.push_stderr(slice),
}
}
}
Err(e) => {
err = Some(anyhow::Error::new(e));
break;
}
}
}
}
if buf.truncated() {
warn!(
stream = ?stream,
bytes_written = buf.total(),
kept = MAX_CAPTURE_BYTES,
"output exceeded the capture cap and was truncated"
);
}
(buf.finish(), err)
}
#[cfg(not(target_os = "macos"))]
async fn run_in_user_session_dispatch(
client: &async_nats::Client,
cmd: &Command,
live: Option<Arc<LiveTail>>,
) -> Result<ExecOutcome> {
#[cfg(not(target_os = "windows"))]
{
let _ = client;
let _ = live;
warn!(
run_as = ?cmd.run_as,
"run_as: user / system_gui is not supported on Linux agents — skipping the script",
);
Ok(ExecOutcome::Completed {
exit_code: 0,
stdout: String::new(),
stderr: format!(
"run_as: {:?} is not supported on Linux agents; the script was skipped.\n",
cmd.run_as
),
})
}
#[cfg(target_os = "windows")]
{
let (kill_tx, kill_rx) = tokio::sync::oneshot::channel::<()>();
let bridge = if let Some(eid) = cmd.exec_id.clone() {
let nats = client.clone();
let subject = subject::kill(&eid);
Some(tokio::spawn(async move {
match nats.subscribe(subject.clone()).await {
Ok(mut sub) => {
nats.flush().await.ok();
debug!(exec_id = %eid, subject = %subject, "kill listener armed (user-session path)");
if sub.next().await.is_some() {
info!(exec_id = %eid, "kill received → forwarding to user-session waiter");
let _ = kill_tx.send(());
}
}
Err(e) => {
warn!(error = %e, %subject, "subscribe kill failed (user-session path)")
}
}
}))
} else {
None
};
let timeout = Duration::from_secs(cmd.timeout_secs.max(1));
let outcome = crate::process_as_user::run_command_in_user_session(
cmd, cmd.run_as, timeout, kill_rx, live,
)
.await;
if let Some(b) = bridge {
b.abort();
}
outcome
}
}
#[cfg(test)]
mod tests {
use tokio::io::AsyncReadExt;
#[tokio::test]
async fn capture_finishes_with_partial_output_while_writer_stays_open() {
use tokio::io::AsyncWriteExt;
let (mut writer, reader) = tokio::io::duplex(64);
writer.write_all(b"launcher done").await.unwrap();
let (stop, rx) = tokio::sync::watch::channel(false);
let out = tokio::spawn(super::drain_to_string(
Some(reader),
None,
super::Stream::Stdout,
rx,
));
let err = tokio::spawn(async { (String::new(), None) });
let (stdout, stderr) = tokio::time::timeout(
std::time::Duration::from_secs(5),
super::finish_capture(out, err, stop),
)
.await
.expect("must not wait for descendant EOF")
.unwrap();
assert_eq!(stdout, "launcher done");
assert!(stderr.contains("stdout capture ended early"));
drop(writer);
}
#[tokio::test]
async fn capture_preserves_clean_eof_without_warning() {
let (stop, rx) = tokio::sync::watch::channel(false);
let out = tokio::spawn(super::drain_to_string(
Some(&b"done"[..]),
None,
super::Stream::Stdout,
rx,
));
let err = tokio::spawn(async { (String::new(), None) });
let (stdout, stderr) = super::finish_capture(out, err, stop).await.unwrap();
assert_eq!(stdout, "done");
assert_eq!(stderr, "");
}
async fn capture_lossy<R: tokio::io::AsyncRead + Unpin>(mut r: R) -> String {
let mut buf = Vec::new();
r.read_to_end(&mut buf).await.unwrap();
String::from_utf8_lossy(&buf).into_owned()
}
#[tokio::test]
async fn cp932_japanese_bytes_are_kept_lossy_not_dropped() {
let raw: Vec<u8> = vec![
b'{', b'"', b'k', b'"', b':', b'"', 0x82, 0xbf, 0x82, 0xc2, b'"', b'}',
];
let captured = capture_lossy(tokio::io::BufReader::new(&raw[..])).await;
assert!(captured.starts_with("{\"k\":\""), "ASCII frame preserved");
assert!(captured.ends_with("\"}"), "ASCII frame preserved");
assert!(captured.contains('\u{FFFD}'), "invalid runs marked");
}
#[tokio::test]
async fn pure_utf8_payload_round_trips() {
let raw = "こんにちは {\"ok\": true}".as_bytes().to_vec();
let captured = capture_lossy(tokio::io::BufReader::new(&raw[..])).await;
assert_eq!(captured, "こんにちは {\"ok\": true}");
}
#[tokio::test]
async fn empty_stream_yields_empty_string() {
let raw: Vec<u8> = Vec::new();
let captured = capture_lossy(tokio::io::BufReader::new(&raw[..])).await;
assert_eq!(captured, "");
}
#[test]
fn powershell_prelude_constant_shape() {
assert!(super::POWERSHELL_UTF8_PRELUDE.ends_with("; "));
assert!(super::POWERSHELL_UTF8_PRELUDE.contains("[Console]::OutputEncoding"));
assert!(super::POWERSHELL_UTF8_PRELUDE.contains("$OutputEncoding"));
}
#[test]
fn temp_powershell_script_writes_bom_then_body_verbatim() {
let script = "[CmdletBinding()] param([string]$X='a'); Write-Output $X";
let staged = super::TempPowerShellScript::write(script).expect("write");
let bytes = std::fs::read(staged.path()).expect("read back");
assert_eq!(
&bytes[..3],
&[0xEF, 0xBB, 0xBF],
"BOM not at start of staged file",
);
let body_bytes = &bytes[3..];
assert_eq!(
std::str::from_utf8(body_bytes).unwrap(),
script,
"user body must be verbatim — no prelude prefix",
);
assert_eq!(
staged.path().extension().and_then(|s| s.to_str()),
Some("ps1"),
);
}
#[test]
fn temp_powershell_script_drop_removes_file() {
let staged = super::TempPowerShellScript::write("Write-Output 'x'").expect("write");
let path = staged.path().to_path_buf();
assert!(path.exists(), "file should exist before drop");
drop(staged);
assert!(!path.exists(), "file should be gone after drop");
}
#[test]
fn temp_powershell_launch_user_file_has_no_prelude() {
let user_script =
"[CmdletBinding()] param([string]$Foo = 'bar'); Write-Output \"got:$Foo\"";
let launch = super::TempPowerShellLaunch::stage(user_script).expect("stage");
let launcher_text = std::fs::read_to_string(launch.launcher_path()).expect("read launcher");
let start = launcher_text
.find("& '")
.expect("launcher should invoke user script via call operator");
let after = &launcher_text[start + 3..];
let end = after.find('\'').expect("launcher path closes its quote");
let user_path_in_launcher: String = after[..end].replace("''", "'");
let user_bytes = std::fs::read(&user_path_in_launcher).expect("read user file");
assert_eq!(&user_bytes[..3], &[0xEF, 0xBB, 0xBF]);
let body = std::str::from_utf8(&user_bytes[3..]).unwrap();
assert_eq!(body, user_script);
assert!(
!body.contains("[Console]::OutputEncoding"),
"user file must not carry the encoding prelude",
);
}
#[test]
fn temp_powershell_launch_launcher_carries_prelude_then_invokes_user() {
let launch = super::TempPowerShellLaunch::stage("Write-Output 'hi'").expect("stage");
let launcher_text = std::fs::read_to_string(launch.launcher_path()).expect("read launcher");
assert!(
launcher_text.contains("[Console]::OutputEncoding"),
"launcher must set console encoding before invoking user",
);
assert!(
launcher_text.contains("& '") && launcher_text.contains("' @args"),
"launcher must invoke user via call operator with @args splat",
);
let prelude_pos = launcher_text.find("[Console]::OutputEncoding").unwrap();
let call_pos = launcher_text.find("& '").unwrap();
assert!(prelude_pos < call_pos);
}
#[test]
fn temp_powershell_launch_drop_removes_both_files() {
let launch = super::TempPowerShellLaunch::stage("Write-Output 'x'").expect("stage");
let launcher_path = launch.launcher_path().to_path_buf();
let launcher_text = std::fs::read_to_string(&launcher_path).expect("read launcher");
let start = launcher_text.find("& '").unwrap() + 3;
let end = launcher_text[start..].find('\'').unwrap();
let user_path =
std::path::PathBuf::from(launcher_text[start..start + end].replace("''", "'"));
assert!(launcher_path.exists());
assert!(user_path.exists());
drop(launch);
assert!(!launcher_path.exists(), "launcher must be removed on drop");
assert!(!user_path.exists(), "user file must be removed on drop");
}
#[cfg(unix)]
#[test]
fn staged_launcher_reports_the_user_scripts_exit_code() {
let Ok(pwsh) = which::which("pwsh") else {
println!("skipping: `pwsh` not on PATH");
return;
};
for (script, want) in [("exit 3", 3), ("/usr/bin/false; Write-Output ok", 0)] {
let launch = super::TempPowerShellLaunch::stage(script).expect("stage");
let status = std::process::Command::new(&pwsh)
.args(["-NoProfile", "-NonInteractive", "-File"])
.arg(launch.launcher_path())
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.status()
.expect("run pwsh");
assert_eq!(status.code(), Some(want), "script: {script}");
}
}
#[cfg(windows)]
#[test]
fn staged_launcher_reports_the_user_scripts_exit_code_on_windows_powershell() {
for (script, want) in [("exit 3", 3), ("cmd /c exit 1; Write-Output ok", 0)] {
let launch = super::TempPowerShellLaunch::stage(script).expect("stage");
let status = std::process::Command::new("powershell")
.args([
"-NoProfile",
"-NonInteractive",
"-ExecutionPolicy",
"Bypass",
"-File",
])
.arg(launch.launcher_path())
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.status()
.expect("run powershell");
assert_eq!(status.code(), Some(want), "script: {script}");
}
}
}
#[cfg(all(test, unix))]
mod process_tree_tests {
use std::path::Path;
use std::time::{Duration, Instant};
use super::{ExecOutcome, run_command_with_kill};
async fn client() -> async_nats::Client {
async_nats::ConnectOptions::new()
.retry_on_initial_connect()
.connect("127.0.0.1:1")
.await
.expect("lazy client")
}
fn sh_job(script: &str, timeout_secs: u64) -> kanade_shared::wire::Command {
serde_json::from_value(serde_json::json!({
"id": "tree-test",
"version": "1",
"request_id": "req",
"exec_id": null,
"shell": "sh",
"script": script,
"timeout_secs": timeout_secs,
"jitter_secs": null,
"run_as": "system",
}))
.expect("command")
}
fn read_pid(file: &Path) -> libc::pid_t {
std::fs::read_to_string(file)
.expect("pid file")
.trim()
.parse()
.expect("pid")
}
fn alive(pid: libc::pid_t) -> bool {
unsafe { libc::kill(pid, 0) == 0 }
}
struct Reap(libc::pid_t);
impl Drop for Reap {
fn drop(&mut self) {
unsafe { libc::kill(self.0, libc::SIGKILL) };
}
}
#[tokio::test]
async fn timeout_kills_the_whole_process_tree() {
let dir = tempfile::tempdir().unwrap();
let pid_file = dir.path().join("grandchild.pid");
let script = format!("sleep 300 & echo $! > '{}'; sleep 300", pid_file.display());
let outcome = run_command_with_kill(&client().await, &sh_job(&script, 1), None)
.await
.expect("run");
assert!(matches!(outcome, ExecOutcome::Timeout { .. }));
let grandchild = read_pid(&pid_file);
let _reap = Reap(grandchild);
let deadline = Instant::now() + Duration::from_secs(5);
while alive(grandchild) && Instant::now() < deadline {
tokio::time::sleep(Duration::from_millis(50)).await;
}
assert!(
!alive(grandchild),
"grandchild {grandchild} survived the timeout"
);
}
#[tokio::test]
async fn clean_exit_leaves_a_detached_daemon_running() {
let dir = tempfile::tempdir().unwrap();
let pid_file = dir.path().join("daemon.pid");
let script = format!(
"sleep 300 >/dev/null 2>&1 </dev/null & echo $! > '{}'; exit 0",
pid_file.display()
);
let outcome = run_command_with_kill(&client().await, &sh_job(&script, 60), None)
.await
.expect("run");
let daemon = read_pid(&pid_file);
let _reap = Reap(daemon);
assert!(matches!(
outcome,
ExecOutcome::Completed { exit_code: 0, .. }
));
assert!(alive(daemon), "daemon {daemon} was killed on a clean exit");
}
}