From 88a8d141ab22e166ba73c48bc337f4d4579fc348 Mon Sep 17 00:00:00 2001 From: Chase Date: Tue, 4 Aug 2026 14:07:47 +0200 Subject: [PATCH] Add jittered Electrum client connection max age Add an opt-in --electrum-rpc-conn-max-age flag (default 0 = unlimited). Each accepted client gets a lifetime jittered between 50% and 100% of the configured maximum, so clients behind a connection-level load balancer reconnect gradually instead of in a synchronized storm. A single RPC-wide reaper thread tracks all connection deadlines in a min-heap and shuts the socket down at the absolute deadline. The socket shutdown unblocks both peer threads even when the writer is stuck in a blocking write to a client that stopped reading, which an in-band expiry check between messages could never catch. Connections are registered with the reaper in the acceptor immediately after the accept, so the deadline is enforced even while a socket waits in the accept queue during a connection burst. The reaper holds only a Weak reference to each stream, so connections that end early release their file descriptor immediately, and an amortized compaction pass drops their queue entries so the queue stays proportional to the number of live connections under high churn. On RPC shutdown the reaper is stopped with an explicit Shutdown message and joined, since the acceptor's sender clone can outlive the RPC thread. Teardown races (reader send against a closed reply channel, disconnect- rooted reader errors after the socket was shut down under it) now log at debug instead of warn/error. Signed-off-by: Chase --- src/config.rs | 14 ++ src/electrum/server.rs | 370 +++++++++++++++++++++++++++++++++++++++-- tests/common.rs | 1 + 3 files changed, 373 insertions(+), 12 deletions(-) diff --git a/src/config.rs b/src/config.rs index 85c591e59..341e3fb2b 100644 --- a/src/config.rs +++ b/src/config.rs @@ -31,6 +31,7 @@ pub struct Config { pub daemon_conn_max_age: Option, pub cookie: Option, pub electrum_rpc_addr: SocketAddr, + pub electrum_rpc_conn_max_age: Option, pub http_addr: SocketAddr, pub http_socket_file: Option, pub monitoring_addr: SocketAddr, @@ -156,6 +157,13 @@ impl Config { .help("Electrum server JSONRPC 'addr:port' to listen on (default: '127.0.0.1:50001' for mainnet, '127.0.0.1:60001' for testnet3, '127.0.0.1:40001' for testnet4 and '127.0.0.1:60401' for regtest)") .takes_value(true), ) + .arg( + Arg::with_name("electrum_rpc_conn_max_age") + .long("electrum-rpc-conn-max-age") + .help("Maximum age (in seconds) of inbound Electrum RPC TCP connections. Each connection is closed at a randomly selected age between 50% and 100% of this value so clients reconnect gradually and load balancers can redistribute them. 0 = unlimited / never disconnect (default)") + .default_value("0") + .takes_value(true), + ) .arg( Arg::with_name("http_addr") .long("http-addr") @@ -451,6 +459,11 @@ impl Config { .unwrap_or(&format!("127.0.0.1:{}", default_electrum_port)), "Electrum RPC", ); + let electrum_rpc_conn_max_age: Option = + match value_t_or_exit!(m, "electrum_rpc_conn_max_age", u64) { + 0 => None, // 0 = unlimited / never disconnect + secs => Some(Duration::from_secs(secs)), + }; let http_addr: SocketAddr = str_to_socketaddr( m.value_of("http_addr") .unwrap_or(&format!("127.0.0.1:{}", default_http_port)), @@ -519,6 +532,7 @@ impl Config { cookie, utxos_limit: value_t_or_exit!(m, "utxos_limit", usize), electrum_rpc_addr, + electrum_rpc_conn_max_age, electrum_txs_limit: value_t_or_exit!(m, "electrum_txs_limit", usize), electrum_banner, rpc_logging: { diff --git a/src/electrum/server.rs b/src/electrum/server.rs index 045ccf185..6887ccb98 100644 --- a/src/electrum/server.rs +++ b/src/electrum/server.rs @@ -1,14 +1,16 @@ -use std::collections::HashMap; +use std::cmp::Ordering; +use std::collections::{BinaryHeap, HashMap}; use std::io::{BufRead, BufReader, Write}; use std::net::{Shutdown, SocketAddr, TcpListener, TcpStream}; -use std::sync::mpsc::{self, Receiver, Sender, SyncSender, TrySendError}; -use std::sync::{Arc, Mutex, RwLock}; +use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender, SyncSender, TrySendError}; +use std::sync::{Arc, Mutex, RwLock, Weak}; use std::thread; -use std::time::Instant; +use std::time::{Duration, Instant}; use bitcoin::hashes::{sha256, sha256d::Hash as Sha256dHash, Hash, HashEngine}; use bitcoin::hex::DisplayHex; use error_chain::ChainedError; +use rand::Rng; use serde_json::{from_str, Value}; use electrs_macros::trace; @@ -158,7 +160,9 @@ struct Connection { query: Arc, last_header_entry: Option, status_hashes: HashMap, // ScriptHash -> StatusHash - stream: TcpStream, + // Shared with the connection reaper (via a Weak), which enforces the + // maximum connection age by shutting the socket down at the deadline. + stream: Arc, addr: SocketAddr, sender: SyncSender, stats: Arc, @@ -179,7 +183,7 @@ fn hash_ip_with_salt(salt: &str, ip: &str) -> String { impl Connection { pub fn new( query: Arc, - stream: TcpStream, + stream: Arc, addr: SocketAddr, sender: SyncSender, stats: Arc, @@ -667,7 +671,7 @@ impl Connection { fn send_values(&mut self, values: &[Value]) -> Result<()> { for value in values { let line = value.to_string() + "\n"; - self.stream + (&*self.stream) .write_all(line.as_bytes()) .chain_err(|| format!("failed to send response ({} bytes)", line.len()))?; } @@ -787,7 +791,9 @@ impl Connection { fn reader_thread(reader: BufReader, tx: SyncSender) -> Result<()> { let result = Connection::parse_requests(reader, &tx); if let Err(e) = tx.send(Message::Done) { - warn!("failed closing channel: {}", e); + // The writer already tore the channel down (e.g. after a write + // error or connection expiry) — expected during teardown races. + debug!("failed closing channel: {}", e); } result } @@ -821,7 +827,142 @@ impl Connection { let _ = self.stream.shutdown(Shutdown::Both); if let Err(err) = child.join().expect("receiver panicked") { - error!("[{}] receiver failed: {}", self.addr, err); + if is_disconnect(&err) || is_channel_closed(&err) { + // Reader failures rooted in a disconnect or in the reply + // channel tearing down are expected when the socket was shut + // down under it (client reset or expiry). + debug!("[{}] receiver closed: {}", self.addr, err); + } else { + error!("[{}] receiver failed: {}", self.addr, err); + } + } + } +} + +fn connection_lifetime(max_age: Option) -> Option { + max_age.and_then(|max_age| { + let max_secs = max_age.as_secs(); + if max_secs == 0 { + return None; + } + let min_secs = max_secs / 2 + max_secs % 2; + Some(Duration::from_secs( + rand::rng().random_range(min_secs..=max_secs), + )) + }) +} + +/// The absolute deadline for a new connection, jittered between 50% and 100% of +/// `max_age`. `None` means the connection never expires — either the max age is +/// disabled, or it is so large that the deadline is not representable. +fn connection_deadline(max_age: Option) -> Option { + connection_lifetime(max_age).and_then(|lifetime| Instant::now().checked_add(lifetime)) +} + +/// A connection registered with the reaper: shut `stream` down at `expires_at`. +struct ConnectionExpiry { + expires_at: Instant, + addr: SocketAddr, + // Weak so the reaper never keeps the socket (and its fd) alive after the + // connection ends on its own before the deadline. + stream: Weak, +} + +enum ReaperMessage { + Register(ConnectionExpiry), + Shutdown, +} + +// Ordered by soonest deadline first, so a BinaryHeap acts as a min-heap. +impl Ord for ConnectionExpiry { + fn cmp(&self, other: &Self) -> Ordering { + other.expires_at.cmp(&self.expires_at) + } +} + +impl PartialOrd for ConnectionExpiry { + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } +} + +impl PartialEq for ConnectionExpiry { + fn eq(&self, other: &Self) -> bool { + self.expires_at == other.expires_at + } +} + +impl Eq for ConnectionExpiry {} + +/// Compact the expiry queue once it reaches this many entries. +const REAPER_COMPACT_MIN: usize = 1024; + +/// Pending connection expiries, ordered by soonest deadline. Entries for +/// connections that ended before their deadline are dropped by an amortized +/// compaction pass on registration, so the queue stays proportional to the +/// number of live connections even under high connection churn combined with +/// a large max age. +struct ExpiryQueue { + expiries: BinaryHeap, + compact_at: usize, +} + +impl ExpiryQueue { + fn new() -> ExpiryQueue { + ExpiryQueue { + expiries: BinaryHeap::new(), + compact_at: REAPER_COMPACT_MIN, + } + } + + fn register(&mut self, expiry: ConnectionExpiry) { + self.expiries.push(expiry); + if self.expiries.len() >= self.compact_at { + self.expiries.retain(|e| e.stream.strong_count() > 0); + self.compact_at = REAPER_COMPACT_MIN.max(self.expiries.len() * 2); + } + } + + /// Shut down every connection whose deadline has passed. + fn reap_due(&mut self, now: Instant) { + while self.expiries.peek().map_or(false, |e| e.expires_at <= now) { + let expiry = self.expiries.pop().unwrap(); + // A connection that already ended no longer upgrades. + if let Some(stream) = expiry.stream.upgrade() { + debug!( + "[{}] maximum connection age reached, closing connection", + expiry.addr + ); + let _ = stream.shutdown(Shutdown::Both); + } + } + } + + fn next_deadline(&self) -> Option { + self.expiries.peek().map(|e| e.expires_at) + } +} + +/// Enforces the maximum connection age for all Electrum RPC connections by +/// shutting each socket down at its absolute deadline. The shutdown unblocks +/// both peer threads even when the writer is stuck in a blocking write to a +/// client that stopped reading, which an in-band expiry check between messages +/// could never catch. Runs until a `Shutdown` message arrives or the +/// registration channel is closed. +fn reap_expired_connections(registrations: Receiver) { + let mut queue = ExpiryQueue::new(); + loop { + queue.reap_due(Instant::now()); + let next = match queue.next_deadline() { + Some(deadline) => { + registrations.recv_timeout(deadline.saturating_duration_since(Instant::now())) + } + None => registrations.recv().map_err(RecvTimeoutError::from), + }; + match next { + Ok(ReaperMessage::Register(registration)) => queue.register(registration), + Ok(ReaperMessage::Shutdown) | Err(RecvTimeoutError::Disconnected) => return, + Err(RecvTimeoutError::Timeout) => {} } } } @@ -845,6 +986,20 @@ fn is_disconnect(err: &Error) -> bool { false } +/// True if the error chain is rooted in the reply channel closing, meaning the +/// writer half tore down first (e.g. on connection expiry or a write error) +/// while the reader still had a request in flight — expected during teardown. +fn is_channel_closed(err: &Error) -> bool { + let mut cause: Option<&(dyn std::error::Error + 'static)> = Some(err); + while let Some(e) = cause { + if e.downcast_ref::>().is_some() { + return true; + } + cause = e.source(); + } + false +} + #[trace] fn get_history( query: &Query, @@ -893,7 +1048,7 @@ impl RPC { fn start_notifier( notification: Channel, senders: Arc>>>, - acceptor: Sender>, + acceptor: Sender, SocketAddr)>>, ) { spawn_thread("notification", move || { for msg in notification.receiver().iter() { @@ -920,7 +1075,11 @@ impl RPC { }); } - fn start_acceptor(addr: SocketAddr) -> Channel> { + fn start_acceptor( + addr: SocketAddr, + conn_max_age: Option, + reaper: Option>, + ) -> Channel, SocketAddr)>> { let chan = Channel::unbounded(); let acceptor = chan.sender(); spawn_thread("acceptor", move || { @@ -938,6 +1097,19 @@ impl RPC { .set_nonblocking(false) .expect("failed to set connection as blocking"); stream.set_nodelay(true).expect("failed to set TCP_NODELAY"); + // Register with the reaper before enqueueing, so the deadline + // is anchored to the accept and enforced even while the socket + // waits behind other new connections during an accept burst. + let stream = Arc::new(stream); + if let Some(reaper) = &reaper { + if let Some(expires_at) = connection_deadline(conn_max_age) { + let _ = reaper.send(ReaperMessage::Register(ConnectionExpiry { + expires_at, + addr, + stream: Arc::downgrade(&stream), + })); + } + } if acceptor.send(Some((stream, addr))).is_err() { break; // receiver dropped, server is shutting down } @@ -994,13 +1166,28 @@ impl RPC { let rpc_addr = config.electrum_rpc_addr; let txs_limit = config.electrum_txs_limit; + let conn_max_age = config.electrum_rpc_conn_max_age; RPC { notification: notification.sender(), server: Some(spawn_thread("rpc", move || { let senders = Arc::new(Mutex::new(Vec::>::new())); - let acceptor = RPC::start_acceptor(rpc_addr); + // The reaper enforces the maximum connection age. It is + // stopped with an explicit Shutdown message below, since the + // acceptor's sender clone can outlive this thread (the + // acceptor stays blocked in accept() during shutdown). + let reaper = conn_max_age.map(|_| { + let (sender, receiver) = mpsc::channel(); + let handle = spawn_thread("reaper", move || reap_expired_connections(receiver)); + (sender, handle) + }); + + let acceptor = RPC::start_acceptor( + rpc_addr, + conn_max_age, + reaper.as_ref().map(|(sender, _)| sender.clone()), + ); RPC::start_notifier(notification, senders.clone(), acceptor.sender()); let mut threads = HashMap::new(); @@ -1064,6 +1251,12 @@ impl RPC { } trace!("RPC connections are closed"); + + if let Some((sender, handle)) = reaper { + let _ = sender.send(ReaperMessage::Shutdown); + handle.join().expect("reaper panicked"); + trace!("reaper stopped"); + } })), } } @@ -1097,4 +1290,157 @@ mod tests { "d474826bbd126d38bdfb1e61bf727a2d9a306ea1645071faf2638cc3891a2b30" ); } + + #[test] + fn connection_lifetime_is_disabled_without_max_age() { + assert_eq!(connection_lifetime(None), None); + assert_eq!(connection_lifetime(Some(Duration::ZERO)), None); + } + + #[test] + fn connection_lifetime_is_jittered_up_to_max_age() { + let max_age = Duration::from_secs(3_600); + for _ in 0..100 { + let lifetime = connection_lifetime(Some(max_age)).unwrap(); + assert!(lifetime >= Duration::from_secs(1_800)); + assert!(lifetime <= max_age); + } + } + + #[test] + fn connection_expiry_does_not_overflow() { + // A max age too large to be representable as a deadline must mean + // "never expires", not a panic. + assert_eq!( + connection_deadline(Some(Duration::from_secs(u64::MAX))), + None + ); + assert_eq!(connection_deadline(None), None); + assert!(connection_deadline(Some(Duration::from_secs(3_600))).is_some()); + } + + /// Spawn a reaper and register `stream` to expire at `expires_at`. + fn spawn_reaper(stream: &Arc, expires_at: Instant) -> mpsc::Sender { + let (registrations, receiver) = mpsc::channel(); + thread::spawn(move || reap_expired_connections(receiver)); + registrations + .send(ReaperMessage::Register(ConnectionExpiry { + expires_at, + addr: stream.peer_addr().unwrap(), + stream: Arc::downgrade(stream), + })) + .unwrap(); + registrations + } + + #[test] + fn reaper_stops_on_shutdown_message() { + let (registrations, receiver) = mpsc::channel(); + let reaper = thread::spawn(move || reap_expired_connections(receiver)); + // A pending far-future registration must not delay the shutdown. + registrations + .send(ReaperMessage::Register(ConnectionExpiry { + expires_at: Instant::now() + Duration::from_secs(3_600), + addr: "127.0.0.1:1".parse().unwrap(), + stream: std::sync::Weak::new(), + })) + .unwrap(); + registrations.send(ReaperMessage::Shutdown).unwrap(); + reaper.join().unwrap(); + } + + #[test] + fn reaper_closes_connection_of_client_that_stops_reading() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + // A client that never reads its socket: writes towards it fill the + // kernel buffers and then block indefinitely. + let client = TcpStream::connect(listener.local_addr().unwrap()).unwrap(); + let stream = Arc::new(listener.accept().unwrap().0); + + let started = Instant::now(); + let expires_at = started + Duration::from_millis(200); + let _registrations = spawn_reaper(&stream, expires_at); + + let (done_sender, done_receiver) = mpsc::channel(); + let writer = Arc::clone(&stream); + thread::spawn(move || { + let chunk = [0u8; 1 << 20]; + while (&*writer).write_all(&chunk).is_ok() {} + let _ = done_sender.send(()); + }); + + // The blocked write must be forced to fail at the deadline, not linger + // for as long as the client keeps the socket open. + done_receiver + .recv_timeout(Duration::from_secs(30)) + .expect("writer stayed blocked past the connection deadline"); + assert!(started.elapsed() >= Duration::from_millis(200)); + drop(client); + } + + #[test] + fn reaper_queue_drops_entries_of_finished_connections() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let _client = TcpStream::connect(listener.local_addr().unwrap()).unwrap(); + let (live_stream, addr) = listener.accept().unwrap(); + let live_stream = Arc::new(live_stream); + + // A Weak whose connection already ended (strong count is zero). + let dead = { + let stream = Arc::new(TcpStream::connect(listener.local_addr().unwrap()).unwrap()); + Arc::downgrade(&stream) + }; + + // Far-future deadlines: without compaction, the entries below would + // all sit in the queue until the deadline no matter that their + // connections are long gone. + let expires_at = Instant::now() + Duration::from_secs(3_600); + let mut queue = ExpiryQueue::new(); + for _ in 0..(REAPER_COMPACT_MIN - 1) { + queue.register(ConnectionExpiry { + expires_at, + addr, + stream: dead.clone(), + }); + } + // This registration reaches the compaction threshold, so the pass runs + // with both the dead entries and this live one in the queue. + queue.register(ConnectionExpiry { + expires_at, + addr, + stream: Arc::downgrade(&live_stream), + }); + + // Compaction dropped the dead entries and kept the live one. + assert_eq!(queue.expiries.len(), 1); + assert!(queue.expiries.peek().unwrap().stream.upgrade().is_some()); + } + + #[test] + fn reply_channel_teardown_is_not_logged_as_error() { + let (sender, receiver) = mpsc::sync_channel(1); + drop(receiver); + let err: Error = sender + .send(Message::Done) + .chain_err(|| "channel closed") + .unwrap_err(); + assert!(is_channel_closed(&err)); + assert!(!is_channel_closed(&"unrelated".into())); + } + + #[test] + fn reaper_does_not_keep_finished_connections_alive() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let _client = TcpStream::connect(listener.local_addr().unwrap()).unwrap(); + let stream = Arc::new(listener.accept().unwrap().0); + + let weak = Arc::downgrade(&stream); + let _registrations = spawn_reaper(&stream, Instant::now() + Duration::from_secs(3_600)); + + // The reaper holds only a Weak: once the connection is done with the + // stream, the socket must close immediately instead of staying open + // (leaking the fd) until the deadline. + drop(stream); + assert!(weak.upgrade().is_none()); + } } diff --git a/tests/common.rs b/tests/common.rs index 99a7730fd..1965b8629 100644 --- a/tests/common.rs +++ b/tests/common.rs @@ -100,6 +100,7 @@ impl TestRunner { daemon_rpc_fallback_addr: None, cookie: None, electrum_rpc_addr: rand_available_addr(), + electrum_rpc_conn_max_age: None, http_addr: rand_available_addr(), http_socket_file: None, // XXX test with socket file or tcp? monitoring_addr: rand_available_addr(),