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/config/src/comments.rs b/config/src/comments.rs index 0e9ac707da..e4a5cbc45b 100644 --- a/config/src/comments.rs +++ b/config/src/comments.rs @@ -398,6 +398,23 @@ fn comments() -> HashMap { .to_string(), ); + retval.insert( + "max_workers".to_string(), + " +#maximum number of concurrent stratum workers +" + .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/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/common/types.rs b/servers/src/common/types.rs index 445a57edab..85c4e40d52 100644 --- a/servers/src/common/types.rs +++ b/servers/src/common/types.rs @@ -246,6 +246,15 @@ 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, + + /// 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, @@ -261,11 +270,21 @@ pub struct StratumServerConfig { pub burn_reward: bool, } +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), @@ -428,3 +447,20 @@ pub enum NetAdapterWorkerMessage { /// Received PIBD segment. PIBDSegment(QueuedPIBDSegment), } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn stratum_config_defaults() { + let mut value = serde_json::to_value(StratumServerConfig::default()).unwrap(); + 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/grin/server.rs b/servers/src/grin/server.rs index f486350345..0445eccc48 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: RwLock>>, } impl Server { @@ -93,6 +94,12 @@ 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(); + 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 { @@ -345,6 +352,7 @@ impl Server { connect_thread, sync_thread, dandelion_thread, + stratum_thread: RwLock::new(None), }) } @@ -375,8 +383,13 @@ impl Server { /// Start a minimal "stratum" mining service on a separate thread 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(); let mut stratum_server = stratumserver::StratumServer::new( config, @@ -384,11 +397,13 @@ impl Server { self.tx_pool.clone(), self.state_info.stratum_stats.clone(), ); - let _ = 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); - }); + 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 @@ -412,6 +427,7 @@ impl Server { stratum_server_addr: None, wallet_listener_url: config_wallet_url, minimum_share_difficulty: 1, + ..StratumServerConfig::default() }; let mut miner = Miner::new( @@ -573,6 +589,13 @@ impl Server { info!("No active connect_and_monitor 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"), + } + } + match self.sync_thread.join() { Err(e) => error!("failed to join to sync thread: {:?}", e), Ok(_) => info!("sync thread stopped"), @@ -609,3 +632,68 @@ 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.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(), + )); + } + 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 = 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()); + + 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..c36eb5d40e 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,45 @@ 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; + 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; + } let mut new_key_id = key_id.to_owned(); match e { self::Error::Chain(c) => match c { @@ -96,7 +134,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 +145,35 @@ 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()); + result = build_block( + chain, + tx_pool, + new_key_id, + wallet_listener_url.clone(), + &should_stop, + ); + } + if should_stop() { + None + } else { + result.ok() } - return result.unwrap(); +} + +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 @@ -121,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()?; @@ -160,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 @@ -227,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 => { @@ -234,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; @@ -251,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", @@ -263,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!( @@ -298,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 b6b775af42..247669323b 100644 --- a/servers/src/mining/stratumserver.rs +++ b/servers/src/mining/stratumserver.rs @@ -14,14 +14,15 @@ //! 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, Semaphore}; +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; @@ -43,7 +44,13 @@ use crate::mining::mine_block; use crate::util::ToHex; use crate::ServerTxPool; -type Tx = mpsc::UnboundedSender; +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; // ---------------------------------------- // http://www.jsonrpc.org/specification @@ -240,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(), @@ -382,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 @@ -418,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 @@ -437,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); @@ -488,7 +497,7 @@ impl Handler { b.header.pow.nonce, params.job_id, scaled_share_difficulty, - state.current_difficulty, + current_difficulty, submitted_by, ); self.workers @@ -525,12 +534,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; @@ -543,29 +557,48 @@ 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 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) - 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; + }; + drop(state); + + let mut state = self.current_state.write(); + head = self.chain.head().unwrap(); + let latest_hash = head.last_block_h; + // 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)); + 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 = (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; @@ -599,72 +632,179 @@ 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, idle_timeout: Duration) { + let peer_addr = socket.peer_addr().ok(); + let (tx, mut rx) = mpsc::channel(WORKER_QUEUE_SIZE); + let worker_id = handler.workers.add_worker(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, mut activity_rx) = mpsc::channel::<()>(1); + let read_activity = activity.clone(); + + let reader_handler = handler.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.send_to(worker_id, resp).await { + warn!("Worker {} outbound queue 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.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 => { + 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; + } + 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, + 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)) => { + let permit = match worker_limit.clone().try_acquire_owned() { + Ok(permit) => permit, + Err(_) => { + debug!( + "Stratum: rejecting connection from {} (max workers: {})", + peer_addr, max_workers + ); + drop(socket); + continue; + } + }; + 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, idle_timeout).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); + } + } + } + } + connections.shutdown().await; +} + +fn accept_connections( + listen_addr: SocketAddr, + 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) }); - server.await + accept_connections_loop(listener, handler, max_workers, idle_timeout, stop_state).await; }; let rt = Runtime::new().unwrap(); @@ -711,27 +851,39 @@ impl WorkersList { pub fn add_worker(&self, tx: Tx) -> 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); 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_some() { + let worker_stats = &mut stratum_stats.worker_stats[worker_id]; + worker_stats.is_connected = false; + worker_stats.last_seen = SystemTime::now(); + } stratum_stats.num_workers = workers_list.len(); } @@ -777,19 +929,35 @@ impl WorkersList { f(&mut stratum_stats.worker_stats[worker_id]); } - pub fn send_to(&self, worker_id: usize, msg: String) { - let _ = self - .workers_list - .read() - .get(&worker_id) - .unwrap() - .tx - .unbounded_send(msg); + 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, + } + }; + tx.send(msg).await.is_ok() } pub fn broadcast(&self, msg: String) { - for worker in self.workers_list.read().values() { - let _ = worker.tx.unbounded_send(msg.clone()); + let workers_list = self.workers_list.read(); + for (worker_id, worker) in workers_list.iter() { + 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(_)) => { + debug!( + "Stratum: skipping broadcast to disconnected worker {}", + worker_id + ); + } + } } } @@ -859,7 +1027,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 @@ -877,9 +1050,18 @@ impl StratumServer { let handler = Arc::new(Handler::from_stratum(&self)); let h = handler.clone(); - - let _listener_th = thread::spawn(move || { - accept_connections(listen_addr, h); + let max_workers = self.config.max_workers; + let idle_timeout = Duration::from_secs(self.config.worker_idle_timeout_secs); + + 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 @@ -896,11 +1078,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 @@ -923,6 +1112,7 @@ mod tests { use crate::core::global::{self, ChainTypes}; use crate::core::pow::Difficulty; use std::fs; + use std::path::{Path, PathBuf}; use std::sync::OnceLock; // ---------------------------------------- @@ -930,11 +1120,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 { @@ -942,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( @@ -962,20 +1147,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 +1176,371 @@ mod tests { serde_json::from_str(json).unwrap() } + struct TestDir { + path: PathBuf, + } + + impl TestDir { + fn new(path: &str) -> Self { + let path = PathBuf::from("target/tmp").join(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::channel(WORKER_QUEUE_SIZE) + } + + fn add_dummy_worker(workers: &WorkersList) -> usize { + let (tx, _rx) = dummy_tx(); + workers.add_worker(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() { + let stats = Arc::new(RwLock::new(StratumStats::default())); + let workers = WorkersList::new(stats.clone()); + + 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); + + workers.remove_worker(id0); + assert_eq!(workers.count(), 0); + assert!(!stats.read().worker_stats[0].is_connected); + + 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); + assert_eq!(workers.count(), 1); + } + + #[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); + + let (tx, mut rx) = mpsc::channel(1); + let id = workers.add_worker(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); + + workers.remove_worker(id); + assert!(!workers.send_to(id, "after-remove".into()).await); + } + + #[test] + fn test_remove_worker_twice() { + let stats = Arc::new(RwLock::new(StratumStats::default())); + let workers = WorkersList::new(stats.clone()); + let (tx, _rx) = dummy_tx(); + let id = workers.add_worker(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_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 worker_id = workers.add_worker(tx); + + workers.broadcast("next job".into()); + + assert_eq!(workers.count(), 1); + assert_eq!(rx.try_recv(), Ok("queued".into())); + workers.remove_worker(worker_id); + } + + #[tokio::test] + async fn test_accept_loop_shutdown() { + 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()); + 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); + } + + #[tokio::test] + async fn test_worker_limit() { + 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(); + 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); + } + + #[tokio::test] + async fn test_sync_idle_timeout() { + 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); + + 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); + } + + #[tokio::test] + 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()); + + 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; + } + }); + + broadcaster.await.unwrap(); + assert!(!task.is_finished()); + + timeout(Duration::from_millis(500), task) + .await + .expect("worker remained connected after outbound activity stopped") + .unwrap(); + assert_eq!(handler.workers.count(), 0); + drop(client); + } + + #[tokio::test] + async fn test_pipelined_backpressure() { + let test_dir = TestDir::new("grin_stratum_pipeline_test"); + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; + + 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_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()); + 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) = dummy_tx(); + handler.workers.add_worker(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 = format!("http://{}", wallet_addr); + 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(()); + }); + + 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(); + 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(())); + let _ = wallet_stop_tx.send(()); + wallet_thread.join().unwrap(); + } + + fn setup_handler(dir: &Path) -> Arc { + global::set_local_chain_type(ChainTypes::AutomatedTesting); + let chain = Arc::new( + chain::Chain::init( + dir.to_string_lossy().into_owned(), + 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 +1745,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 +1775,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 +1806,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 +1819,19 @@ 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 (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!( - 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!(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()); } @@ -1316,8 +1859,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 +1872,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 +1886,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 +1908,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 +1920,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 +1928,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 +1941,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 +1963,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 +1987,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 +2011,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 +2032,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 +2044,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 +2070,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