daemon: add configurable max-age to recycle RPC connections

Long-lived daemon RPC connections stay pinned to a single backend for
their whole lifetime. When electrs connects through a load balancer such
as a Kubernetes ClusterSetIP (`*.clusterset.local`), a connection
established before a node rotation keeps routing to the original backend
via the existing TCP/conntrack flow, even after healthier/closer
backends become available. The connection is only re-established on
error, so a still-working-but-stale endpoint is never rebalanced.

Add a `--daemon-rpc-conn-max-age` option (seconds). When a connection
exceeds the configured age it is proactively recycled before the next
request, re-establishing the TCP connection so the load balancer can
re-select a backend. Defaults to 0 = unlimited (never recycle), so
behavior is unchanged unless explicitly enabled. The age check is also
applied to the per-thread connections used for parallel RPC requests.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: DeviaVir

src/bin/electrs.rs

@@ -74,6 +74,7 @@ fn run_server(config: Arc<Config>, salt_rwlock: Arc<RwLock<String>>) -> Result<(
         config.network_type,
         signal.clone(),
         &metrics,
+        config.daemon_conn_max_age,
     )?);
     info!("opening database at {}", config.db_path.display());
     let store = Arc::new(Store::open(&config, &metrics, true));

src/bin/tx-fingerprint-stats.rs

@@ -40,6 +40,7 @@ fn main() {
             config.network_type,
             signal,
             &metrics,
+            config.daemon_conn_max_age,
         )
         .unwrap(),
     );

src/config.rs

@@ -5,6 +5,7 @@ use std::net::SocketAddr;
 use std::net::ToSocketAddrs;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
+use std::time::Duration;
 use stderrlog;
 
 use crate::chain::Network;
