diff --git a/.changeset/tidy-ravens-bail.md b/.changeset/tidy-ravens-bail.md new file mode 100644 index 0000000000..c5a5062b94 --- /dev/null +++ b/.changeset/tidy-ravens-bail.md @@ -0,0 +1,5 @@ +--- +"pacquet": patch +--- + +Stop in-flight recursive `run` and `exec` commands when bailing after the first failure. diff --git a/Cargo.lock b/Cargo.lock index 045ee603ae..d49fd3e9f4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4897,6 +4897,7 @@ version = "0.0.1" dependencies = [ "deno_task_shell", "derive_more", + "libc", "miette 7.6.0", "pnpm-package-manifest", "pnpm-reporter", @@ -4905,6 +4906,7 @@ dependencies = [ "tempfile", "tokio", "tracing", + "windows-sys 0.61.2", ] [[package]] diff --git a/pnpm/crates/cli/src/cli_args/exec.rs b/pnpm/crates/cli/src/cli_args/exec.rs index edd8450f14..aa4835721a 100644 --- a/pnpm/crates/cli/src/cli_args/exec.rs +++ b/pnpm/crates/cli/src/cli_args/exec.rs @@ -5,7 +5,9 @@ use clap::Args; use derive_more::{Display, Error}; use miette::Diagnostic; use pnpm_config::Config; -use pnpm_executor::{ScriptOutput, StreamedScript, push_script_arg, select_shell}; +use pnpm_executor::{ + ProcessTracker, ScriptOutput, StreamedScript, push_script_arg, select_shell, spawn_child, +}; use pnpm_package_manager::{ make_node_package_map_option, make_node_require_option, package_map_path_for_execution, pnp_path_for_execution, @@ -103,7 +105,8 @@ impl ExecArgs { pub fn run(self, dir: &Path, config: &Config, reporter: ReporterType) -> miette::Result<()> { let command = prepare_command(self.command)?; super::verify_deps::verify_deps_before_run(dir, config, reporter)?; - let status = spawn_in_dir(&command, dir, config, self.shell_mode, ScriptOutput::Inherit)?; + let status = + spawn_in_dir(&command, dir, config, self.shell_mode, ScriptOutput::Inherit, None)?; if !status.success() { // Propagate the child's exit code. A signal-terminated child // has no code; fall back to 1, matching pnpm's `exitCode ?? 1`. @@ -159,22 +162,23 @@ pub(super) fn spawn_in_dir( config: &Config, shell_mode: bool, output: ScriptOutput<'_>, + process_tracker: Option<&ProcessTracker>, ) -> Result { let mut cmd = command_in_dir(command, dir, config, shell_mode)?; let ScriptOutput::Streamed { dep_path, emit } = output else { - return cmd - .status() + let mut child = spawn_child(&mut cmd, process_tracker) + .map_err(|source| ExecError::Spawn { command: command[0].clone(), source })?; + return child + .wait() .map_err(|source| ExecError::Spawn { command: command[0].clone(), source }); }; let wd = dir.to_string_lossy(); let streamed = StreamedScript { dep_path, stage: EXEC_STAGE, wd: &wd, emit }; - let mut child = cmd - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn() + cmd.stdout(Stdio::piped()).stderr(Stdio::piped()); + let mut child = spawn_child(&mut cmd, process_tracker) .map_err(|source| ExecError::Spawn { command: command[0].clone(), source })?; let status = streamed - .pump(&mut child) + .pump(child.child_mut()) .map_err(|source| ExecError::Spawn { command: command[0].clone(), source })?; streamed.finished(status.code().unwrap_or(-1)); Ok(status) diff --git a/pnpm/crates/cli/src/cli_args/exec/recursive.rs b/pnpm/crates/cli/src/cli_args/exec/recursive.rs index da66b47db0..19df1e5ca0 100644 --- a/pnpm/crates/cli/src/cli_args/exec/recursive.rs +++ b/pnpm/crates/cli/src/cli_args/exec/recursive.rs @@ -21,7 +21,7 @@ use derive_more::{Display, Error}; use indexmap::IndexMap; use miette::Diagnostic; use pnpm_config::Config; -use pnpm_executor::ScriptOutput; +use pnpm_executor::{ProcessTracker, ScriptOutput}; use pnpm_reporter::LogEvent; use pnpm_workspace_task_scheduler::{ ScheduleTasksOptions, SequenceTasksOptions, TaskCompletion, TaskGraph, TaskKey, TaskNode, @@ -163,6 +163,7 @@ pub async fn exec_recursive( .collect(), ); let first_failure: Mutex> = Mutex::new(None); + let process_tracker = bail.then(ProcessTracker::default); let run_task = |node: &TaskNode| -> TaskCompletion { let root = node.project.as_path(); @@ -171,10 +172,16 @@ pub async fn exec_recursive( let start = Instant::now(); let dep_path = project_dep_path(root, dir, show_prefix); let output = project_output(dep_path.as_deref(), emit); - let outcome = spawn_in_dir(&command, root, config, args.shell_mode, output); + let outcome = + spawn_in_dir(&command, root, config, args.shell_mode, output, process_tracker.as_ref()); let execution = project_execution(start, outcome); let mut result = result.lock().expect("summary lock is not poisoned"); let entry = &mut result[&prefix]; + if process_tracker.as_ref().is_some_and(ProcessTracker::is_cancelled) + && execution.message.is_none() + { + return TaskCompletion::Cancelled; + } entry.duration = Some(execution.duration); match execution.message { None => { @@ -182,6 +189,9 @@ pub async fn exec_recursive( TaskCompletion::Passed } Some(message) => { + if process_tracker.as_ref().is_some_and(|tracker| !tracker.cancel()) { + return TaskCompletion::Cancelled; + } entry.status = Status::Failure; entry.message = Some(message); entry.prefix = Some(prefix.clone()); diff --git a/pnpm/crates/cli/src/cli_args/run.rs b/pnpm/crates/cli/src/cli_args/run.rs index 09e0691f80..3e26e7ece2 100644 --- a/pnpm/crates/cli/src/cli_args/run.rs +++ b/pnpm/crates/cli/src/cli_args/run.rs @@ -6,7 +6,9 @@ use clap::Args; use derive_more::{Display, Error}; use miette::Diagnostic; use pnpm_config::Config; -use pnpm_executor::{RunScript, ScriptExit, ScriptOutput, ScriptsPrependNodePath, run_script}; +use pnpm_executor::{ + ProcessTracker, RunScript, ScriptExit, ScriptOutput, ScriptsPrependNodePath, run_script, +}; use pnpm_injected_deps_syncer::{SyncInjectedDeps, sync_injected_deps}; use pnpm_package_manager::{ make_node_package_map_option, make_node_require_option, package_map_path_for_execution, @@ -260,6 +262,7 @@ impl RunArgs { extra_env: &extra_env, silent, output: ScriptOutput::Inherit, + process_tracker: None, }; for name in &specified { // Resolve the main body (with `start` → `node server.js` @@ -338,6 +341,7 @@ pub(super) struct RunContext<'a> { pub(super) extra_env: &'a HashMap, pub(super) silent: bool, pub(super) output: ScriptOutput<'a>, + pub(super) process_tracker: Option<&'a ProcessTracker>, } /// Resolve `name` to a runnable main script body, or `Ok(None)` when @@ -502,6 +506,7 @@ pub(super) fn run_stage( extra_env: ctx.extra_env, silent: ctx.silent, output: ctx.output, + process_tracker: ctx.process_tracker, }) .map_err(miette::Report::new)?; diff --git a/pnpm/crates/cli/src/cli_args/run/recursive.rs b/pnpm/crates/cli/src/cli_args/run/recursive.rs index 92b8e1a774..06fdef253b 100644 --- a/pnpm/crates/cli/src/cli_args/run/recursive.rs +++ b/pnpm/crates/cli/src/cli_args/run/recursive.rs @@ -25,7 +25,7 @@ use derive_more::{Display, Error}; use indexmap::IndexMap; use miette::{Diagnostic, IntoDiagnostic}; use pnpm_config::Config; -use pnpm_executor::ScriptOutput; +use pnpm_executor::{ProcessTracker, ScriptOutput}; use pnpm_package_manager::{ make_node_package_map_option, make_node_require_option, package_map_path_for_execution, pnp_path_for_execution, @@ -219,6 +219,7 @@ pub fn run_recursive( let has_command = AtomicUsize::new(0); let first_failure: Mutex> = Mutex::new(None); let abort: Mutex> = Mutex::new(None); + let process_tracker = bail.then(ProcessTracker::default); // Compute the shared lifecycle env once. Each project adds loader // options for its own root before reusing that environment across @@ -227,6 +228,11 @@ pub fn run_recursive( let extra_env: HashMap = config.extra_env_with_node_options(); let run_task = |node: &TaskNode| -> TaskCompletion { + let summary_key = task_summary_key(node); + let on_started = || { + result.lock().expect("summary lock is not poisoned")[&summary_key].status = + Status::Running; + }; let execution = match run_project(RunProjectOptions { node, graph, @@ -238,6 +244,8 @@ pub fn run_recursive( silent, inherit_output, emit, + process_tracker: process_tracker.as_ref(), + on_started: &on_started, }) { Ok(execution) => execution, Err(error) => { @@ -245,6 +253,9 @@ pub fn run_recursive( if abort.is_none() { *abort = Some(error); } + if let Some(process_tracker) = &process_tracker { + process_tracker.cancel(); + } return TaskCompletion::Aborted; } }; @@ -252,8 +263,11 @@ pub fn run_recursive( has_command.fetch_add(execution.has_command, Ordering::Relaxed); } let failed = execution.status.status == Status::Failure; - result.lock().expect("summary lock is not poisoned")[&task_summary_key(node)] = - execution.status; + let cancelled = execution.cancelled; + result.lock().expect("summary lock is not poisoned")[&summary_key] = execution.status; + if cancelled { + return TaskCompletion::Cancelled; + } if failed { let mut first_failure = first_failure.lock().expect("first-failure slot lock is not poisoned"); @@ -396,6 +410,7 @@ fn build_run_task_graph( struct ProjectExecution { status: ExecutionStatus, has_command: usize, + cancelled: bool, } #[derive(Clone, Copy)] @@ -410,6 +425,8 @@ struct RunProjectOptions<'a, 'project> { silent: bool, inherit_output: bool, emit: fn(&LogEvent), + process_tracker: Option<&'a ProcessTracker>, + on_started: &'a dyn Fn(), } fn run_project(options: RunProjectOptions<'_, '_>) -> miette::Result { @@ -424,6 +441,8 @@ fn run_project(options: RunProjectOptions<'_, '_>) -> miette::Result) -> miette::Result) -> miette::Result) -> miette::Result {}, 5000)""#, ), ), ( - "passes", + "b-fails", manifest( - "passes", - "touch ran.txt; i=0; while [ ! -f ../fails/failed.txt ] && [ $i -lt 1000 ]; do i=$((i + 1)); sleep 0.01; done; [ -f ../fails/failed.txt ]", + "b-fails", + r#"node -e "const fs = require('fs'); const wait = () => fs.existsSync('../a-slow-1/ran.txt') && fs.existsSync('../c-slow-2/ran.txt') ? process.exit(1) : setTimeout(wait, 10); wait()""#, ), ), + ( + "c-slow-2", + manifest( + "c-slow-2", + r#"node -e "require('fs').writeFileSync('ran.txt', ''); setTimeout(() => {}, 5000)""#, + ), + ), + ("z-queued", manifest("z-queued", "touch ran.txt")), ], ); + if shell_emulator { + fs::write( + workspace.join("pnpm-workspace.yaml"), + "packages:\n - a-slow-1\n - b-fails\n - c-slow-2\n - z-queued\nshellEmulator: true\n", + ) + .expect("enable the shell emulator"); + } - pacquet + let start = Instant::now(); + let output = pacquet .with_args([ - "--workspace-concurrency=2", + "--workspace-concurrency=3", "--no-sort", "--report-summary", "-r", "run", "build", ]) - .assert() - .failure(); + .output() + .expect("spawn pacquet"); + let elapsed = start.elapsed(); + let stderr = String::from_utf8_lossy(&output.stderr); + eprintln!("STDERR:\n{stderr}\n"); + assert!(!output.status.success(), "the failing project should fail the run"); + eprintln!("recursive run elapsed: {elapsed:?}"); + assert!( + elapsed < Duration::from_secs(4), + "bail should interrupt the five-second in-flight scripts", + ); let statuses = summary_statuses(&workspace); - assert_eq!(statuses.get("fails").map(String::as_str), Some("failure")); - assert_eq!(statuses.get("passes").map(String::as_str), Some("passed")); + dbg!(&statuses); + assert_eq!(statuses.get("a-slow-1").map(String::as_str), Some("running")); + assert_eq!(statuses.get("b-fails").map(String::as_str), Some("failure")); + assert_eq!(statuses.get("c-slow-2").map(String::as_str), Some("running")); + assert_eq!(statuses.get("z-queued").map(String::as_str), Some("queued")); + assert!(!workspace.join("z-queued").join("ran.txt").exists()); drop(root); } diff --git a/pnpm/crates/executor/Cargo.toml b/pnpm/crates/executor/Cargo.toml index f388b7ce33..ba8972e87a 100644 --- a/pnpm/crates/executor/Cargo.toml +++ b/pnpm/crates/executor/Cargo.toml @@ -21,6 +21,12 @@ serde_json = { workspace = true } tokio = { workspace = true, features = ["process"] } tracing = { workspace = true } +[target.'cfg(unix)'.dependencies] +libc = { workspace = true } + +[target.'cfg(windows)'.dependencies] +windows-sys = { workspace = true, features = ["Win32_System_SystemInformation"] } + [dev-dependencies] pretty_assertions = { workspace = true } tempfile = { workspace = true } diff --git a/pnpm/crates/executor/src/lib.rs b/pnpm/crates/executor/src/lib.rs index c51427e5b8..8a800fb78e 100644 --- a/pnpm/crates/executor/src/lib.rs +++ b/pnpm/crates/executor/src/lib.rs @@ -2,6 +2,7 @@ mod bundled_node_gyp; mod extend_path; mod lifecycle; mod make_env; +mod process_tracker; mod run_script; mod script_exit; mod shell; @@ -16,6 +17,7 @@ pub use lifecycle::{ run_project_lifecycle_scripts, }; pub use make_env::{EnvBuild, EnvOptions, VERIFY_DEPS_BEFORE_RUN_ENV, build_env}; +pub use process_tracker::{ProcessTracker, SpawnedChild, spawn_child}; pub use run_script::{RunScript, RunScriptError, ScriptOutput, run_script}; pub use script_exit::ScriptExit; pub use shell::{ScriptShellError, SelectedShell, select_shell}; diff --git a/pnpm/crates/executor/src/lifecycle.rs b/pnpm/crates/executor/src/lifecycle.rs index 4880f22dc3..a5e699f081 100644 --- a/pnpm/crates/executor/src/lifecycle.rs +++ b/pnpm/crates/executor/src/lifecycle.rs @@ -445,7 +445,7 @@ fn run_in_emulator( ) -> Result { let target = StreamedScript { dep_path: opts.dep_path, stage, wd, emit: Reporter::emit }; let emit_line = |stdio, line| target.emit_line(stdio, line); - execute_emulated(script, opts.pkg_root, env, EmulatedOutput::Lines(&emit_line)) + execute_emulated(script, opts.pkg_root, env, EmulatedOutput::Lines(&emit_line), None) .map(ScriptExit::Emulated) .map_err(LifecycleScriptError::ShellEmulator) } diff --git a/pnpm/crates/executor/src/process_tracker.rs b/pnpm/crates/executor/src/process_tracker.rs new file mode 100644 index 0000000000..04a7db35fb --- /dev/null +++ b/pnpm/crates/executor/src/process_tracker.rs @@ -0,0 +1,276 @@ +use std::{ + collections::HashMap, + io, + process::{Child, Command}, + sync::Mutex, +}; +use tokio::sync::watch; + +#[cfg(unix)] +use std::{io::Read, os::unix::process::CommandExt, process::Stdio, time::Duration}; + +/// Tracks the processes started by one recursive command so a bailing task +/// can stop the other work that is still in flight. +#[derive(Default)] +pub struct ProcessTracker { + state: Mutex, +} + +#[derive(Default)] +struct TrackerState { + cancelled: bool, + next_id: usize, + executions: HashMap, +} + +#[derive(Clone)] +enum RunningExecution { + Process(u32), + Emulated(watch::Sender), +} + +impl RunningExecution { + fn cancel(&self) { + match self { + Self::Process(pid) => terminate_process_tree(*pid), + Self::Emulated(sender) => { + let _ = sender.send(true); + } + } + } +} + +impl ProcessTracker { + /// Cancel every registered execution. Returns `true` only to the caller + /// that initiated cancellation. + pub fn cancel(&self) -> bool { + let executions = { + let mut state = self.state.lock().expect("process tracker lock is not poisoned"); + if state.cancelled { + return false; + } + state.cancelled = true; + state.executions.values().cloned().collect::>() + }; + #[cfg(unix)] + let descendants = descendant_processes(std::process::id()); + for execution in executions { + execution.cancel(); + } + #[cfg(unix)] + for pid in descendants { + terminate_process(pid); + } + true + } + + #[must_use] + pub fn is_cancelled(&self) -> bool { + self.state.lock().expect("process tracker lock is not poisoned").cancelled + } + + pub(crate) fn track_emulated(&self) -> EmulatedCancellation<'_> { + let (sender, receiver) = watch::channel(false); + let registration = self.register(RunningExecution::Emulated(sender)); + EmulatedCancellation { receiver, _registration: registration } + } + + fn register(&self, execution: RunningExecution) -> Registration<'_> { + let mut state = self.state.lock().expect("process tracker lock is not poisoned"); + if state.cancelled { + drop(state); + execution.cancel(); + return Registration { tracker: self, id: None }; + } + let id = state.next_id; + state.next_id += 1; + state.executions.insert(id, execution); + Registration { tracker: self, id: Some(id) } + } +} + +/// Spawn a child and optionally register it for recursive-command +/// cancellation. On Unix a tracked child starts a process group so +/// cancellation also reaches descendants. +pub fn spawn_child<'tracker>( + command: &mut Command, + process_tracker: Option<&'tracker ProcessTracker>, +) -> io::Result> { + if process_tracker.is_some() { + prepare_command(command); + } + let child = command.spawn()?; + let registration = + process_tracker.map(|tracker| tracker.register(RunningExecution::Process(child.id()))); + Ok(SpawnedChild { child, _registration: registration }) +} + +pub struct SpawnedChild<'tracker> { + child: Child, + _registration: Option>, +} + +impl SpawnedChild<'_> { + pub fn child_mut(&mut self) -> &mut Child { + &mut self.child + } + + pub fn wait(&mut self) -> io::Result { + self.child.wait() + } +} + +pub(crate) struct EmulatedCancellation<'tracker> { + receiver: watch::Receiver, + _registration: Registration<'tracker>, +} + +impl EmulatedCancellation<'_> { + pub(crate) fn receiver(&self) -> watch::Receiver { + self.receiver.clone() + } +} + +struct Registration<'tracker> { + tracker: &'tracker ProcessTracker, + id: Option, +} + +impl Drop for Registration<'_> { + fn drop(&mut self) { + let Some(id) = self.id else { return }; + self.tracker + .state + .lock() + .expect("process tracker lock is not poisoned") + .executions + .remove(&id); + } +} + +#[cfg(unix)] +fn prepare_command(command: &mut Command) { + command.process_group(0); +} + +#[cfg(not(unix))] +fn prepare_command(_: &mut Command) {} + +#[cfg(unix)] +fn terminate_process_tree(pid: u32) { + let Ok(pid) = i32::try_from(pid) else { return }; + // SAFETY: a negative PID asks `kill` to signal the process group whose + // ID is the tracked child's PID. The child was placed in that group by + // `prepare_command`; ESRCH is harmless when it exited concurrently. + unsafe { + libc::kill(-pid, libc::SIGKILL); + } +} + +#[cfg(unix)] +fn terminate_process(pid: i32) { + // SAFETY: every PID comes from the OS process list as a descendant of + // this process. ESRCH is harmless when it exited concurrently. + unsafe { + libc::kill(pid, libc::SIGKILL); + } +} + +#[cfg(unix)] +fn descendant_processes(root: u32) -> Vec { + let mut command = Command::new("/bin/ps"); + command.args(["-A", "-o", "pid=", "-o", "ppid="]).stdout(Stdio::piped()); + let Ok(mut child) = command.spawn() else { + return Vec::new(); + }; + let Some(mut stdout) = child.stdout.take() else { + return Vec::new(); + }; + let output = std::thread::spawn(move || { + let mut listing = String::new(); + stdout.read_to_string(&mut listing).map(|_| listing) + }); + let mut completed = false; + for _ in 0..50 { + match child.try_wait() { + Ok(Some(_)) => { + completed = true; + break; + } + Ok(None) => std::thread::sleep(Duration::from_millis(10)), + Err(_) => break, + } + } + if !completed { + let _ = child.kill(); + let _ = child.wait(); + } + let Ok(Ok(listing)) = output.join() else { + return Vec::new(); + }; + if !completed { + return Vec::new(); + } + + let mut children: HashMap> = HashMap::new(); + for line in listing.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(), parent.parse()) else { + continue; + }; + children.entry(parent).or_default().push(pid); + } + let mut descendants = Vec::new(); + let mut stack = vec![root]; + while let Some(parent) = stack.pop() { + for &pid in children.get(&parent).into_iter().flatten() { + stack.push(pid); + if let Ok(pid) = i32::try_from(pid) { + descendants.push(pid); + } + } + } + descendants +} + +#[cfg(windows)] +fn terminate_process_tree(pid: u32) { + use std::{os::windows::process::CommandExt, process::Stdio}; + + let Some(taskkill) = taskkill_path() else { return }; + let _ = Command::new(taskkill) + .args(["/pid", &pid.to_string(), "/T", "/F"]) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .creation_flags(0x0800_0000) + .spawn(); +} + +#[cfg(windows)] +fn taskkill_path() -> Option { + use std::{ffi::OsString, os::windows::ffi::OsStringExt, ptr}; + use windows_sys::Win32::System::SystemInformation::GetSystemDirectoryW; + + // SAFETY: the first call requests the required UTF-16 buffer length. + // The second call writes into a buffer of that size, and its returned + // length is checked before constructing the path. + unsafe { + let required = GetSystemDirectoryW(ptr::null_mut(), 0); + if required == 0 { + return None; + } + let mut buffer = vec![0_u16; required as usize]; + let length = GetSystemDirectoryW(buffer.as_mut_ptr(), required); + if length == 0 || length >= required { + return None; + } + Some( + std::path::PathBuf::from(OsString::from_wide(&buffer[..length as usize])) + .join("taskkill.exe"), + ) + } +} diff --git a/pnpm/crates/executor/src/run_script.rs b/pnpm/crates/executor/src/run_script.rs index b1dacf8bb7..5831177465 100644 --- a/pnpm/crates/executor/src/run_script.rs +++ b/pnpm/crates/executor/src/run_script.rs @@ -2,6 +2,7 @@ use crate::{ extend_path::{ScriptsPrependNodePath, extend_path}, lifecycle::{StreamedScript, push_script_arg}, make_env::{EnvOptions, build_env, path_value}, + process_tracker::{ProcessTracker, spawn_child}, script_exit::ScriptExit, shell::{ScriptShellError, SelectedShell, select_shell}, shell_emulator::{EmulatedOutput, ShellEmulatorError, execute_emulated}, @@ -103,6 +104,8 @@ pub struct RunScript<'a> { pub silent: bool, /// Where the script's output goes. pub output: ScriptOutput<'a>, + /// Tracks this script for cancellation when a recursive command bails. + pub process_tracker: Option<&'a ProcessTracker>, } /// Run a single user script in the foreground, sending its output where @@ -154,11 +157,17 @@ pub fn run_script(opts: &RunScript<'_>) -> Result { streamed.started(&command); let status = if opts.shell_emulator { let emit_line = |stdio, line| streamed.emit_line(stdio, line); - execute_emulated(&command, opts.pkg_root, &child_env, EmulatedOutput::Lines(&emit_line)) - .map(ScriptExit::Emulated) - .map_err(RunScriptError::ShellEmulator)? + execute_emulated( + &command, + opts.pkg_root, + &child_env, + EmulatedOutput::Lines(&emit_line), + opts.process_tracker, + ) + .map(ScriptExit::Emulated) + .map_err(RunScriptError::ShellEmulator)? } else { - run_piped(&shell, &command, opts.pkg_root, &child_env, streamed)? + run_piped(&shell, &command, opts.pkg_root, &child_env, streamed, opts.process_tracker)? }; streamed.finished(status.code().unwrap_or(-1)); return Ok(status); @@ -172,9 +181,15 @@ pub fn run_script(opts: &RunScript<'_>) -> Result { } if opts.shell_emulator { - return execute_emulated(&command, opts.pkg_root, &child_env, EmulatedOutput::Inherit) - .map(ScriptExit::Emulated) - .map_err(RunScriptError::ShellEmulator); + return execute_emulated( + &command, + opts.pkg_root, + &child_env, + EmulatedOutput::Inherit, + opts.process_tracker, + ) + .map(ScriptExit::Emulated) + .map_err(RunScriptError::ShellEmulator); } // The script is appended through `push_script_arg` (not a chained @@ -184,12 +199,11 @@ pub fn run_script(opts: &RunScript<'_>) -> Result { let mut cmd = Command::new(&shell.program); cmd.args(&shell.args); push_script_arg(&mut cmd, &command, shell.windows_verbatim_args); - let status = cmd - .current_dir(opts.pkg_root) - .env_clear() - .envs(&child_env) - .status() + cmd.current_dir(opts.pkg_root).env_clear().envs(&child_env); + let mut child = spawn_child(&mut cmd, opts.process_tracker) .map_err(|source| RunScriptError::Spawn { script: command.clone(), source })?; + let status = + child.wait().map_err(|source| RunScriptError::Wait { script: command.clone(), source })?; Ok(ScriptExit::Process(status)) } @@ -202,21 +216,21 @@ fn run_piped( pkg_root: &Path, child_env: &HashMap, streamed: StreamedScript<'_>, + process_tracker: Option<&ProcessTracker>, ) -> Result { let mut cmd = Command::new(&shell.program); cmd.args(&shell.args); push_script_arg(&mut cmd, command, shell.windows_verbatim_args); - let mut child = cmd - .current_dir(pkg_root) + cmd.current_dir(pkg_root) .env_clear() .envs(child_env) .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn() + .stderr(Stdio::piped()); + let mut child = spawn_child(&mut cmd, process_tracker) .map_err(|source| RunScriptError::Spawn { script: command.to_string(), source })?; streamed - .pump(&mut child) + .pump(child.child_mut()) .map(ScriptExit::Process) .map_err(|source| RunScriptError::Wait { script: command.to_string(), source }) } diff --git a/pnpm/crates/executor/src/run_script/tests.rs b/pnpm/crates/executor/src/run_script/tests.rs index 225a0a0e54..3e4e7ba0d1 100644 --- a/pnpm/crates/executor/src/run_script/tests.rs +++ b/pnpm/crates/executor/src/run_script/tests.rs @@ -56,6 +56,7 @@ fn run(pkg_root: &Path, stage: &str, script: &str, args: &[String]) -> ScriptExi extra_env: &extra_env, silent: true, output: ScriptOutput::Inherit, + process_tracker: None, }) .expect("run the script") } diff --git a/pnpm/crates/executor/src/shell_emulator.rs b/pnpm/crates/executor/src/shell_emulator.rs index 16eaddbb07..8ffb628278 100644 --- a/pnpm/crates/executor/src/shell_emulator.rs +++ b/pnpm/crates/executor/src/shell_emulator.rs @@ -1,6 +1,6 @@ use deno_task_shell::{ - KillSignal, ShellPipeReader, ShellPipeWriter, ShellState, execute_with_pipes, parser, - parser::SequentialList, pipe, + KillSignal, ShellPipeReader, ShellPipeWriter, ShellState, SignalKind, execute_with_pipes, + parser, parser::SequentialList, pipe, }; use derive_more::{Display, Error}; use miette::Diagnostic; @@ -14,6 +14,8 @@ use std::{ }; use tokio::{runtime::Builder, task::LocalSet}; +use crate::process_tracker::{EmulatedCancellation, ProcessTracker}; + /// Failure to run a script under the `shellEmulator` setting. A script /// that runs and exits non-zero is not an error here — the exit code is /// returned to the caller, which decides what a failure means. @@ -56,6 +58,7 @@ pub fn execute_emulated( cwd: &Path, env: &HashMap, output: EmulatedOutput<'_>, + process_tracker: Option<&ProcessTracker>, ) -> Result { let list = parser::parse(script).map_err(|error| ShellEmulatorError::Parse { script: script.to_string(), @@ -67,6 +70,8 @@ pub fn execute_emulated( let cwd = path::absolute(cwd) .map_err(|source| ShellEmulatorError::Start { script: script.to_string(), source })?; let env = env.iter().map(|(key, value)| (OsString::from(key), OsString::from(value))).collect(); + let cancellation = process_tracker.map(ProcessTracker::track_emulated); + let receiver = cancellation.as_ref().map(EmulatedCancellation::receiver); match output { EmulatedOutput::Inherit => run_to_completion( @@ -76,6 +81,7 @@ pub fn execute_emulated( cwd, ShellPipeWriter::stdout(), ShellPipeWriter::stderr(), + receiver, ), EmulatedOutput::Lines(sink) => thread::scope(|scope| { let (stdout_reader, stdout_writer) = pipe(); @@ -87,7 +93,8 @@ pub fn execute_emulated( // Both writers are consumed by the run, so the pumps see EOF // as soon as it returns and the joins below finish promptly. - let code = run_to_completion(script, list, env, cwd, stdout_writer, stderr_writer); + let code = + run_to_completion(script, list, env, cwd, stdout_writer, stderr_writer, receiver); let _ = stdout_pump.join(); let _ = stderr_pump.join(); code @@ -108,13 +115,23 @@ fn run_to_completion( cwd: PathBuf, stdout: ShellPipeWriter, stderr: ShellPipeWriter, + cancellation: Option>, ) -> Result { let run = thread::spawn(move || { let runtime = Builder::new_current_thread().enable_all().build()?; - let state = ShellState::new(env, cwd, HashMap::new(), KillSignal::default()); + let kill_signal = KillSignal::default(); + let local_set = LocalSet::new(); + if let Some(mut cancellation) = cancellation { + let cancellation_signal = kill_signal.clone(); + local_set.spawn_local(async move { + if *cancellation.borrow_and_update() || cancellation.changed().await.is_ok() { + cancellation_signal.send(SignalKind::SIGKILL); + } + }); + } + let state = ShellState::new(env, cwd, HashMap::new(), kill_signal); let stdin = ShellPipeReader::stdin(); - Ok(LocalSet::new() - .block_on(&runtime, execute_with_pipes(list, state, stdin, stdout, stderr))) + Ok(local_set.block_on(&runtime, execute_with_pipes(list, state, stdin, stdout, stderr))) }); match run.join() { diff --git a/pnpm/crates/executor/src/shell_emulator/tests.rs b/pnpm/crates/executor/src/shell_emulator/tests.rs index 2e2478124b..d9b55880f5 100644 --- a/pnpm/crates/executor/src/shell_emulator/tests.rs +++ b/pnpm/crates/executor/src/shell_emulator/tests.rs @@ -12,7 +12,7 @@ fn run( ) -> (i32, Vec<(LifecycleStdio, String)>) { let lines = Mutex::new(Vec::new()); let sink = |stdio, line| lines.lock().expect("the sink is never poisoned").push((stdio, line)); - let code = execute_emulated(script, cwd, env, EmulatedOutput::Lines(&sink)) + let code = execute_emulated(script, cwd, env, EmulatedOutput::Lines(&sink), None) .expect("run the script under the emulator"); (code, lines.into_inner().expect("the sink is never poisoned")) } @@ -104,6 +104,7 @@ fn rejects_a_script_the_shell_cannot_parse() { dir.path(), &HashMap::new(), EmulatedOutput::Inherit, + None, ) .expect_err("an unparsable script is an error, not an exit code"); dbg!(&error); diff --git a/pnpm/crates/workspace-task-scheduler/src/lib.rs b/pnpm/crates/workspace-task-scheduler/src/lib.rs index 5dbae237a3..cc1f268b00 100644 --- a/pnpm/crates/workspace-task-scheduler/src/lib.rs +++ b/pnpm/crates/workspace-task-scheduler/src/lib.rs @@ -433,6 +433,8 @@ pub fn render_task_graph_dry_run( pub enum TaskCompletion { Passed, Failed, + /// The task was interrupted because another task caused the run to bail. + Cancelled, /// The task's work errored before it could run — an infrastructure /// failure, not a script failure. Stops dispatch like a bail. Aborted, @@ -779,7 +781,7 @@ where block(&mut guard, index); } } - TaskCompletion::Aborted => { + TaskCompletion::Aborted | TaskCompletion::Cancelled => { guard.settled[index] = true; guard.unsettled -= 1; guard.stop_dispatch = true; @@ -860,7 +862,7 @@ pub async fn schedule_graph_async( } } TaskCompletion::Failed if options.bail => stop_dispatch = true, - TaskCompletion::Aborted => stop_dispatch = true, + TaskCompletion::Aborted | TaskCompletion::Cancelled => stop_dispatch = true, TaskCompletion::Failed if options.continue_on_failure => { for &dependent in &dependents[index] { pending_dependencies[dependent] -= 1;