fix bad SIGINT patterns

integration-tests/lib/mining_device/mod.rs

@@ -37,6 +37,7 @@ use stratum_apps::{
     sync::SharedLock,
 };
 use tokio::net::TcpStream;
+use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, info};
 
 // Fast SHA256d midstate hasher
@@ -134,9 +135,13 @@ pub async fn connect(
     info!("Pool tcp connection established at {}", address);
     let address = socket.peer_addr().unwrap();
     let initiator = Initiator::new(pub_key.map(|e| e.0));
-    let (receiver, sender) = Connection::new(socket, HandshakeRole::Initiator(initiator))
-        .await
-        .unwrap();
+    let (receiver, sender) = Connection::new(
+        socket,
+        HandshakeRole::Initiator(initiator),
+        CancellationToken::new(),
+    )
+    .await
+    .unwrap();
     info!("Pool noise connection established at {}", address);
     Device::start(
         receiver,

integration-tests/lib/sniffer.rs

@@ -107,7 +107,6 @@ impl<'a> Sniffer<'a> {
             .await
             .expect("Failed to create upstream");
             select! {
-                _ = tokio::signal::ctrl_c() => { },
                 _ = recv_from_down_send_to_up(downstream_receiver, upstream_sender, messages_from_downstream, action.clone(), &identifier, negotiated_extensions.clone()) => { },
                 _ = recv_from_up_send_to_down(upstream_receiver, downstream_sender, messages_from_upstream, action, &identifier, negotiated_extensions.clone()) => { },
             };

integration-tests/lib/sv1_sniffer.rs

@@ -77,7 +77,6 @@ impl SnifferSV1 {
             let downstream_to_sniffer_connection =
                 ConnectionSV1::new(downstream_stream, CancellationToken::new()).await;
             select! {
-                _ = tokio::signal::ctrl_c() => { },
                 _ = Self::recv_from_down_send_to_up_sv1(
                     downstream_to_sniffer_connection.receiver(),
                     sniffer_to_upstream_connection.sender(),
@@ -98,67 +97,54 @@ impl SnifferSV1 {
             panic!("Message cannot be empty");
         }
         let now = std::time::Instant::now();
-        tokio::select!(
-            _ = tokio::signal::ctrl_c() => { },
-            _ = async {
-                loop {
-                    match direction {
-                        MessageDirection::ToUpstream => {
-                            if self.messages_from_downstream.has_message(message).await {
-                                break;
-                            }
-                        }
-                        MessageDirection::ToDownstream => {
-                            if self.messages_from_upstream.has_message(message).await {
-                                break;
-                            }
-                        }
+        loop {
+            match direction {
+                MessageDirection::ToUpstream => {
+                    if self.messages_from_downstream.has_message(message).await {
+                        break;
                     }
-                    if now.elapsed().as_secs() > 60 {
-                        panic!( "Timeout: SV1 message {} not found", message.first().unwrap());
-                    } else {
-                        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
-                        continue;
+                }
+                MessageDirection::ToDownstream => {
+                    if self.messages_from_upstream.has_message(message).await {
+                        break;
                     }
                 }
-            } => {}
-        );
+            }
+            if now.elapsed().as_secs() > 60 {
+                panic!(
+                    "Timeout: SV1 message {} not found",
+                    message.first().unwrap()
+                );
+            } else {
+                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+                continue;
+            }
+        }
     }
 
     /// Wait for a mining.notify message with a job_id that is a keepalive job.
     /// Keepalive job IDs contain the '#' delimiter (format: `{original_job_id}#{counter}`).
     pub async fn wait_for_keepalive_notify(&self, direction: MessageDirection) {
         let now = std::time::Instant::now();
-        tokio::select!(
-            _ = tokio::signal::ctrl_c() => { },
-            _ = async {
-                loop {
-                    let has_notify = match direction {
-                        MessageDirection::ToUpstream => {
-                            self.messages_from_downstream
-                                .has_keepalive_notify()
-                                .await
-                        }
-                        MessageDirection::ToDownstream => {
-                            self.messages_from_upstream
-                                .has_keepalive_notify()
-                                .await
-                        }
-                    };
-                    if has_notify {
-                        break;
-                    }
-                    if now.elapsed().as_secs() > 60 {
-                        panic!(
-                            "Timeout: keepalive mining.notify (job_id containing '#') not found"
-                        );
-                    } else {
-                        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
-                        continue;
-                    }
+        loop {
+            let has_notify = match direction {
+                MessageDirection::ToUpstream => {
+                    self.messages_from_downstream.has_keepalive_notify().await
+                }
+                MessageDirection::ToDownstream => {
+                    self.messages_from_upstream.has_keepalive_notify().await
                 }
-            } => {}
-        );
+            };
+            if has_notify {
+                break;
+            }
+            if now.elapsed().as_secs() > 60 {
+                panic!("Timeout: keepalive mining.notify (job_id containing '#') not found");
+            } else {
+                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+                continue;
+            }
+        }
     }
 
     /// Waits for a message and executes an assertion closure on it.

integration-tests/lib/utils.rs

@@ -25,6 +25,7 @@ use stratum_apps::{
         },
     },
 };
+use tokio_util::sync::CancellationToken;
 
 // prevents get_available_port from ever returning the same port twice
 static UNIQUE_PORTS: Lazy<Mutex<HashSet<u16>>> = Lazy::new(|| Mutex::new(HashSet::new()));
@@ -77,8 +78,12 @@ pub async fn create_downstream(
         Responder::from_authority_kp(&pub_key, &prv_key, std::time::Duration::from_secs(10000))
             .unwrap();
 
-    if let Ok((receiver_from_client, sender_to_client)) =
-        Connection::new::<AnyMessageOwned>(stream, HandshakeRole::Responder(responder)).await
+    if let Ok((receiver_from_client, sender_to_client)) = Connection::new::<AnyMessageOwned>(
+        stream,
+        HandshakeRole::Responder(responder),
+        CancellationToken::new(),
+    )
+    .await
     {
         Some((receiver_from_client, sender_to_client))
     } else {
@@ -90,9 +95,13 @@ pub async fn create_upstream(
     stream: tokio::net::TcpStream,
 ) -> Option<(Receiver<MessageFrame>, Sender<MessageFrame>)> {
     let initiator = Initiator::without_pk().expect("This fn call can not fail");
-    Connection::new::<AnyMessageOwned>(stream, HandshakeRole::Initiator(initiator))
-        .await
-        .ok()
+    Connection::new::<AnyMessageOwned>(
+        stream,
+        HandshakeRole::Initiator(initiator),
+        CancellationToken::new(),
+    )
+    .await
+    .ok()
 }
 
 pub async fn recv_from_down_send_to_up(

miner-apps/jd-client/src/lib/jdc_runtime.rs

@@ -877,10 +877,6 @@ impl JdcRuntime<Running> {
                 warn!("Upstream/Job Declarator connection dropped — attempting reconnection...");
                 RuntimeEvent::Fallback
             }
-            _ = tokio::signal::ctrl_c() => {
-                info!("Ctrl+C received — initiating graceful shutdown...");
-                RuntimeEvent::Shutdown
-            }
         }
     }
 

miner-apps/jd-client/src/main.rs

@@ -25,7 +25,19 @@ async fn inner_main() {
     });
 
     init_logging(jdc_config.log_file());
-    if JobDeclaratorClient::new(jdc_config).start().await.is_err() {
+
+    let jdc = JobDeclaratorClient::new(jdc_config);
+    tokio::spawn({
+        let jdc = jdc.clone();
+        async move {
+            if tokio::signal::ctrl_c().await.is_ok() {
+                tracing::info!("Ctrl+C received — initiating graceful shutdown...");
+                jdc.shutdown().await;
+            }
+        }
+    });
+
+    if jdc.start().await.is_err() {
         std::process::exit(1);
     }
 }

miner-apps/translator/src/lib/mod.rs

@@ -297,11 +297,6 @@ impl TranslatorSv2 {
 
                                 info!("Upstream and ChannelManager restarted successfully.");
                 }
-                _ = tokio::signal::ctrl_c() => {
-                    info!("Ctrl+C received — initiating graceful shutdown...");
-                    cancellation_token.cancel();
-                    break;
-                }
             }
         }
 

miner-apps/translator/src/main.rs

@@ -29,5 +29,16 @@ async fn inner_main() {
 
     init_logging(proxy_config.log_dir());
 
-    TranslatorSv2::new(proxy_config).start().await;
+    let translator = TranslatorSv2::new(proxy_config);
+    tokio::spawn({
+        let translator = translator.clone();
+        async move {
+            if tokio::signal::ctrl_c().await.is_ok() {
+                tracing::info!("Ctrl+C received — initiating graceful shutdown...");
+                translator.shutdown().await;
+            }
+        }
+    });
+
+    translator.start().await;
 }

pool-apps/pool/src/lib/pool_runtime.rs

@@ -618,13 +618,6 @@ impl PoolRuntime<ChannelManagerReady> {
 
 impl PoolRuntime<Running> {
     pub(super) async fn wait_for_shutdown(&self) {
-        let cancellation_token = self.pool.cancellation_token.clone();
-        tokio::select! {
-            _ = tokio::signal::ctrl_c() => {
-                info!("Ctrl+C received — initiating graceful shutdown...");
-                cancellation_token.cancel();
-            }
-            _ = cancellation_token.cancelled() => {}
-        }
+        self.pool.cancellation_token.cancelled().await;
     }
 }

pool-apps/pool/src/main.rs

@@ -24,7 +24,19 @@ async fn inner_main() {
         std::process::exit(1);
     });
     init_logging(config.log_dir());
-    if let Err(e) = PoolSv2::new(config).start().await {
+
+    let pool = PoolSv2::new(config);
+    tokio::spawn({
+        let pool = pool.clone();
+        async move {
+            if tokio::signal::ctrl_c().await.is_ok() {
+                tracing::info!("Ctrl+C received — initiating graceful shutdown...");
+                pool.shutdown().await;
+            }
+        }
+    });
+
+    if let Err(e) = pool.start().await {
         tracing::error!("Pool Error'ed out: {e}");
         std::process::exit(1);
     };

stratum-apps/README.md

@@ -75,10 +75,12 @@ stratum-apps = { version = "0.4.0", features = ["pool"] }
 ```
 
 ```rust
-use stratum_apps::{network_helpers, config_helpers};
+use stratum_apps::{network_helpers, config_helpers, tokio_util::sync::CancellationToken};
 
 // Use networking
-let connection = network_helpers::Connection::new(stream, HandshakeRole::Responder).await?;
+let cancellation_token = CancellationToken::new();
+let connection =
+    network_helpers::Connection::new(stream, HandshakeRole::Responder, cancellation_token).await?;
 
 // Use configuration
 let config: PoolConfig = config_helpers::parse_config("pool.toml")?;

stratum-apps/src/lib.rs

@@ -39,6 +39,11 @@ pub use stratum_core;
 #[cfg(feature = "bitcoin-core-sv2")]
 pub use bitcoin_core_sv2;
 
+/// Re-export `tokio_util`, for the [`tokio_util::sync::CancellationToken`] required by the
+/// networking helpers
+#[cfg(feature = "tokio-util")]
+pub use tokio_util;
+
 /// High-level networking utilities for SV2 connections
 ///
 /// Provides connection management, encrypted streams, and protocol handling.