fix(task): forward signals to spawned sub-processes on unix (#27141)

Closes https://github.com/denoland/deno/issues/18445

cli/task_runner.rs

@@ -14,6 +14,7 @@ use deno_runtime::deno_node::NodeResolver;
 use deno_semver::package::PackageNv;
 use deno_task_shell::ExecutableCommand;
 use deno_task_shell::ExecuteResult;
+use deno_task_shell::KillSignal;
 use deno_task_shell::ShellCommand;
 use deno_task_shell::ShellCommandContext;
 use deno_task_shell::ShellPipeReader;
@@ -22,6 +23,7 @@ use lazy_regex::Lazy;
 use regex::Regex;
 use tokio::task::JoinHandle;
 use tokio::task::LocalSet;
+use tokio_util::sync::CancellationToken;
 
 use crate::npm::CliNpmResolver;
 use crate::npm::InnerCliNpmResolverRef;
@@ -45,9 +47,11 @@ impl TaskStdio {
   pub fn stdout() -> Self {
     Self(None, ShellPipeWriter::stdout())
   }
+
   pub fn stderr() -> Self {
     Self(None, ShellPipeWriter::stderr())
   }
+
   pub fn piped() -> Self {
     let (r, w) = deno_task_shell::pipe();
     Self(Some(r), w)
@@ -62,8 +66,8 @@ pub struct TaskIo {
 impl Default for TaskIo {
   fn default() -> Self {
     Self {
-      stderr: TaskStdio::stderr(),
       stdout: TaskStdio::stdout(),
+      stderr: TaskStdio::stderr(),
     }
   }
 }
@@ -78,6 +82,7 @@ pub struct RunTaskOptions<'a> {
   pub custom_commands: HashMap<String, Rc<dyn ShellCommand>>,
   pub root_node_modules_dir: Option<&'a Path>,
   pub stdio: Option<TaskIo>,
+  pub kill_signal: KillSignal,
 }
 
 pub type TaskCustomCommands = HashMap<String, Rc<dyn ShellCommand>>;
@@ -96,8 +101,12 @@ pub async fn run_task(
     .with_context(|| format!("Error parsing script '{}'.", opts.task_name))?;
   let env_vars =
     prepare_env_vars(opts.env_vars, opts.init_cwd, opts.root_node_modules_dir);
-  let state =
-    deno_task_shell::ShellState::new(env_vars, opts.cwd, opts.custom_commands);
+  let state = deno_task_shell::ShellState::new(
+    env_vars,
+    opts.cwd,
+    opts.custom_commands,
+    opts.kill_signal,
+  );
   let stdio = opts.stdio.unwrap_or_default();
   let (
     TaskStdio(stdout_read, stdout_write),
@@ -537,6 +546,80 @@ fn resolve_managed_npm_commands(
   Ok(result)
 }
 
+/// Runs a deno task future forwarding any signals received
+/// to the process.
+///
+/// Signal listeners and ctrl+c listening will be setup.
+pub async fn run_future_forwarding_signals<TOutput>(
+  kill_signal: KillSignal,
+  future: impl std::future::Future<Output = TOutput>,
+) -> TOutput {
+  fn spawn_future_with_cancellation(
+    future: impl std::future::Future<Output = ()> + 'static,
+    token: CancellationToken,
+  ) {
+    deno_core::unsync::spawn(async move {
+      tokio::select! {
+        _ = future => {}
+        _ = token.cancelled() => {}
+      }
+    });
+  }
+
+  let token = CancellationToken::new();
+  let _token_drop_guard = token.clone().drop_guard();
+  let _drop_guard = kill_signal.clone().drop_guard();
+
+  spawn_future_with_cancellation(
+    listen_ctrl_c(kill_signal.clone()),
+    token.clone(),
+  );
+  #[cfg(unix)]
+  spawn_future_with_cancellation(
+    listen_and_forward_all_signals(kill_signal),
+    token,
+  );
+
+  future.await
+}
+
+async fn listen_ctrl_c(kill_signal: KillSignal) {
+  while let Ok(()) = tokio::signal::ctrl_c().await {
+    kill_signal.send(deno_task_shell::SignalKind::SIGINT)
+  }
+}
+
+#[cfg(unix)]
+async fn listen_and_forward_all_signals(kill_signal: KillSignal) {
+  use deno_core::futures::FutureExt;
+  use deno_runtime::signal::SIGNAL_NUMS;
+
+  // listen and forward every signal we support
+  let mut futures = Vec::with_capacity(SIGNAL_NUMS.len());
+  for signo in SIGNAL_NUMS.iter().copied() {
+    if signo == libc::SIGKILL || signo == libc::SIGSTOP {
+      continue; // skip, can't listen to these
+    }
+
+    let kill_signal = kill_signal.clone();
+    futures.push(
+      async move {
+        let Ok(mut stream) = tokio::signal::unix::signal(
+          tokio::signal::unix::SignalKind::from_raw(signo),
+        ) else {
+          return;
+        };
+        let signal_kind: deno_task_shell::SignalKind = signo.into();
+        while let Some(()) = stream.recv().await {
+          kill_signal.send(signal_kind);
+        }
+      }
+      .boxed_local(),
+    )
+  }
+  futures::future::join_all(futures).await;
+}
+
 #[cfg(test)]
 mod test {