use std::fs::{self, File, OpenOptions};
use std::io::{Read, Write};
#[cfg(unix)]
use std::os::unix::fs::OpenOptionsExt;
#[cfg(unix)]
use std::os::unix::process::CommandExt;
#[cfg(windows)]
use std::os::windows::process::CommandExt;
use std::path::PathBuf;
use std::process::{Command, Stdio};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};
use crate::backends::typed::CommandInvocation;
use crate::cancellation;
use crate::error::ForgeError;
use crate::execution::windows_job::{ProcessJob, job_ref};
use crate::execution::{EventSink, ExecutionEvent, UiEventSink};
use crate::paths::app_home;
use crate::ui::progress::CommandStatus;
use crate::util::{now_secs, resolve_command, shell_quote_str};
const PROGRESS_OUTPUT_LIMIT: usize = 8 * 1024;
const COMMAND_POLL_INTERVAL: Duration = Duration::from_millis(100);
static LOG_SEQUENCE: AtomicU64 = AtomicU64::new(0);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ShellDisplayMode {
Plain,
StatusBar,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct ShellStep {
pub(crate) current: usize,
pub(crate) total: usize,
}
#[derive(Debug, Clone, Copy)]
struct ShellRunOptions {
include_command_on_error: bool,
display_mode: ShellDisplayMode,
step: Option<ShellStep>,
timeout: Option<Duration>,
inactivity_timeout: Option<Duration>,
}
pub(crate) fn run_shell_labeled(label: &str, command: &str) -> Result<(), ForgeError> {
run_shell_labeled_with_options(
label,
command,
ShellRunOptions {
include_command_on_error: true,
display_mode: ShellDisplayMode::Plain,
step: None,
timeout: None,
inactivity_timeout: None,
},
)
}
pub(crate) fn run_shell_labeled_limits_display_step(
label: &str,
command: &str,
mode: ShellDisplayMode,
step: Option<ShellStep>,
timeout: Option<Duration>,
inactivity_timeout: Option<Duration>,
) -> Result<(), ForgeError> {
run_shell_labeled_with_options(
label,
command,
ShellRunOptions {
include_command_on_error: true,
display_mode: mode,
step,
timeout,
inactivity_timeout,
},
)
}
pub(crate) fn run_invocation_labeled(
label: &str,
invocation: &CommandInvocation,
display_mode: ShellDisplayMode,
step: Option<ShellStep>,
) -> Result<(), ForgeError> {
run_invocation_labeled_measured(label, invocation, display_mode, step).map(|_| ())
}
#[derive(Debug, Clone, Copy, Default)]
pub(crate) struct CommandMetrics {
pub(crate) peak_rss_mib: u64,
}
pub(crate) fn run_invocation_labeled_measured(
label: &str,
invocation: &CommandInvocation,
display_mode: ShellDisplayMode,
step: Option<ShellStep>,
) -> Result<CommandMetrics, ForgeError> {
let mut command = Command::new(resolve_command(&invocation.program));
command.args(&invocation.args);
if invocation.clear_env {
command.env_clear();
}
if invocation.null_stdin {
command.stdin(Stdio::null());
}
command.envs(&invocation.env);
if let Some(directory) = &invocation.current_dir {
command.current_dir(directory);
}
configure_process_group(&mut command);
command.stdout(Stdio::piped()).stderr(Stdio::piped());
let display = format_invocation(invocation);
let mut child = command.spawn().map_err(|source| ForgeError::Io {
path: PathBuf::from(&invocation.program),
source,
})?;
#[cfg(windows)]
let job = ProcessJob::assign(&child)?;
#[cfg(not(windows))]
let job = ();
let output = Arc::new(Mutex::new(CommandOutput::new(label, &display)?));
let stdout_handle = child
.stdout
.take()
.map(|stdout| collect_command_output(stdout, Arc::clone(&output)));
let stderr_handle = child
.stderr
.take()
.map(|stderr| collect_command_output(stderr, Arc::clone(&output)));
let started = Instant::now();
let mut last_sample = Instant::now() - Duration::from_secs(1);
let mut metrics = CommandMetrics::default();
let mut status_bar = CommandStatus::new(
label,
step.map(|step| (step.current, step.total)),
matches!(display_mode, ShellDisplayMode::StatusBar),
);
let status = loop {
if cancellation::requested() {
terminate_child_process(&mut child, job_ref(&job));
if cancellation::forced() {
let _ = child.kill();
}
return Err(ForgeError::Command(
"operation interrupted by Ctrl-C".into(),
));
}
if invocation
.timeout_secs
.is_some_and(|seconds| started.elapsed() >= Duration::from_secs(seconds))
{
terminate_child_process(&mut child, job_ref(&job));
return Err(ForgeError::Command(format!("command timed out: {display}")));
}
if invocation.inactivity_timeout_secs.is_some_and(|seconds| {
output_snapshot(&output).2.elapsed() >= Duration::from_secs(seconds)
}) {
terminate_child_process(&mut child, job_ref(&job));
return Err(ForgeError::Command(format!(
"command produced no output for too long: {display}"
)));
}
if let Some(status) = child.try_wait().map_err(|source| ForgeError::Io {
path: PathBuf::from(&invocation.program),
source,
})? {
break status;
}
if last_sample.elapsed() >= Duration::from_millis(500) {
metrics.peak_rss_mib = metrics
.peak_rss_mib
.max(process_tree_rss_mib(child.id(), job_ref(&job)).unwrap_or(0));
last_sample = Instant::now();
}
status_bar.update(
started.elapsed().as_secs(),
recent_output_line(&output_snapshot(&output).0),
);
thread::sleep(COMMAND_POLL_INTERVAL);
};
join_output_thread(stdout_handle, &display)?;
join_output_thread(stderr_handle, &display)?;
status_bar.finish();
metrics.peak_rss_mib = metrics
.peak_rss_mib
.max(process_tree_rss_mib(child.id(), job_ref(&job)).unwrap_or(0));
if status
.code()
.is_some_and(|code| invocation.success_codes.contains(&code))
{
Ok(metrics)
} else {
let (tail, log, _) = output_snapshot(&output);
Err(ForgeError::Command(format!(
"{display}\n{tail}\nFull log: {}",
log.display()
)))
}
}
#[cfg(target_os = "macos")]
fn process_tree_rss_mib(pid: u32, _: Option<&ProcessJob>) -> Option<u64> {
process_group_rss_mib(&["-o", "rss=", "-g", &pid.to_string()])
}
#[cfg(target_os = "linux")]
fn process_tree_rss_mib(pid: u32, _: Option<&ProcessJob>) -> Option<u64> {
process_group_rss_mib(&["-o", "rss=", "--pgroup", &pid.to_string()])
}
#[cfg(any(target_os = "macos", target_os = "linux"))]
fn process_group_rss_mib(args: &[&str]) -> Option<u64> {
let output = Command::new("/bin/ps").args(args).output().ok()?;
output.status.success().then(|| {
String::from_utf8_lossy(&output.stdout)
.split_whitespace()
.filter_map(|value| value.parse::<u64>().ok())
.sum::<u64>()
.div_ceil(1024)
})
}
#[cfg(windows)]
fn process_tree_rss_mib(_: u32, job: Option<&ProcessJob>) -> Option<u64> {
job.and_then(ProcessJob::peak_memory_mib)
}
#[cfg(not(any(target_os = "macos", target_os = "linux", windows)))]
fn process_tree_rss_mib(_: u32, _: Option<&ProcessJob>) -> Option<u64> {
None
}
pub(crate) fn run_invocation_capture(invocation: &CommandInvocation) -> Result<String, ForgeError> {
let mut command = Command::new(resolve_command(&invocation.program));
command.args(&invocation.args);
if invocation.clear_env {
command.env_clear();
}
if invocation.null_stdin {
command.stdin(Stdio::null());
}
command.envs(&invocation.env);
if let Some(directory) = &invocation.current_dir {
command.current_dir(directory);
}
configure_process_group(&mut command);
command.stdout(Stdio::piped()).stderr(Stdio::piped());
let display = format_invocation(invocation);
let mut child = command.spawn().map_err(|source| ForgeError::Io {
path: PathBuf::from(&invocation.program),
source,
})?;
#[cfg(windows)]
let job = ProcessJob::assign(&child)?;
#[cfg(not(windows))]
let job = ();
let output = Arc::new(Mutex::new(String::new()));
let stdout_handle = child
.stdout
.take()
.map(|stdout| collect_capture_output(stdout, Arc::clone(&output)));
let stderr_handle = child
.stderr
.take()
.map(|stderr| collect_capture_output(stderr, Arc::clone(&output)));
let started = Instant::now();
let status = loop {
if cancellation::requested() {
terminate_child_process(&mut child, job_ref(&job));
return Err(ForgeError::Command(
"operation interrupted by Ctrl-C".into(),
));
}
if invocation
.timeout_secs
.is_some_and(|seconds| started.elapsed() >= Duration::from_secs(seconds))
{
terminate_child_process(&mut child, job_ref(&job));
return Err(ForgeError::Command(format!("command timed out: {display}")));
}
if let Some(status) = child.try_wait().map_err(|source| ForgeError::Io {
path: PathBuf::from(&invocation.program),
source,
})? {
break status;
}
thread::sleep(COMMAND_POLL_INTERVAL);
};
join_output_thread(stdout_handle, &display)?;
join_output_thread(stderr_handle, &display)?;
let output = output.lock().map(|value| value.clone()).unwrap_or_default();
if status
.code()
.is_some_and(|code| invocation.success_codes.contains(&code))
&& invocation
.stdout_contains
.as_ref()
.is_none_or(|needle| output.contains(needle))
{
Ok(output)
} else {
Err(ForgeError::Command(format!("{display}\n{output}")))
}
}
fn format_invocation(invocation: &CommandInvocation) -> String {
std::iter::once(invocation.program.as_str())
.chain(invocation.args.iter().map(String::as_str))
.map(shell_quote_str)
.collect::<Vec<_>>()
.join(" ")
}
fn run_shell_labeled_with_options(
label: &str,
command: &str,
options: ShellRunOptions,
) -> Result<(), ForgeError> {
let ShellRunOptions {
include_command_on_error,
display_mode,
step,
timeout,
inactivity_timeout,
} = options;
let command = command_for_current_user(command);
let event_step = step.unwrap_or(ShellStep {
current: 1,
total: 1,
});
let mut events = UiEventSink;
events.emit(&ExecutionEvent::Started {
step: event_step.current,
total: event_step.total,
component: label.to_string(),
});
let mut status_bar = CommandStatus::new(
label,
step.map(|step| (step.current, step.total)),
matches!(display_mode, ShellDisplayMode::StatusBar),
);
let mut child = shell_command(&command)
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.map_err(|source| ForgeError::Io {
path: PathBuf::from(&command),
source,
})?;
#[cfg(windows)]
let job = ProcessJob::assign(&child)?;
#[cfg(not(windows))]
let job = ();
let output = Arc::new(Mutex::new(CommandOutput::new(label, &command)?));
let stdout_handle = child
.stdout
.take()
.map(|stdout| collect_command_output(stdout, Arc::clone(&output)));
let stderr_handle = child
.stderr
.take()
.map(|stderr| collect_command_output(stderr, Arc::clone(&output)));
let started = Instant::now();
let status = loop {
if cancellation::requested() {
terminate_child_process(&mut child, job_ref(&job));
join_output_thread(stdout_handle, &command)?;
join_output_thread(stderr_handle, &command)?;
status_bar.finish();
return Err(ForgeError::Command(
"operation interrupted by Ctrl-C".to_string(),
));
}
if timeout.is_some_and(|timeout| started.elapsed() >= timeout) {
terminate_child_process(&mut child, job_ref(&job));
join_output_thread(stdout_handle, &command)?;
join_output_thread(stderr_handle, &command)?;
status_bar.finish();
let (output, log_path, _) = output_snapshot(&output);
return Err(ForgeError::Command(format!(
"command timed out after {} seconds\n{}\nFull log: {}\nHint: check the network and proxy, or increase timeout_secs",
timeout.unwrap_or_default().as_secs(),
if output.trim().is_empty() {
"the command produced no error details"
} else {
&output
},
log_path.display()
)));
}
let (_, log_path, last_activity) = output_snapshot(&output);
if inactivity_timeout.is_some_and(|limit| last_activity.elapsed() >= limit) {
terminate_child_process(&mut child, job_ref(&job));
join_output_thread(stdout_handle, &command)?;
join_output_thread(stderr_handle, &command)?;
status_bar.finish();
let (recent, _, _) = output_snapshot(&output);
return Err(ForgeError::Command(format!(
"command produced no output for {} consecutive seconds\n{}\nFull log: {}\nHint: check whether the command is waiting for input, or increase inactivity_timeout_secs",
inactivity_timeout.unwrap_or_default().as_secs(),
if recent.trim().is_empty() {
"the command produced no error details"
} else {
&recent
},
log_path.display()
)));
}
if let Some(status) = child.try_wait().map_err(|source| ForgeError::Io {
path: PathBuf::from(&command),
source,
})? {
break status;
}
let elapsed = started.elapsed().as_secs();
let (recent, _, _) = output_snapshot(&output);
let recent = recent_output_line(&recent);
status_bar.update(elapsed, recent.clone());
events.emit(&ExecutionEvent::Progress {
step: event_step.current,
total: event_step.total,
component: label.to_string(),
elapsed_secs: elapsed,
recent,
});
thread::sleep(COMMAND_POLL_INTERVAL);
};
join_output_thread(stdout_handle, &command)?;
join_output_thread(stderr_handle, &command)?;
status_bar.finish();
if status.success() {
events.emit(&ExecutionEvent::Completed {
step: event_step.current,
total: event_step.total,
component: label.to_string(),
elapsed_secs: started.elapsed().as_secs(),
});
Ok(())
} else {
let (output, log_path, _) = output_snapshot(&output);
let detail = if output.trim().is_empty() {
"the command produced no error details".to_string()
} else {
output
};
if include_command_on_error {
Err(ForgeError::Command(format!(
"{}\n{}\nFull log: {}\nHint: fix dependency or permission issues based on the recent output, then retry",
command,
detail,
log_path.display()
)))
} else {
Err(ForgeError::Command(detail))
}
}
}
pub(crate) fn run_shell_capture(command: &str) -> Result<String, ForgeError> {
run_shell_capture_timeout(command, None, &[])
}
pub(crate) fn run_shell_capture_timeout(
command: &str,
timeout_secs: Option<u64>,
environment: &[(&str, &str)],
) -> Result<String, ForgeError> {
let mut child = shell_command(command)
.envs(environment.iter().copied())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.map_err(|source| ForgeError::Io {
path: PathBuf::from(command),
source,
})?;
#[cfg(windows)]
let job = ProcessJob::assign(&child)?;
#[cfg(not(windows))]
let job = ();
let output = Arc::new(Mutex::new(String::new()));
let stdout_handle = child
.stdout
.take()
.map(|stdout| collect_capture_output(stdout, Arc::clone(&output)));
let stderr_handle = child
.stderr
.take()
.map(|stderr| collect_capture_output(stderr, Arc::clone(&output)));
let started = Instant::now();
let status = loop {
if timeout_secs.is_some_and(|seconds| started.elapsed() >= Duration::from_secs(seconds)) {
terminate_child_process(&mut child, job_ref(&job));
return Err(ForgeError::Command(format!("command timed out: {command}")));
}
if let Some(status) = child.try_wait().map_err(|source| ForgeError::Io {
path: PathBuf::from(command),
source,
})? {
break status;
}
thread::sleep(COMMAND_POLL_INTERVAL);
};
join_output_thread(stdout_handle, command)?;
join_output_thread(stderr_handle, command)?;
let output = output.lock().map(|value| value.clone()).unwrap_or_default();
if status.success() {
Ok(output)
} else {
Err(ForgeError::Command(format!("{command}\n{output}")))
}
}
fn shell_command(command: &str) -> Command {
#[cfg(windows)]
{
let mut cmd = Command::new("cmd");
cmd.arg("/C").raw_arg(command);
configure_process_group(&mut cmd);
cmd
}
#[cfg(not(windows))]
if command_exists("bash") {
let mut cmd = Command::new("bash");
cmd.arg("-lc").arg(command);
configure_process_group(&mut cmd);
cmd
} else {
let mut cmd = Command::new("sh");
cmd.arg("-c").arg(command);
configure_process_group(&mut cmd);
cmd
}
}
#[cfg(unix)]
fn configure_process_group(command: &mut Command) {
command.process_group(0);
}
#[cfg(windows)]
fn configure_process_group(command: &mut Command) {
const CREATE_NEW_PROCESS_GROUP: u32 = 0x0000_0200;
command.creation_flags(CREATE_NEW_PROCESS_GROUP);
}
#[cfg(not(windows))]
fn command_exists(command: &str) -> bool {
Command::new(resolve_command(command))
.arg("--version")
.output()
.map(|output| output.status.success())
.unwrap_or(false)
}
fn command_for_current_user(command: &str) -> String {
if cfg!(target_os = "linux") && is_linux_root() {
strip_sudo_from_apt_commands(command)
} else {
command.to_string()
}
}
fn is_linux_root() -> bool {
if !cfg!(target_os = "linux") {
return false;
}
Command::new("id")
.arg("-u")
.output()
.ok()
.filter(|output| output.status.success())
.map(|output| String::from_utf8_lossy(&output.stdout).trim() == "0")
.unwrap_or(false)
}
fn strip_sudo_from_apt_commands(command: &str) -> String {
command
.replace("sudo apt-get ", "apt-get ")
.replace("sudo apt ", "apt ")
.replace("sudo -E apt-get ", "apt-get ")
.replace("sudo -E apt ", "apt ")
}
struct CommandOutput {
tail: String,
log: File,
log_path: PathBuf,
last_activity: Instant,
}
impl CommandOutput {
fn new(label: &str, command: &str) -> Result<Self, ForgeError> {
let directory = app_home().join("logs").join("commands");
fs::create_dir_all(&directory).map_err(|source| ForgeError::Io {
path: directory.clone(),
source,
})?;
let sequence = LOG_SEQUENCE.fetch_add(1, Ordering::Relaxed);
let safe_label: String = label
.chars()
.map(|ch| {
if ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_') {
ch
} else {
'-'
}
})
.collect();
let path = directory.join(format!(
"{}-{}-{}-{}.log",
now_secs(),
std::process::id(),
sequence,
safe_label
));
let mut options = OpenOptions::new();
options.create_new(true).write(true);
#[cfg(unix)]
{
options.mode(0o600);
}
let mut log = options.open(&path).map_err(|source| ForgeError::Io {
path: path.clone(),
source,
})?;
writeln!(log, "$ {command}").map_err(|source| ForgeError::Io {
path: path.clone(),
source,
})?;
Ok(Self {
tail: String::new(),
log,
log_path: path,
last_activity: Instant::now(),
})
}
}
fn collect_command_output<R>(
mut reader: R,
output: Arc<Mutex<CommandOutput>>,
) -> thread::JoinHandle<()>
where
R: Read + Send + 'static,
{
thread::spawn(move || {
let mut buffer = [0; 1024];
loop {
match reader.read(&mut buffer) {
Ok(0) | Err(_) => break,
Ok(size) => push_output(&output, &String::from_utf8_lossy(&buffer[..size])),
}
}
})
}
fn collect_capture_output<R>(mut reader: R, output: Arc<Mutex<String>>) -> thread::JoinHandle<()>
where
R: Read + Send + 'static,
{
thread::spawn(move || {
let mut buffer = [0; 1024];
loop {
match reader.read(&mut buffer) {
Ok(0) | Err(_) => break,
Ok(size) => {
if let Ok(mut output) = output.lock() {
push_tail(&mut output, &String::from_utf8_lossy(&buffer[..size]));
}
}
}
}
})
}
fn push_output(output: &Arc<Mutex<CommandOutput>>, text: &str) {
let Ok(mut output) = output.lock() else {
return;
};
let _ = output.log.write_all(text.as_bytes());
output.last_activity = Instant::now();
push_tail(&mut output.tail, text);
}
fn push_tail(output: &mut String, text: &str) {
output.push_str(text);
if output.len() <= PROGRESS_OUTPUT_LIMIT {
return;
}
let mut keep_from = output.len() - PROGRESS_OUTPUT_LIMIT;
while !output.is_char_boundary(keep_from) {
keep_from += 1;
}
output.drain(..keep_from);
}
fn recent_output_line(output: &str) -> String {
let output = output.replace('\r', "\n");
if output.trim().is_empty() {
return "no command output".to_string();
}
output
.lines()
.map(str::trim_end)
.rev()
.find(|line| !line.trim().is_empty())
.unwrap_or("no command output")
.to_string()
}
fn terminate_child_process(child: &mut std::process::Child, job: Option<&ProcessJob>) {
let pid = child.id();
#[cfg(not(windows))]
let descendants = descendant_processes(pid);
terminate_process_tree(pid, job);
#[cfg(not(windows))]
{
signal_processes(&descendants, "-TERM");
for _ in 0..10 {
if child.try_wait().ok().flatten().is_some() {
break;
}
thread::sleep(Duration::from_millis(25));
}
kill_process_tree(pid);
signal_processes(&descendants, "-KILL");
}
let _ = child.kill();
let _ = child.wait();
}
#[cfg(windows)]
fn terminate_process_tree(_: u32, job: Option<&ProcessJob>) {
if let Some(job) = job {
job.terminate();
}
}
#[cfg(not(windows))]
fn terminate_process_tree(pid: u32, _: Option<&ProcessJob>) {
signal_process_tree(pid, "-TERM");
}
#[cfg(not(windows))]
fn kill_process_tree(pid: u32) {
signal_process_tree(pid, "-KILL");
}
#[cfg(not(windows))]
fn signal_process_tree(pid: u32, signal: &str) {
let group = format!("-{pid}");
let _ = Command::new("/bin/kill")
.arg(signal)
.arg("--")
.arg(&group)
.stdout(Stdio::null())
.stderr(Stdio::null())
.status();
}
#[cfg(not(windows))]
fn descendant_processes(root: u32) -> Vec<u32> {
let Ok(output) = Command::new("ps")
.args(["-eo", "pid=,ppid="])
.stdout(Stdio::piped())
.stderr(Stdio::null())
.output()
else {
return Vec::new();
};
let mut children = std::collections::BTreeMap::<u32, Vec<u32>>::new();
for line in String::from_utf8_lossy(&output.stdout).lines() {
let mut fields = line.split_whitespace();
let (Some(pid), Some(parent)) = (fields.next(), fields.next()) else {
continue;
};
let (Ok(pid), Ok(parent)) = (pid.parse::<u32>(), parent.parse::<u32>()) else {
continue;
};
children.entry(parent).or_default().push(pid);
}
let mut descendants = Vec::new();
let mut pending = children.remove(&root).unwrap_or_default();
while let Some(pid) = pending.pop() {
if let Some(mut nested) = children.remove(&pid) {
pending.append(&mut nested);
}
descendants.push(pid);
}
descendants
}
#[cfg(not(windows))]
fn signal_processes(processes: &[u32], signal: &str) {
for pid in processes.iter().rev() {
let _ = Command::new("/bin/kill")
.arg(signal)
.arg(pid.to_string())
.stdout(Stdio::null())
.stderr(Stdio::null())
.status();
}
}
fn output_snapshot(output: &Arc<Mutex<CommandOutput>>) -> (String, PathBuf, Instant) {
output
.lock()
.map(|output| {
(
output.tail.clone(),
output.log_path.clone(),
output.last_activity,
)
})
.unwrap_or_else(|_| (String::new(), PathBuf::new(), Instant::now()))
}
fn join_output_thread(
handle: Option<thread::JoinHandle<()>>,
command: &str,
) -> Result<(), ForgeError> {
if let Some(handle) = handle {
handle.join().map_err(|_| {
ForgeError::Command(format!(
"{command}\ninternal error while reading command output"
))
})?;
}
Ok(())
}
#[cfg(test)]
mod tests;