From 7ebe4a05d6f7ab57cdb126ac6b979f097a463009 Mon Sep 17 00:00:00 2001 From: plebhash Date: Wed, 5 Aug 2026 19:01:22 -0300 Subject: [PATCH 1/3] fix(stratum-apps): use CancellationToken instead of ctrl_c in noise Connection `Connection::spawn_reader` and `Connection::spawn_writer` both selected on `tokio::signal::ctrl_c()` as their shutdown arm. The first poll of either future makes tokio install a process-global SIGINT handler, so from that point on SIGINT no longer terminates the process, it only completes those two futures. Any binary opening a single noise Connection silently lost default Ctrl+C behavior, and could not be interrupted if it later hung. Signal handling belongs to the application, not to a transport helper. Replace both arms with a caller-supplied CancellationToken, mirroring the sibling helper `ConnectionSV1::new`. `tokio_util` is re-exported from the crate root so consumers can build the token without declaring their own dependency on it. Note this is a breaking change to `Connection::new`. --- integration-tests/lib/mining_device/mod.rs | 11 ++- integration-tests/lib/utils.rs | 18 ++++- stratum-apps/README.md | 6 +- stratum-apps/src/lib.rs | 5 ++ .../src/network_helpers/noise_connection.rs | 75 ++++++++++++++++++- 5 files changed, 102 insertions(+), 13 deletions(-) diff --git a/integration-tests/lib/mining_device/mod.rs b/integration-tests/lib/mining_device/mod.rs index d4df6fc3e..cdc45be4d 100644 --- a/integration-tests/lib/mining_device/mod.rs +++ b/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, diff --git a/integration-tests/lib/utils.rs b/integration-tests/lib/utils.rs index 388327018..e0a44affd 100644 --- a/integration-tests/lib/utils.rs +++ b/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>> = Lazy::new(|| Mutex::new(HashSet::new())); @@ -78,7 +79,12 @@ pub async fn create_downstream( .unwrap(); if let Ok((receiver_from_client, sender_to_client)) = - Connection::new::(stream, HandshakeRole::Responder(responder)).await + Connection::new::( + stream, + HandshakeRole::Responder(responder), + CancellationToken::new(), + ) + .await { Some((receiver_from_client, sender_to_client)) } else { @@ -90,9 +96,13 @@ pub async fn create_upstream( stream: tokio::net::TcpStream, ) -> Option<(Receiver, Sender)> { let initiator = Initiator::without_pk().expect("This fn call can not fail"); - Connection::new::(stream, HandshakeRole::Initiator(initiator)) - .await - .ok() + Connection::new::( + stream, + HandshakeRole::Initiator(initiator), + CancellationToken::new(), + ) + .await + .ok() } pub async fn recv_from_down_send_to_up( diff --git a/stratum-apps/README.md b/stratum-apps/README.md index 1463054e6..f67cdb8ed 100644 --- a/stratum-apps/README.md +++ b/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")?; diff --git a/stratum-apps/src/lib.rs b/stratum-apps/src/lib.rs index f2f27c99b..bb9274e9b 100644 --- a/stratum-apps/src/lib.rs +++ b/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. diff --git a/stratum-apps/src/network_helpers/noise_connection.rs b/stratum-apps/src/network_helpers/noise_connection.rs index cbfb01209..a34b8f295 100644 --- a/stratum-apps/src/network_helpers/noise_connection.rs +++ b/stratum-apps/src/network_helpers/noise_connection.rs @@ -10,6 +10,7 @@ use stratum_core::{ codec_sv2::{HandshakeRole, StandardEitherFrame}, }; use tokio::{net::TcpStream, task}; +use tokio_util::sync::CancellationToken; use tracing::{debug, error}; pub struct Connection; @@ -31,9 +32,13 @@ impl ConnectionState { } impl Connection { + /// Performs the Noise handshake and spawns the reader and writer tasks. + /// + /// Cancelling `cancellation_token` makes both tasks exit and closes all channels. pub async fn new( stream: TcpStream, role: HandshakeRole, + cancellation_token: CancellationToken, ) -> Result< ( Receiver>, @@ -59,14 +64,19 @@ impl Connection { .await? .into_split(); - Self::spawn_reader(read_half, Arc::clone(&conn_state)); - Self::spawn_writer(write_half, conn_state); + Self::spawn_reader( + read_half, + Arc::clone(&conn_state), + cancellation_token.clone(), + ); + Self::spawn_writer(write_half, conn_state, cancellation_token); Ok((receiver_incoming, sender_outgoing)) } fn spawn_reader( mut read_half: NoiseTcpReadHalf, conn_state: Arc>, + cancellation_token: CancellationToken, ) -> task::JoinHandle<()> where Message: Serialize + for<'decoder> Deserialize<'decoder> + GetSize + Send + 'static, @@ -76,7 +86,7 @@ impl Connection { task::spawn(async move { loop { tokio::select! { - _ = tokio::signal::ctrl_c() => { + _ = cancellation_token.cancelled() => { debug!("Reader received shutdown signal."); break; } @@ -102,6 +112,7 @@ impl Connection { fn spawn_writer( mut write_half: NoiseTcpWriteHalf, conn_state: Arc>, + cancellation_token: CancellationToken, ) -> task::JoinHandle<()> where Message: Serialize + for<'decoder> Deserialize<'decoder> + GetSize + Send + 'static, @@ -111,7 +122,7 @@ impl Connection { task::spawn(async move { loop { tokio::select! { - _ = tokio::signal::ctrl_c() => { + _ = cancellation_token.cancelled() => { debug!("Writer received shutdown signal."); break; } @@ -138,3 +149,59 @@ impl Connection { }) } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::key_utils::{Secp256k1PublicKey, Secp256k1SecretKey}; + use std::time::Duration; + use stratum_core::{ + noise_sv2::{Initiator, Responder}, + parsers_sv2::AnyMessageOwned, + }; + use tokio::net::TcpListener; + + // same test authority keypair used by the integration tests + const PUB_KEY: &str = "9auqWEzQDVyd2oe1JVGFLMLHZtCo2FFqZwtKA5gd9xbuEu7PH72"; + const PRV_KEY: &str = "mkDLTBBRxdBv998612qipDYoTK3YUrqLe8uWw7gu3iXbSrn2n"; + + #[tokio::test] + async fn cancellation_shuts_down_connection() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + + let responder_task = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let pub_key = PUB_KEY.parse::().unwrap().into_bytes(); + let prv_key = PRV_KEY.parse::().unwrap().into_bytes(); + let responder = + Responder::from_authority_kp(&pub_key, &prv_key, Duration::from_secs(10_000)) + .unwrap(); + Connection::new::( + stream, + HandshakeRole::Responder(responder), + CancellationToken::new(), + ) + .await + .unwrap() + }); + + let stream = TcpStream::connect(addr).await.unwrap(); + let initiator = Initiator::without_pk().unwrap(); + let cancellation_token = CancellationToken::new(); + let (receiver, sender) = Connection::new::( + stream, + HandshakeRole::Initiator(initiator), + cancellation_token.clone(), + ) + .await + .unwrap(); + let _responder_side = responder_task.await.unwrap(); + + cancellation_token.cancel(); + + // both tasks must exit and close every channel + assert!(receiver.recv().await.is_err()); + assert!(sender.is_closed()); + } +} From b2b9ffe06c765a41d573fd5a7be610351c79c61d Mon Sep 17 00:00:00 2001 From: plebhash Date: Wed, 5 Aug 2026 19:03:40 -0300 Subject: [PATCH 2/3] fix(integration-tests): stop registering process-wide SIGINT handlers The sniffers selected on `tokio::signal::ctrl_c()` in arms that do nothing. Their only real effect was installing a process-global SIGINT handler, which stops SIGINT from terminating the test binary. Drop those arms so the sniffers no longer claim the signal. In `wait_for_message` and `wait_for_keepalive_notify` this leaves a single-arm `select!`, so the remaining block is awaited directly. Note this is not sufficient on its own to make Ctrl+C kill a hung test: any test that starts a role in-process still installs that role's own SIGINT handler (e.g. `pool_runtime::wait_for_shutdown`), which is correct behavior for the role binaries but leaks into the test process. --- integration-tests/lib/sniffer.rs | 1 - integration-tests/lib/sv1_sniffer.rs | 90 ++++++++++++---------------- integration-tests/lib/utils.rs | 13 ++-- 3 files changed, 44 insertions(+), 60 deletions(-) diff --git a/integration-tests/lib/sniffer.rs b/integration-tests/lib/sniffer.rs index 4754b7831..36984277b 100644 --- a/integration-tests/lib/sniffer.rs +++ b/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()) => { }, }; diff --git a/integration-tests/lib/sv1_sniffer.rs b/integration-tests/lib/sv1_sniffer.rs index f46afca69..f41d4a409 100644 --- a/integration-tests/lib/sv1_sniffer.rs +++ b/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. diff --git a/integration-tests/lib/utils.rs b/integration-tests/lib/utils.rs index e0a44affd..4367e71ea 100644 --- a/integration-tests/lib/utils.rs +++ b/integration-tests/lib/utils.rs @@ -78,13 +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::( - stream, - HandshakeRole::Responder(responder), - CancellationToken::new(), - ) - .await + if let Ok((receiver_from_client, sender_to_client)) = Connection::new::( + stream, + HandshakeRole::Responder(responder), + CancellationToken::new(), + ) + .await { Some((receiver_from_client, sender_to_client)) } else { From 7992938e6fa30842171bb4f1fcc6219d8aa81747 Mon Sep 17 00:00:00 2001 From: plebhash Date: Wed, 5 Aug 2026 19:18:31 -0300 Subject: [PATCH 3/3] fix(apps): move SIGINT handling from the role libraries into the binaries `PoolRuntime::wait_for_shutdown`, `TranslatorSv2::start` and `JobDeclaratorClient::start` each selected on `tokio::signal::ctrl_c()`. That made the libraries install a process-global SIGINT handler, so any process embedding a role lost default Ctrl+C behavior: in the integration tests, a SIGINT would gracefully stop the role but leave the test harness running, making a hung test impossible to interrupt. All three run loops already break on their own CancellationToken, and all three types expose `shutdown()`, so the ctrl_c arms are removed and each binary now watches for the signal itself and calls `shutdown()`. Behavior of the binaries is unchanged: Ctrl+C still triggers the same graceful shutdown path, it is just initiated by the binary rather than by the library. --- miner-apps/jd-client/src/lib/jdc_runtime.rs | 4 ---- miner-apps/jd-client/src/main.rs | 14 +++++++++++++- miner-apps/translator/src/lib/mod.rs | 5 ----- miner-apps/translator/src/main.rs | 13 ++++++++++++- pool-apps/pool/src/lib/pool_runtime.rs | 9 +-------- pool-apps/pool/src/main.rs | 14 +++++++++++++- 6 files changed, 39 insertions(+), 20 deletions(-) diff --git a/miner-apps/jd-client/src/lib/jdc_runtime.rs b/miner-apps/jd-client/src/lib/jdc_runtime.rs index 5c1d4b2f5..b30546ce8 100644 --- a/miner-apps/jd-client/src/lib/jdc_runtime.rs +++ b/miner-apps/jd-client/src/lib/jdc_runtime.rs @@ -877,10 +877,6 @@ impl JdcRuntime { warn!("Upstream/Job Declarator connection dropped — attempting reconnection..."); RuntimeEvent::Fallback } - _ = tokio::signal::ctrl_c() => { - info!("Ctrl+C received — initiating graceful shutdown..."); - RuntimeEvent::Shutdown - } } } diff --git a/miner-apps/jd-client/src/main.rs b/miner-apps/jd-client/src/main.rs index 7f63af23c..b1b1cee81 100644 --- a/miner-apps/jd-client/src/main.rs +++ b/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); } } diff --git a/miner-apps/translator/src/lib/mod.rs b/miner-apps/translator/src/lib/mod.rs index ff0b5c3d0..5612ce30a 100644 --- a/miner-apps/translator/src/lib/mod.rs +++ b/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; - } } } diff --git a/miner-apps/translator/src/main.rs b/miner-apps/translator/src/main.rs index 3e49f3878..f6640e3f0 100644 --- a/miner-apps/translator/src/main.rs +++ b/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; } diff --git a/pool-apps/pool/src/lib/pool_runtime.rs b/pool-apps/pool/src/lib/pool_runtime.rs index 5418269f4..84a4f9b66 100644 --- a/pool-apps/pool/src/lib/pool_runtime.rs +++ b/pool-apps/pool/src/lib/pool_runtime.rs @@ -618,13 +618,6 @@ impl PoolRuntime { impl PoolRuntime { 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; } } diff --git a/pool-apps/pool/src/main.rs b/pool-apps/pool/src/main.rs index 2a015bc3a..6813f04d2 100644 --- a/pool-apps/pool/src/main.rs +++ b/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); };