From ccec32c609125b4e44c4e531aa90a5dcbdd55647 Mon Sep 17 00:00:00 2001 From: syntaxjak Date: Fri, 19 Jun 2026 11:15:48 -0500 Subject: [PATCH 01/13] Refactor stratum server leak handling --- Cargo.lock | 23 -- servers/Cargo.toml | 1 - servers/src/mining/stratumserver.rs | 580 +++++++++++++++++++++------- 3 files changed, 448 insertions(+), 156 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index df63b8c02f..8798764fc5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -96,28 +96,6 @@ version = "0.7.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50" -[[package]] -name = "async-stream" -version = "0.3.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0b5a71a6f37880a80d1d7f19efd781e4b5de42c88f0722cc13bcb6cc2cfe8476" -dependencies = [ - "async-stream-impl", - "futures-core", - "pin-project-lite", -] - -[[package]] -name = "async-stream-impl" -version = "0.3.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d" -dependencies = [ - "proc-macro2 1.0.106", - "quote 1.0.45", - "syn 2.0.118", -] - [[package]] name = "atomic-waker" version = "1.1.2" @@ -1224,7 +1202,6 @@ dependencies = [ name = "grin_servers" version = "5.5.1-alpha.0" dependencies = [ - "async-stream", "chrono", "fs2", "futures 0.3.32", diff --git a/servers/Cargo.toml b/servers/Cargo.toml index 69f792aca2..645533008c 100644 --- a/servers/Cargo.toml +++ b/servers/Cargo.toml @@ -21,7 +21,6 @@ serde_json = "1" chrono = "0.4.11" tokio = { version = "1", features = ["full"] } tokio-util = { version = "0.7", features = ["codec"] } -async-stream = "0.3" walkdir = "2.3.1" hyper-util = { version = "0.1.20", features = ["client-legacy"] } http-body-util = "0.1.3" diff --git a/servers/src/mining/stratumserver.rs b/servers/src/mining/stratumserver.rs index b6b775af42..6b04cababf 100644 --- a/servers/src/mining/stratumserver.rs +++ b/servers/src/mining/stratumserver.rs @@ -14,11 +14,12 @@ //! Mining Stratum Server -use futures::channel::mpsc; -use futures::pin_mut; use futures::{SinkExt, StreamExt, TryStreamExt}; -use tokio::net::TcpListener; +use tokio::net::{TcpListener, TcpStream}; use tokio::runtime::Runtime; +use tokio::sync::{mpsc, OwnedSemaphorePermit, Semaphore}; +use tokio::task::JoinSet; +use tokio::time::{timeout, Instant}; use tokio_util::codec::{Framed, LinesCodec}; use crate::util::RwLock; @@ -43,7 +44,14 @@ use crate::mining::mine_block; use crate::util::ToHex; use crate::ServerTxPool; -type Tx = mpsc::UnboundedSender; +type Tx = mpsc::Sender; + +const MAX_STRATUM_WORKERS: usize = 256; +const WORKER_QUEUE_SIZE: usize = 64; +const WORKER_IDLE_TIMEOUT: Duration = Duration::from_secs(5 * 60); +const WORKER_WRITE_TIMEOUT: Duration = Duration::from_secs(30); +const ACCEPT_ERROR_BACKOFF: Duration = Duration::from_millis(100); +const MAX_RPC_LINE_BYTES: usize = 64 * 1024; // ---------------------------------------- // http://www.jsonrpc.org/specification @@ -599,72 +607,169 @@ impl Handler { // ---------------------------------------- // Worker Factory Thread Function -fn accept_connections(listen_addr: SocketAddr, handler: Arc) { - info!("Start tokio stratum server"); - let task = async move { - let listener = TcpListener::bind(&listen_addr).await.unwrap_or_else(|_| { - panic!("Stratum: Failed to bind to listen address {}", listen_addr) - }); - let server = async_stream::stream! { - loop { - match listener.accept().await { - Ok((socket, _)) => yield socket, - Err(e) => { - error!("accept error = {:?}", e); - continue; - } + +struct WorkerCleanup { + worker_id: usize, + workers: Arc, + peer_addr: Option, +} + +impl Drop for WorkerCleanup { + fn drop(&mut self) { + self.workers.remove_worker(self.worker_id); + match self.peer_addr { + Some(peer_addr) => info!("Worker {} disconnected from {}", self.worker_id, peer_addr), + None => info!("Worker {} disconnected", self.worker_id), + } + } +} + +async fn handle_connection( + socket: TcpStream, + handler: Arc, + _permit: OwnedSemaphorePermit, +) { + handle_connection_with_idle_timeout(socket, handler, WORKER_IDLE_TIMEOUT).await; +} + +async fn handle_connection_with_idle_timeout( + socket: TcpStream, + handler: Arc, + idle_timeout: Duration, +) { + let peer_addr = socket.peer_addr().ok(); + let (tx, mut rx) = mpsc::channel(WORKER_QUEUE_SIZE); + let (shutdown_tx, mut shutdown_rx) = mpsc::channel::<()>(1); + let worker_id = handler.workers.add_worker(tx, shutdown_tx); + let _cleanup = WorkerCleanup { + worker_id, + workers: handler.workers.clone(), + peer_addr, + }; + + match peer_addr { + Some(peer_addr) => info!("Worker {} connected from {}", worker_id, peer_addr), + None => info!("Worker {} connected", worker_id), + } + + let framed = Framed::new(socket, LinesCodec::new_with_max_length(MAX_RPC_LINE_BYTES)); + let (mut writer, mut reader) = framed.split(); + let (activity_tx, mut activity_rx) = mpsc::channel::<()>(1); + + let reader_handler = handler.clone(); + let read_activity = activity_tx.clone(); + let read = async move { + while let Some(line) = reader + .try_next() + .await + .map_err(|e| error!("Worker {} read error: {}", worker_id, e))? + { + let _ = read_activity.try_send(()); + let request: RpcRequest = serde_json::from_str(&line).map_err(|e| { + error!("Worker {} invalid JSON: {}", worker_id, e); + })?; + let resp = reader_handler.handle_rpc_requests(request, worker_id); + if !reader_handler.workers.try_send_to(worker_id, resp) { + warn!("Worker {} outbound queue full or closed", worker_id); + return Err(()); + } + } + Result::<_, ()>::Ok(()) + }; + + let write = async move { + while let Some(line) = rx.recv().await { + match timeout(WORKER_WRITE_TIMEOUT, writer.send(line)).await { + Ok(Ok(())) => { + let _ = activity_tx.try_send(()); + } + Ok(Err(e)) => { + error!("Worker {} write error: {}", worker_id, e); + return Err(()); + } + Err(_) => { + warn!("Worker {} write timed out", worker_id); + return Err(()); } } } - .for_each(move |socket| { - let handler = handler.clone(); - async move { - // Spawn a task to process the connection - let (tx, mut rx) = mpsc::unbounded(); - - let worker_id = handler.workers.add_worker(tx); - info!("Worker {} connected", worker_id); - - let framed = Framed::new(socket, LinesCodec::new()); - let (mut writer, mut reader) = framed.split(); - - let h = handler.clone(); - let read = async move { - while let Some(line) = reader - .try_next() - .await - .map_err(|e| error!("error reading line: {}", e))? - { - let request = serde_json::from_str(&line) - .map_err(|e| error!("error serializing line: {}", e))?; - let resp = h.handle_rpc_requests(request, worker_id); - h.workers.send_to(worker_id, resp); - } + Result::<_, ()>::Ok(()) + }; - Result::<_, ()>::Ok(()) - }; + tokio::pin!(read); + tokio::pin!(write); + let idle_sleep = tokio::time::sleep_until(Instant::now() + idle_timeout); + tokio::pin!(idle_sleep); + + loop { + tokio::select! { + _ = &mut read => break, + _ = &mut write => break, + _ = &mut idle_sleep => { + warn!("Worker {} idle for {:?}; disconnecting", worker_id, idle_timeout); + break; + } + _ = shutdown_rx.recv() => break, + activity = activity_rx.recv() => { + if activity.is_some() { + idle_sleep.as_mut().reset(Instant::now() + idle_timeout); + } else { + break; + } + } + } + } +} - let write = async move { - while let Some(line) = rx.next().await { - writer - .send(line) - .await - .map_err(|e| error!("error writing line: {}", e))?; +async fn accept_connections_loop(listener: TcpListener, handler: Arc) { + let mut connections = JoinSet::new(); + let worker_limit = Arc::new(Semaphore::new(MAX_STRATUM_WORKERS)); + loop { + tokio::select! { + accepted = listener.accept() => { + match accepted { + Ok((socket, peer_addr)) => { + let permit = match worker_limit.clone().try_acquire_owned() { + Ok(permit) => permit, + Err(_) => { + warn!( + "Stratum: rejecting connection from {} (max workers: {})", + peer_addr, MAX_STRATUM_WORKERS + ); + drop(socket); + continue; + } + }; + let handler = handler.clone(); + connections.spawn(async move { + if let Err(e) = socket.set_nodelay(true) { + debug!("Stratum: set_nodelay failed for {}: {}", peer_addr, e); + } + handle_connection(socket, handler, permit).await; + }); } - - Result::<_, ()>::Ok(()) - }; - - let task = async move { - pin_mut!(read, write); - futures::future::select(read, write).await; - handler.workers.remove_worker(worker_id); - info!("Worker {} disconnected", worker_id); - }; - tokio::spawn(task); + Err(e) => { + error!("accept error = {:?}", e); + tokio::time::sleep(ACCEPT_ERROR_BACKOFF).await; + } + } + } + Some(joined) = connections.join_next(), if !connections.is_empty() => { + if let Err(e) = joined { + error!("stratum connection task failed: {}", e); + } } + } + } +} + +fn accept_connections(listen_addr: SocketAddr, handler: Arc) { + info!("Start tokio stratum server"); + let task = async move { + let listener = TcpListener::bind(&listen_addr).await.unwrap_or_else(|_| { + panic!("Stratum: Failed to bind to listen address {}", listen_addr) }); - server.await + accept_connections_loop(listener, handler).await; }; let rt = Runtime::new().unwrap(); @@ -681,17 +786,19 @@ pub struct Worker { login: Option, authenticated: bool, tx: Tx, + shutdown_tx: mpsc::Sender<()>, } impl Worker { /// Creates a new Stratum Worker. - pub fn new(id: usize, tx: Tx) -> Worker { + pub fn new(id: usize, tx: Tx, shutdown_tx: mpsc::Sender<()>) -> Worker { Worker { id: id, agent: String::from(""), login: None, authenticated: false, tx: tx, + shutdown_tx, } } } // impl Worker @@ -709,30 +816,47 @@ impl WorkersList { } } - pub fn add_worker(&self, tx: Tx) -> usize { + pub fn add_worker(&self, tx: Tx, shutdown_tx: mpsc::Sender<()>) -> usize { let mut stratum_stats = self.stratum_stats.write(); - let worker_id = stratum_stats.worker_stats.len(); - let worker = Worker::new(worker_id, tx); let mut workers_list = self.workers_list.write(); + let worker_id = match stratum_stats + .worker_stats + .iter() + .position(|ws| !ws.is_connected) + { + Some(id) => id, + None => { + let id = stratum_stats.worker_stats.len(); + stratum_stats.worker_stats.push(WorkerStats::default()); + id + } + }; + let worker = Worker::new(worker_id, tx, shutdown_tx); workers_list.insert(worker_id, worker); let mut worker_stats = WorkerStats::default(); worker_stats.is_connected = true; worker_stats.id = worker_id.to_string(); worker_stats.pow_difficulty = stratum_stats.minimum_share_difficulty; - stratum_stats.worker_stats.push(worker_stats); + stratum_stats.worker_stats[worker_id] = worker_stats; stratum_stats.num_workers = workers_list.len(); worker_id } pub fn remove_worker(&self, worker_id: usize) { - self.update_stats(worker_id, |ws| ws.is_connected = false); - let mut stratum_stats = self.stratum_stats.write(); let mut workers_list = self.workers_list.write(); - workers_list - .remove(&worker_id) - .expect("Stratum: no such addr in map"); + if workers_list.remove(&worker_id).is_none() { + let mut stratum_stats = self.stratum_stats.write(); + stratum_stats.num_workers = workers_list.len(); + return; + } + drop(workers_list); - stratum_stats.num_workers = workers_list.len(); + self.update_stats(worker_id, |ws| { + ws.is_connected = false; + ws.last_seen = SystemTime::now(); + }); + let mut stratum_stats = self.stratum_stats.write(); + stratum_stats.num_workers = self.workers_list.read().len(); } pub fn login(&self, worker_id: usize, login: String, agent: String) -> Result<(), RpcError> { @@ -777,19 +901,42 @@ impl WorkersList { f(&mut stratum_stats.worker_stats[worker_id]); } - pub fn send_to(&self, worker_id: usize, msg: String) { - let _ = self + pub fn try_send_to(&self, worker_id: usize, msg: String) -> bool { + let workers_list = self.workers_list.read(); + let worker = match workers_list.get(&worker_id) { + Some(worker) => worker, + None => return false, + }; + worker.tx.try_send(msg).is_ok() + } + + pub fn disconnect_worker(&self, worker_id: usize) { + let shutdown_tx = self .workers_list .read() .get(&worker_id) - .unwrap() - .tx - .unbounded_send(msg); + .map(|worker| worker.shutdown_tx.clone()); + if let Some(shutdown_tx) = shutdown_tx { + let _ = shutdown_tx.try_send(()); + } } pub fn broadcast(&self, msg: String) { - for worker in self.workers_list.read().values() { - let _ = worker.tx.unbounded_send(msg.clone()); + let mut slow_workers = Vec::new(); + { + let workers_list = self.workers_list.read(); + for (worker_id, worker) in workers_list.iter() { + if worker.tx.try_send(msg.clone()).is_err() { + slow_workers.push(*worker_id); + } + } + } + for worker_id in slow_workers { + warn!( + "Stratum: dropping slow or disconnected worker {}", + worker_id + ); + self.disconnect_worker(worker_id); } } @@ -923,6 +1070,7 @@ mod tests { use crate::core::global::{self, ChainTypes}; use crate::core::pow::Difficulty; use std::fs; + use std::net::TcpListener as StdTcpListener; use std::sync::OnceLock; // ---------------------------------------- @@ -930,11 +1078,6 @@ mod tests { const TEST_MINIMUM_SHARE_DIFFICULTY: u64 = 1; - fn dummy_tx() -> Tx { - let (tx, _rx) = mpsc::unbounded(); - tx - } - /// Read-only chain shared by the RPC routing tests below, so the suite /// opens a single LMDB env. Tests that write to the chain need their own. fn shared_test_chain() -> Arc { @@ -962,20 +1105,20 @@ mod tests { } /// Build a Handler backed by the shared test chain for RPC routing tests. - fn setup_handler() -> Handler { + fn shared_handler() -> Arc { global::set_local_chain_type(ChainTypes::AutomatedTesting); let chain = shared_test_chain(); let stratum_stats = Arc::new(RwLock::new(StratumStats::default())); let sync_state = Arc::new(SyncState::new()); // Default SyncState is Initial (syncing); mark as fully synced for most tests. sync_state.update(SyncStatus::NoSync); - Handler::new( + Arc::new(Handler::new( String::from("test"), stratum_stats, sync_state, TEST_MINIMUM_SHARE_DIFFICULTY, chain, - ) + )) } fn rpc_request(method: &str, params: Option) -> RpcRequest { @@ -991,6 +1134,188 @@ mod tests { serde_json::from_str(json).unwrap() } + fn dummy_tx() -> (Tx, mpsc::Receiver, mpsc::Sender<()>) { + let (tx, rx) = mpsc::channel(WORKER_QUEUE_SIZE); + let (shutdown_tx, _shutdown_rx) = mpsc::channel(1); + (tx, rx, shutdown_tx) + } + + fn add_dummy_worker(workers: &WorkersList) -> usize { + let (tx, _rx, shutdown_tx) = dummy_tx(); + workers.add_worker(tx, shutdown_tx) + } + + #[test] + fn test_worker_slot_reuse_after_disconnect() { + let stats = Arc::new(RwLock::new(StratumStats::default())); + let workers = WorkersList::new(stats.clone()); + + let (tx0, _rx0, shutdown_tx0) = dummy_tx(); + let id0 = workers.add_worker(tx0, shutdown_tx0); + assert_eq!(id0, 0); + assert_eq!(workers.count(), 1); + assert_eq!(stats.read().worker_stats.len(), 1); + + workers.remove_worker(id0); + assert_eq!(workers.count(), 0); + assert!(!stats.read().worker_stats[0].is_connected); + + let (tx1, _rx1, shutdown_tx1) = dummy_tx(); + let id1 = workers.add_worker(tx1, shutdown_tx1); + assert_eq!(id1, 0); + assert_eq!(stats.read().worker_stats.len(), 1); + assert!(stats.read().worker_stats[0].is_connected); + assert_eq!(workers.count(), 1); + } + + #[test] + fn test_try_send_to_missing_full_and_ok() { + let stats = Arc::new(RwLock::new(StratumStats::default())); + let workers = WorkersList::new(stats); + + assert!(!workers.try_send_to(0, "missing".into())); + + let (tx, mut rx) = mpsc::channel(1); + let (shutdown_tx, _shutdown_rx) = mpsc::channel(1); + let id = workers.add_worker(tx, shutdown_tx); + assert!(workers.try_send_to(id, "one".into())); + assert!(!workers.try_send_to(id, "two".into())); + assert_eq!(rx.try_recv().unwrap(), "one"); + assert!(workers.try_send_to(id, "three".into())); + + workers.remove_worker(id); + assert!(!workers.try_send_to(id, "after-remove".into())); + } + + #[test] + fn test_remove_worker_is_idempotent() { + let stats = Arc::new(RwLock::new(StratumStats::default())); + let workers = WorkersList::new(stats); + let (tx, _rx, shutdown_tx) = dummy_tx(); + let id = workers.add_worker(tx, shutdown_tx); + workers.remove_worker(id); + workers.remove_worker(id); + assert_eq!(workers.count(), 0); + } + + #[test] + fn test_accept_loop_tracks_connection_tasks() { + let rt = Runtime::new().unwrap(); + rt.block_on(async { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let handler = setup_handler(".grin_stratum_accept_loop_test"); + + let task = tokio::spawn(accept_connections_loop(listener, handler.clone())); + let client = tokio::net::TcpStream::connect(addr).await.unwrap(); + for _ in 0..100 { + if handler.workers.count() == 1 { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert_eq!(handler.workers.count(), 1); + drop(client); + for _ in 0..100 { + if handler.workers.count() == 0 { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert_eq!(handler.workers.count(), 0); + task.abort(); + }); + } + + #[test] + fn test_accept_loop_enforces_max_connection_limit() { + let rt = Runtime::new().unwrap(); + rt.block_on(async { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let handler = setup_handler(".grin_stratum_max_workers_test"); + let task = tokio::spawn(accept_connections_loop(listener, handler.clone())); + let mut clients = Vec::new(); + for _ in 0..(MAX_STRATUM_WORKERS + 8) { + if let Ok(client) = tokio::net::TcpStream::connect(addr).await { + clients.push(client); + } + } + for _ in 0..100 { + if handler.workers.count() == MAX_STRATUM_WORKERS { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert_eq!(handler.workers.count(), MAX_STRATUM_WORKERS); + drop(clients); + for _ in 0..100 { + if handler.workers.count() == 0 { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert_eq!(handler.workers.count(), 0); + task.abort(); + }); + } + + #[test] + fn test_idle_connection_is_disconnected() { + let rt = Runtime::new().unwrap(); + rt.block_on(async { + let server = StdTcpListener::bind("127.0.0.1:0").unwrap(); + let addr = server.local_addr().unwrap(); + let client = TcpStream::connect(addr).await.unwrap(); + let (server_socket, _) = server.accept().unwrap(); + server_socket.set_nonblocking(true).unwrap(); + let server_socket = TcpStream::from_std(server_socket).unwrap(); + let handler = setup_handler(".grin_stratum_idle_test"); + + let task = tokio::spawn(handle_connection_with_idle_timeout( + server_socket, + handler.clone(), + Duration::from_millis(50), + )); + for _ in 0..100 { + if handler.workers.count() == 1 { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert_eq!(handler.workers.count(), 1); + task.await.unwrap(); + assert_eq!(handler.workers.count(), 0); + drop(client); + }); + } + + fn setup_handler(dir: &str) -> Arc { + global::set_local_chain_type(ChainTypes::AutomatedTesting); + let _ = std::fs::remove_dir_all(dir); + let chain = Arc::new( + chain::Chain::init( + dir.to_string(), + Arc::new(NoopAdapter {}), + genesis::genesis_dev(), + pow::verify_size, + false, + None, + ) + .unwrap(), + ); + let stratum_stats = Arc::new(RwLock::new(StratumStats::default())); + let sync_state = Arc::new(SyncState::new()); + sync_state.update(SyncStatus::NoSync); + Arc::new(Handler::new( + String::from("test"), + stratum_stats, + sync_state, + 1, + chain, + )) + } + // ---------------------------------------- // RpcRequest / RpcResponse serde @@ -1195,8 +1520,8 @@ mod tests { assert_eq!(workers.count(), 0); - let id0 = workers.add_worker(dummy_tx()); - let id1 = workers.add_worker(dummy_tx()); + let id0 = add_dummy_worker(&workers); + let id1 = add_dummy_worker(&workers); assert_eq!(id0, 0); assert_eq!(id1, 1); assert_eq!(workers.count(), 2); @@ -1225,7 +1550,7 @@ mod tests { fn test_workers_list_relogin_replaces_login_and_agent() { let stats = Arc::new(RwLock::new(StratumStats::default())); let workers = WorkersList::new(stats); - let id0 = workers.add_worker(dummy_tx()); + let id0 = add_dummy_worker(&workers); workers .login(id0, "alice".to_string(), "agent-a".to_string()) @@ -1256,7 +1581,7 @@ mod tests { fn test_workers_list_get_stats_missing_worker() { let stats = Arc::new(RwLock::new(StratumStats::default())); let workers = WorkersList::new(stats); - let _ = workers.add_worker(dummy_tx()); + let _ = add_dummy_worker(&workers); // Index past the end of the stats vec: `get_stats` reports it as a // clean RpcError rather than panicking. @@ -1269,26 +1594,17 @@ mod tests { let stats = Arc::new(RwLock::new(StratumStats::default())); let workers = WorkersList::new(stats); - let (tx0, mut rx0) = mpsc::unbounded(); - let (tx1, mut rx1) = mpsc::unbounded(); - let id0 = workers.add_worker(tx0); - let _id1 = workers.add_worker(tx1); + let (tx0, mut rx0, shutdown_tx0) = dummy_tx(); + let (tx1, mut rx1, shutdown_tx1) = dummy_tx(); + let id0 = workers.add_worker(tx0, shutdown_tx0); + let _id1 = workers.add_worker(tx1, shutdown_tx1); workers.broadcast("hello-all".to_string()); - assert_eq!( - futures::executor::block_on(rx0.next()).unwrap(), - "hello-all" - ); - assert_eq!( - futures::executor::block_on(rx1.next()).unwrap(), - "hello-all" - ); + assert_eq!(rx0.try_recv().unwrap(), "hello-all"); + assert_eq!(rx1.try_recv().unwrap(), "hello-all"); workers.send_to(id0, "hello-one".to_string()); - assert_eq!( - futures::executor::block_on(rx0.next()).unwrap(), - "hello-one" - ); + assert_eq!(rx0.try_recv().unwrap(), "hello-one"); // Unicast must not deliver to the other worker (channel open but empty). assert!(rx1.try_recv().is_err()); } @@ -1316,8 +1632,8 @@ mod tests { #[test] fn test_handle_keepalive() { - let handler = setup_handler(); - let worker_id = handler.workers.add_worker(dummy_tx()); + let handler = shared_handler(); + let worker_id = add_dummy_worker(&handler.workers); let resp = parse_rpc_response( &handler.handle_rpc_requests(rpc_request("keepalive", None), worker_id), @@ -1329,8 +1645,8 @@ mod tests { #[test] fn test_handle_method_not_found() { - let handler = setup_handler(); - let worker_id = handler.workers.add_worker(dummy_tx()); + let handler = shared_handler(); + let worker_id = add_dummy_worker(&handler.workers); let resp = parse_rpc_response( &handler.handle_rpc_requests(rpc_request("does_not_exist", None), worker_id), @@ -1343,8 +1659,8 @@ mod tests { #[test] fn test_handle_login_ok() { - let handler = setup_handler(); - let worker_id = handler.workers.add_worker(dummy_tx()); + let handler = shared_handler(); + let worker_id = add_dummy_worker(&handler.workers); let params = serde_json::json!({ "login": "bob", @@ -1365,8 +1681,8 @@ mod tests { #[test] fn test_handle_login_invalid_params() { - let handler = setup_handler(); - let worker_id = handler.workers.add_worker(dummy_tx()); + let handler = shared_handler(); + let worker_id = add_dummy_worker(&handler.workers); let resp = parse_rpc_response(&handler.handle_rpc_requests(rpc_request("login", None), worker_id)); @@ -1377,7 +1693,7 @@ mod tests { #[test] fn test_handle_getjobtemplate_while_syncing() { - let handler = setup_handler(); + let handler = shared_handler(); // Force syncing state handler.sync_state.update(SyncStatus::HeaderSync { sync_head: handler.chain.head().unwrap(), @@ -1385,7 +1701,7 @@ mod tests { highest_height: 100, highest_diff: Difficulty::from_num(1000), }); - let worker_id = handler.workers.add_worker(dummy_tx()); + let worker_id = add_dummy_worker(&handler.workers); let resp = parse_rpc_response( &handler.handle_rpc_requests(rpc_request("getjobtemplate", None), worker_id), @@ -1398,11 +1714,11 @@ mod tests { #[test] fn test_handle_getjobtemplate_ok() { - let handler = setup_handler(); + let handler = shared_handler(); // Non-default difficulty so the template path is not only asserting the const. let job_difficulty = 7; handler.current_state.write().minimum_share_difficulty = job_difficulty; - let worker_id = handler.workers.add_worker(dummy_tx()); + let worker_id = add_dummy_worker(&handler.workers); let resp = parse_rpc_response( &handler.handle_rpc_requests(rpc_request("getjobtemplate", None), worker_id), @@ -1420,8 +1736,8 @@ mod tests { #[test] fn test_handle_status() { - let handler = setup_handler(); - let worker_id = handler.workers.add_worker(dummy_tx()); + let handler = shared_handler(); + let worker_id = add_dummy_worker(&handler.workers); handler.workers.update_stats(worker_id, |ws| { ws.num_accepted = 10; ws.num_rejected = 2; @@ -1444,8 +1760,8 @@ mod tests { #[test] fn test_handle_submit_too_late() { - let handler = setup_handler(); - let worker_id = handler.workers.add_worker(dummy_tx()); + let handler = shared_handler(); + let worker_id = add_dummy_worker(&handler.workers); // Wrong height vs current block version (height 0) => stale share let params = serde_json::json!({ @@ -1468,8 +1784,8 @@ mod tests { #[test] fn test_handle_submit_invalid_job_id() { - let handler = setup_handler(); - let worker_id = handler.workers.add_worker(dummy_tx()); + let handler = shared_handler(); + let worker_id = add_dummy_worker(&handler.workers); // job_id out of range of current_block_versions let params = serde_json::json!({ @@ -1489,8 +1805,8 @@ mod tests { #[test] fn test_handle_submit_missing_params() { - let handler = setup_handler(); - let worker_id = handler.workers.add_worker(dummy_tx()); + let handler = shared_handler(); + let worker_id = add_dummy_worker(&handler.workers); let resp = parse_rpc_response( &handler.handle_rpc_requests(rpc_request("submit", None), worker_id), @@ -1501,8 +1817,8 @@ mod tests { #[test] fn test_handle_submit_invalid_edge_bits() { - let handler = setup_handler(); - let worker_id = handler.workers.add_worker(dummy_tx()); + let handler = shared_handler(); + let worker_id = add_dummy_worker(&handler.workers); // edge_bits below the AutomatedTesting minimum (10) and not the // secondary size (29): the proof is neither primary nor secondary, so @@ -1527,8 +1843,8 @@ mod tests { #[test] fn test_last_seen_updates() { - let handler = setup_handler(); - let worker_id = handler.workers.add_worker(dummy_tx()); + let handler = shared_handler(); + let worker_id = add_dummy_worker(&handler.workers); // Force a known baseline instead of racing a real clock read against // the update below. handler From 4bc1cb1eb789d50c782b0ebbdf1eaad10b8b4d30 Mon Sep 17 00:00:00 2001 From: wiesche Date: Sun, 26 Jul 2026 15:28:56 +0200 Subject: [PATCH 02/13] apply backpressure to worker responses --- servers/src/mining/stratumserver.rs | 99 +++++++++++++++++++++++------ 1 file changed, 78 insertions(+), 21 deletions(-) diff --git a/servers/src/mining/stratumserver.rs b/servers/src/mining/stratumserver.rs index 6b04cababf..6cf1cae14f 100644 --- a/servers/src/mining/stratumserver.rs +++ b/servers/src/mining/stratumserver.rs @@ -669,8 +669,8 @@ async fn handle_connection_with_idle_timeout( error!("Worker {} invalid JSON: {}", worker_id, e); })?; let resp = reader_handler.handle_rpc_requests(request, worker_id); - if !reader_handler.workers.try_send_to(worker_id, resp) { - warn!("Worker {} outbound queue full or closed", worker_id); + if !reader_handler.workers.send_to(worker_id, resp).await { + warn!("Worker {} outbound queue closed", worker_id); return Err(()); } } @@ -901,13 +901,15 @@ impl WorkersList { f(&mut stratum_stats.worker_stats[worker_id]); } - pub fn try_send_to(&self, worker_id: usize, msg: String) -> bool { - let workers_list = self.workers_list.read(); - let worker = match workers_list.get(&worker_id) { - Some(worker) => worker, - None => return false, + pub async fn send_to(&self, worker_id: usize, msg: String) -> bool { + let tx = { + let workers_list = self.workers_list.read(); + match workers_list.get(&worker_id) { + Some(worker) => worker.tx.clone(), + None => return false, + } }; - worker.tx.try_send(msg).is_ok() + tx.send(msg).await.is_ok() } pub fn disconnect_worker(&self, worker_id: usize) { @@ -1169,22 +1171,26 @@ mod tests { } #[test] - fn test_try_send_to_missing_full_and_ok() { - let stats = Arc::new(RwLock::new(StratumStats::default())); - let workers = WorkersList::new(stats); + fn test_send_to_missing_closed_and_ok() { + let rt = Runtime::new().unwrap(); + rt.block_on(async { + let stats = Arc::new(RwLock::new(StratumStats::default())); + let workers = WorkersList::new(stats); - assert!(!workers.try_send_to(0, "missing".into())); + assert!(!workers.send_to(0, "missing".into()).await); - let (tx, mut rx) = mpsc::channel(1); - let (shutdown_tx, _shutdown_rx) = mpsc::channel(1); - let id = workers.add_worker(tx, shutdown_tx); - assert!(workers.try_send_to(id, "one".into())); - assert!(!workers.try_send_to(id, "two".into())); - assert_eq!(rx.try_recv().unwrap(), "one"); - assert!(workers.try_send_to(id, "three".into())); + let (tx, mut rx) = mpsc::channel(1); + let (shutdown_tx, _shutdown_rx) = mpsc::channel(1); + let id = workers.add_worker(tx, shutdown_tx); + assert!(workers.send_to(id, "one".into()).await); + assert_eq!(rx.try_recv().unwrap(), "one"); - workers.remove_worker(id); - assert!(!workers.try_send_to(id, "after-remove".into())); + drop(rx); + assert!(!workers.send_to(id, "closed".into()).await); + + workers.remove_worker(id); + assert!(!workers.send_to(id, "after-remove".into()).await); + }); } #[test] @@ -1290,6 +1296,57 @@ mod tests { }); } + #[test] + fn test_pipelined_requests_apply_backpressure() { + let rt = Runtime::new().unwrap(); + rt.block_on(async { + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; + + const REQUEST_COUNT: usize = 200; + + let server = StdTcpListener::bind("127.0.0.1:0").unwrap(); + let addr = server.local_addr().unwrap(); + let mut client = TcpStream::connect(addr).await.unwrap(); + let (server_socket, _) = server.accept().unwrap(); + server_socket.set_nonblocking(true).unwrap(); + let server_socket = TcpStream::from_std(server_socket).unwrap(); + let handler = setup_handler(".grin_stratum_pipeline_test"); + + let task = tokio::spawn(handle_connection_with_idle_timeout( + server_socket, + handler.clone(), + Duration::from_secs(5), + )); + + let requests = (0..REQUEST_COUNT) + .map(|id| { + format!( + r#"{{"id":{},"jsonrpc":"2.0","method":"keepalive","params":null}}"#, + id + ) + }) + .collect::>() + .join("\n") + "\n"; + client.write_all(requests.as_bytes()).await.unwrap(); + + let mut lines = BufReader::new(client).lines(); + for expected_id in 0..REQUEST_COUNT { + let line = timeout(Duration::from_secs(5), lines.next_line()) + .await + .unwrap() + .unwrap() + .unwrap(); + let response: Value = serde_json::from_str(&line).unwrap(); + assert_eq!(response["id"], expected_id); + } + assert_eq!(handler.workers.count(), 1); + + drop(lines); + task.await.unwrap(); + assert_eq!(handler.workers.count(), 0); + }); + } + fn setup_handler(dir: &str) -> Arc { global::set_local_chain_type(ChainTypes::AutomatedTesting); let _ = std::fs::remove_dir_all(dir); From 2ec5b4a306ca9eacf5991fa65924dec7779fbca8 Mon Sep 17 00:00:00 2001 From: wiesche Date: Sun, 26 Jul 2026 16:02:54 +0200 Subject: [PATCH 03/13] fix worker removal and disconnect handling --- servers/src/mining/stratumserver.rs | 78 +++++++++++++++++------------ 1 file changed, 46 insertions(+), 32 deletions(-) diff --git a/servers/src/mining/stratumserver.rs b/servers/src/mining/stratumserver.rs index 6cf1cae14f..0479a70011 100644 --- a/servers/src/mining/stratumserver.rs +++ b/servers/src/mining/stratumserver.rs @@ -842,21 +842,16 @@ impl WorkersList { stratum_stats.num_workers = workers_list.len(); worker_id } + pub fn remove_worker(&self, worker_id: usize) { + let mut stratum_stats = self.stratum_stats.write(); let mut workers_list = self.workers_list.write(); - if workers_list.remove(&worker_id).is_none() { - let mut stratum_stats = self.stratum_stats.write(); - stratum_stats.num_workers = workers_list.len(); - return; + if workers_list.remove(&worker_id).is_some() { + let worker_stats = &mut stratum_stats.worker_stats[worker_id]; + worker_stats.is_connected = false; + worker_stats.last_seen = SystemTime::now(); } - drop(workers_list); - - self.update_stats(worker_id, |ws| { - ws.is_connected = false; - ws.last_seen = SystemTime::now(); - }); - let mut stratum_stats = self.stratum_stats.write(); - stratum_stats.num_workers = self.workers_list.read().len(); + stratum_stats.num_workers = workers_list.len(); } pub fn login(&self, worker_id: usize, login: String, agent: String) -> Result<(), RpcError> { @@ -912,33 +907,24 @@ impl WorkersList { tx.send(msg).await.is_ok() } - pub fn disconnect_worker(&self, worker_id: usize) { - let shutdown_tx = self - .workers_list - .read() - .get(&worker_id) - .map(|worker| worker.shutdown_tx.clone()); - if let Some(shutdown_tx) = shutdown_tx { - let _ = shutdown_tx.try_send(()); + fn queue_broadcast(&self, msg: &str) -> Vec<(usize, mpsc::Sender<()>)> { + let mut slow_workers = Vec::new(); + let workers_list = self.workers_list.read(); + for (worker_id, worker) in workers_list.iter() { + if worker.tx.try_send(msg.to_owned()).is_err() { + slow_workers.push((*worker_id, worker.shutdown_tx.clone())); + } } + slow_workers } pub fn broadcast(&self, msg: String) { - let mut slow_workers = Vec::new(); - { - let workers_list = self.workers_list.read(); - for (worker_id, worker) in workers_list.iter() { - if worker.tx.try_send(msg.clone()).is_err() { - slow_workers.push(*worker_id); - } - } - } - for worker_id in slow_workers { + for (worker_id, shutdown_tx) in self.queue_broadcast(&msg) { warn!( "Stratum: dropping slow or disconnected worker {}", worker_id ); - self.disconnect_worker(worker_id); + let _ = shutdown_tx.try_send(()); } } @@ -1196,12 +1182,40 @@ mod tests { #[test] fn test_remove_worker_is_idempotent() { let stats = Arc::new(RwLock::new(StratumStats::default())); - let workers = WorkersList::new(stats); + let workers = WorkersList::new(stats.clone()); let (tx, _rx, shutdown_tx) = dummy_tx(); let id = workers.add_worker(tx, shutdown_tx); workers.remove_worker(id); workers.remove_worker(id); assert_eq!(workers.count(), 0); + assert_eq!(stats.read().num_workers, 0); + assert!(!stats.read().worker_stats[id].is_connected); + } + + #[test] + fn test_slow_worker_slot_reuse() { + let stats = Arc::new(RwLock::new(StratumStats::default())); + let workers = WorkersList::new(stats); + + let (old_tx, _old_rx) = mpsc::channel(1); + old_tx.try_send("queued".into()).unwrap(); + let (old_shutdown_tx, mut old_shutdown_rx) = mpsc::channel(1); + let old_id = workers.add_worker(old_tx, old_shutdown_tx); + + let slow_workers = workers.queue_broadcast("next job"); + assert_eq!(slow_workers.len(), 1); + + workers.remove_worker(old_id); + let (new_tx, _new_rx) = mpsc::channel(1); + let (new_shutdown_tx, mut new_shutdown_rx) = mpsc::channel(1); + let new_id = workers.add_worker(new_tx, new_shutdown_tx); + assert_eq!(new_id, old_id); + + let (slow_worker_id, shutdown_tx) = slow_workers.into_iter().next().unwrap(); + assert_eq!(slow_worker_id, old_id); + shutdown_tx.try_send(()).unwrap(); + assert_eq!(old_shutdown_rx.try_recv(), Ok(())); + assert!(new_shutdown_rx.try_recv().is_err()); } #[test] From 979ede248b96c2a22d0720f6c317e3e305784671 Mon Sep 17 00:00:00 2001 From: wiesche Date: Sun, 26 Jul 2026 16:14:25 +0200 Subject: [PATCH 04/13] make stratum connection limit configurable --- config/src/comments.rs | 8 +++++++ servers/src/common/types.rs | 23 +++++++++++++++++++ servers/src/grin/server.rs | 1 + servers/src/mining/stratumserver.rs | 34 ++++++++++++++++------------- 4 files changed, 51 insertions(+), 15 deletions(-) diff --git a/config/src/comments.rs b/config/src/comments.rs index 0e9ac707da..be489f6d63 100644 --- a/config/src/comments.rs +++ b/config/src/comments.rs @@ -398,6 +398,14 @@ fn comments() -> HashMap { .to_string(), ); + retval.insert( + "max_workers".to_string(), + " +#maximum number of concurrent stratum workers +" + .to_string(), + ); + retval.insert( "attempt_time_per_block".to_string(), " diff --git a/servers/src/common/types.rs b/servers/src/common/types.rs index 445a57edab..bb85399664 100644 --- a/servers/src/common/types.rs +++ b/servers/src/common/types.rs @@ -246,6 +246,10 @@ pub struct StratumServerConfig { /// If enabled, the address and port to listen on pub stratum_server_addr: Option, + /// Maximum number of concurrent stratum workers + #[serde(default = "default_stratum_max_workers")] + pub max_workers: usize, + /// How long to wait before stopping the miner, recollecting transactions /// and starting again pub attempt_time_per_block: u32, @@ -261,11 +265,16 @@ pub struct StratumServerConfig { pub burn_reward: bool, } +fn default_stratum_max_workers() -> usize { + 256 +} + impl Default for StratumServerConfig { fn default() -> StratumServerConfig { StratumServerConfig { wallet_listener_url: "http://127.0.0.1:3415".to_string(), burn_reward: false, + max_workers: default_stratum_max_workers(), attempt_time_per_block: 15, minimum_share_difficulty: 1, enable_stratum_server: Some(false), @@ -428,3 +437,17 @@ pub enum NetAdapterWorkerMessage { /// Received PIBD segment. PIBDSegment(QueuedPIBDSegment), } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn stratum_max_workers_default() { + let mut value = serde_json::to_value(StratumServerConfig::default()).unwrap(); + value.as_object_mut().unwrap().remove("max_workers"); + + let config: StratumServerConfig = serde_json::from_value(value).unwrap(); + assert_eq!(config.max_workers, 256); + } +} diff --git a/servers/src/grin/server.rs b/servers/src/grin/server.rs index f486350345..f0abfad383 100644 --- a/servers/src/grin/server.rs +++ b/servers/src/grin/server.rs @@ -412,6 +412,7 @@ impl Server { stratum_server_addr: None, wallet_listener_url: config_wallet_url, minimum_share_difficulty: 1, + ..StratumServerConfig::default() }; let mut miner = Miner::new( diff --git a/servers/src/mining/stratumserver.rs b/servers/src/mining/stratumserver.rs index 0479a70011..fff3607e16 100644 --- a/servers/src/mining/stratumserver.rs +++ b/servers/src/mining/stratumserver.rs @@ -46,7 +46,6 @@ use crate::ServerTxPool; type Tx = mpsc::Sender; -const MAX_STRATUM_WORKERS: usize = 256; const WORKER_QUEUE_SIZE: usize = 64; const WORKER_IDLE_TIMEOUT: Duration = Duration::from_secs(5 * 60); const WORKER_WRITE_TIMEOUT: Duration = Duration::from_secs(30); @@ -721,9 +720,9 @@ async fn handle_connection_with_idle_timeout( } } -async fn accept_connections_loop(listener: TcpListener, handler: Arc) { +async fn accept_connections_loop(listener: TcpListener, handler: Arc, max_workers: usize) { let mut connections = JoinSet::new(); - let worker_limit = Arc::new(Semaphore::new(MAX_STRATUM_WORKERS)); + let worker_limit = Arc::new(Semaphore::new(max_workers)); loop { tokio::select! { accepted = listener.accept() => { @@ -734,7 +733,7 @@ async fn accept_connections_loop(listener: TcpListener, handler: Arc) { Err(_) => { warn!( "Stratum: rejecting connection from {} (max workers: {})", - peer_addr, MAX_STRATUM_WORKERS + peer_addr, max_workers ); drop(socket); continue; @@ -763,13 +762,13 @@ async fn accept_connections_loop(listener: TcpListener, handler: Arc) { } } -fn accept_connections(listen_addr: SocketAddr, handler: Arc) { +fn accept_connections(listen_addr: SocketAddr, handler: Arc, max_workers: usize) { info!("Start tokio stratum server"); let task = async move { let listener = TcpListener::bind(&listen_addr).await.unwrap_or_else(|_| { panic!("Stratum: Failed to bind to listen address {}", listen_addr) }); - accept_connections_loop(listener, handler).await; + accept_connections_loop(listener, handler, max_workers).await; }; let rt = Runtime::new().unwrap(); @@ -1012,9 +1011,10 @@ impl StratumServer { let handler = Arc::new(Handler::from_stratum(&self)); let h = handler.clone(); + let max_workers = self.config.max_workers; let _listener_th = thread::spawn(move || { - accept_connections(listen_addr, h); + accept_connections(listen_addr, h, max_workers); }); // We have started @@ -1226,7 +1226,7 @@ mod tests { let addr = listener.local_addr().unwrap(); let handler = setup_handler(".grin_stratum_accept_loop_test"); - let task = tokio::spawn(accept_connections_loop(listener, handler.clone())); + let task = tokio::spawn(accept_connections_loop(listener, handler.clone(), 1)); let client = tokio::net::TcpStream::connect(addr).await.unwrap(); for _ in 0..100 { if handler.workers.count() == 1 { @@ -1251,23 +1251,27 @@ mod tests { fn test_accept_loop_enforces_max_connection_limit() { let rt = Runtime::new().unwrap(); rt.block_on(async { + const MAX_TEST_WORKERS: usize = 8; + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); let handler = setup_handler(".grin_stratum_max_workers_test"); - let task = tokio::spawn(accept_connections_loop(listener, handler.clone())); + let task = tokio::spawn(accept_connections_loop( + listener, + handler.clone(), + MAX_TEST_WORKERS, + )); let mut clients = Vec::new(); - for _ in 0..(MAX_STRATUM_WORKERS + 8) { - if let Ok(client) = tokio::net::TcpStream::connect(addr).await { - clients.push(client); - } + for _ in 0..(MAX_TEST_WORKERS + 8) { + clients.push(tokio::net::TcpStream::connect(addr).await.unwrap()); } for _ in 0..100 { - if handler.workers.count() == MAX_STRATUM_WORKERS { + if handler.workers.count() == MAX_TEST_WORKERS { break; } tokio::time::sleep(Duration::from_millis(10)).await; } - assert_eq!(handler.workers.count(), MAX_STRATUM_WORKERS); + assert_eq!(handler.workers.count(), MAX_TEST_WORKERS); drop(clients); for _ in 0..100 { if handler.workers.count() == 0 { From a2693d0418dbbb39011b8934c2a8d257f3e512a3 Mon Sep 17 00:00:00 2001 From: wiesche Date: Sun, 26 Jul 2026 16:31:48 +0200 Subject: [PATCH 05/13] align stratum idle handling with node sync --- config/src/comments.rs | 9 +++++ servers/src/common/types.rs | 17 +++++++-- servers/src/mining/stratumserver.rs | 53 +++++++++++++++++++++++------ 3 files changed, 67 insertions(+), 12 deletions(-) diff --git a/config/src/comments.rs b/config/src/comments.rs index be489f6d63..e4a5cbc45b 100644 --- a/config/src/comments.rs +++ b/config/src/comments.rs @@ -406,6 +406,15 @@ fn comments() -> HashMap { .to_string(), ); + retval.insert( + "worker_idle_timeout_secs".to_string(), + " +#disconnect workers after this many seconds without traffic +#must be greater than zero and should exceed attempt_time_per_block +" + .to_string(), + ); + retval.insert( "attempt_time_per_block".to_string(), " diff --git a/servers/src/common/types.rs b/servers/src/common/types.rs index bb85399664..85c4e40d52 100644 --- a/servers/src/common/types.rs +++ b/servers/src/common/types.rs @@ -250,6 +250,11 @@ pub struct StratumServerConfig { #[serde(default = "default_stratum_max_workers")] pub max_workers: usize, + /// Disconnect workers after this many seconds without traffic. Must be + /// greater than zero. + #[serde(default = "default_stratum_worker_idle_timeout_secs")] + pub worker_idle_timeout_secs: u64, + /// How long to wait before stopping the miner, recollecting transactions /// and starting again pub attempt_time_per_block: u32, @@ -269,12 +274,17 @@ fn default_stratum_max_workers() -> usize { 256 } +fn default_stratum_worker_idle_timeout_secs() -> u64 { + 5 * 60 +} + impl Default for StratumServerConfig { fn default() -> StratumServerConfig { StratumServerConfig { wallet_listener_url: "http://127.0.0.1:3415".to_string(), burn_reward: false, max_workers: default_stratum_max_workers(), + worker_idle_timeout_secs: default_stratum_worker_idle_timeout_secs(), attempt_time_per_block: 15, minimum_share_difficulty: 1, enable_stratum_server: Some(false), @@ -443,11 +453,14 @@ mod tests { use super::*; #[test] - fn stratum_max_workers_default() { + fn stratum_config_defaults() { let mut value = serde_json::to_value(StratumServerConfig::default()).unwrap(); - value.as_object_mut().unwrap().remove("max_workers"); + let config = value.as_object_mut().unwrap(); + config.remove("max_workers"); + config.remove("worker_idle_timeout_secs"); let config: StratumServerConfig = serde_json::from_value(value).unwrap(); assert_eq!(config.max_workers, 256); + assert_eq!(config.worker_idle_timeout_secs, 5 * 60); } } diff --git a/servers/src/mining/stratumserver.rs b/servers/src/mining/stratumserver.rs index fff3607e16..a5afd5d645 100644 --- a/servers/src/mining/stratumserver.rs +++ b/servers/src/mining/stratumserver.rs @@ -47,7 +47,6 @@ use crate::ServerTxPool; type Tx = mpsc::Sender; const WORKER_QUEUE_SIZE: usize = 64; -const WORKER_IDLE_TIMEOUT: Duration = Duration::from_secs(5 * 60); const WORKER_WRITE_TIMEOUT: Duration = Duration::from_secs(30); const ACCEPT_ERROR_BACKOFF: Duration = Duration::from_millis(100); const MAX_RPC_LINE_BYTES: usize = 64 * 1024; @@ -627,8 +626,9 @@ async fn handle_connection( socket: TcpStream, handler: Arc, _permit: OwnedSemaphorePermit, + idle_timeout: Duration, ) { - handle_connection_with_idle_timeout(socket, handler, WORKER_IDLE_TIMEOUT).await; + handle_connection_with_idle_timeout(socket, handler, idle_timeout).await; } async fn handle_connection_with_idle_timeout( @@ -705,6 +705,10 @@ async fn handle_connection_with_idle_timeout( _ = &mut read => break, _ = &mut write => break, _ = &mut idle_sleep => { + if handler.sync_state.is_syncing() { + idle_sleep.as_mut().reset(Instant::now() + idle_timeout); + continue; + } warn!("Worker {} idle for {:?}; disconnecting", worker_id, idle_timeout); break; } @@ -720,7 +724,12 @@ async fn handle_connection_with_idle_timeout( } } -async fn accept_connections_loop(listener: TcpListener, handler: Arc, max_workers: usize) { +async fn accept_connections_loop( + listener: TcpListener, + handler: Arc, + max_workers: usize, + idle_timeout: Duration, +) { let mut connections = JoinSet::new(); let worker_limit = Arc::new(Semaphore::new(max_workers)); loop { @@ -744,7 +753,7 @@ async fn accept_connections_loop(listener: TcpListener, handler: Arc, m if let Err(e) = socket.set_nodelay(true) { debug!("Stratum: set_nodelay failed for {}: {}", peer_addr, e); } - handle_connection(socket, handler, permit).await; + handle_connection(socket, handler, permit, idle_timeout).await; }); } Err(e) => { @@ -762,13 +771,18 @@ async fn accept_connections_loop(listener: TcpListener, handler: Arc, m } } -fn accept_connections(listen_addr: SocketAddr, handler: Arc, max_workers: usize) { +fn accept_connections( + listen_addr: SocketAddr, + handler: Arc, + max_workers: usize, + idle_timeout: Duration, +) { info!("Start tokio stratum server"); let task = async move { let listener = TcpListener::bind(&listen_addr).await.unwrap_or_else(|_| { panic!("Stratum: Failed to bind to listen address {}", listen_addr) }); - accept_connections_loop(listener, handler, max_workers).await; + accept_connections_loop(listener, handler, max_workers, idle_timeout).await; }; let rt = Runtime::new().unwrap(); @@ -1012,9 +1026,14 @@ impl StratumServer { let handler = Arc::new(Handler::from_stratum(&self)); let h = handler.clone(); let max_workers = self.config.max_workers; + let idle_timeout = Duration::from_secs(self.config.worker_idle_timeout_secs); + assert!( + !idle_timeout.is_zero(), + "Stratum: worker_idle_timeout_secs must be greater than zero" + ); let _listener_th = thread::spawn(move || { - accept_connections(listen_addr, h, max_workers); + accept_connections(listen_addr, h, max_workers, idle_timeout); }); // We have started @@ -1226,7 +1245,12 @@ mod tests { let addr = listener.local_addr().unwrap(); let handler = setup_handler(".grin_stratum_accept_loop_test"); - let task = tokio::spawn(accept_connections_loop(listener, handler.clone(), 1)); + let task = tokio::spawn(accept_connections_loop( + listener, + handler.clone(), + 1, + Duration::from_secs(StratumServerConfig::default().worker_idle_timeout_secs), + )); let client = tokio::net::TcpStream::connect(addr).await.unwrap(); for _ in 0..100 { if handler.workers.count() == 1 { @@ -1260,6 +1284,7 @@ mod tests { listener, handler.clone(), MAX_TEST_WORKERS, + Duration::from_secs(StratumServerConfig::default().worker_idle_timeout_secs), )); let mut clients = Vec::new(); for _ in 0..(MAX_TEST_WORKERS + 8) { @@ -1285,7 +1310,7 @@ mod tests { } #[test] - fn test_idle_connection_is_disconnected() { + fn test_idle_timeout_after_sync() { let rt = Runtime::new().unwrap(); rt.block_on(async { let server = StdTcpListener::bind("127.0.0.1:0").unwrap(); @@ -1295,6 +1320,7 @@ mod tests { server_socket.set_nonblocking(true).unwrap(); let server_socket = TcpStream::from_std(server_socket).unwrap(); let handler = setup_handler(".grin_stratum_idle_test"); + handler.sync_state.update(SyncStatus::Initial); let task = tokio::spawn(handle_connection_with_idle_timeout( server_socket, @@ -1308,7 +1334,14 @@ mod tests { tokio::time::sleep(Duration::from_millis(10)).await; } assert_eq!(handler.workers.count(), 1); - task.await.unwrap(); + tokio::time::sleep(Duration::from_millis(120)).await; + assert_eq!(handler.workers.count(), 1); + + handler.sync_state.update(SyncStatus::NoSync); + timeout(Duration::from_millis(500), task) + .await + .unwrap() + .unwrap(); assert_eq!(handler.workers.count(), 0); drop(client); }); From 5f1c89849372738d45b20853891a266aa9d3cef1 Mon Sep 17 00:00:00 2001 From: wiesche Date: Sun, 26 Jul 2026 16:44:50 +0200 Subject: [PATCH 06/13] stop stratum tasks on server shutdown --- servers/src/grin/server.rs | 22 +++-- servers/src/mining/stratumserver.rs | 130 +++++++++++++++++++++------- 2 files changed, 114 insertions(+), 38 deletions(-) diff --git a/servers/src/grin/server.rs b/servers/src/grin/server.rs index f0abfad383..d511f079a8 100644 --- a/servers/src/grin/server.rs +++ b/servers/src/grin/server.rs @@ -76,6 +76,7 @@ pub struct Server { connect_thread: Option>, sync_thread: JoinHandle<()>, dandelion_thread: JoinHandle<()>, + stratum_thread: Option>, } impl Server { @@ -93,7 +94,7 @@ impl Server { let mining_config = config.stratum_mining_config.clone(); let enable_test_miner = config.run_test_miner; let test_miner_wallet_url = config.test_miner_wallet_url.clone(); - let serv = Server::new(config, stop_state, server_tx, api_chan)?; + let mut serv = Server::new(config, stop_state, server_tx, api_chan)?; if let Some(c) = mining_config { let enable_stratum_server = c.enable_stratum_server; @@ -345,6 +346,7 @@ impl Server { connect_thread, sync_thread, dandelion_thread, + stratum_thread: None, }) } @@ -374,9 +376,10 @@ impl Server { } /// Start a minimal "stratum" mining service on a separate thread - pub fn start_stratum_server(&self, config: StratumServerConfig) { + pub fn start_stratum_server(&mut self, config: StratumServerConfig) { let proof_size = global::proofsize(); let sync_state = self.sync_state.clone(); + let stop_state = self.stop_state.clone(); let mut stratum_server = stratumserver::StratumServer::new( config, @@ -384,11 +387,13 @@ impl Server { self.tx_pool.clone(), self.state_info.stratum_stats.clone(), ); - let _ = thread::Builder::new() + self.stratum_thread = thread::Builder::new() .name("stratum_server".to_string()) .spawn(move || { - stratum_server.run_loop(proof_size, sync_state); - }); + stratum_server.run_loop(proof_size, sync_state, stop_state); + }) + .map_err(|e| error!("Failed to start stratum server thread: {}", e)) + .ok(); } /// Start mining for blocks internally on a separate thread. Relies on @@ -574,6 +579,13 @@ impl Server { info!("No active connect_and_monitor thread") } + if let Some(stratum_thread) = self.stratum_thread { + match stratum_thread.join() { + Err(e) => error!("failed to join stratum server thread: {:?}", e), + Ok(_) => info!("stratum server thread stopped"), + } + } + match self.sync_thread.join() { Err(e) => error!("failed to join to sync thread: {:?}", e), Ok(_) => info!("sync thread stopped"), diff --git a/servers/src/mining/stratumserver.rs b/servers/src/mining/stratumserver.rs index a5afd5d645..b9468b20f1 100644 --- a/servers/src/mining/stratumserver.rs +++ b/servers/src/mining/stratumserver.rs @@ -22,7 +22,7 @@ use tokio::task::JoinSet; use tokio::time::{timeout, Instant}; use tokio_util::codec::{Framed, LinesCodec}; -use crate::util::RwLock; +use crate::util::{RwLock, StopState}; use chrono::prelude::Utc; use serde_json::Value; use std::collections::HashMap; @@ -49,6 +49,7 @@ type Tx = mpsc::Sender; const WORKER_QUEUE_SIZE: usize = 64; const WORKER_WRITE_TIMEOUT: Duration = Duration::from_secs(30); const ACCEPT_ERROR_BACKOFF: Duration = Duration::from_millis(100); +const SHUTDOWN_POLL_INTERVAL: Duration = Duration::from_millis(100); const MAX_RPC_LINE_BYTES: usize = 64 * 1024; // ---------------------------------------- @@ -531,12 +532,17 @@ impl Handler { self.workers.broadcast(job_request_json); } - pub fn run(&self, config: &StratumServerConfig, tx_pool: &ServerTxPool) { + pub fn run( + &self, + config: &StratumServerConfig, + tx_pool: &ServerTxPool, + stop_state: Arc, + ) { debug!("Run main loop"); let mut deadline: i64 = 0; let mut head = self.chain.head().unwrap(); let mut current_hash = head.prev_block_h; - loop { + while !stop_state.is_stopped() { // get the latest chain state head = self.chain.head().unwrap(); let latest_hash = head.last_block_h; @@ -729,11 +735,18 @@ async fn accept_connections_loop( handler: Arc, max_workers: usize, idle_timeout: Duration, + stop_state: Arc, ) { let mut connections = JoinSet::new(); let worker_limit = Arc::new(Semaphore::new(max_workers)); + let mut shutdown_poll = tokio::time::interval(SHUTDOWN_POLL_INTERVAL); loop { tokio::select! { + _ = shutdown_poll.tick() => { + if stop_state.is_stopped() { + break; + } + } accepted = listener.accept() => { match accepted { Ok((socket, peer_addr)) => { @@ -769,6 +782,7 @@ async fn accept_connections_loop( } } } + connections.shutdown().await; } fn accept_connections( @@ -776,13 +790,14 @@ fn accept_connections( handler: Arc, max_workers: usize, idle_timeout: Duration, + stop_state: Arc, ) { info!("Start tokio stratum server"); let task = async move { let listener = TcpListener::bind(&listen_addr).await.unwrap_or_else(|_| { panic!("Stratum: Failed to bind to listen address {}", listen_addr) }); - accept_connections_loop(listener, handler, max_workers, idle_timeout).await; + accept_connections_loop(listener, handler, max_workers, idle_timeout, stop_state).await; }; let rt = Runtime::new().unwrap(); @@ -1007,7 +1022,12 @@ impl StratumServer { /// existing chain anytime required and sending that to the connected /// stratum miner, proxy, or pool, and accepts full solutions to /// be submitted. - pub fn run_loop(&mut self, proof_size: usize, sync_state: Arc) { + pub fn run_loop( + &mut self, + proof_size: usize, + sync_state: Arc, + stop_state: Arc, + ) { info!( "(Server ID: {}) Starting stratum server with proof_size = {}", self.id, proof_size @@ -1032,8 +1052,15 @@ impl StratumServer { "Stratum: worker_idle_timeout_secs must be greater than zero" ); - let _listener_th = thread::spawn(move || { - accept_connections(listen_addr, h, max_workers, idle_timeout); + let listener_stop_state = stop_state.clone(); + let listener_th = thread::spawn(move || { + accept_connections( + listen_addr, + h, + max_workers, + idle_timeout, + listener_stop_state, + ); }); // We have started @@ -1050,11 +1077,18 @@ impl StratumServer { ); // Initial Loop. Waiting node complete syncing - while self.sync_state.is_syncing() { + while self.sync_state.is_syncing() && !stop_state.is_stopped() { thread::sleep(Duration::from_millis(50)); } - handler.run(&self.config, &self.tx_pool); + if !stop_state.is_stopped() { + handler.run(&self.config, &self.tx_pool, stop_state); + } + + if let Err(e) = listener_th.join() { + error!("failed to join stratum listener thread: {:?}", e); + } + self.stratum_stats.write().is_running = false; } // fn run_loop() } // StratumServer @@ -1078,6 +1112,7 @@ mod tests { use crate::core::pow::Difficulty; use std::fs; use std::net::TcpListener as StdTcpListener; + use std::path::{Path, PathBuf}; use std::sync::OnceLock; // ---------------------------------------- @@ -1141,6 +1176,32 @@ mod tests { serde_json::from_str(json).unwrap() } + struct TestDir { + path: PathBuf, + } + + impl TestDir { + fn new(path: &str) -> Self { + let path = PathBuf::from(path); + let _ = std::fs::remove_dir_all(&path); + Self { path } + } + + fn path(&self) -> &Path { + &self.path + } + } + + impl Drop for TestDir { + fn drop(&mut self) { + if let Err(e) = std::fs::remove_dir_all(&self.path) { + if e.kind() != std::io::ErrorKind::NotFound && !thread::panicking() { + panic!("failed to remove test directory {:?}: {}", self.path, e); + } + } + } + } + fn dummy_tx() -> (Tx, mpsc::Receiver, mpsc::Sender<()>) { let (tx, rx) = mpsc::channel(WORKER_QUEUE_SIZE); let (shutdown_tx, _shutdown_rx) = mpsc::channel(1); @@ -1238,18 +1299,21 @@ mod tests { } #[test] - fn test_accept_loop_tracks_connection_tasks() { + fn test_accept_loop_stops_connection_tasks() { + let test_dir = TestDir::new(".grin_stratum_accept_loop_test"); let rt = Runtime::new().unwrap(); rt.block_on(async { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); - let handler = setup_handler(".grin_stratum_accept_loop_test"); + let handler = setup_handler(test_dir.path()); + let stop_state = Arc::new(StopState::new()); let task = tokio::spawn(accept_connections_loop( listener, handler.clone(), 1, Duration::from_secs(StratumServerConfig::default().worker_idle_timeout_secs), + stop_state.clone(), )); let client = tokio::net::TcpStream::connect(addr).await.unwrap(); for _ in 0..100 { @@ -1259,32 +1323,33 @@ mod tests { tokio::time::sleep(Duration::from_millis(10)).await; } assert_eq!(handler.workers.count(), 1); - drop(client); - for _ in 0..100 { - if handler.workers.count() == 0 { - break; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } + stop_state.stop(); + timeout(Duration::from_millis(500), task) + .await + .unwrap() + .unwrap(); assert_eq!(handler.workers.count(), 0); - task.abort(); + drop(client); }); } #[test] fn test_accept_loop_enforces_max_connection_limit() { + let test_dir = TestDir::new(".grin_stratum_max_workers_test"); let rt = Runtime::new().unwrap(); rt.block_on(async { const MAX_TEST_WORKERS: usize = 8; let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); - let handler = setup_handler(".grin_stratum_max_workers_test"); + let handler = setup_handler(test_dir.path()); + let stop_state = Arc::new(StopState::new()); let task = tokio::spawn(accept_connections_loop( listener, handler.clone(), MAX_TEST_WORKERS, Duration::from_secs(StratumServerConfig::default().worker_idle_timeout_secs), + stop_state.clone(), )); let mut clients = Vec::new(); for _ in 0..(MAX_TEST_WORKERS + 8) { @@ -1297,20 +1362,19 @@ mod tests { tokio::time::sleep(Duration::from_millis(10)).await; } assert_eq!(handler.workers.count(), MAX_TEST_WORKERS); - drop(clients); - for _ in 0..100 { - if handler.workers.count() == 0 { - break; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } + stop_state.stop(); + timeout(Duration::from_millis(500), task) + .await + .unwrap() + .unwrap(); assert_eq!(handler.workers.count(), 0); - task.abort(); + drop(clients); }); } #[test] fn test_idle_timeout_after_sync() { + let test_dir = TestDir::new(".grin_stratum_idle_test"); let rt = Runtime::new().unwrap(); rt.block_on(async { let server = StdTcpListener::bind("127.0.0.1:0").unwrap(); @@ -1319,7 +1383,7 @@ mod tests { let (server_socket, _) = server.accept().unwrap(); server_socket.set_nonblocking(true).unwrap(); let server_socket = TcpStream::from_std(server_socket).unwrap(); - let handler = setup_handler(".grin_stratum_idle_test"); + let handler = setup_handler(test_dir.path()); handler.sync_state.update(SyncStatus::Initial); let task = tokio::spawn(handle_connection_with_idle_timeout( @@ -1349,6 +1413,7 @@ mod tests { #[test] fn test_pipelined_requests_apply_backpressure() { + let test_dir = TestDir::new(".grin_stratum_pipeline_test"); let rt = Runtime::new().unwrap(); rt.block_on(async { use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; @@ -1361,7 +1426,7 @@ mod tests { let (server_socket, _) = server.accept().unwrap(); server_socket.set_nonblocking(true).unwrap(); let server_socket = TcpStream::from_std(server_socket).unwrap(); - let handler = setup_handler(".grin_stratum_pipeline_test"); + let handler = setup_handler(test_dir.path()); let task = tokio::spawn(handle_connection_with_idle_timeout( server_socket, @@ -1398,12 +1463,11 @@ mod tests { }); } - fn setup_handler(dir: &str) -> Arc { + fn setup_handler(dir: &Path) -> Arc { global::set_local_chain_type(ChainTypes::AutomatedTesting); - let _ = std::fs::remove_dir_all(dir); let chain = Arc::new( chain::Chain::init( - dir.to_string(), + dir.to_string_lossy().into_owned(), Arc::new(NoopAdapter {}), genesis::genesis_dev(), pow::verify_size, From 7fd95b322f7522384424241672e7cd31dd5db7fe Mon Sep 17 00:00:00 2001 From: wiesche Date: Sun, 26 Jul 2026 19:11:47 +0200 Subject: [PATCH 07/13] fix stratum block retries --- servers/src/grin/server.rs | 78 ++++++++++++++++++++++++++--- servers/src/mining/mine_block.rs | 53 ++++++++++++++++++-- servers/src/mining/stratumserver.rs | 73 +++++++++++++++++++++++---- 3 files changed, 182 insertions(+), 22 deletions(-) diff --git a/servers/src/grin/server.rs b/servers/src/grin/server.rs index d511f079a8..f881842654 100644 --- a/servers/src/grin/server.rs +++ b/servers/src/grin/server.rs @@ -76,7 +76,7 @@ pub struct Server { connect_thread: Option>, sync_thread: JoinHandle<()>, dandelion_thread: JoinHandle<()>, - stratum_thread: Option>, + stratum_thread: RwLock>>, } impl Server { @@ -94,7 +94,13 @@ impl Server { let mining_config = config.stratum_mining_config.clone(); let enable_test_miner = config.run_test_miner; let test_miner_wallet_url = config.test_miner_wallet_url.clone(); - let mut serv = Server::new(config, stop_state, server_tx, api_chan)?; + if let Some(c) = mining_config + .as_ref() + .filter(|c| c.enable_stratum_server == Some(true)) + { + validate_stratum_config(c)?; + } + let serv = Server::new(config, stop_state, server_tx, api_chan)?; if let Some(c) = mining_config { let enable_stratum_server = c.enable_stratum_server; @@ -346,7 +352,7 @@ impl Server { connect_thread, sync_thread, dandelion_thread, - stratum_thread: None, + stratum_thread: RwLock::new(None), }) } @@ -376,7 +382,11 @@ impl Server { } /// Start a minimal "stratum" mining service on a separate thread - pub fn start_stratum_server(&mut self, config: StratumServerConfig) { + pub fn start_stratum_server(&self, config: StratumServerConfig) { + if let Err(e) = validate_stratum_config(&config) { + error!("Invalid stratum server configuration: {:?}", e); + return; + } let proof_size = global::proofsize(); let sync_state = self.sync_state.clone(); let stop_state = self.stop_state.clone(); @@ -387,7 +397,7 @@ impl Server { self.tx_pool.clone(), self.state_info.stratum_stats.clone(), ); - self.stratum_thread = thread::Builder::new() + *self.stratum_thread.write() = thread::Builder::new() .name("stratum_server".to_string()) .spawn(move || { stratum_server.run_loop(proof_size, sync_state, stop_state); @@ -579,7 +589,7 @@ impl Server { info!("No active connect_and_monitor thread") } - if let Some(stratum_thread) = self.stratum_thread { + if let Some(stratum_thread) = self.stratum_thread.into_inner() { match stratum_thread.join() { Err(e) => error!("failed to join stratum server thread: {:?}", e), Ok(_) => info!("stratum server thread stopped"), @@ -622,3 +632,59 @@ impl Server { info!("stop_test_miner - stop",); } } + +fn validate_stratum_config(config: &StratumServerConfig) -> Result<(), Error> { + if config.max_workers == 0 { + return Err(Error::Configuration( + "stratum max_workers must be greater than zero".to_string(), + )); + } + if config.worker_idle_timeout_secs == 0 { + return Err(Error::Configuration( + "stratum worker_idle_timeout_secs must be greater than zero".to_string(), + )); + } + if time::Instant::now() + .checked_add(Duration::from_secs(config.worker_idle_timeout_secs)) + .is_none() + { + return Err(Error::Configuration( + "stratum worker_idle_timeout_secs is too large".to_string(), + )); + } + let address = config.stratum_server_addr.as_deref().ok_or_else(|| { + Error::Configuration("stratum_server_addr must be configured".to_string()) + })?; + address.parse::().map_err(|e| { + Error::Configuration(format!("invalid stratum_server_addr '{}': {}", address, e)) + })?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_stratum_config_validation() { + assert!(validate_stratum_config(&StratumServerConfig::default()).is_ok()); + + let mut config = StratumServerConfig::default(); + config.max_workers = 0; + assert!(validate_stratum_config(&config).is_err()); + + config.max_workers = 1; + config.worker_idle_timeout_secs = 0; + assert!(validate_stratum_config(&config).is_err()); + + config.worker_idle_timeout_secs = u64::MAX; + assert!(validate_stratum_config(&config).is_err()); + + config.worker_idle_timeout_secs = 1; + config.stratum_server_addr = Some("invalid".to_string()); + assert!(validate_stratum_config(&config).is_err()); + + config.stratum_server_addr = None; + assert!(validate_stratum_config(&config).is_err()); + } +} diff --git a/servers/src/mining/mine_block.rs b/servers/src/mining/mine_block.rs index 16c2bc1816..8eb4d0585b 100644 --- a/servers/src/mining/mine_block.rs +++ b/servers/src/mining/mine_block.rs @@ -20,7 +20,7 @@ use rand::{thread_rng, Rng}; use serde_json::{json, Value}; use std::sync::Arc; use std::thread; -use std::time::Duration; +use std::time::{Duration, Instant}; use crate::api; use crate::chain; @@ -30,8 +30,11 @@ use crate::core::libtx::secp_ser; use crate::core::libtx::ProofBuilder; use crate::core::{consensus, core, global}; use crate::keychain::{ExtKeychain, Identifier, Keychain}; +use crate::util::StopState; use crate::ServerTxPool; +const STOP_POLL_INTERVAL: Duration = Duration::from_millis(100); + /// Fees in block to use for coinbase amount calculation /// (Duplicated from Grin wallet project) #[derive(Serialize, Deserialize, Debug, Clone)] @@ -73,10 +76,36 @@ pub fn get_block( key_id: Option, wallet_listener_url: Option, ) -> (core::Block, BlockFees) { + get_block_internal(chain, tx_pool, key_id, wallet_listener_url, || false).unwrap() +} + +/// Build a block unless shutdown is requested while retrying. +pub(crate) fn get_block_with_stop( + chain: &Arc, + tx_pool: &ServerTxPool, + key_id: Option, + wallet_listener_url: Option, + stop_state: &StopState, +) -> Option<(core::Block, BlockFees)> { + get_block_internal(chain, tx_pool, key_id, wallet_listener_url, || { + stop_state.is_stopped() + }) +} + +fn get_block_internal( + chain: &Arc, + tx_pool: &ServerTxPool, + key_id: Option, + wallet_listener_url: Option, + should_stop: impl Fn() -> bool, +) -> Option<(core::Block, BlockFees)> { let wallet_retry_interval = 5; // get the latest chain state and build a block on top of it let mut result = build_block(chain, tx_pool, key_id.clone(), wallet_listener_url.clone()); while let Err(e) = result { + if should_stop() { + return None; + } let mut new_key_id = key_id.to_owned(); match e { self::Error::Chain(c) => match c { @@ -96,7 +125,9 @@ pub fn get_block( "Error building new block: Can't connect to wallet listener at {:?}; will retry", wallet_listener_url.as_ref().unwrap() ); - thread::sleep(Duration::from_secs(wallet_retry_interval)); + if sleep_or_stop(Duration::from_secs(wallet_retry_interval), &should_stop) { + return None; + } } ae => { warn!("Error building new block: {:?}. Retrying.", ae); @@ -105,13 +136,25 @@ pub fn get_block( // only wait if we are still using the same key: a different coinbase commitment is unlikely // to have duplication - if new_key_id.is_some() { - thread::sleep(Duration::from_millis(100)); + if new_key_id.is_some() && sleep_or_stop(Duration::from_millis(100), &should_stop) { + return None; } result = build_block(chain, tx_pool, new_key_id, wallet_listener_url.clone()); } - return result.unwrap(); + result.ok() +} + +fn sleep_or_stop(duration: Duration, should_stop: &impl Fn() -> bool) -> bool { + let deadline = Instant::now() + duration; + while !should_stop() { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return false; + } + thread::sleep(remaining.min(STOP_POLL_INTERVAL)); + } + true } /// Builds a new block with the chain head as previous and eligible diff --git a/servers/src/mining/stratumserver.rs b/servers/src/mining/stratumserver.rs index b9468b20f1..3cadc295dc 100644 --- a/servers/src/mining/stratumserver.rs +++ b/servers/src/mining/stratumserver.rs @@ -555,22 +555,33 @@ impl Handler { { { debug!("resend updated block"); - let mut state = self.current_state.write(); let wallet_listener_url = if !config.burn_reward { Some(config.wallet_listener_url.clone()) } else { None }; - // If this is a new block we will clear the current_block version history - let clear_blocks = current_hash != latest_hash; + let key_id = self.current_state.read().current_key_id.clone(); // Build the new block (version) - let (new_block, block_fees) = mine_block::get_block( + let Some((new_block, block_fees)) = mine_block::get_block_with_stop( &self.chain, tx_pool, - state.current_key_id.clone(), + key_id, wallet_listener_url, - ); + &stop_state, + ) else { + return; + }; + + let mut state = self.current_state.write(); + head = self.chain.head().unwrap(); + let latest_hash = head.last_block_h; + if new_block.header.prev_hash != latest_hash { + continue; + } + + // If this is a new block we will clear the current_block version history + let clear_blocks = current_hash != latest_hash; // scaled difficulty state.current_difficulty = @@ -753,7 +764,7 @@ async fn accept_connections_loop( let permit = match worker_limit.clone().try_acquire_owned() { Ok(permit) => permit, Err(_) => { - warn!( + debug!( "Stratum: rejecting connection from {} (max workers: {})", peer_addr, max_workers ); @@ -1047,10 +1058,6 @@ impl StratumServer { let h = handler.clone(); let max_workers = self.config.max_workers; let idle_timeout = Duration::from_secs(self.config.worker_idle_timeout_secs); - assert!( - !idle_timeout.is_zero(), - "Stratum: worker_idle_timeout_secs must be greater than zero" - ); let listener_stop_state = stop_state.clone(); let listener_th = thread::spawn(move || { @@ -1463,6 +1470,50 @@ mod tests { }); } + #[test] + fn test_block_retry_stays_responsive() { + use crate::common::adapters::{PoolToChainAdapter, PoolToNetAdapter}; + + let test_dir = TestDir::new(".grin_stratum_shutdown_retry_test"); + let handler = setup_handler(test_dir.path()); + let pool_adapter = Arc::new(PoolToChainAdapter::new()); + pool_adapter.set_chain(handler.chain.clone()); + let pool_net_adapter = Arc::new(PoolToNetAdapter::new( + crate::pool::DandelionConfig::default(), + )); + let tx_pool = Arc::new(RwLock::new(crate::pool::TransactionPool::new( + crate::pool::PoolConfig::default(), + pool_adapter, + pool_net_adapter, + ))); + let (tx, _rx, shutdown_tx) = dummy_tx(); + handler.workers.add_worker(tx, shutdown_tx); + + let mut config = StratumServerConfig::default(); + config.burn_reward = false; + config.wallet_listener_url = "http://127.0.0.1:1".to_string(); + let stop_state = Arc::new(StopState::new()); + let run_stop_state = stop_state.clone(); + let request_handler = handler.clone(); + let (done_tx, done_rx) = std::sync::mpsc::channel(); + thread::spawn(move || { + global::set_local_chain_type(ChainTypes::AutomatedTesting); + handler.run(&config, &tx_pool, run_stop_state); + let _ = done_tx.send(()); + }); + + thread::sleep(Duration::from_millis(250)); + let (request_tx, request_rx) = std::sync::mpsc::channel(); + thread::spawn(move || { + request_handler.build_block_template(); + let _ = request_tx.send(()); + }); + assert_eq!(request_rx.recv_timeout(Duration::from_secs(1)), Ok(())); + + stop_state.stop(); + assert_eq!(done_rx.recv_timeout(Duration::from_secs(1)), Ok(())); + } + fn setup_handler(dir: &Path) -> Arc { global::set_local_chain_type(ChainTypes::AutomatedTesting); let chain = Arc::new( From 96954c373072fb63e98a68d0c63abaced935eaf4 Mon Sep 17 00:00:00 2001 From: wiesche Date: Sun, 26 Jul 2026 19:30:12 +0200 Subject: [PATCH 08/13] simplify stratum tests --- servers/src/mining/stratumserver.rs | 342 +++++++++++++--------------- 1 file changed, 154 insertions(+), 188 deletions(-) diff --git a/servers/src/mining/stratumserver.rs b/servers/src/mining/stratumserver.rs index 3cadc295dc..31385ed7b7 100644 --- a/servers/src/mining/stratumserver.rs +++ b/servers/src/mining/stratumserver.rs @@ -17,7 +17,7 @@ use futures::{SinkExt, StreamExt, TryStreamExt}; use tokio::net::{TcpListener, TcpStream}; use tokio::runtime::Runtime; -use tokio::sync::{mpsc, OwnedSemaphorePermit, Semaphore}; +use tokio::sync::{mpsc, Semaphore}; use tokio::task::JoinSet; use tokio::time::{timeout, Instant}; use tokio_util::codec::{Framed, LinesCodec}; @@ -639,20 +639,7 @@ impl Drop for WorkerCleanup { } } -async fn handle_connection( - socket: TcpStream, - handler: Arc, - _permit: OwnedSemaphorePermit, - idle_timeout: Duration, -) { - handle_connection_with_idle_timeout(socket, handler, idle_timeout).await; -} - -async fn handle_connection_with_idle_timeout( - socket: TcpStream, - handler: Arc, - idle_timeout: Duration, -) { +async fn handle_connection(socket: TcpStream, handler: Arc, idle_timeout: Duration) { let peer_addr = socket.peer_addr().ok(); let (tx, mut rx) = mpsc::channel(WORKER_QUEUE_SIZE); let (shutdown_tx, mut shutdown_rx) = mpsc::channel::<()>(1); @@ -774,10 +761,11 @@ async fn accept_connections_loop( }; let handler = handler.clone(); connections.spawn(async move { + let _permit = permit; if let Err(e) = socket.set_nodelay(true) { debug!("Stratum: set_nodelay failed for {}: {}", peer_addr, e); } - handle_connection(socket, handler, permit, idle_timeout).await; + handle_connection(socket, handler, idle_timeout).await; }); } Err(e) => { @@ -1220,8 +1208,28 @@ mod tests { workers.add_worker(tx, shutdown_tx) } + async fn tcp_pair() -> (TcpStream, TcpStream) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let client = TcpStream::connect(listener.local_addr().unwrap()) + .await + .unwrap(); + let (server, _) = listener.accept().await.unwrap(); + (client, server) + } + + async fn wait_for_worker_count(handler: &Handler, expected: usize) { + timeout(Duration::from_secs(1), async { + while handler.workers.count() != expected { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("worker count did not change in time"); + assert_eq!(handler.workers.count(), expected); + } + #[test] - fn test_worker_slot_reuse_after_disconnect() { + fn test_worker_slot_reuse() { let stats = Arc::new(RwLock::new(StratumStats::default())); let workers = WorkersList::new(stats.clone()); @@ -1243,31 +1251,28 @@ mod tests { assert_eq!(workers.count(), 1); } - #[test] - fn test_send_to_missing_closed_and_ok() { - let rt = Runtime::new().unwrap(); - rt.block_on(async { - let stats = Arc::new(RwLock::new(StratumStats::default())); - let workers = WorkersList::new(stats); + #[tokio::test] + async fn test_worker_send() { + let stats = Arc::new(RwLock::new(StratumStats::default())); + let workers = WorkersList::new(stats); - assert!(!workers.send_to(0, "missing".into()).await); + assert!(!workers.send_to(0, "missing".into()).await); - let (tx, mut rx) = mpsc::channel(1); - let (shutdown_tx, _shutdown_rx) = mpsc::channel(1); - let id = workers.add_worker(tx, shutdown_tx); - assert!(workers.send_to(id, "one".into()).await); - assert_eq!(rx.try_recv().unwrap(), "one"); + let (tx, mut rx) = mpsc::channel(1); + let (shutdown_tx, _shutdown_rx) = mpsc::channel(1); + let id = workers.add_worker(tx, shutdown_tx); + assert!(workers.send_to(id, "one".into()).await); + assert_eq!(rx.try_recv().unwrap(), "one"); - drop(rx); - assert!(!workers.send_to(id, "closed".into()).await); + drop(rx); + assert!(!workers.send_to(id, "closed".into()).await); - workers.remove_worker(id); - assert!(!workers.send_to(id, "after-remove".into()).await); - }); + workers.remove_worker(id); + assert!(!workers.send_to(id, "after-remove".into()).await); } #[test] - fn test_remove_worker_is_idempotent() { + fn test_remove_worker_twice() { let stats = Arc::new(RwLock::new(StratumStats::default())); let workers = WorkersList::new(stats.clone()); let (tx, _rx, shutdown_tx) = dummy_tx(); @@ -1280,7 +1285,7 @@ mod tests { } #[test] - fn test_slow_worker_slot_reuse() { + fn test_stale_worker_shutdown() { let stats = Arc::new(RwLock::new(StratumStats::default())); let workers = WorkersList::new(stats); @@ -1305,173 +1310,134 @@ mod tests { assert!(new_shutdown_rx.try_recv().is_err()); } - #[test] - fn test_accept_loop_stops_connection_tasks() { + #[tokio::test] + async fn test_accept_loop_shutdown() { let test_dir = TestDir::new(".grin_stratum_accept_loop_test"); - let rt = Runtime::new().unwrap(); - rt.block_on(async { - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let addr = listener.local_addr().unwrap(); - let handler = setup_handler(test_dir.path()); - let stop_state = Arc::new(StopState::new()); - - let task = tokio::spawn(accept_connections_loop( - listener, - handler.clone(), - 1, - Duration::from_secs(StratumServerConfig::default().worker_idle_timeout_secs), - stop_state.clone(), - )); - let client = tokio::net::TcpStream::connect(addr).await.unwrap(); - for _ in 0..100 { - if handler.workers.count() == 1 { - break; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } - assert_eq!(handler.workers.count(), 1); - stop_state.stop(); - timeout(Duration::from_millis(500), task) - .await - .unwrap() - .unwrap(); - assert_eq!(handler.workers.count(), 0); - drop(client); - }); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let handler = setup_handler(test_dir.path()); + let stop_state = Arc::new(StopState::new()); + + let task = tokio::spawn(accept_connections_loop( + listener, + handler.clone(), + 1, + Duration::from_secs(StratumServerConfig::default().worker_idle_timeout_secs), + stop_state.clone(), + )); + let client = tokio::net::TcpStream::connect(addr).await.unwrap(); + wait_for_worker_count(&handler, 1).await; + stop_state.stop(); + timeout(Duration::from_millis(500), task) + .await + .unwrap() + .unwrap(); + assert_eq!(handler.workers.count(), 0); + drop(client); } - #[test] - fn test_accept_loop_enforces_max_connection_limit() { + #[tokio::test] + async fn test_worker_limit() { let test_dir = TestDir::new(".grin_stratum_max_workers_test"); - let rt = Runtime::new().unwrap(); - rt.block_on(async { - const MAX_TEST_WORKERS: usize = 8; - - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let addr = listener.local_addr().unwrap(); - let handler = setup_handler(test_dir.path()); - let stop_state = Arc::new(StopState::new()); - let task = tokio::spawn(accept_connections_loop( - listener, - handler.clone(), - MAX_TEST_WORKERS, - Duration::from_secs(StratumServerConfig::default().worker_idle_timeout_secs), - stop_state.clone(), - )); - let mut clients = Vec::new(); - for _ in 0..(MAX_TEST_WORKERS + 8) { - clients.push(tokio::net::TcpStream::connect(addr).await.unwrap()); - } - for _ in 0..100 { - if handler.workers.count() == MAX_TEST_WORKERS { - break; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } - assert_eq!(handler.workers.count(), MAX_TEST_WORKERS); - stop_state.stop(); - timeout(Duration::from_millis(500), task) - .await - .unwrap() - .unwrap(); - assert_eq!(handler.workers.count(), 0); - drop(clients); - }); + const MAX_TEST_WORKERS: usize = 8; + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let handler = setup_handler(test_dir.path()); + let stop_state = Arc::new(StopState::new()); + let task = tokio::spawn(accept_connections_loop( + listener, + handler.clone(), + MAX_TEST_WORKERS, + Duration::from_secs(StratumServerConfig::default().worker_idle_timeout_secs), + stop_state.clone(), + )); + let mut clients = Vec::new(); + for _ in 0..(MAX_TEST_WORKERS + 8) { + clients.push(tokio::net::TcpStream::connect(addr).await.unwrap()); + } + wait_for_worker_count(&handler, MAX_TEST_WORKERS).await; + stop_state.stop(); + timeout(Duration::from_millis(500), task) + .await + .unwrap() + .unwrap(); + assert_eq!(handler.workers.count(), 0); + drop(clients); } - #[test] - fn test_idle_timeout_after_sync() { + #[tokio::test] + async fn test_sync_idle_timeout() { let test_dir = TestDir::new(".grin_stratum_idle_test"); - let rt = Runtime::new().unwrap(); - rt.block_on(async { - let server = StdTcpListener::bind("127.0.0.1:0").unwrap(); - let addr = server.local_addr().unwrap(); - let client = TcpStream::connect(addr).await.unwrap(); - let (server_socket, _) = server.accept().unwrap(); - server_socket.set_nonblocking(true).unwrap(); - let server_socket = TcpStream::from_std(server_socket).unwrap(); - let handler = setup_handler(test_dir.path()); - handler.sync_state.update(SyncStatus::Initial); - - let task = tokio::spawn(handle_connection_with_idle_timeout( - server_socket, - handler.clone(), - Duration::from_millis(50), - )); - for _ in 0..100 { - if handler.workers.count() == 1 { - break; - } - tokio::time::sleep(Duration::from_millis(10)).await; - } - assert_eq!(handler.workers.count(), 1); - tokio::time::sleep(Duration::from_millis(120)).await; - assert_eq!(handler.workers.count(), 1); + let (client, server_socket) = tcp_pair().await; + let handler = setup_handler(test_dir.path()); + handler.sync_state.update(SyncStatus::Initial); - handler.sync_state.update(SyncStatus::NoSync); - timeout(Duration::from_millis(500), task) - .await - .unwrap() - .unwrap(); - assert_eq!(handler.workers.count(), 0); - drop(client); - }); + let task = tokio::spawn(handle_connection( + server_socket, + handler.clone(), + Duration::from_millis(50), + )); + wait_for_worker_count(&handler, 1).await; + tokio::time::sleep(Duration::from_millis(120)).await; + assert_eq!(handler.workers.count(), 1); + + handler.sync_state.update(SyncStatus::NoSync); + timeout(Duration::from_millis(500), task) + .await + .unwrap() + .unwrap(); + assert_eq!(handler.workers.count(), 0); + drop(client); } - #[test] - fn test_pipelined_requests_apply_backpressure() { + #[tokio::test] + async fn test_pipelined_backpressure() { let test_dir = TestDir::new(".grin_stratum_pipeline_test"); - let rt = Runtime::new().unwrap(); - rt.block_on(async { - use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; - - const REQUEST_COUNT: usize = 200; - - let server = StdTcpListener::bind("127.0.0.1:0").unwrap(); - let addr = server.local_addr().unwrap(); - let mut client = TcpStream::connect(addr).await.unwrap(); - let (server_socket, _) = server.accept().unwrap(); - server_socket.set_nonblocking(true).unwrap(); - let server_socket = TcpStream::from_std(server_socket).unwrap(); - let handler = setup_handler(test_dir.path()); - - let task = tokio::spawn(handle_connection_with_idle_timeout( - server_socket, - handler.clone(), - Duration::from_secs(5), - )); - - let requests = (0..REQUEST_COUNT) - .map(|id| { - format!( - r#"{{"id":{},"jsonrpc":"2.0","method":"keepalive","params":null}}"#, - id - ) - }) - .collect::>() - .join("\n") + "\n"; - client.write_all(requests.as_bytes()).await.unwrap(); - - let mut lines = BufReader::new(client).lines(); - for expected_id in 0..REQUEST_COUNT { - let line = timeout(Duration::from_secs(5), lines.next_line()) - .await - .unwrap() - .unwrap() - .unwrap(); - let response: Value = serde_json::from_str(&line).unwrap(); - assert_eq!(response["id"], expected_id); - } - assert_eq!(handler.workers.count(), 1); + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; - drop(lines); - task.await.unwrap(); - assert_eq!(handler.workers.count(), 0); - }); + const REQUEST_COUNT: usize = 200; + + let (mut client, server_socket) = tcp_pair().await; + let handler = setup_handler(test_dir.path()); + + let task = tokio::spawn(handle_connection( + server_socket, + handler.clone(), + Duration::from_secs(5), + )); + + let requests = (0..REQUEST_COUNT) + .map(|id| { + format!( + r#"{{"id":{},"jsonrpc":"2.0","method":"keepalive","params":null}}"#, + id + ) + }) + .collect::>() + .join("\n") + + "\n"; + client.write_all(requests.as_bytes()).await.unwrap(); + + let mut lines = BufReader::new(client).lines(); + for expected_id in 0..REQUEST_COUNT { + let line = timeout(Duration::from_secs(5), lines.next_line()) + .await + .unwrap() + .unwrap() + .unwrap(); + let response: Value = serde_json::from_str(&line).unwrap(); + assert_eq!(response["id"], expected_id); + } + assert_eq!(handler.workers.count(), 1); + + drop(lines); + task.await.unwrap(); + assert_eq!(handler.workers.count(), 0); } #[test] - fn test_block_retry_stays_responsive() { + fn test_block_retry_shutdown() { use crate::common::adapters::{PoolToChainAdapter, PoolToNetAdapter}; let test_dir = TestDir::new(".grin_stratum_shutdown_retry_test"); From c71a2a85d020ed5469668423e7b32f74b22b28f6 Mon Sep 17 00:00:00 2001 From: wiesche Date: Sun, 26 Jul 2026 19:53:42 +0200 Subject: [PATCH 09/13] cancel wallet request on shutdown --- servers/src/mining/mine_block.rs | 75 ++++++++++++++++++++++------- servers/src/mining/stratumserver.rs | 17 ++++++- 2 files changed, 73 insertions(+), 19 deletions(-) diff --git a/servers/src/mining/mine_block.rs b/servers/src/mining/mine_block.rs index 8eb4d0585b..c36eb5d40e 100644 --- a/servers/src/mining/mine_block.rs +++ b/servers/src/mining/mine_block.rs @@ -100,8 +100,17 @@ fn get_block_internal( should_stop: impl Fn() -> bool, ) -> Option<(core::Block, BlockFees)> { let wallet_retry_interval = 5; + if should_stop() { + return None; + } // get the latest chain state and build a block on top of it - let mut result = build_block(chain, tx_pool, key_id.clone(), wallet_listener_url.clone()); + let mut result = build_block( + chain, + tx_pool, + key_id.clone(), + wallet_listener_url.clone(), + &should_stop, + ); while let Err(e) = result { if should_stop() { return None; @@ -140,9 +149,19 @@ fn get_block_internal( return None; } - result = build_block(chain, tx_pool, new_key_id, wallet_listener_url.clone()); + result = build_block( + chain, + tx_pool, + new_key_id, + wallet_listener_url.clone(), + &should_stop, + ); + } + if should_stop() { + None + } else { + result.ok() } - result.ok() } fn sleep_or_stop(duration: Duration, should_stop: &impl Fn() -> bool) -> bool { @@ -164,6 +183,7 @@ fn build_block( tx_pool: &ServerTxPool, key_id: Option, wallet_listener_url: Option, + should_stop: &impl Fn() -> bool, ) -> Result<(core::Block, BlockFees), Error> { let head = chain.head_header()?; @@ -203,7 +223,7 @@ fn build_block( height, }; - let (output, kernel, block_fees) = get_coinbase(wallet_listener_url, block_fees)?; + let (output, kernel, block_fees) = get_coinbase(wallet_listener_url, block_fees, should_stop)?; let mut b = core::Block::from_reward(&head, &txs, output, kernel, difficulty.difficulty)?; // making sure we're not spending time mining a useless block @@ -270,6 +290,7 @@ fn burn_reward(block_fees: BlockFees) -> Result<(core::Output, core::TxKernel, B fn get_coinbase( wallet_listener_url: Option, block_fees: BlockFees, + should_stop: &impl Fn() -> bool, ) -> Result<(core::Output, core::TxKernel, BlockFees), Error> { match wallet_listener_url { None => { @@ -277,7 +298,7 @@ fn get_coinbase( return burn_reward(block_fees); } Some(wallet_listener_url) => { - let res = create_coinbase(&wallet_listener_url, &block_fees)?; + let res = create_coinbase(&wallet_listener_url, &block_fees, should_stop)?; let output = res.output; let kernel = res.kernel; let key_id = res.key_id; @@ -294,7 +315,11 @@ fn get_coinbase( /// Call the wallet API to create a coinbase output for the given block_fees. /// Will retry based on default "retry forever with backoff" behavior. -fn create_coinbase(dest: &str, block_fees: &BlockFees) -> Result { +fn create_coinbase( + dest: &str, + block_fees: &BlockFees, + should_stop: &impl Fn() -> bool, +) -> Result { let url = format!("{}/v2/foreign", dest); let req_body = json!({ "jsonrpc": "2.0", @@ -306,18 +331,28 @@ fn create_coinbase(dest: &str, block_fees: &BlockFees) -> Result }); trace!("Sending build_coinbase request: {}", req_body); - let req = api::client::create_post_request(url.as_str(), None, &req_body)?; - let timeout = api::client::TimeOut::default(); - let res: String = api::client::send_request(req, timeout).map_err(|e| { - let report = format!( - "Failed to get coinbase from {}. Is the wallet listening? {}", - dest, e - ); - error!("{}", report); - Error::WalletComm(report) - })?; + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .map_err(|e| Error::WalletComm(format!("Failed to start wallet request: {}", e)))?; + let res = runtime.block_on(async { + tokio::select! { + res = api::client::post_async::<_, Value>(url.as_str(), &req_body, None) => Some(res), + _ = wait_for_stop(should_stop) => None, + } + }); + let res = match res { + Some(res) => res.map_err(|e| { + let report = format!( + "Failed to get coinbase from {}. Is the wallet listening? {}", + dest, e + ); + error!("{}", report); + Error::WalletComm(report) + })?, + None => return Err(Error::General("Block building stopped".into())), + }; - let res: Value = serde_json::from_str(&res).unwrap(); trace!("Response: {}", res); if res["error"] != json!(null) { let report = format!( @@ -341,3 +376,9 @@ fn create_coinbase(dest: &str, block_fees: &BlockFees) -> Result Ok(ret_val) } + +async fn wait_for_stop(should_stop: &impl Fn() -> bool) { + while !should_stop() { + tokio::time::sleep(STOP_POLL_INTERVAL).await; + } +} diff --git a/servers/src/mining/stratumserver.rs b/servers/src/mining/stratumserver.rs index 31385ed7b7..f39eb31cf8 100644 --- a/servers/src/mining/stratumserver.rs +++ b/servers/src/mining/stratumserver.rs @@ -1439,6 +1439,7 @@ mod tests { #[test] fn test_block_retry_shutdown() { use crate::common::adapters::{PoolToChainAdapter, PoolToNetAdapter}; + use std::net::TcpListener; let test_dir = TestDir::new(".grin_stratum_shutdown_retry_test"); let handler = setup_handler(test_dir.path()); @@ -1455,9 +1456,19 @@ mod tests { let (tx, _rx, shutdown_tx) = dummy_tx(); handler.workers.add_worker(tx, shutdown_tx); + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let wallet_addr = listener.local_addr().unwrap(); + let (accepted_tx, accepted_rx) = std::sync::mpsc::channel(); + let (wallet_stop_tx, wallet_stop_rx) = std::sync::mpsc::channel(); + let wallet_thread = thread::spawn(move || { + let (_socket, _) = listener.accept().unwrap(); + let _ = accepted_tx.send(()); + let _ = wallet_stop_rx.recv(); + }); + let mut config = StratumServerConfig::default(); config.burn_reward = false; - config.wallet_listener_url = "http://127.0.0.1:1".to_string(); + config.wallet_listener_url = format!("http://{}", wallet_addr); let stop_state = Arc::new(StopState::new()); let run_stop_state = stop_state.clone(); let request_handler = handler.clone(); @@ -1468,7 +1479,7 @@ mod tests { let _ = done_tx.send(()); }); - thread::sleep(Duration::from_millis(250)); + assert_eq!(accepted_rx.recv_timeout(Duration::from_secs(1)), Ok(())); let (request_tx, request_rx) = std::sync::mpsc::channel(); thread::spawn(move || { request_handler.build_block_template(); @@ -1478,6 +1489,8 @@ mod tests { stop_state.stop(); assert_eq!(done_rx.recv_timeout(Duration::from_secs(1)), Ok(())); + let _ = wallet_stop_tx.send(()); + wallet_thread.join().unwrap(); } fn setup_handler(dir: &Path) -> Arc { From b7f96c59e19d12a80f675553f9826f9d88c812dd Mon Sep 17 00:00:00 2001 From: wiesche Date: Sun, 26 Jul 2026 20:52:44 +0200 Subject: [PATCH 10/13] adapt stratum tests --- servers/src/mining/stratumserver.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/servers/src/mining/stratumserver.rs b/servers/src/mining/stratumserver.rs index f39eb31cf8..88ab8596b2 100644 --- a/servers/src/mining/stratumserver.rs +++ b/servers/src/mining/stratumserver.rs @@ -1106,7 +1106,6 @@ mod tests { use crate::core::global::{self, ChainTypes}; use crate::core::pow::Difficulty; use std::fs; - use std::net::TcpListener as StdTcpListener; use std::path::{Path, PathBuf}; use std::sync::OnceLock; @@ -1805,7 +1804,9 @@ mod tests { assert_eq!(rx0.try_recv().unwrap(), "hello-all"); assert_eq!(rx1.try_recv().unwrap(), "hello-all"); - workers.send_to(id0, "hello-one".to_string()); + assert!(futures::executor::block_on( + workers.send_to(id0, "hello-one".to_string()) + )); assert_eq!(rx0.try_recv().unwrap(), "hello-one"); // Unicast must not deliver to the other worker (channel open but empty). assert!(rx1.try_recv().is_err()); From f29745ce455c6d08207d0438dd36fbd708845e76 Mon Sep 17 00:00:00 2001 From: wiesche Date: Sun, 26 Jul 2026 21:28:06 +0200 Subject: [PATCH 11/13] fix stratum idle and broadcast handling --- servers/src/mining/stratumserver.rs | 112 +++++++++++++++++++++------- 1 file changed, 83 insertions(+), 29 deletions(-) diff --git a/servers/src/mining/stratumserver.rs b/servers/src/mining/stratumserver.rs index 88ab8596b2..ed8ac2bc60 100644 --- a/servers/src/mining/stratumserver.rs +++ b/servers/src/mining/stratumserver.rs @@ -576,7 +576,12 @@ impl Handler { let mut state = self.current_state.write(); head = self.chain.head().unwrap(); let latest_hash = head.last_block_h; + // Preserve the wallet-provided key even if the chain advanced while + // the block was being built. The next attempt must reuse that key. + state.current_key_id = block_fees.key_id(); if new_block.header.prev_hash != latest_hash { + drop(state); + thread::sleep(Duration::from_millis(5)); continue; } @@ -587,8 +592,6 @@ impl Handler { state.current_difficulty = (new_block.header.total_difficulty() - head.total_difficulty).to_num(); - state.current_key_id = block_fees.key_id(); - current_hash = latest_hash; // set the minimum acceptable share unscaled difficulty for this block state.minimum_share_difficulty = config.minimum_share_difficulty; @@ -657,10 +660,9 @@ async fn handle_connection(socket: TcpStream, handler: Arc, idle_timeou let framed = Framed::new(socket, LinesCodec::new_with_max_length(MAX_RPC_LINE_BYTES)); let (mut writer, mut reader) = framed.split(); - let (activity_tx, mut activity_rx) = mpsc::channel::<()>(1); + let (read_activity, mut activity_rx) = mpsc::channel::<()>(1); let reader_handler = handler.clone(); - let read_activity = activity_tx.clone(); let read = async move { while let Some(line) = reader .try_next() @@ -683,9 +685,7 @@ async fn handle_connection(socket: TcpStream, handler: Arc, idle_timeou let write = async move { while let Some(line) = rx.recv().await { match timeout(WORKER_WRITE_TIMEOUT, writer.send(line)).await { - Ok(Ok(())) => { - let _ = activity_tx.try_send(()); - } + Ok(Ok(())) => {} Ok(Err(e)) => { error!("Worker {} write error: {}", worker_id, e); return Err(()); @@ -935,22 +935,28 @@ impl WorkersList { } fn queue_broadcast(&self, msg: &str) -> Vec<(usize, mpsc::Sender<()>)> { - let mut slow_workers = Vec::new(); + let mut disconnected_workers = Vec::new(); let workers_list = self.workers_list.read(); for (worker_id, worker) in workers_list.iter() { - if worker.tx.try_send(msg.to_owned()).is_err() { - slow_workers.push((*worker_id, worker.shutdown_tx.clone())); + match worker.tx.try_send(msg.to_owned()) { + Ok(()) => {} + Err(mpsc::error::TrySendError::Full(_)) => { + debug!( + "Stratum: skipping broadcast to worker {} with a full queue", + worker_id + ); + } + Err(mpsc::error::TrySendError::Closed(_)) => { + disconnected_workers.push((*worker_id, worker.shutdown_tx.clone())); + } } } - slow_workers + disconnected_workers } pub fn broadcast(&self, msg: String) { for (worker_id, shutdown_tx) in self.queue_broadcast(&msg) { - warn!( - "Stratum: dropping slow or disconnected worker {}", - worker_id - ); + warn!("Stratum: dropping disconnected worker {}", worker_id); let _ = shutdown_tx.try_send(()); } } @@ -1121,8 +1127,8 @@ mod tests { CHAIN .get_or_init(|| { global::set_local_chain_type(ChainTypes::AutomatedTesting); - // Under the crate-local target directory (servers/target/tmp), not - // the workspace target/, so interrupted tests do not litter the repo root. + // Keep interrupted test data under Cargo's target directory instead of + // littering the repository root. let dir = "target/tmp/grin_stratum_test_shared_chain"; let _ = fs::remove_dir_all(dir); Arc::new( @@ -1176,7 +1182,7 @@ mod tests { impl TestDir { fn new(path: &str) -> Self { - let path = PathBuf::from(path); + let path = PathBuf::from("target/tmp").join(path); let _ = std::fs::remove_dir_all(&path); Self { path } } @@ -1288,13 +1294,13 @@ mod tests { let stats = Arc::new(RwLock::new(StratumStats::default())); let workers = WorkersList::new(stats); - let (old_tx, _old_rx) = mpsc::channel(1); - old_tx.try_send("queued".into()).unwrap(); + let (old_tx, old_rx) = mpsc::channel(1); + drop(old_rx); let (old_shutdown_tx, mut old_shutdown_rx) = mpsc::channel(1); let old_id = workers.add_worker(old_tx, old_shutdown_tx); - let slow_workers = workers.queue_broadcast("next job"); - assert_eq!(slow_workers.len(), 1); + let disconnected_workers = workers.queue_broadcast("next job"); + assert_eq!(disconnected_workers.len(), 1); workers.remove_worker(old_id); let (new_tx, _new_rx) = mpsc::channel(1); @@ -1302,16 +1308,34 @@ mod tests { let new_id = workers.add_worker(new_tx, new_shutdown_tx); assert_eq!(new_id, old_id); - let (slow_worker_id, shutdown_tx) = slow_workers.into_iter().next().unwrap(); - assert_eq!(slow_worker_id, old_id); + let (disconnected_worker_id, shutdown_tx) = + disconnected_workers.into_iter().next().unwrap(); + assert_eq!(disconnected_worker_id, old_id); shutdown_tx.try_send(()).unwrap(); assert_eq!(old_shutdown_rx.try_recv(), Ok(())); assert!(new_shutdown_rx.try_recv().is_err()); } + #[test] + fn test_full_queue_keeps_worker() { + let stats = Arc::new(RwLock::new(StratumStats::default())); + let workers = WorkersList::new(stats); + let (tx, mut rx) = mpsc::channel(1); + tx.try_send("queued".into()).unwrap(); + let (shutdown_tx, mut shutdown_rx) = mpsc::channel(1); + let worker_id = workers.add_worker(tx, shutdown_tx); + + workers.broadcast("next job".into()); + + assert_eq!(workers.count(), 1); + assert_eq!(rx.try_recv(), Ok("queued".into())); + assert!(shutdown_rx.try_recv().is_err()); + workers.remove_worker(worker_id); + } + #[tokio::test] async fn test_accept_loop_shutdown() { - let test_dir = TestDir::new(".grin_stratum_accept_loop_test"); + let test_dir = TestDir::new("grin_stratum_accept_loop_test"); let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); let handler = setup_handler(test_dir.path()); @@ -1337,7 +1361,7 @@ mod tests { #[tokio::test] async fn test_worker_limit() { - let test_dir = TestDir::new(".grin_stratum_max_workers_test"); + let test_dir = TestDir::new("grin_stratum_max_workers_test"); const MAX_TEST_WORKERS: usize = 8; let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -1367,7 +1391,7 @@ mod tests { #[tokio::test] async fn test_sync_idle_timeout() { - let test_dir = TestDir::new(".grin_stratum_idle_test"); + let test_dir = TestDir::new("grin_stratum_idle_test"); let (client, server_socket) = tcp_pair().await; let handler = setup_handler(test_dir.path()); handler.sync_state.update(SyncStatus::Initial); @@ -1390,9 +1414,39 @@ mod tests { drop(client); } + #[tokio::test] + async fn test_outbound_does_not_reset_idle() { + let test_dir = TestDir::new("grin_stratum_outbound_idle_test"); + let (client, server_socket) = tcp_pair().await; + let handler = setup_handler(test_dir.path()); + + let task = tokio::spawn(handle_connection( + server_socket, + handler.clone(), + Duration::from_millis(50), + )); + wait_for_worker_count(&handler, 1).await; + + let broadcast_handler = handler.clone(); + let broadcaster = tokio::spawn(async move { + for _ in 0..20 { + broadcast_handler.workers.broadcast("job".into()); + tokio::time::sleep(Duration::from_millis(10)).await; + } + }); + + timeout(Duration::from_millis(150), task) + .await + .expect("outbound broadcasts kept an idle worker connected") + .unwrap(); + assert_eq!(handler.workers.count(), 0); + broadcaster.await.unwrap(); + drop(client); + } + #[tokio::test] async fn test_pipelined_backpressure() { - let test_dir = TestDir::new(".grin_stratum_pipeline_test"); + let test_dir = TestDir::new("grin_stratum_pipeline_test"); use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; const REQUEST_COUNT: usize = 200; @@ -1440,7 +1494,7 @@ mod tests { use crate::common::adapters::{PoolToChainAdapter, PoolToNetAdapter}; use std::net::TcpListener; - let test_dir = TestDir::new(".grin_stratum_shutdown_retry_test"); + let test_dir = TestDir::new("grin_stratum_shutdown_retry_test"); let handler = setup_handler(test_dir.path()); let pool_adapter = Arc::new(PoolToChainAdapter::new()); pool_adapter.set_chain(handler.chain.clone()); From 5e072472f579ced6b020d7ddc40e07c14ff52657 Mon Sep 17 00:00:00 2001 From: wiesche Date: Sun, 26 Jul 2026 22:51:37 +0200 Subject: [PATCH 12/13] preserve stratum compatibility and simplify cleanup --- servers/src/grin/server.rs | 9 ++ servers/src/mining/stratumserver.rs | 122 ++++++++++------------------ 2 files changed, 53 insertions(+), 78 deletions(-) diff --git a/servers/src/grin/server.rs b/servers/src/grin/server.rs index f881842654..0445eccc48 100644 --- a/servers/src/grin/server.rs +++ b/servers/src/grin/server.rs @@ -639,6 +639,12 @@ fn validate_stratum_config(config: &StratumServerConfig) -> Result<(), Error> { "stratum max_workers must be greater than zero".to_string(), )); } + if config.max_workers > tokio::sync::Semaphore::MAX_PERMITS { + return Err(Error::Configuration(format!( + "stratum max_workers must not exceed {}", + tokio::sync::Semaphore::MAX_PERMITS + ))); + } if config.worker_idle_timeout_secs == 0 { return Err(Error::Configuration( "stratum worker_idle_timeout_secs must be greater than zero".to_string(), @@ -673,6 +679,9 @@ mod tests { config.max_workers = 0; assert!(validate_stratum_config(&config).is_err()); + config.max_workers = tokio::sync::Semaphore::MAX_PERMITS + 1; + assert!(validate_stratum_config(&config).is_err()); + config.max_workers = 1; config.worker_idle_timeout_secs = 0; assert!(validate_stratum_config(&config).is_err()); diff --git a/servers/src/mining/stratumserver.rs b/servers/src/mining/stratumserver.rs index ed8ac2bc60..5dc82e4be0 100644 --- a/servers/src/mining/stratumserver.rs +++ b/servers/src/mining/stratumserver.rs @@ -561,6 +561,7 @@ impl Handler { None }; let key_id = self.current_state.read().current_key_id.clone(); + let requested_key_id = key_id.clone(); // Build the new block (version) let Some((new_block, block_fees)) = mine_block::get_block_with_stop( @@ -576,9 +577,11 @@ impl Handler { let mut state = self.current_state.write(); head = self.chain.head().unwrap(); let latest_hash = head.last_block_h; - // Preserve the wallet-provided key even if the chain advanced while - // the block was being built. The next attempt must reuse that key. - state.current_key_id = block_fees.key_id(); + // Preserve the wallet-provided key for a stale build unless a winning + // submission reset the key while the block was being built. + if state.current_key_id == requested_key_id { + state.current_key_id = block_fees.key_id(); + } if new_block.header.prev_hash != latest_hash { drop(state); thread::sleep(Duration::from_millis(5)); @@ -645,8 +648,7 @@ impl Drop for WorkerCleanup { async fn handle_connection(socket: TcpStream, handler: Arc, idle_timeout: Duration) { let peer_addr = socket.peer_addr().ok(); let (tx, mut rx) = mpsc::channel(WORKER_QUEUE_SIZE); - let (shutdown_tx, mut shutdown_rx) = mpsc::channel::<()>(1); - let worker_id = handler.workers.add_worker(tx, shutdown_tx); + let worker_id = handler.workers.add_worker(tx); let _cleanup = WorkerCleanup { worker_id, workers: handler.workers.clone(), @@ -660,7 +662,8 @@ async fn handle_connection(socket: TcpStream, handler: Arc, idle_timeou let framed = Framed::new(socket, LinesCodec::new_with_max_length(MAX_RPC_LINE_BYTES)); let (mut writer, mut reader) = framed.split(); - let (read_activity, mut activity_rx) = mpsc::channel::<()>(1); + let (activity, mut activity_rx) = mpsc::channel::<()>(1); + let read_activity = activity.clone(); let reader_handler = handler.clone(); let read = async move { @@ -685,7 +688,9 @@ async fn handle_connection(socket: TcpStream, handler: Arc, idle_timeou let write = async move { while let Some(line) = rx.recv().await { match timeout(WORKER_WRITE_TIMEOUT, writer.send(line)).await { - Ok(Ok(())) => {} + Ok(Ok(())) => { + let _ = activity.try_send(()); + } Ok(Err(e)) => { error!("Worker {} write error: {}", worker_id, e); return Err(()); @@ -716,7 +721,6 @@ async fn handle_connection(socket: TcpStream, handler: Arc, idle_timeou warn!("Worker {} idle for {:?}; disconnecting", worker_id, idle_timeout); break; } - _ = shutdown_rx.recv() => break, activity = activity_rx.recv() => { if activity.is_some() { idle_sleep.as_mut().reset(Instant::now() + idle_timeout); @@ -813,19 +817,17 @@ pub struct Worker { login: Option, authenticated: bool, tx: Tx, - shutdown_tx: mpsc::Sender<()>, } impl Worker { /// Creates a new Stratum Worker. - pub fn new(id: usize, tx: Tx, shutdown_tx: mpsc::Sender<()>) -> Worker { + pub fn new(id: usize, tx: Tx) -> Worker { Worker { id: id, agent: String::from(""), login: None, authenticated: false, tx: tx, - shutdown_tx, } } } // impl Worker @@ -843,7 +845,7 @@ impl WorkersList { } } - pub fn add_worker(&self, tx: Tx, shutdown_tx: mpsc::Sender<()>) -> usize { + pub fn add_worker(&self, tx: Tx) -> usize { let mut stratum_stats = self.stratum_stats.write(); let mut workers_list = self.workers_list.write(); let worker_id = match stratum_stats @@ -858,7 +860,7 @@ impl WorkersList { id } }; - let worker = Worker::new(worker_id, tx, shutdown_tx); + let worker = Worker::new(worker_id, tx); workers_list.insert(worker_id, worker); let mut worker_stats = WorkerStats::default(); @@ -934,8 +936,7 @@ impl WorkersList { tx.send(msg).await.is_ok() } - fn queue_broadcast(&self, msg: &str) -> Vec<(usize, mpsc::Sender<()>)> { - let mut disconnected_workers = Vec::new(); + pub fn broadcast(&self, msg: String) { let workers_list = self.workers_list.read(); for (worker_id, worker) in workers_list.iter() { match worker.tx.try_send(msg.to_owned()) { @@ -947,18 +948,13 @@ impl WorkersList { ); } Err(mpsc::error::TrySendError::Closed(_)) => { - disconnected_workers.push((*worker_id, worker.shutdown_tx.clone())); + debug!( + "Stratum: skipping broadcast to disconnected worker {}", + worker_id + ); } } } - disconnected_workers - } - - pub fn broadcast(&self, msg: String) { - for (worker_id, shutdown_tx) in self.queue_broadcast(&msg) { - warn!("Stratum: dropping disconnected worker {}", worker_id); - let _ = shutdown_tx.try_send(()); - } } pub fn count(&self) -> usize { @@ -1202,15 +1198,13 @@ mod tests { } } - fn dummy_tx() -> (Tx, mpsc::Receiver, mpsc::Sender<()>) { - let (tx, rx) = mpsc::channel(WORKER_QUEUE_SIZE); - let (shutdown_tx, _shutdown_rx) = mpsc::channel(1); - (tx, rx, shutdown_tx) + fn dummy_tx() -> (Tx, mpsc::Receiver) { + mpsc::channel(WORKER_QUEUE_SIZE) } fn add_dummy_worker(workers: &WorkersList) -> usize { - let (tx, _rx, shutdown_tx) = dummy_tx(); - workers.add_worker(tx, shutdown_tx) + let (tx, _rx) = dummy_tx(); + workers.add_worker(tx) } async fn tcp_pair() -> (TcpStream, TcpStream) { @@ -1238,8 +1232,8 @@ mod tests { let stats = Arc::new(RwLock::new(StratumStats::default())); let workers = WorkersList::new(stats.clone()); - let (tx0, _rx0, shutdown_tx0) = dummy_tx(); - let id0 = workers.add_worker(tx0, shutdown_tx0); + let (tx0, _rx0) = dummy_tx(); + let id0 = workers.add_worker(tx0); assert_eq!(id0, 0); assert_eq!(workers.count(), 1); assert_eq!(stats.read().worker_stats.len(), 1); @@ -1248,8 +1242,8 @@ mod tests { assert_eq!(workers.count(), 0); assert!(!stats.read().worker_stats[0].is_connected); - let (tx1, _rx1, shutdown_tx1) = dummy_tx(); - let id1 = workers.add_worker(tx1, shutdown_tx1); + let (tx1, _rx1) = dummy_tx(); + let id1 = workers.add_worker(tx1); assert_eq!(id1, 0); assert_eq!(stats.read().worker_stats.len(), 1); assert!(stats.read().worker_stats[0].is_connected); @@ -1264,8 +1258,7 @@ mod tests { assert!(!workers.send_to(0, "missing".into()).await); let (tx, mut rx) = mpsc::channel(1); - let (shutdown_tx, _shutdown_rx) = mpsc::channel(1); - let id = workers.add_worker(tx, shutdown_tx); + let id = workers.add_worker(tx); assert!(workers.send_to(id, "one".into()).await); assert_eq!(rx.try_recv().unwrap(), "one"); @@ -1280,8 +1273,8 @@ mod tests { fn test_remove_worker_twice() { let stats = Arc::new(RwLock::new(StratumStats::default())); let workers = WorkersList::new(stats.clone()); - let (tx, _rx, shutdown_tx) = dummy_tx(); - let id = workers.add_worker(tx, shutdown_tx); + let (tx, _rx) = dummy_tx(); + let id = workers.add_worker(tx); workers.remove_worker(id); workers.remove_worker(id); assert_eq!(workers.count(), 0); @@ -1289,47 +1282,18 @@ mod tests { assert!(!stats.read().worker_stats[id].is_connected); } - #[test] - fn test_stale_worker_shutdown() { - let stats = Arc::new(RwLock::new(StratumStats::default())); - let workers = WorkersList::new(stats); - - let (old_tx, old_rx) = mpsc::channel(1); - drop(old_rx); - let (old_shutdown_tx, mut old_shutdown_rx) = mpsc::channel(1); - let old_id = workers.add_worker(old_tx, old_shutdown_tx); - - let disconnected_workers = workers.queue_broadcast("next job"); - assert_eq!(disconnected_workers.len(), 1); - - workers.remove_worker(old_id); - let (new_tx, _new_rx) = mpsc::channel(1); - let (new_shutdown_tx, mut new_shutdown_rx) = mpsc::channel(1); - let new_id = workers.add_worker(new_tx, new_shutdown_tx); - assert_eq!(new_id, old_id); - - let (disconnected_worker_id, shutdown_tx) = - disconnected_workers.into_iter().next().unwrap(); - assert_eq!(disconnected_worker_id, old_id); - shutdown_tx.try_send(()).unwrap(); - assert_eq!(old_shutdown_rx.try_recv(), Ok(())); - assert!(new_shutdown_rx.try_recv().is_err()); - } - #[test] fn test_full_queue_keeps_worker() { let stats = Arc::new(RwLock::new(StratumStats::default())); let workers = WorkersList::new(stats); let (tx, mut rx) = mpsc::channel(1); tx.try_send("queued".into()).unwrap(); - let (shutdown_tx, mut shutdown_rx) = mpsc::channel(1); - let worker_id = workers.add_worker(tx, shutdown_tx); + let worker_id = workers.add_worker(tx); workers.broadcast("next job".into()); assert_eq!(workers.count(), 1); assert_eq!(rx.try_recv(), Ok("queued".into())); - assert!(shutdown_rx.try_recv().is_err()); workers.remove_worker(worker_id); } @@ -1415,7 +1379,7 @@ mod tests { } #[tokio::test] - async fn test_outbound_does_not_reset_idle() { + async fn test_outbound_resets_idle() { let test_dir = TestDir::new("grin_stratum_outbound_idle_test"); let (client, server_socket) = tcp_pair().await; let handler = setup_handler(test_dir.path()); @@ -1435,12 +1399,14 @@ mod tests { } }); - timeout(Duration::from_millis(150), task) + broadcaster.await.unwrap(); + assert!(!task.is_finished()); + + timeout(Duration::from_millis(500), task) .await - .expect("outbound broadcasts kept an idle worker connected") + .expect("worker remained connected after outbound activity stopped") .unwrap(); assert_eq!(handler.workers.count(), 0); - broadcaster.await.unwrap(); drop(client); } @@ -1506,8 +1472,8 @@ mod tests { pool_adapter, pool_net_adapter, ))); - let (tx, _rx, shutdown_tx) = dummy_tx(); - handler.workers.add_worker(tx, shutdown_tx); + let (tx, _rx) = dummy_tx(); + handler.workers.add_worker(tx); let listener = TcpListener::bind("127.0.0.1:0").unwrap(); let wallet_addr = listener.local_addr().unwrap(); @@ -1849,10 +1815,10 @@ mod tests { let stats = Arc::new(RwLock::new(StratumStats::default())); let workers = WorkersList::new(stats); - let (tx0, mut rx0, shutdown_tx0) = dummy_tx(); - let (tx1, mut rx1, shutdown_tx1) = dummy_tx(); - let id0 = workers.add_worker(tx0, shutdown_tx0); - let _id1 = workers.add_worker(tx1, shutdown_tx1); + let (tx0, mut rx0) = dummy_tx(); + let (tx1, mut rx1) = dummy_tx(); + let id0 = workers.add_worker(tx0); + let _id1 = workers.add_worker(tx1); workers.broadcast("hello-all".to_string()); assert_eq!(rx0.try_recv().unwrap(), "hello-all"); From 39a282bd4f042c799aae842a18b96210084ea2b8 Mon Sep 17 00:00:00 2001 From: wiesche Date: Sun, 26 Jul 2026 23:15:56 +0200 Subject: [PATCH 13/13] fix coinbase key race --- servers/src/mining/stratumserver.rs | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/servers/src/mining/stratumserver.rs b/servers/src/mining/stratumserver.rs index 5dc82e4be0..247669323b 100644 --- a/servers/src/mining/stratumserver.rs +++ b/servers/src/mining/stratumserver.rs @@ -247,10 +247,6 @@ impl Handler { "login" => self.handle_login(request.params, worker_id), "submit" => { let res = self.handle_submit(request.params, worker_id); - // this key_id has been used now, reset - if let Ok((_, true)) = res { - self.current_state.write().current_key_id = None; - } res.map(|(v, _)| v) } "keepalive" => self.handle_keepalive(), @@ -389,6 +385,7 @@ impl Handler { let scaled_share_difficulty: u64; let unscaled_share_difficulty: u64; let mut share_is_block = false; + let current_difficulty = state.current_difficulty; let mut b: Block = b.unwrap().clone(); // Reconstruct the blocks header with this nonce and pow added @@ -425,8 +422,10 @@ impl Handler { } // If the difficulty is high enough, submit it (which also validates it) - if scaled_share_difficulty >= state.current_difficulty { + if scaled_share_difficulty >= current_difficulty { // This is a full solution, submit it to the network + drop(state); + let mut state = self.current_state.write(); let res = self.chain.process_block(b.clone(), chain::Options::MINE); if let Err(e) = res { // Return error status @@ -444,6 +443,9 @@ impl Handler { .update_stats(worker_id, |worker_stats| worker_stats.num_rejected += 1); return Err(RpcError::cannot_validate()); } + // Reset the key before allowing the block builder to observe the new head. + state.current_key_id = None; + drop(state); share_is_block = true; self.workers .update_stats(worker_id, |worker_stats| worker_stats.num_blocks_found += 1); @@ -495,7 +497,7 @@ impl Handler { b.header.pow.nonce, params.job_id, scaled_share_difficulty, - state.current_difficulty, + current_difficulty, submitted_by, ); self.workers @@ -560,7 +562,8 @@ impl Handler { } else { None }; - let key_id = self.current_state.read().current_key_id.clone(); + let state = self.current_state.read(); + let key_id = state.current_key_id.clone(); let requested_key_id = key_id.clone(); // Build the new block (version) @@ -573,6 +576,7 @@ impl Handler { ) else { return; }; + drop(state); let mut state = self.current_state.write(); head = self.chain.head().unwrap();