From c1c665b8c356b30144f57fd7c93762633e19ab95 Mon Sep 17 00:00:00 2001 From: iho Date: Thu, 9 Jul 2026 10:34:34 +0300 Subject: [PATCH 1/6] fix: stop stratum TCP connection leak under reconnect load Rework stratum accept/session handling so abandoned clients cannot pin file descriptors indefinitely: - Always free the worker slot via Drop guard on disconnect - Idle timeout (5m) for silent/half-open sessions - Cap concurrent workers (256) and reject excess accepts - Bounded per-worker write queue; drop slow/full peers - Write timeout; max RPC line length - Reuse disconnected worker_stats slots instead of growing forever - Idempotent remove_worker (no panic on double cleanup) Addresses #3867. --- servers/src/mining/stratumserver.rs | 352 ++++++++++++++++++++++------ 1 file changed, 276 insertions(+), 76 deletions(-) diff --git a/servers/src/mining/stratumserver.rs b/servers/src/mining/stratumserver.rs index d39312a560..03add7e119 100644 --- a/servers/src/mining/stratumserver.rs +++ b/servers/src/mining/stratumserver.rs @@ -14,11 +14,11 @@ //! 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; +use tokio::time::{timeout, Instant}; use tokio_util::codec::{Framed, LinesCodec}; use crate::util::RwLock; @@ -43,7 +43,24 @@ use crate::mining::mine_block; use crate::util::ToHex; use crate::ServerTxPool; -type Tx = mpsc::UnboundedSender; +/// Tokio bounded sender: `try_send` is `&self` and enforces capacity without +/// clone-to-bypass issues that futures::mpsc has when cloning senders. +type Tx = mpsc::Sender; + +/// Max concurrent stratum worker connections. Beyond this, new accepts are +/// closed immediately so we do not exhaust file descriptors. +const MAX_STRATUM_WORKERS: usize = 256; + +/// Bound outbound per-worker queue. If a worker cannot keep up, the connection +/// is dropped rather than buffering forever. +const WORKER_QUEUE_SIZE: usize = 64; + +/// Drop connections with no successful read/write for this long. Prevents +/// half-open / abandoned TCP sessions from accumulating (see #3867). +const WORKER_IDLE_TIMEOUT: Duration = Duration::from_secs(5 * 60); + +/// Cap a single RPC line to avoid unbounded memory from a bad client. +const MAX_RPC_LINE_BYTES: usize = 64 * 1024; // ---------------------------------------- // http://www.jsonrpc.org/specification @@ -605,70 +622,156 @@ fn accept_connections(listen_addr: SocketAddr, handler: Arc) { 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); + + loop { + match listener.accept().await { + Ok((socket, peer_addr)) => { + // Hard cap concurrent workers to avoid FD exhaustion. + if handler.workers.count() >= MAX_STRATUM_WORKERS { + warn!( + "Stratum: rejecting connection from {} (at max {} workers)", + peer_addr, MAX_STRATUM_WORKERS + ); + // Drop socket immediately (close FD). + drop(socket); continue; } + + let handler = handler.clone(); + tokio::spawn(async move { + if let Err(e) = socket.set_nodelay(true) { + debug!("Stratum: set_nodelay failed for {}: {}", peer_addr, e); + } + handle_connection(socket, peer_addr, handler).await; + }); + } + Err(e) => { + error!("Stratum accept error = {:?}", e); + // Avoid busy-looping on EMFILE / transient accept failures. + tokio::time::sleep(Duration::from_millis(100)).await; } } } - .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(()) - }; + let rt = Runtime::new().unwrap(); + rt.block_on(task); +} - let write = async move { - while let Some(line) = rx.next().await { - writer - .send(line) - .await - .map_err(|e| error!("error writing line: {}", e))?; - } +/// Run a single stratum client connection until it ends, then always free the worker slot. +async fn handle_connection(socket: TcpStream, peer_addr: SocketAddr, handler: Arc) { + let (tx, mut rx) = mpsc::channel(WORKER_QUEUE_SIZE); + let worker_id = handler.workers.add_worker(tx); + info!("Worker {} connected from {}", worker_id, peer_addr); - Result::<_, ()>::Ok(()) - }; + // Ensure the worker is always removed, even if the session future panics + // or returns early (connection leak fix). + struct WorkerGuard { + handler: Arc, + worker_id: usize, + peer_addr: SocketAddr, + } + impl Drop for WorkerGuard { + fn drop(&mut self) { + self.handler.workers.remove_worker(self.worker_id); + info!( + "Worker {} disconnected ({})", + self.worker_id, self.peer_addr + ); + } + } + let _guard = WorkerGuard { + handler: handler.clone(), + worker_id, + peer_addr, + }; - 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); + let framed = Framed::new(socket, LinesCodec::new_with_max_length(MAX_RPC_LINE_BYTES)); + let (mut writer, mut reader) = framed.split(); + + let mut idle_deadline = Instant::now() + WORKER_IDLE_TIMEOUT; + + loop { + tokio::select! { + biased; + + line = reader.try_next() => { + match line { + Ok(Some(line)) => { + idle_deadline = Instant::now() + WORKER_IDLE_TIMEOUT; + let request = match serde_json::from_str(&line) { + Ok(r) => r, + Err(e) => { + error!( + "Worker {}: invalid JSON from {}: {}", + worker_id, peer_addr, e + ); + break; + } + }; + let resp = handler.handle_rpc_requests(request, worker_id); + if !handler.workers.try_send_to(worker_id, resp) { + // Queue full or worker gone — drop the connection. + warn!( + "Worker {}: outbound queue full or closed, dropping {}", + worker_id, peer_addr + ); + break; + } + } + Ok(None) => { + // Clean peer EOF. + break; + } + Err(e) => { + error!("Worker {}: read error from {}: {}", worker_id, peer_addr, e); + break; + } + } } - }); - server.await - }; - let rt = Runtime::new().unwrap(); - rt.block_on(task); + msg = rx.recv() => { + match msg { + Some(line) => { + idle_deadline = Instant::now() + WORKER_IDLE_TIMEOUT; + // Bound write time so a stalled peer cannot pin the task forever. + match timeout(Duration::from_secs(30), writer.send(line)).await { + Ok(Ok(())) => {} + Ok(Err(e)) => { + error!( + "Worker {}: write error to {}: {}", + worker_id, peer_addr, e + ); + break; + } + Err(_) => { + warn!( + "Worker {}: write timeout to {}, dropping", + worker_id, peer_addr + ); + break; + } + } + } + None => { + // All senders dropped. + break; + } + } + } + + _ = tokio::time::sleep_until(idle_deadline) => { + warn!( + "Worker {}: idle timeout ({}s) from {}, closing", + worker_id, + WORKER_IDLE_TIMEOUT.as_secs(), + peer_addr + ); + break; + } + } + } + // WorkerGuard drops here → remove_worker + FD released with Framed. } // ---------------------------------------- @@ -711,28 +814,52 @@ 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(); + + // Reuse a free worker_stats slot so reconnect storms do not grow + // the stats vector without bound. + 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_none() { + // Already removed (e.g. concurrent cleanup); still refresh counts. + 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(); + // Mark slot free for reuse; keep historical counters. + 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 +904,31 @@ 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); + /// Queue a message for a single worker. Returns false if the worker is gone + /// or its outbound queue is full (caller should drop the connection). + pub fn try_send_to(&self, worker_id: usize, msg: String) -> bool { + let workers = self.workers_list.read(); + let worker = match workers.get(&worker_id) { + Some(w) => w, + None => return false, + }; + worker.tx.try_send(msg).is_ok() } pub fn broadcast(&self, msg: String) { - for worker in self.workers_list.read().values() { - let _ = worker.tx.unbounded_send(msg.clone()); + // Collect ids of workers whose queue is full so we can drop them. + let mut slow = Vec::new(); + { + let workers = self.workers_list.read(); + for (id, worker) in workers.iter() { + if worker.tx.try_send(msg.clone()).is_err() { + slow.push(*id); + } + } + } + for id in slow { + warn!("Stratum: dropping slow/disconnected worker {}", id); + self.remove_worker(id); } } @@ -919,6 +1058,65 @@ where mod tests { use super::*; + fn dummy_tx() -> (Tx, mpsc::Receiver) { + mpsc::channel(WORKER_QUEUE_SIZE) + } + + #[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) = 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); + + // Next connection must reuse slot 0 rather than growing the vec. + 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); + } + + #[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, "nope".into())); + + let (tx, mut rx) = mpsc::channel(1); + let id = workers.add_worker(tx); + assert!(workers.try_send_to(id, "one".into())); + // Tokio bounded channel: second send fails while first is pending. + 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) = dummy_tx(); + let id = workers.add_worker(tx); + workers.remove_worker(id); + // Second remove must not panic. + workers.remove_worker(id); + assert_eq!(workers.count(), 0); + } + /// Tests deserializing an `RpcRequest` given a String as the id. #[test] fn test_request_deserialize_str() { @@ -1059,3 +1257,5 @@ mod tests { assert_eq!(expected_deserialized, actual_deserialized); } } + +// temp From 864dc7ce4df588a3faab125494e29c98fffa3264 Mon Sep 17 00:00:00 2001 From: iho Date: Thu, 9 Jul 2026 10:36:25 +0300 Subject: [PATCH 2/6] chore: remove accidental trailing comment from stratumserver tests --- servers/src/mining/stratumserver.rs | 2 -- 1 file changed, 2 deletions(-) diff --git a/servers/src/mining/stratumserver.rs b/servers/src/mining/stratumserver.rs index 03add7e119..60f5acf487 100644 --- a/servers/src/mining/stratumserver.rs +++ b/servers/src/mining/stratumserver.rs @@ -1257,5 +1257,3 @@ mod tests { assert_eq!(expected_deserialized, actual_deserialized); } } - -// temp From ec1d4943fc6c30ce45d8b068ca68aeb7d2113fd7 Mon Sep 17 00:00:00 2001 From: iho Date: Thu, 9 Jul 2026 10:40:01 +0300 Subject: [PATCH 3/6] test: live stratum reconnect storm and max-worker cap Add integration-style tests that spin a real accept loop on an ephemeral port, simulate reconnecting miners, assert worker count and FD growth stay bounded, and verify the max concurrent worker cap rejects surplus clients. --- servers/src/mining/stratumserver.rs | 197 ++++++++++++++++++++++++++++ 1 file changed, 197 insertions(+) diff --git a/servers/src/mining/stratumserver.rs b/servers/src/mining/stratumserver.rs index 60f5acf487..b9c738ddc4 100644 --- a/servers/src/mining/stratumserver.rs +++ b/servers/src/mining/stratumserver.rs @@ -1057,11 +1057,208 @@ where #[cfg(test)] mod tests { use super::*; + use crate::chain::types::{NoopAdapter, SyncStatus}; + use crate::core::genesis; + use crate::core::global::{self, ChainTypes}; + use std::io::{Read, Write}; + use std::net::TcpStream as StdTcpStream; + use std::path::Path; + use std::sync::mpsc::sync_channel; fn dummy_tx() -> (Tx, mpsc::Receiver) { mpsc::channel(WORKER_QUEUE_SIZE) } + fn count_open_fds() -> Option { + // macOS / Linux: count process FDs. Best-effort leak signal. + std::fs::read_dir("/dev/fd").ok().map(|d| d.count()) + } + + 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, + )) + } + + /// Start the real accept loop on an ephemeral port; return the bound address. + fn start_test_stratum(handler: Arc) -> SocketAddr { + let (addr_tx, addr_rx) = sync_channel(1); + thread::spawn(move || { + let rt = Runtime::new().unwrap(); + rt.block_on(async move { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + addr_tx.send(addr).unwrap(); + loop { + match listener.accept().await { + Ok((socket, peer_addr)) => { + if handler.workers.count() >= MAX_STRATUM_WORKERS { + drop(socket); + continue; + } + let handler = handler.clone(); + tokio::spawn(async move { + let _ = socket.set_nodelay(true); + handle_connection(socket, peer_addr, handler).await; + }); + } + Err(_) => { + tokio::time::sleep(Duration::from_millis(50)).await; + } + } + } + }); + }); + addr_rx + .recv_timeout(Duration::from_secs(5)) + .expect("stratum test listener failed to start") + } + + fn stratum_login(stream: &mut StdTcpStream) { + let req = r#"{"id":1,"jsonrpc":"2.0","method":"login","params":{"login":"miner","pass":"x","agent":"test"}}"#; + stream.write_all(req.as_bytes()).unwrap(); + stream.write_all(b"\n").unwrap(); + stream.flush().unwrap(); + // Best-effort read of the login response (do not block forever). + let _ = stream.set_read_timeout(Some(Duration::from_millis(500))); + let mut buf = [0u8; 512]; + let _ = stream.read(&mut buf); + } + + fn wait_workers(handler: &Handler, pred: impl Fn(usize) -> bool, label: &str) -> usize { + for _ in 0..100 { + let n = handler.workers.count(); + if pred(n) { + return n; + } + thread::sleep(Duration::from_millis(50)); + } + let n = handler.workers.count(); + panic!("{}: last worker count was {}", label, n); + } + + /// Live stratum listener: reconnect storm must not leave workers or FDs behind. + #[test] + fn test_live_reconnect_storm_workers_and_fds_bounded() { + let dir = ".grin_stratum_live_reconnect"; + let handler = setup_handler(dir); + let addr = start_test_stratum(handler.clone()); + + let fd_before = count_open_fds(); + + // Reconnecting miner simulation: open, login, drop — many times. + const CYCLES: usize = 150; + for _ in 0..CYCLES { + let mut stream = StdTcpStream::connect_timeout(&addr, Duration::from_secs(2)) + .expect("connect"); + stratum_login(&mut stream); + drop(stream); + } + + wait_workers(&handler, |n| n == 0, "drain after reconnect storm"); + + // Concurrent holds then release. + let mut held = Vec::new(); + const CONCURRENT: usize = 40; + for _ in 0..CONCURRENT { + let mut stream = StdTcpStream::connect_timeout(&addr, Duration::from_secs(2)) + .expect("connect concurrent"); + stratum_login(&mut stream); + held.push(stream); + } + let concurrent = wait_workers( + &handler, + |n| n > 0 && n <= CONCURRENT, + "concurrent workers", + ); + assert!(concurrent <= CONCURRENT); + drop(held); + + wait_workers(&handler, |n| n == 0, "drain after concurrent release"); + + if let (Some(before), Some(after)) = (fd_before, count_open_fds()) { + // Listener + misc FDs may grow slightly; must not track ~CYCLES connections. + assert!( + after < before + 80, + "possible FD leak: before={} after={} (cycles={})", + before, + after, + CYCLES + ); + } + + // worker_stats must not grow with every reconnect (slot reuse). + assert!( + handler.workers.stratum_stats.read().worker_stats.len() <= CONCURRENT + 5, + "worker_stats grew unboundedly: {}", + handler.workers.stratum_stats.read().worker_stats.len() + ); + + let _ = std::fs::remove_dir_all(Path::new(dir)); + } + + /// When at max workers, further accepts are closed immediately and count stays capped. + #[test] + fn test_live_max_workers_cap() { + let dir = ".grin_stratum_live_max"; + let handler = setup_handler(dir); + let addr = start_test_stratum(handler.clone()); + + // Hold more connections than the cap. + let mut held = Vec::new(); + for i in 0..(MAX_STRATUM_WORKERS + 16) { + match StdTcpStream::connect_timeout(&addr, Duration::from_secs(2)) { + Ok(mut stream) => { + let _ = stream.set_nodelay(true); + if i < MAX_STRATUM_WORKERS { + stratum_login(&mut stream); + } + held.push(stream); + } + Err(_) => break, + } + } + // Give accept loop time to reject surplus. + thread::sleep(Duration::from_millis(500)); + let count = handler.workers.count(); + assert!( + count <= MAX_STRATUM_WORKERS, + "worker count exceeded cap: {} > {}", + count, + MAX_STRATUM_WORKERS + ); + assert!( + count >= MAX_STRATUM_WORKERS.saturating_sub(5), + "expected near-cap workers, got {}", + count + ); + + drop(held); + wait_workers(&handler, |n| n == 0, "drain after max-cap release"); + + let _ = std::fs::remove_dir_all(Path::new(dir)); + } + #[test] fn test_worker_slot_reuse_after_disconnect() { let stats = Arc::new(RwLock::new(StratumStats::default())); From 6b5f4723ba03d56fc16a8f2af5b29a6339d34667 Mon Sep 17 00:00:00 2001 From: iho Date: Thu, 9 Jul 2026 10:46:05 +0300 Subject: [PATCH 4/6] test: add full-node stratum reconnect soak script Script boots grin --usernet with stratum enabled, drives reconnecting TCP "miners", and asserts process FD count and established stratum sockets stay bounded after clients disconnect (issue #3867). --- scripts/stratum_reconnect_soak.sh | 306 ++++++++++++++++++++++++++++++ 1 file changed, 306 insertions(+) create mode 100755 scripts/stratum_reconnect_soak.sh diff --git a/scripts/stratum_reconnect_soak.sh b/scripts/stratum_reconnect_soak.sh new file mode 100755 index 0000000000..ec575b627c --- /dev/null +++ b/scripts/stratum_reconnect_soak.sh @@ -0,0 +1,306 @@ +#!/usr/bin/env bash +# Full-node soak: run grin with stratum enabled + reconnecting "miner" clients. +# Verifies worker connections and process FDs stay bounded (issue #3867 / PR #3889). +# +# Usage (from repo root): +# ./scripts/stratum_reconnect_soak.sh +# +# Env overrides: +# GRIN_BIN path to grin binary (default: target/debug/grin) +# CYCLES connect/login/drop cycles (default: 120) +# CONCURRENT concurrent held connections (default: 32) +# FD_GROWTH_MAX max allowed FD growth over soak (default: 100) + +set -euo pipefail + +ROOT="$(cd "$(dirname "$0")/.." && pwd)" +cd "$ROOT" + +GRIN_BIN="${GRIN_BIN:-$ROOT/target/debug/grin}" +CYCLES="${CYCLES:-120}" +CONCURRENT="${CONCURRENT:-32}" +FD_GROWTH_MAX="${FD_GROWTH_MAX:-100}" +STRATUM_HOST="127.0.0.1" +# Ephemeral high port avoids collisions with leftover usernet nodes. +STRATUM_PORT="${STRATUM_PORT:-$(( 24000 + (RANDOM % 1000) ))}" +API_PORT="${API_PORT:-$(( 25000 + (RANDOM % 1000) ))}" +P2P_PORT="${P2P_PORT:-$(( 26000 + (RANDOM % 1000) ))}" +WORKDIR="$(mktemp -d "${TMPDIR:-/tmp}/grin-stratum-soak.XXXXXX")" +LOG="$WORKDIR/grin.log" +PID="" + +cleanup() { + if [[ -n "${PID}" ]] && kill -0 "$PID" 2>/dev/null; then + echo "==> Stopping grin (pid $PID)" + kill -TERM "$PID" 2>/dev/null || true + for _ in $(seq 1 30); do + kill -0 "$PID" 2>/dev/null || break + sleep 0.2 + done + kill -KILL "$PID" 2>/dev/null || true + wait "$PID" 2>/dev/null || true + fi +} +trap cleanup EXIT + +count_fds() { + local pid="$1" + if command -v lsof >/dev/null 2>&1; then + # Count open files for the process (includes sockets). + lsof -nP -p "$pid" 2>/dev/null | tail -n +2 | wc -l | tr -d ' ' + elif [[ -d "/proc/$pid/fd" ]]; then + ls -1 "/proc/$pid/fd" 2>/dev/null | wc -l | tr -d ' ' + else + echo "0" + fi +} + +count_tcp_established_stratum() { + local pid="$1" + if command -v lsof >/dev/null 2>&1; then + # ESTABLISHED only (exclude LISTEN). Match local or remote stratum port. + lsof -nP -p "$pid" -iTCP 2>/dev/null \ + | awk -v p=":${STRATUM_PORT}" ' + NR>1 && $0 ~ /ESTABLISHED/ && index($0, p) { c++ } + END { print c+0 } + ' + else + echo "0" + fi +} + +wait_port() { + local host="$1" port="$2" tries="${3:-60}" + for i in $(seq 1 "$tries"); do + if (echo >/dev/tcp/"$host"/"$port") >/dev/null 2>&1; then + return 0 + fi + # bash /dev/tcp may be unavailable; fall back to python + if python3 - "$host" "$port" <<'PY' 2>/dev/null; then +import socket, sys +s = socket.socket() +s.settimeout(0.3) +try: + s.connect((sys.argv[1], int(sys.argv[2]))) + sys.exit(0) +except Exception: + sys.exit(1) +finally: + s.close() +PY + return 0 + fi + sleep 0.5 + done + return 1 +} + +echo "==> Workdir: $WORKDIR" +echo "==> Building grin if needed" +if [[ ! -x "$GRIN_BIN" ]]; then + cargo build -p grin +fi + +echo "==> Generating usernet config with stratum enabled" +mkdir -p "$WORKDIR" +( + cd "$WORKDIR" + "$GRIN_BIN" --usernet server config >/dev/null +) + +CFG="$WORKDIR/grin-server.toml" +if [[ ! -f "$CFG" ]]; then + echo "ERROR: expected $CFG after 'grin server config'" >&2 + exit 1 +fi + +# Kill any leftover listener on our chosen ports (best-effort). +if command -v lsof >/dev/null 2>&1; then + for p in "$STRATUM_PORT" "$API_PORT" "$P2P_PORT"; do + lsof -nP -iTCP:"$p" -sTCP:LISTEN -t 2>/dev/null | xargs kill -9 2>/dev/null || true + done +fi + +# Enable stratum, disable TUI, skip sync wait, burn rewards (no wallet needed). +export STRATUM_PORT API_PORT P2P_PORT +python3 - "$CFG" <<'PY' +import re, sys +path = sys.argv[1] +text = open(path).read() + +def set_key(text, key, value): + # Match key = ... at line start (optional spaces) + pat = re.compile(rf'(?m)^(\s*{re.escape(key)}\s*=\s*).*$') + if pat.search(text): + return pat.sub(rf'\g<1>{value}', text) + return text + +import os +stratum = os.environ["STRATUM_PORT"] +api = os.environ["API_PORT"] +p2p = os.environ["P2P_PORT"] +text = set_key(text, 'run_tui', 'false') +text = set_key(text, 'skip_sync_wait', 'true') +text = set_key(text, 'enable_stratum_server', 'true') +text = set_key(text, 'burn_reward', 'true') +text = set_key(text, 'stratum_server_addr', f'"127.0.0.1:{stratum}"') +text = set_key(text, 'api_http_addr', f'"127.0.0.1:{api}"') +# p2p listen port field is just `port` under [server.p2p_config] +# Prefer the commented-uncommented style: replace first bare port = under p2p section via line walk. +lines = text.splitlines(True) +out = [] +in_p2p = False +for line in lines: + if line.strip().startswith('[server.p2p_config]'): + in_p2p = True + out.append(line) + continue + if in_p2p and line.strip().startswith('['): + in_p2p = False + if in_p2p and re.match(r'^\s*port\s*=', line): + out.append(f'port = {p2p}\n') + continue + out.append(line) +text = ''.join(out) +# Keep usernet local-only (must remain a quoted TOML string) +text = set_key(text, 'seeding_type', '"None"') +open(path, 'w').write(text) +print(f"config patched stratum=127.0.0.1:{stratum} api={api} p2p={p2p}") +PY + +echo "==> Starting grin --usernet --no-tui server run" +# Run grin as the background job itself (not a wrapper subshell) so FD accounting +# via lsof -p targets the real server process. +( + cd "$WORKDIR" + exec "$GRIN_BIN" --usernet --no-tui server run +) >"$LOG" 2>&1 & +PID=$! +echo " pid=$PID log=$LOG" + +echo "==> Waiting for stratum port ${STRATUM_HOST}:${STRATUM_PORT}" +if ! wait_port "$STRATUM_HOST" "$STRATUM_PORT" 90; then + echo "ERROR: stratum port never opened" >&2 + echo "---- grin log (tail) ----" >&2 + tail -80 "$LOG" >&2 || true + exit 1 +fi +echo " stratum is accepting connections" + +# Allow a moment for server threads to settle. +sleep 1 +FD_BEFORE=$(count_fds "$PID") +TCP_BEFORE=$(count_tcp_established_stratum "$PID") +echo "==> Baseline FDs=$FD_BEFORE established-stratum-TCP=$TCP_BEFORE" + +echo "==> Reconnect storm: $CYCLES connect/login/drop cycles" +python3 - "$STRATUM_HOST" "$STRATUM_PORT" "$CYCLES" "$CONCURRENT" <<'PY' +import socket, sys, time + +host, port = sys.argv[1], int(sys.argv[2]) +cycles, concurrent = int(sys.argv[3]), int(sys.argv[4]) +login = ( + b'{"id":1,"jsonrpc":"2.0","method":"login",' + b'"params":{"login":"soak","pass":"x","agent":"soak-script"}}\n' +) +keepalive = b'{"id":2,"jsonrpc":"2.0","method":"keepalive","params":null}\n' + +def connect_login(): + s = socket.create_connection((host, port), timeout=3) + s.settimeout(1.0) + s.sendall(login) + try: + s.recv(1024) + except Exception: + pass + return s + +# Phase 1: reconnect storm +for i in range(cycles): + s = connect_login() + s.close() + if (i + 1) % 40 == 0: + print(f" cycle {i+1}/{cycles}", flush=True) + +# Phase 2: hold concurrent connections, then drop +print(f" holding {concurrent} concurrent connections", flush=True) +held = [] +for _ in range(concurrent): + held.append(connect_login()) +time.sleep(1.0) +# Send keepalive on each so idle path is exercised a bit +for s in held: + try: + s.sendall(keepalive) + s.recv(1024) + except Exception: + pass +time.sleep(0.5) +for s in held: + s.close() +print(" concurrent connections closed", flush=True) +# Allow server tasks to observe EOF and free workers +time.sleep(1.5) +print("miner simulation done", flush=True) +PY + +FD_AFTER=$(count_fds "$PID") +TCP_AFTER=$(count_tcp_established_stratum "$PID") +FD_DELTA=$((FD_AFTER - FD_BEFORE)) +echo "==> After soak: FDs=$FD_AFTER (delta=$FD_DELTA) established-stratum-TCP=$TCP_AFTER" + +# Give a little more time for cleanup +sleep 1 +FD_FINAL=$(count_fds "$PID") +TCP_FINAL=$(count_tcp_established_stratum "$PID") +echo "==> Final: FDs=$FD_FINAL established-stratum-TCP=$TCP_FINAL" + +echo "==> Checking bounds" +FAIL=0 +if [[ "$FD_DELTA" -gt "$FD_GROWTH_MAX" ]]; then + echo "FAIL: FD growth $FD_DELTA exceeds limit $FD_GROWTH_MAX" >&2 + FAIL=1 +else + echo "OK: FD growth $FD_DELTA <= $FD_GROWTH_MAX" +fi + +# After drop, should not still hold client ESTABLISHED sockets to stratum. +if [[ "$TCP_FINAL" -gt 4 ]]; then + echo "FAIL: still have $TCP_FINAL ESTABLISHED TCP to stratum after clients closed" >&2 + FAIL=1 +else + echo "OK: established stratum TCP after close = $TCP_FINAL (expected ~0)" +fi + +if ! kill -0 "$PID" 2>/dev/null; then + echo "FAIL: grin process died during soak" >&2 + tail -40 "$LOG" >&2 || true + FAIL=1 +else + echo "OK: grin still running" +fi + +# Ignore bind panics from concurrent leftover nodes on shared default ports; +# with unique ports a bind panic is a real failure. +if grep -qi "Too many open files" "$LOG"; then + echo "FAIL: EMFILE in log" >&2 + grep -i "Too many open files" "$LOG" | tail -20 >&2 || true + FAIL=1 +elif grep -qi "panicked" "$LOG"; then + echo "FAIL: panic in log" >&2 + grep -i "panicked" "$LOG" | tail -20 >&2 || true + FAIL=1 +else + echo "OK: no panic/EMFILE in log" +fi + +if [[ "$FAIL" -ne 0 ]]; then + echo "---- grin log (tail) ----" >&2 + tail -60 "$LOG" >&2 || true + exit 1 +fi + +echo "" +echo "PASS: full-node stratum reconnect soak bounded" +echo " workdir=$WORKDIR" +echo " cycles=$CYCLES concurrent=$CONCURRENT fd_before=$FD_BEFORE fd_after=$FD_AFTER" From 0f5210fb4971122a5cae0b70ee715d766a7031f6 Mon Sep 17 00:00:00 2001 From: iho Date: Thu, 9 Jul 2026 10:48:07 +0300 Subject: [PATCH 5/6] test: make max-workers stratum test non-flaky Pre-fill WorkersList to MAX_STRATUM_WORKERS with dummy channels instead of opening 256+ concurrent TCP sockets, which was unreliable under load (only ~117 workers registered). Still asserts surplus TCP accepts do not push the count past the cap. --- servers/src/mining/stratumserver.rs | 79 ++++++++++++++++++----------- 1 file changed, 50 insertions(+), 29 deletions(-) diff --git a/servers/src/mining/stratumserver.rs b/servers/src/mining/stratumserver.rs index b9c738ddc4..11910aba65 100644 --- a/servers/src/mining/stratumserver.rs +++ b/servers/src/mining/stratumserver.rs @@ -1169,8 +1169,8 @@ mod tests { // Reconnecting miner simulation: open, login, drop — many times. const CYCLES: usize = 150; for _ in 0..CYCLES { - let mut stream = StdTcpStream::connect_timeout(&addr, Duration::from_secs(2)) - .expect("connect"); + let mut stream = + StdTcpStream::connect_timeout(&addr, Duration::from_secs(2)).expect("connect"); stratum_login(&mut stream); drop(stream); } @@ -1186,11 +1186,7 @@ mod tests { stratum_login(&mut stream); held.push(stream); } - let concurrent = wait_workers( - &handler, - |n| n > 0 && n <= CONCURRENT, - "concurrent workers", - ); + let concurrent = wait_workers(&handler, |n| n > 0 && n <= CONCURRENT, "concurrent workers"); assert!(concurrent <= CONCURRENT); drop(held); @@ -1218,43 +1214,68 @@ mod tests { } /// When at max workers, further accepts are closed immediately and count stays capped. + /// + /// We pre-fill the worker table (no real sockets) so the test does not depend on + /// opening 256 concurrent TCP connections — which is flaky under CI/load. #[test] fn test_live_max_workers_cap() { let dir = ".grin_stratum_live_max"; let handler = setup_handler(dir); let addr = start_test_stratum(handler.clone()); - // Hold more connections than the cap. - let mut held = Vec::new(); - for i in 0..(MAX_STRATUM_WORKERS + 16) { + // Fill to the hard cap with dummy channels (kept alive so slots stay "connected"). + let mut keep_alive: Vec> = Vec::with_capacity(MAX_STRATUM_WORKERS); + for _ in 0..MAX_STRATUM_WORKERS { + let (tx, rx) = mpsc::channel(1); + handler.workers.add_worker(tx); + keep_alive.push(rx); + } + assert_eq!( + handler.workers.count(), + MAX_STRATUM_WORKERS, + "pre-fill should reach exact cap" + ); + + // Extra TCP connects must not register more workers (accept drops them). + let mut rejected = Vec::new(); + for _ in 0..16 { match StdTcpStream::connect_timeout(&addr, Duration::from_secs(2)) { - Ok(mut stream) => { + Ok(stream) => { let _ = stream.set_nodelay(true); - if i < MAX_STRATUM_WORKERS { - stratum_login(&mut stream); - } - held.push(stream); + rejected.push(stream); } Err(_) => break, } } - // Give accept loop time to reject surplus. - thread::sleep(Duration::from_millis(500)); - let count = handler.workers.count(); - assert!( - count <= MAX_STRATUM_WORKERS, - "worker count exceeded cap: {} > {}", - count, - MAX_STRATUM_WORKERS - ); - assert!( - count >= MAX_STRATUM_WORKERS.saturating_sub(5), - "expected near-cap workers, got {}", - count + thread::sleep(Duration::from_millis(300)); + assert_eq!( + handler.workers.count(), + MAX_STRATUM_WORKERS, + "accept path must not exceed MAX_STRATUM_WORKERS under surplus connects" ); - drop(held); + // Free pre-filled slots; surplus TCP clients (if any still open) should not + // leave the table permanently non-empty after we drop everything. + drop(keep_alive); + // Manually remove dummy workers (Drop on Receiver does not call remove_worker). + // Re-read ids from stats / map: + let ids: Vec = handler + .workers + .workers_list + .read() + .keys() + .copied() + .collect(); + for id in ids { + handler.workers.remove_worker(id); + } + drop(rejected); + wait_workers(&handler, |n| n == 0, "drain after max-cap release"); + assert!( + handler.workers.count() <= MAX_STRATUM_WORKERS, + "cap invariant" + ); let _ = std::fs::remove_dir_all(Path::new(dir)); } From 86aaf7899106fc128ab3f01799ef08482a62d54f Mon Sep 17 00:00:00 2001 From: iho Date: Thu, 9 Jul 2026 10:50:28 +0300 Subject: [PATCH 6/6] test: confirm idle stratum miners are disconnected Parameterize handle_connection idle timeout and add a live test that connects, stays silent past a short timeout, and asserts the worker is removed and the TCP session is closed. Production still uses the 5-minute WORKER_IDLE_TIMEOUT constant. --- servers/src/mining/stratumserver.rs | 76 ++++++++++++++++++++++++++--- 1 file changed, 68 insertions(+), 8 deletions(-) diff --git a/servers/src/mining/stratumserver.rs b/servers/src/mining/stratumserver.rs index 11910aba65..52ebf6267c 100644 --- a/servers/src/mining/stratumserver.rs +++ b/servers/src/mining/stratumserver.rs @@ -642,7 +642,7 @@ fn accept_connections(listen_addr: SocketAddr, handler: Arc) { if let Err(e) = socket.set_nodelay(true) { debug!("Stratum: set_nodelay failed for {}: {}", peer_addr, e); } - handle_connection(socket, peer_addr, handler).await; + handle_connection(socket, peer_addr, handler, WORKER_IDLE_TIMEOUT).await; }); } Err(e) => { @@ -659,7 +659,15 @@ fn accept_connections(listen_addr: SocketAddr, handler: Arc) { } /// Run a single stratum client connection until it ends, then always free the worker slot. -async fn handle_connection(socket: TcpStream, peer_addr: SocketAddr, handler: Arc) { +/// +/// `idle_timeout` is how long a session may sit with no successful read/write before +/// being closed (production uses [`WORKER_IDLE_TIMEOUT`]; tests may pass a shorter value). +async fn handle_connection( + socket: TcpStream, + peer_addr: SocketAddr, + handler: Arc, + idle_timeout: Duration, +) { let (tx, mut rx) = mpsc::channel(WORKER_QUEUE_SIZE); let worker_id = handler.workers.add_worker(tx); info!("Worker {} connected from {}", worker_id, peer_addr); @@ -689,7 +697,7 @@ async fn handle_connection(socket: TcpStream, peer_addr: SocketAddr, handler: Ar let framed = Framed::new(socket, LinesCodec::new_with_max_length(MAX_RPC_LINE_BYTES)); let (mut writer, mut reader) = framed.split(); - let mut idle_deadline = Instant::now() + WORKER_IDLE_TIMEOUT; + let mut idle_deadline = Instant::now() + idle_timeout; loop { tokio::select! { @@ -698,7 +706,7 @@ async fn handle_connection(socket: TcpStream, peer_addr: SocketAddr, handler: Ar line = reader.try_next() => { match line { Ok(Some(line)) => { - idle_deadline = Instant::now() + WORKER_IDLE_TIMEOUT; + idle_deadline = Instant::now() + idle_timeout; let request = match serde_json::from_str(&line) { Ok(r) => r, Err(e) => { @@ -733,7 +741,7 @@ async fn handle_connection(socket: TcpStream, peer_addr: SocketAddr, handler: Ar msg = rx.recv() => { match msg { Some(line) => { - idle_deadline = Instant::now() + WORKER_IDLE_TIMEOUT; + idle_deadline = Instant::now() + idle_timeout; // Bound write time so a stalled peer cannot pin the task forever. match timeout(Duration::from_secs(30), writer.send(line)).await { Ok(Ok(())) => {} @@ -762,9 +770,9 @@ async fn handle_connection(socket: TcpStream, peer_addr: SocketAddr, handler: Ar _ = tokio::time::sleep_until(idle_deadline) => { warn!( - "Worker {}: idle timeout ({}s) from {}, closing", + "Worker {}: idle timeout ({:?}) from {}, closing", worker_id, - WORKER_IDLE_TIMEOUT.as_secs(), + idle_timeout, peer_addr ); break; @@ -1102,6 +1110,10 @@ mod tests { /// Start the real accept loop on an ephemeral port; return the bound address. fn start_test_stratum(handler: Arc) -> SocketAddr { + start_test_stratum_with_idle(handler, WORKER_IDLE_TIMEOUT) + } + + fn start_test_stratum_with_idle(handler: Arc, idle_timeout: Duration) -> SocketAddr { let (addr_tx, addr_rx) = sync_channel(1); thread::spawn(move || { let rt = Runtime::new().unwrap(); @@ -1119,7 +1131,7 @@ mod tests { let handler = handler.clone(); tokio::spawn(async move { let _ = socket.set_nodelay(true); - handle_connection(socket, peer_addr, handler).await; + handle_connection(socket, peer_addr, handler, idle_timeout).await; }); } Err(_) => { @@ -1157,6 +1169,54 @@ mod tests { panic!("{}: last worker count was {}", label, n); } + /// Idle sessions are closed after the idle timeout (production: 5 minutes). + /// Uses a short timeout so the test does not wait wall-clock 5 minutes. + #[test] + fn test_live_idle_miner_disconnected() { + let dir = ".grin_stratum_live_idle"; + let handler = setup_handler(dir); + // Short idle timeout for the test; same code path as WORKER_IDLE_TIMEOUT. + let idle = Duration::from_millis(800); + let addr = start_test_stratum_with_idle(handler.clone(), idle); + + // Connect and complete login (counts as activity), then go silent. + let mut stream = + StdTcpStream::connect_timeout(&addr, Duration::from_secs(2)).expect("connect"); + stratum_login(&mut stream); + // Keep the socket open but send nothing further (idle miner). + let _ = stream.set_read_timeout(Some(Duration::from_millis(200))); + + wait_workers(&handler, |n| n >= 1, "worker should register after login"); + + // Before idle timeout, worker must still be present. + thread::sleep(idle / 2); + assert!( + handler.workers.count() >= 1, + "worker should still be connected before idle timeout" + ); + + // After idle timeout (+ slack for task scheduling), worker must be gone. + wait_workers( + &handler, + |n| n == 0, + "worker should be removed after idle timeout", + ); + + // Peer should observe the server closed the connection (read EOF / error). + let mut buf = [0u8; 64]; + let read_res = stream.read(&mut buf); + assert!( + matches!(read_res, Ok(0) | Err(_)), + "expected EOF or error after idle disconnect, got {:?}", + read_res + ); + + // Production constant is 5 minutes (document the contract this test stands in for). + assert_eq!(WORKER_IDLE_TIMEOUT, Duration::from_secs(5 * 60)); + + let _ = std::fs::remove_dir_all(Path::new(dir)); + } + /// Live stratum listener: reconnect storm must not leave workers or FDs behind. #[test] fn test_live_reconnect_storm_workers_and_fds_bounded() {