@@ -27,6 +28,7 @@ pub struct Config {
     pub daemon_rpc_addr: SocketAddr,
     pub daemon_rpc_fallback_addr: Option<SocketAddr>,
     pub daemon_parallelism: usize,
+    pub daemon_conn_max_age: Option<Duration>,
     pub cookie: Option<String>,
     pub electrum_rpc_addr: SocketAddr,
     pub http_addr: SocketAddr,
@@ -177,6 +179,13 @@ impl Config {
                     .help("Number of JSONRPC requests to send in parallel")
                     .default_value("4")
             )
+            .arg(
+                Arg::with_name("daemon_rpc_conn_max_age")
+                    .long("daemon-rpc-conn-max-age")
+                    .help("Max age (in seconds) of a daemon RPC TCP connection before it is proactively recycled. Recycling re-establishes the connection, letting a load balancer (e.g. a Kubernetes ClusterSetIP) re-select a backend after node rotations. 0 = unlimited / never recycle (default)")
+                    .default_value("0")
+                    .takes_value(true),
+            )
             .arg(
                 Arg::with_name("monitoring_addr")
                     .long("monitoring-addr")
@@ -425,6 +434,15 @@ impl Config {
             .value_of("daemon_rpc_fallback_addr")
             .map(|e| str_to_socketaddr(e, "Bitcoin Fallback RPC"));
 
+        let daemon_conn_max_age: Option<Duration> = m
+            .value_of("daemon_rpc_conn_max_age")
+            .map(|s| {
+                s.parse::<u64>()
+                    .unwrap_or_else(|_| panic!("invalid daemon-rpc-conn-max-age: {}", s))
+            })
+            .filter(|&secs| secs > 0)
+            .map(Duration::from_secs);
+
         let electrum_rpc_addr: SocketAddr = str_to_socketaddr(
             m.value_of("electrum_rpc_addr")
                 .unwrap_or(&format!("127.0.0.1:{}", default_electrum_port)),
@@ -494,6 +512,7 @@ impl Config {
             daemon_rpc_addr,
             daemon_rpc_fallback_addr,
             daemon_parallelism: value_t_or_exit!(m, "daemon_parallelism", usize),
+            daemon_conn_max_age,
             cookie,
             utxos_limit: value_t_or_exit!(m, "utxos_limit", usize),
             electrum_rpc_addr,

src/daemon.rs

@@ -6,7 +6,7 @@ use std::net::{SocketAddr, TcpStream};
 use std::path::PathBuf;
 use std::str::FromStr;
 use std::sync::{Arc, Mutex};
-use std::time::Duration;
+use std::time::{Duration, Instant};
 use std::{env, fs, io};
 
 use base64::prelude::{Engine, BASE64_STANDARD};
@@ -188,6 +188,11 @@ struct Connection {
     addr: SocketAddr,
     fallback: Option<SocketAddr>,
     signal: Waiter,
+    // When the TCP connection was (re)established, used together with `max_age` to
+    // proactively recycle long-lived connections (see `is_expired`).
+    established: Instant,
+    // Maximum age of a connection before it is proactively recycled, or None for unlimited.
+    max_age: Option<Duration>,
 }
 
 #[trace]
@@ -242,6 +247,7 @@ impl Connection {
         fallback: Option<SocketAddr>,
         cookie_getter: Arc<dyn CookieGetter>,
         signal: Waiter,
+        max_age: Option<Duration>,
     ) -> Result<Connection> {
         let (conn, active_addr) = tcp_connect(addr, fallback, &signal)?;
         debug!("connected to bitcoind at {}", active_addr);
@@ -256,6 +262,8 @@ impl Connection {
             addr,
             fallback,
             signal,
+            established: Instant::now(),
+            max_age,
         })
     }
 
@@ -266,9 +274,17 @@ impl Connection {
             self.fallback,
             self.cookie_getter.clone(),
             self.signal.clone(),
+            self.max_age,
         )
     }
 
+    /// Whether this connection has exceeded its configured `max_age` and should be recycled.
+    /// Always false when no max age is configured (unlimited).
+    fn is_expired(&self) -> bool {
+        self.max_age
+            .map_or(false, |max_age| self.established.elapsed() >= max_age)
+    }
+
     #[trace]
     fn send(&mut self, request: &str) -> Result<()> {
         let cookie = &self.cookie_getter.get()?;
@@ -372,6 +388,7 @@ pub struct Daemon {
     conn: Mutex<Connection>,
     message_id: Counter, // for monotonic JSONRPC 'id'
     signal: Waiter,
+    conn_max_age: Option<Duration>,
 
     rpc_threads: Arc<rayon::ThreadPool>,
 
@@ -391,6 +408,7 @@ impl Daemon {
         network: Network,
         signal: Waiter,
         metrics: &Metrics,
+        conn_max_age: Option<Duration>,
     ) -> Result<Daemon> {
         let daemon = Daemon {
             daemon_dir: daemon_dir.clone(),
@@ -401,9 +419,11 @@ impl Daemon {
                 daemon_rpc_fallback_addr,
                 cookie_getter,
                 signal.clone(),
+                conn_max_age,
             )?),
             message_id: Counter::new(),
             signal: signal.clone(),
+            conn_max_age,
             rpc_threads: Arc::new(
                 rayon::ThreadPoolBuilder::new()
                     .num_threads(daemon_parallelism)
@@ -460,6 +480,7 @@ impl Daemon {
             conn: Mutex::new(self.conn.lock().unwrap().reconnect()?),
             message_id: Counter::new(),
             signal: self.signal.clone(),
+            conn_max_age: self.conn_max_age,
             rpc_threads: self.rpc_threads.clone(),
             latency: self.latency.clone(),
             size: self.size.clone(),
@@ -505,6 +526,18 @@ impl Daemon {
     #[trace]
     fn call_jsonrpc(&self, method: &str, request: &Value) -> Result<Value> {
         let mut conn = self.conn.lock().unwrap();
+        // Proactively recycle connections older than the configured max age. Re-establishing
+        // the TCP connection lets a fronting load balancer (e.g. a Kubernetes ClusterSetIP)
+        // re-select a backend, so a long-lived connection does not stay pinned to a stale
+        // endpoint after node rotations. No-op when no max age is configured (the default).
+        if conn.is_expired() {
+            debug!(
+                "recycling daemon RPC connection to {} after {:?}",
+                conn.addr,
+                conn.established.elapsed()
+            );
+            *conn = conn.reconnect()?;
+        }
         let timer = self.latency.with_label_values(&[method]).start_timer();
         let request = request.to_string();
         conn.send(&request)?;