use std::process::Output;
use std::time::{Duration, Instant};
use tokio::process::Command;
pub const TIMEOUT_QUICK: Duration = Duration::from_secs(15);
pub const TIMEOUT_MEDIUM: Duration = Duration::from_secs(30);
pub const TIMEOUT_SLOW: Duration = Duration::from_secs(60);
pub async fn run_cmd(mut cmd: Command, timeout: Duration) -> Result<Output, String> {
let label = format!("{:?}", cmd.as_std().get_program());
match tokio::time::timeout(timeout, cmd.output()).await {
Ok(Ok(output)) => Ok(output),
Ok(Err(e)) => Err(format!("{} failed: {}", label, e)),
Err(_) => Err(format!("{} timed out after {}s", label, timeout.as_secs())),
}
}
pub(crate) async fn run_owned_mutation(
mut cmd: Command,
timeout: Duration,
) -> Result<Output, String> {
use tokio::sync::oneshot;
static MUTATION_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
let label = format!("{:?}", cmd.as_std().get_program());
cmd.kill_on_drop(true);
let mutation_guard = MUTATION_LOCK.lock().await;
let (cancel_tx, cancel_rx) = oneshot::channel();
let supervisor = tokio::spawn(async move {
let _mutation_guard = mutation_guard;
run_owned_mutation_child(cmd, timeout, cancel_rx).await
});
let mut cancel_guard = MutationCancelGuard {
sender: Some(cancel_tx),
};
let result = supervisor
.await
.map_err(|error| format!("{} mutation supervisor failed: {}", label, error))?;
cancel_guard.sender.take();
result.map_err(|error| format!("{} {}", label, error))
}
#[cfg(any(windows, target_os = "linux", test))]
pub(crate) struct MutationPairOutput {
pub first: Result<Output, String>,
pub inverse: Result<Output, String>,
}
#[cfg(any(windows, target_os = "linux", test))]
pub(crate) async fn run_owned_mutation_pair(
first: Command,
first_timeout: Duration,
dwell: Duration,
inverse: Command,
inverse_timeout: Duration,
) -> Result<MutationPairOutput, String> {
use tokio::sync::oneshot;
static PAIR_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
let (cancel_tx, cancel_rx) = oneshot::channel();
let supervisor = tokio::spawn(async move {
let _pair_guard = PAIR_LOCK.lock().await;
run_owned_mutation_pair_child(
first,
first_timeout,
dwell,
inverse,
inverse_timeout,
cancel_rx,
)
.await
});
let mut cancel_guard = PairCancelGuard {
sender: Some(cancel_tx),
};
let result = supervisor
.await
.map_err(|error| format!("mutation-pair supervisor failed: {error}"))?;
cancel_guard.sender.take();
Ok(result)
}
#[cfg(any(windows, target_os = "linux", test))]
struct PairCancelGuard {
sender: Option<tokio::sync::oneshot::Sender<()>>,
}
#[cfg(any(windows, target_os = "linux", test))]
impl Drop for PairCancelGuard {
fn drop(&mut self) {
if let Some(sender) = self.sender.take() {
let _ = sender.send(());
}
}
}
#[cfg(any(windows, target_os = "linux", test))]
async fn run_owned_mutation_pair_child(
first: Command,
first_timeout: Duration,
dwell: Duration,
inverse: Command,
inverse_timeout: Duration,
mut cancelled: tokio::sync::oneshot::Receiver<()>,
) -> MutationPairOutput {
let first_result = {
let first_run = run_owned_mutation(first, first_timeout);
tokio::pin!(first_run);
tokio::select! {
biased;
_ = &mut cancelled => None,
result = &mut first_run => Some(result),
}
};
let first_succeeded = first_result
.as_ref()
.is_some_and(|result| result.as_ref().is_ok_and(|output| output.status.success()));
if first_succeeded && !dwell.is_zero() {
tokio::select! {
biased;
_ = &mut cancelled => {}
_ = tokio::time::sleep(dwell) => {}
}
}
let inverse = run_owned_mutation(inverse, inverse_timeout).await;
MutationPairOutput {
first: first_result.unwrap_or_else(|| Err("was cancelled before completing".to_string())),
inverse,
}
}
#[cfg(target_os = "macos")]
pub(crate) async fn run_macos_mutation(cmd: Command, timeout: Duration) -> Result<Output, String> {
run_owned_mutation(cmd, timeout).await
}
struct MutationCancelGuard {
sender: Option<tokio::sync::oneshot::Sender<()>>,
}
impl Drop for MutationCancelGuard {
fn drop(&mut self) {
if let Some(sender) = self.sender.take() {
let _ = sender.send(());
}
}
}
async fn run_owned_mutation_child(
mut cmd: Command,
timeout: Duration,
mut cancelled: tokio::sync::oneshot::Receiver<()>,
) -> Result<Output, String> {
use std::process::Stdio;
use tokio::io::AsyncReadExt;
cmd.stdout(Stdio::piped()).stderr(Stdio::piped());
let mut child = cmd
.spawn()
.map_err(|error| format!("spawn failed: {error}"))?;
let mut stdout = child
.stdout
.take()
.ok_or_else(|| "could not capture stdout".to_string())?;
let mut stderr = child
.stderr
.take()
.ok_or_else(|| "could not capture stderr".to_string())?;
enum Completion {
Exited(Result<(std::process::ExitStatus, Vec<u8>, Vec<u8>), String>),
TimedOut,
Cancelled,
}
let completion = {
let completed = async {
let mut stdout_bytes = Vec::new();
let mut stderr_bytes = Vec::new();
let (status, stdout_result, stderr_result) = tokio::join!(
child.wait(),
stdout.read_to_end(&mut stdout_bytes),
stderr.read_to_end(&mut stderr_bytes),
);
let status = status.map_err(|error| format!("wait failed: {error}"))?;
stdout_result.map_err(|error| format!("stdout read failed: {error}"))?;
stderr_result.map_err(|error| format!("stderr read failed: {error}"))?;
Ok((status, stdout_bytes, stderr_bytes))
};
tokio::pin!(completed);
tokio::select! {
biased;
_ = &mut cancelled => Completion::Cancelled,
result = &mut completed => Completion::Exited(result),
_ = tokio::time::sleep(timeout) => Completion::TimedOut,
}
};
match completion {
Completion::Exited(Ok((status, stdout, stderr))) => Ok(Output {
status,
stdout,
stderr,
}),
Completion::Exited(Err(error)) => {
terminate_and_reap_mutation(&mut child).await;
Err(error)
}
Completion::TimedOut => {
terminate_and_reap_mutation(&mut child).await;
Err(format!("timed out after {}s", timeout.as_secs_f64()))
}
Completion::Cancelled => {
terminate_and_reap_mutation(&mut child).await;
Err("was cancelled and safely terminated".to_string())
}
}
}
async fn terminate_and_reap_mutation(child: &mut tokio::process::Child) {
let _ = child.start_kill();
let _ = tokio::time::timeout(Duration::from_secs(5), child.wait()).await;
}
#[derive(Debug, Clone)]
pub struct CmdOutcome {
pub command: String,
pub args: Vec<String>,
pub exit_code: Option<i32>,
pub stdout: String,
pub stderr: String,
pub duration: Duration,
pub ok: bool,
pub error: Option<String>,
}
impl CmdOutcome {
pub fn summary(&self) -> String {
if self.ok {
format!("ok ({:.1}s)", self.duration.as_secs_f64())
} else if let Some(err) = &self.error {
format!("failed: {}", err)
} else if let Some(code) = self.exit_code {
format!("exit {} ({:.1}s)", code, self.duration.as_secs_f64())
} else {
"failed".to_string()
}
}
pub fn cmdline(&self) -> String {
let mut s = self.command.clone();
for a in &self.args {
s.push(' ');
if a.contains(' ') {
s.push('"');
s.push_str(a);
s.push('"');
} else {
s.push_str(a);
}
}
s
}
}
pub async fn run_cmd_capture(mut cmd: Command, timeout: Duration) -> CmdOutcome {
let std_cmd = cmd.as_std();
let command = std_cmd.get_program().to_string_lossy().into_owned();
let args: Vec<String> = std_cmd
.get_args()
.map(|s| s.to_string_lossy().into_owned())
.collect();
let started = Instant::now();
let raced = tokio::time::timeout(timeout, cmd.output()).await;
let duration = started.elapsed();
match raced {
Ok(Ok(output)) => {
let exit_code = output.status.code();
let stdout = String::from_utf8_lossy(&output.stdout).into_owned();
let stderr = String::from_utf8_lossy(&output.stderr).into_owned();
let ok = output.status.success();
CmdOutcome {
command,
args,
exit_code,
stdout,
stderr,
duration,
ok,
error: None,
}
}
Ok(Err(e)) => CmdOutcome {
command,
args,
exit_code: None,
stdout: String::new(),
stderr: String::new(),
duration,
ok: false,
error: Some(format!("spawn failed: {}", e)),
},
Err(_) => CmdOutcome {
command,
args,
exit_code: None,
stdout: String::new(),
stderr: String::new(),
duration,
ok: false,
error: Some(format!("timed out after {}s", timeout.as_secs())),
},
}
}
#[cfg(test)]
mod portable_owned_mutation_tests {
use super::*;
use std::path::{Path, PathBuf};
const HELPER_TEST: &str =
"actions::fix::cmd::portable_owned_mutation_tests::mutation_child_helper";
fn test_path(label: &str, suffix: &str) -> PathBuf {
std::env::temp_dir().join(format!(
"nd300-portable-mutation-{label}-{}-{suffix}",
std::process::id()
))
}
fn helper_command(
state_file: &Path,
started_file: &Path,
delay_ms: u64,
value: &str,
) -> Command {
let mut command = Command::new(std::env::current_exe().expect("current test executable"));
command.args(["--exact", HELPER_TEST, "--nocapture"]);
command.env("ND300_MUTATION_HELPER", "1");
command.env("ND300_MUTATION_HELPER_STATE", state_file);
command.env("ND300_MUTATION_HELPER_STARTED", started_file);
command.env("ND300_MUTATION_HELPER_DELAY_MS", delay_ms.to_string());
command.env("ND300_MUTATION_HELPER_VALUE", value);
command
}
async fn wait_for_file(path: &Path) {
let deadline = tokio::time::Instant::now() + Duration::from_secs(3);
while !path.exists() && tokio::time::Instant::now() < deadline {
tokio::time::sleep(Duration::from_millis(10)).await;
}
assert!(path.exists(), "helper did not create {}", path.display());
}
async fn wait_for_contents(path: &Path, expected: &[u8]) {
let deadline = tokio::time::Instant::now() + Duration::from_secs(3);
loop {
if std::fs::read(path).is_ok_and(|contents| contents == expected) {
return;
}
assert!(
tokio::time::Instant::now() < deadline,
"{} never reached expected state {:?}",
path.display(),
String::from_utf8_lossy(expected)
);
tokio::time::sleep(Duration::from_millis(10)).await;
}
}
fn cleanup(paths: &[&Path]) {
for path in paths {
let _ = std::fs::remove_file(path);
}
}
#[test]
fn mutation_child_helper() {
if std::env::var_os("ND300_MUTATION_HELPER").is_none() {
return;
}
let state = PathBuf::from(
std::env::var_os("ND300_MUTATION_HELPER_STATE").expect("helper state path"),
);
let started = PathBuf::from(
std::env::var_os("ND300_MUTATION_HELPER_STARTED").expect("helper started path"),
);
let delay_ms = std::env::var("ND300_MUTATION_HELPER_DELAY_MS")
.expect("helper delay")
.parse::<u64>()
.expect("numeric helper delay");
let value = std::env::var("ND300_MUTATION_HELPER_VALUE").expect("helper value");
std::fs::write(started, b"started").expect("write helper started marker");
std::thread::sleep(Duration::from_millis(delay_ms));
std::fs::write(state, value).expect("write helper state");
}
#[tokio::test]
async fn owned_timeout_prevents_late_mutation_on_every_platform() {
let state = test_path("timeout", "state");
let started = test_path("timeout", "started");
cleanup(&[&state, &started]);
std::fs::write(&state, b"up").unwrap();
let error = run_owned_mutation(
helper_command(&state, &started, 600, "down"),
Duration::from_millis(50),
)
.await
.unwrap_err();
assert!(error.contains("timed out"), "unexpected error: {error}");
tokio::time::sleep(Duration::from_millis(800)).await;
assert_eq!(std::fs::read(&state).unwrap(), b"up");
cleanup(&[&state, &started]);
}
#[tokio::test]
async fn completed_pair_reports_both_transitions() {
let state = test_path("pair-complete", "state");
let down_started = test_path("pair-complete", "down-started");
let up_started = test_path("pair-complete", "up-started");
cleanup(&[&state, &down_started, &up_started]);
let pair = run_owned_mutation_pair(
helper_command(&state, &down_started, 0, "down"),
Duration::from_secs(5),
Duration::ZERO,
helper_command(&state, &up_started, 0, "up"),
Duration::from_secs(5),
)
.await
.unwrap();
assert!(pair.first.unwrap().status.success());
assert!(pair.inverse.unwrap().status.success());
assert_eq!(std::fs::read(&state).unwrap(), b"up");
cleanup(&[&state, &down_started, &up_started]);
}
#[tokio::test]
async fn timed_out_pair_reaps_first_before_running_inverse() {
let state = test_path("pair-timeout", "state");
let down_started = test_path("pair-timeout", "down-started");
let up_started = test_path("pair-timeout", "up-started");
cleanup(&[&state, &down_started, &up_started]);
std::fs::write(&state, b"initial").unwrap();
let pair = run_owned_mutation_pair(
helper_command(&state, &down_started, 600, "down"),
Duration::from_millis(50),
Duration::ZERO,
helper_command(&state, &up_started, 0, "up"),
Duration::from_secs(5),
)
.await
.unwrap();
assert!(
pair.first.unwrap_err().contains("timed out"),
"the first mutation should have timed out"
);
assert!(pair.inverse.unwrap().status.success());
wait_for_contents(&state, b"up").await;
tokio::time::sleep(Duration::from_millis(800)).await;
assert_eq!(
std::fs::read(&state).unwrap(),
b"up",
"timed-out first command mutated state after its inverse"
);
cleanup(&[&state, &down_started, &up_started]);
}
#[tokio::test]
async fn cancelled_pair_runs_inverse_and_cannot_mutate_late() {
let state = test_path("pair-cancel", "state");
let down_started = test_path("pair-cancel", "down-started");
let up_started = test_path("pair-cancel", "up-started");
cleanup(&[&state, &down_started, &up_started]);
std::fs::write(&state, b"initial").unwrap();
let task = tokio::spawn(run_owned_mutation_pair(
helper_command(&state, &down_started, 600, "down"),
Duration::from_secs(5),
Duration::from_secs(2),
helper_command(&state, &up_started, 0, "up"),
Duration::from_secs(5),
));
wait_for_file(&down_started).await;
task.abort();
let _ = task.await;
wait_for_file(&up_started).await;
wait_for_contents(&state, b"up").await;
tokio::time::sleep(Duration::from_millis(800)).await;
assert_eq!(
std::fs::read(&state).unwrap(),
b"up",
"cancelled first command mutated state after its inverse"
);
cleanup(&[&state, &down_started, &up_started]);
}
}
#[cfg(all(test, unix))]
mod owned_mutation_tests {
use super::*;
use std::path::{Path, PathBuf};
fn test_path(label: &str, suffix: &str) -> PathBuf {
std::env::temp_dir().join(format!(
"nd300-owned-mutation-{label}-{}-{suffix}",
std::process::id()
))
}
fn late_down_command(pid_file: &Path, started_file: &Path, state_file: &Path) -> Command {
let mut command = Command::new("sh");
command.args([
"-c",
"printf '%s' \"$$\" > \"$1\"; printf started > \"$2\"; sleep 1; printf down > \"$3\"",
"nd300-owned-mutation-test",
&pid_file.to_string_lossy(),
&started_file.to_string_lossy(),
&state_file.to_string_lossy(),
]);
command
}
fn restore_up_command(state_file: &Path) -> Command {
let mut command = Command::new("sh");
command.args([
"-c",
"printf up > \"$1\"",
"nd300-owned-mutation-restore",
&state_file.to_string_lossy(),
]);
command
}
async fn wait_for_file(path: &Path) {
let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
while !path.exists() && tokio::time::Instant::now() < deadline {
tokio::time::sleep(Duration::from_millis(10)).await;
}
assert!(
path.exists(),
"test child did not create {}",
path.display()
);
}
fn child_is_gone(pid_file: &Path) -> bool {
let pid = std::fs::read_to_string(pid_file)
.ok()
.and_then(|value| value.parse::<libc::pid_t>().ok())
.expect("test child PID");
(unsafe { libc::kill(pid, 0) }) != 0
}
fn cleanup(paths: &[&Path]) {
for path in paths {
let _ = std::fs::remove_file(path);
}
}
#[tokio::test]
async fn timed_out_down_is_killed_reaped_before_restore_returns() {
let pid_file = test_path("timeout", "pid");
let started_file = test_path("timeout", "started");
let state_file = test_path("timeout", "state");
cleanup(&[&pid_file, &started_file, &state_file]);
std::fs::write(&state_file, b"up").unwrap();
let error = run_owned_mutation(
late_down_command(&pid_file, &started_file, &state_file),
Duration::from_millis(50),
)
.await
.unwrap_err();
assert!(error.contains("timed out"), "unexpected error: {error}");
assert!(started_file.exists());
assert!(
child_is_gone(&pid_file),
"timed-out mutation child was not reaped before return"
);
run_owned_mutation(restore_up_command(&state_file), Duration::from_secs(2))
.await
.unwrap();
tokio::time::sleep(Duration::from_millis(1200)).await;
assert_eq!(std::fs::read(&state_file).unwrap(), b"up");
cleanup(&[&pid_file, &started_file, &state_file]);
}
#[tokio::test]
async fn cancelled_down_supervisor_finishes_before_serialized_restore() {
let pid_file = test_path("cancel", "pid");
let started_file = test_path("cancel", "started");
let state_file = test_path("cancel", "state");
cleanup(&[&pid_file, &started_file, &state_file]);
std::fs::write(&state_file, b"up").unwrap();
let task = tokio::spawn(run_owned_mutation(
late_down_command(&pid_file, &started_file, &state_file),
Duration::from_secs(10),
));
wait_for_file(&started_file).await;
task.abort();
let _ = task.await;
run_owned_mutation(restore_up_command(&state_file), Duration::from_secs(2))
.await
.unwrap();
assert!(
child_is_gone(&pid_file),
"cancelled mutation child was not reaped before restore"
);
tokio::time::sleep(Duration::from_millis(1200)).await;
assert_eq!(
std::fs::read(&state_file).unwrap(),
b"up",
"late cancelled mutation overwrote restored state"
);
cleanup(&[&pid_file, &started_file, &state_file]);
}
}