Refactor stratum server leak handling

config/src/comments.rs

@@ -398,6 +398,23 @@ fn comments() -> HashMap<String, String> {
 		.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(),
 		"

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"

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<String>,
 
+	/// 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);
+	}
+}

servers/src/grin/server.rs

@@ -76,6 +76,7 @@ pub struct Server {
 	connect_thread: Option<JoinHandle<()>>,
 	sync_thread: JoinHandle<()>,
 	dandelion_thread: JoinHandle<()>,
+	stratum_thread: RwLock<Option<JoinHandle<()>>>,
 }
 
 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,20 +383,27 @@ 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,
 			self.chain.clone(),
 			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::<std::net::SocketAddr>().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());
+	}
+}