chore(signals): prefactor (#9995)

### Description

Prefactors:
 - moved signal listening to own crate
 - return signal received to subscribers
 - support streams instead of single futures for signal sources

PR that prefactors the signals code to prepare for:
 - Subscribers getting all signals sent to `turbo`
 - Additional signal handlers to be sent transparently to children

There are no behavior changes in this PR. I highly suggest reviewing
each commit on its own.

### Testing Instructions

Existing unit tests.


Quick manual verification that existing `SIGINT` behavior remains the
same:
```
$ turbo_dev watch build
...
│    Finalizing page optimization  .   Collecting build traces  .
└────>
  × Watch interrupted due to signal
```

Author: chris-olszewski

Cargo.lock

@@ -65,6 +65,7 @@ turborepo-repository = { path = "crates/turborepo-repository" }
 turborepo-ui = { path = "crates/turborepo-ui" }
 turborepo-unescape = { path = "crates/turborepo-unescape" }
 turborepo-scm = { path = "crates/turborepo-scm" }
+turborepo-signals = { path = "crates/turborepo-signals" }
 wax = { path = "crates/turborepo-wax" }
 turborepo-vercel-api = { path = "crates/turborepo-vercel-api" }
 turborepo-vercel-api-mock = { path = "crates/turborepo-vercel-api-mock" }

Cargo.toml

@@ -139,6 +139,7 @@ turborepo-lockfiles = { workspace = true }
 turborepo-microfrontends = { workspace = true }
 turborepo-repository = { path = "../turborepo-repository" }
 turborepo-scm = { workspace = true }
+turborepo-signals = { workspace = true }
 turborepo-telemetry = { path = "../turborepo-telemetry" }
 turborepo-ui = { workspace = true }
 turborepo-unescape = { workspace = true }

crates/turborepo-lib/Cargo.toml

@@ -4,17 +4,17 @@ use itertools::Itertools;
 use miette::Diagnostic;
 use thiserror::Error;
 use turborepo_repository::package_graph;
+use turborepo_signals::{listeners::get_signal, SignalHandler};
 use turborepo_telemetry::events::command::CommandEventBuilder;
 use turborepo_ui::{color, BOLD, GREY};
 
 use crate::{
-    commands::{bin, generate, link, login, ls, prune, run::get_signal, CommandBase},
+    commands::{bin, generate, link, login, ls, prune, CommandBase},
     daemon::DaemonError,
     query,
     rewrite_json::RewriteError,
     run,
     run::{builder::RunBuilder, watch},
-    signal::SignalHandler,
 };
 
 #[derive(Debug, Error, Diagnostic)]
@@ -78,7 +78,7 @@ pub async fn print_potential_tasks(
     base: CommandBase,
     telemetry: CommandEventBuilder,
 ) -> Result<(), Error> {
-    let signal = get_signal()?;
+    let signal = get_signal().map_err(run::Error::from)?;
     let handler = SignalHandler::new(signal);
     let color_config = base.color_config;
 

crates/turborepo-lib/src/cli/error.rs

@@ -1,14 +1,10 @@
+use turborepo_signals::{listeners::get_signal, SignalHandler};
 use turborepo_telemetry::events::command::CommandEventBuilder;
 
-use crate::{
-    cli,
-    commands::{run::get_signal, CommandBase},
-    run::builder::RunBuilder,
-    signal::SignalHandler,
-};
+use crate::{cli, commands::CommandBase, run::builder::RunBuilder};
 
 pub async fn run(base: CommandBase, telemetry: CommandEventBuilder) -> Result<i32, cli::Error> {
-    let signal = get_signal()?;
+    let signal = get_signal().map_err(crate::run::Error::from)?;
     let handler = SignalHandler::new(signal);
 
     let run = RunBuilder::new(base)?

crates/turborepo-lib/src/commands/boundaries.rs

@@ -5,15 +5,15 @@ use serde::Serialize;
 use thiserror::Error;
 use turbopath::AnchoredSystemPath;
 use turborepo_repository::package_graph::{PackageName, PackageNode};
+use turborepo_signals::{listeners::get_signal, SignalHandler};
 use turborepo_telemetry::events::command::CommandEventBuilder;
 use turborepo_ui::{color, cprint, cprintln, ColorConfig, BOLD, BOLD_GREEN, GREY};
 
 use crate::{
     cli,
     cli::OutputFormat,
-    commands::{run::get_signal, CommandBase},
+    commands::CommandBase,
     run::{builder::RunBuilder, Run},
-    signal::SignalHandler,
 };
 
 #[derive(Debug, Error, Diagnostic)]
@@ -115,7 +115,7 @@ pub async fn run(
     telemetry: CommandEventBuilder,
     output: Option<OutputFormat>,
 ) -> Result<(), cli::Error> {
-    let signal = get_signal()?;
+    let signal = get_signal().map_err(crate::run::Error::from)?;
     let handler = SignalHandler::new(signal);
 
     let run_builder = RunBuilder::new(base)?;

crates/turborepo-lib/src/commands/ls.rs

@@ -5,14 +5,14 @@ use camino::Utf8Path;
 use miette::{Diagnostic, Report, SourceSpan};
 use thiserror::Error;
 use turbopath::AbsoluteSystemPathBuf;
+use turborepo_signals::{listeners::get_signal, SignalHandler};
 use turborepo_telemetry::events::command::CommandEventBuilder;
 
 use crate::{
-    commands::{run::get_signal, CommandBase},
+    commands::CommandBase,
     query,
     query::{Error, RepositoryQuery},
     run::builder::RunBuilder,
-    signal::SignalHandler,
 };
 
 const SCHEMA_QUERY: &str = "query IntrospectionQuery {
@@ -161,7 +161,7 @@ pub async fn run(
     variables_path: Option<&Utf8Path>,
     include_schema: bool,
 ) -> Result<i32, Error> {
-    let signal = get_signal()?;
+    let signal = get_signal().map_err(crate::run::Error::from)?;
     let handler = SignalHandler::new(signal);
 
     let run_builder = RunBuilder::new(base)?

crates/turborepo-lib/src/commands/query.rs

@@ -1,36 +1,11 @@
-use std::{future::Future, sync::Arc};
+use std::sync::Arc;
 
 use tracing::error;
+use turborepo_signals::{listeners::get_signal, SignalHandler};
 use turborepo_telemetry::events::command::CommandEventBuilder;
 use turborepo_ui::sender::UISender;
 
-use crate::{commands::CommandBase, run, run::builder::RunBuilder, signal::SignalHandler};
-
-#[cfg(windows)]
-pub fn get_signal() -> Result<impl Future<Output = Option<()>>, run::Error> {
-    let mut ctrl_c = tokio::signal::windows::ctrl_c().map_err(run::Error::SignalHandler)?;
-    Ok(async move { ctrl_c.recv().await })
-}
-
-#[cfg(not(windows))]
-pub fn get_signal() -> Result<impl Future<Output = Option<()>>, run::Error> {
-    use tokio::signal::unix;
-    let mut sigint =
-        unix::signal(unix::SignalKind::interrupt()).map_err(run::Error::SignalHandler)?;
-    let mut sigterm =
-        unix::signal(unix::SignalKind::terminate()).map_err(run::Error::SignalHandler)?;
-
-    Ok(async move {
-        tokio::select! {
-            res = sigint.recv() => {
-                res
-            }
-            res = sigterm.recv() => {
-                res
-            }
-        }
-    })
-}
+use crate::{commands::CommandBase, run, run::builder::RunBuilder};
 
 pub async fn run(base: CommandBase, telemetry: CommandEventBuilder) -> Result<i32, run::Error> {
     let signal = get_signal()?;

crates/turborepo-lib/src/commands/run.rs

@@ -34,7 +34,6 @@ mod query;
 mod rewrite_json;
 mod run;
 mod shim;
-mod signal;
 mod task_graph;
 mod task_hash;
 mod tracing;

crates/turborepo-lib/src/lib.rs

@@ -24,12 +24,12 @@ use tokio::select;
 use turbo_trace::TraceError;
 use turbopath::AbsoluteSystemPathBuf;
 use turborepo_repository::{change_mapper::AllPackageChangeReason, package_graph::PackageName};
+use turborepo_signals::SignalHandler;
 
 use crate::{
     get_version,
     query::{file::File, task::RepositoryTask},
     run::{builder::RunBuilder, Run},
-    signal::SignalHandler,
 };
 
 #[derive(Error, Debug, miette::Diagnostic)]

crates/turborepo-lib/src/query/mod.rs

@@ -20,6 +20,7 @@ use turborepo_repository::{
     package_json::PackageJson,
 };
 use turborepo_scm::SCM;
+use turborepo_signals::{SignalHandler, SignalSubscriber};
 use turborepo_telemetry::events::{
     command::CommandEventBuilder,
     generic::{DaemonInitStatus, GenericEventBuilder},
@@ -46,7 +47,6 @@ use crate::{
     process::ProcessManager,
     run::{scope, task_access::TaskAccess, task_id::TaskName, Error, Run, RunCache},
     shim::TurboState,
-    signal::{SignalHandler, SignalSubscriber},
     turbo_json::{TurboJson, TurboJsonLoader, UIMode},
     DaemonConnector,
 };

crates/turborepo-lib/src/run/builder.rs

@@ -52,8 +52,8 @@ pub enum Error {
     #[error(transparent)]
     #[diagnostic(transparent)]
     Visitor(#[from] task_graph::VisitorError),
-    #[error("Failed to register signal handler: {0}")]
-    SignalHandler(std::io::Error),
+    #[error(transparent)]
+    SignalHandler(#[from] turborepo_signals::listeners::Error),
     #[error(transparent)]
     Daemon(#[from] daemon::DaemonError),
     #[error(transparent)]

crates/turborepo-lib/src/run/error.rs

@@ -22,15 +22,17 @@ use std::{
 
 pub use cache::{CacheOutput, ConfigCache, Error as CacheError, RunCache, TaskCache};
 use chrono::{DateTime, Local};
+use futures::StreamExt;
 use rayon::iter::ParallelBridge;
-use tokio::{select, task::JoinHandle};
+use tokio::{pin, select, task::JoinHandle};
 use tracing::{debug, instrument};
 use turbopath::{AbsoluteSystemPath, AbsoluteSystemPathBuf};
 use turborepo_api_client::{APIAuth, APIClient};
 use turborepo_ci::Vendor;
 use turborepo_env::EnvironmentVariableMap;
 use turborepo_repository::package_graph::{PackageGraph, PackageName, PackageNode};
 use turborepo_scm::SCM;
+use turborepo_signals::{listeners::get_signal, SignalHandler};
 use turborepo_telemetry::events::generic::GenericEventBuilder;
 use turborepo_ui::{
     cprint, cprintln, sender::UISender, tui, tui::TuiSender, wui::sender::WebUISender, ColorConfig,
@@ -45,7 +47,6 @@ use crate::{
     opts::Opts,
     process::ProcessManager,
     run::{global_hash::get_global_hash_inputs, summary::RunTracker, task_access::TaskAccess},
-    signal::SignalHandler,
     task_graph::Visitor,
     task_hash::{get_external_deps_hash, get_internal_deps_hash, PackageInputsHashes},
     turbo_json::{TurboJson, TurboJsonLoader, UIMode},
@@ -336,8 +337,9 @@ impl Run {
                     };
 
                     let interrupt = async {
-                        if let Ok(fut) = crate::commands::run::get_signal() {
-                            fut.await;
+                        if let Ok(fut) = get_signal() {
+                            pin!(fut);
+                            fut.next().await;
                         } else {
                             tracing::warn!("could not register ctrl-c handler");
                             // wait forever

crates/turborepo-lib/src/run/mod.rs

@@ -10,15 +10,15 @@ use thiserror::Error;
 use tokio::{select, sync::Notify, task::JoinHandle};
 use tracing::{instrument, trace, warn};
 use turborepo_repository::package_graph::PackageName;
+use turborepo_signals::{listeners::get_signal, SignalHandler};
 use turborepo_telemetry::events::command::CommandEventBuilder;
 use turborepo_ui::sender::UISender;
 
 use crate::{
-    commands::{self, CommandBase},
+    commands::CommandBase,
     daemon::{proto, DaemonConnectorError, DaemonError},
     get_version, opts,
     run::{self, builder::RunBuilder, scope::target_selector::InvalidSelectorError, Run},
-    signal::SignalHandler,
     turbo_json::CONFIG_FILE,
     DaemonConnector, DaemonPaths,
 };
@@ -115,7 +115,7 @@ impl WatchClient {
         experimental_write_cache: bool,
         telemetry: CommandEventBuilder,
     ) -> Result<Self, Error> {
-        let signal = commands::run::get_signal()?;
+        let signal = get_signal().map_err(crate::run::Error::from)?;
         let handler = SignalHandler::new(signal);
 
         if base.opts.repo_opts.root_turbo_json_path != base.repo_root.join_component(CONFIG_FILE) {

crates/turborepo-lib/src/run/watch.rs

@@ -0,0 +1,15 @@
+[package]
+name = "turborepo-signals"
+version = "0.1.0"
+edition = "2021"
+license = "MIT"
+
+[dependencies]
+futures = "0.3.30"
+thiserror = { workspace = true }
+tokio = { workspace = true, features = ["full", "time"] }
+
+[dev-dependencies]
+
+[lints]
+workspace = true