Merge pull request #147 from mempool/junderw/electrum-limits

Fix: Add Electrum timeout to connections

src/config.rs

@@ -68,6 +68,7 @@ pub struct Config {
     pub electrum_max_line_size: usize,
     pub electrum_max_subscriptions: usize,
     pub electrum_max_clients: usize,
+    pub electrum_idle_timeout: u64,
 
     #[cfg(feature = "liquid")]
     pub parent_network: BNetwork,
@@ -296,6 +297,11 @@ impl Config {
                     .long("electrum-max-clients")
                     .help("Maximum number of concurrent Electrum client connections.")
                     .default_value("10")
+            ).arg(
+                Arg::with_name("electrum_idle_timeout")
+                    .long("electrum-idle-timeout")
+                    .help("Maximum idle time in seconds since the last client request before disconnecting the Electrum connection.")
+                    .default_value("600")
             );
 
         #[cfg(unix)]
@@ -568,6 +574,7 @@ impl Config {
             electrum_max_line_size: value_t_or_exit!(m, "electrum_max_line_size", usize),
             electrum_max_subscriptions: value_t_or_exit!(m, "electrum_max_subscriptions", usize),
             electrum_max_clients: value_t_or_exit!(m, "electrum_max_clients", usize),
+            electrum_idle_timeout: value_t_or_exit!(m, "electrum_idle_timeout", u64),
             jsonrpc_import: m.is_present("jsonrpc_import"),
             light_mode: m.is_present("light_mode"),
             main_loop_delay: value_t_or_exit!(m, "main_loop_delay", u64),

src/electrum/server.rs

@@ -12,6 +12,7 @@ use std::sync::atomic::AtomicBool;
 use std::sync::mpsc::{Receiver, Sender};
 use std::sync::{Arc, Mutex};
 use std::thread;
+use std::time::{Duration, Instant};
 
 use bitcoin::hashes::sha256d::Hash as Sha256dHash;
 use error_chain::ChainedError;
@@ -124,6 +125,8 @@ struct Connection {
     txs_limit: usize,
     max_line_size: usize,
     max_subscriptions: usize,
+    idle_timeout: u64,
+    last_request_at: Instant,
     die_please: Option<Receiver<()>>,
     #[cfg(feature = "electrum-discovery")]
     discovery: Option<Arc<DiscoveryManager>>,
@@ -138,6 +141,7 @@ impl Connection {
         txs_limit: usize,
         max_line_size: usize,
         max_subscriptions: usize,
+        idle_timeout: u64,
         die_please: Receiver<()>,
         #[cfg(feature = "electrum-discovery")] discovery: Option<Arc<DiscoveryManager>>,
     ) -> Connection {
@@ -151,6 +155,8 @@ impl Connection {
             txs_limit,
             max_line_size,
             max_subscriptions,
+            idle_timeout,
+            last_request_at: Instant::now(),
             die_please: Some(die_please),
             #[cfg(feature = "electrum-discovery")]
             discovery,
@@ -562,14 +568,34 @@ impl Connection {
         Ok(())
     }
 
+    fn close_idle_connection(&mut self, idle_for: Duration) {
+        info!(
+            "[{}] closing idle connection after {} seconds without requests (timeout: {} seconds)",
+            self.stream.addr_string(),
+            idle_for.as_secs(),
+            self.idle_timeout,
+        );
+        self.chan.close();
+    }
+
     fn handle_replies(&mut self, shutdown: crossbeam_channel::Receiver<()>) -> Result<()> {
+        let idle_timeout = Duration::from_secs(self.idle_timeout);
         loop {
+            let elapsed = self.last_request_at.elapsed();
+            if elapsed > idle_timeout {
+                self.close_idle_connection(elapsed);
+                return Ok(());
+            }
+            let remaining = idle_timeout.saturating_sub(elapsed);
+            let idle_deadline = crossbeam_channel::after(remaining);
+
             crossbeam_channel::select! {
                 recv(self.chan.receiver()) -> msg => {
                     let msg = msg.chain_err(|| "channel closed")?;
                     trace!("RPC {:?}", msg);
                     match msg {
                         Message::Request(line) => {
+                            self.last_request_at = Instant::now();
                             let result = self.handle_line(&line);
                             self.send_values(&[result])?
                         }
@@ -589,6 +615,11 @@ impl Connection {
                     self.chan.close();
                     return Ok(());
                 }
+                recv(idle_deadline) -> _ => {
+                    let idle_for = self.last_request_at.elapsed();
+                    self.close_idle_connection(idle_for);
+                    return Ok(());
+                }
             }
         }
     }
@@ -888,6 +919,7 @@ impl RPC {
         let max_line_size = config.electrum_max_line_size;
         let max_subscriptions = config.electrum_max_subscriptions;
         let max_clients = config.electrum_max_clients;
+        let idle_timeout = config.electrum_idle_timeout;
 
         RPC {
             notification: notification.sender(),
@@ -956,6 +988,7 @@ impl RPC {
                             txs_limit,
                             max_line_size,
                             max_subscriptions,
+                            idle_timeout,
                             peace_receiver,
                             #[cfg(feature = "electrum-discovery")]
                             discovery,