Merge pull request #216 from EddieHouston/feat/dual-daemon-rpc

Add --daemon-rpc-fallback-addr for backup bitcoind RPC

Author: EddieHouston

src/bin/electrs.rs

@@ -68,6 +68,7 @@ fn run_server(config: Arc<Config>, salt_rwlock: Arc<RwLock<String>>) -> Result<(
         &config.daemon_dir,
         &config.blocks_dir,
         config.daemon_rpc_addr,
+        config.daemon_rpc_fallback_addr,
         config.daemon_parallelism,
         config.cookie_getter(),
         config.network_type,

src/bin/tx-fingerprint-stats.rs

@@ -34,6 +34,7 @@ fn main() {
             &config.daemon_dir,
             &config.blocks_dir,
             config.daemon_rpc_addr,
+            config.daemon_rpc_fallback_addr,
             config.daemon_parallelism,
             config.cookie_getter(),
             config.network_type,

src/config.rs

@@ -25,6 +25,7 @@ pub struct Config {
     pub daemon_dir: PathBuf,
     pub blocks_dir: PathBuf,
     pub daemon_rpc_addr: SocketAddr,
+    pub daemon_rpc_fallback_addr: Option<SocketAddr>,
     pub daemon_parallelism: usize,
     pub cookie: Option<String>,
     pub electrum_rpc_addr: SocketAddr,
@@ -164,6 +165,12 @@ impl Config {
                     .help("Bitcoin daemon JSONRPC 'addr:port' to connect (default: 127.0.0.1:8332 for mainnet, 127.0.0.1:18332 for testnet3 and 127.0.0.1:48332 for testnet4 and 127.0.0.1:18443 for regtest)")
                     .takes_value(true),
             )
+            .arg(
+                Arg::with_name("daemon_rpc_fallback_addr")
+                    .long("daemon-rpc-fallback-addr")
+                    .help("Fallback Bitcoin daemon JSONRPC 'addr:port' to connect if the primary fails")
+                    .takes_value(true),
+            )
             .arg(
                 Arg::with_name("daemon_parallelism")
                     .long("daemon-parallelism")
@@ -414,6 +421,10 @@ impl Config {
                 .unwrap_or(&format!("127.0.0.1:{}", default_daemon_port)),
             "Bitcoin RPC",
         );
+        let daemon_rpc_fallback_addr: Option<SocketAddr> = m
+            .value_of("daemon_rpc_fallback_addr")
+            .map(|e| str_to_socketaddr(e, "Bitcoin Fallback RPC"));
+
         let electrum_rpc_addr: SocketAddr = str_to_socketaddr(
             m.value_of("electrum_rpc_addr")
                 .unwrap_or(&format!("127.0.0.1:{}", default_electrum_port)),
@@ -481,6 +492,7 @@ impl Config {
             daemon_dir,
             blocks_dir,
             daemon_rpc_addr,
+            daemon_rpc_fallback_addr,
             daemon_parallelism: value_t_or_exit!(m, "daemon_parallelism", usize),
             cookie,
             utxos_limit: value_t_or_exit!(m, "utxos_limit", usize),

src/daemon.rs

@@ -186,24 +186,48 @@ struct Connection {
     rx: Lines<BufReader<TcpStream>>,
     cookie_getter: Arc<dyn CookieGetter>,
     addr: SocketAddr,
+    fallback: Option<SocketAddr>,
     signal: Waiter,
 }
 
 #[trace]
-fn tcp_connect(addr: SocketAddr, signal: &Waiter) -> Result<TcpStream> {
+fn tcp_connect(
+    primary: SocketAddr,
+    fallback: Option<SocketAddr>,
+    signal: &Waiter,
+) -> Result<(TcpStream, SocketAddr)> {
     loop {
-        match TcpStream::connect_timeout(&addr, *DAEMON_CONNECTION_TIMEOUT) {
+        match TcpStream::connect_timeout(&primary, *DAEMON_CONNECTION_TIMEOUT) {
             Ok(conn) => {
                 // can only fail if DAEMON_TIMEOUT is 0
                 conn.set_read_timeout(Some(*DAEMON_READ_TIMEOUT)).unwrap();
                 conn.set_write_timeout(Some(*DAEMON_WRITE_TIMEOUT)).unwrap();
-                return Ok(conn);
+                return Ok((conn, primary));
             }
             Err(err) => {
+                let suffix = if fallback.is_some() {
+                    " (trying fallback...)"
+                } else {
+                    ""
+                };
                 warn!(
-                    "failed to connect daemon at {}: {} (backoff 3 seconds)",
-                    addr, err
+                    "failed to connect to primary daemon at {}: {}{}",
+                    primary, err, suffix
                 );
+                if let Some(fallback_addr) = fallback {
+                    match TcpStream::connect_timeout(&fallback_addr, *DAEMON_CONNECTION_TIMEOUT) {
+                        Ok(conn) => {
+                            info!("connected to fallback daemon at {}", fallback_addr);
+                            conn.set_read_timeout(Some(*DAEMON_READ_TIMEOUT)).unwrap();
+                            conn.set_write_timeout(Some(*DAEMON_WRITE_TIMEOUT)).unwrap();
+                            return Ok((conn, fallback_addr));
+                        }
+                        Err(err) => {
+                            warn!("failed to connect to fallback daemon at {}: {}", fallback_addr, err);
+                        }
+                    }
+                }
+                warn!("backoff 3 seconds before next attempt");
                 signal.wait(Duration::from_secs(3), false)?;
                 continue;
             }
@@ -215,10 +239,12 @@ impl Connection {
     #[trace]
     fn new(
         addr: SocketAddr,
+        fallback: Option<SocketAddr>,
         cookie_getter: Arc<dyn CookieGetter>,
         signal: Waiter,
     ) -> Result<Connection> {
-        let conn = tcp_connect(addr, &signal)?;
+        let (conn, active_addr) = tcp_connect(addr, fallback, &signal)?;
+        debug!("connected to bitcoind at {}", active_addr);
         let reader = BufReader::new(
             conn.try_clone()
                 .chain_err(|| format!("failed to clone {:?}", conn))?,
@@ -228,13 +254,19 @@ impl Connection {
             rx: reader.lines(),
             cookie_getter,
             addr,
+            fallback,
             signal,
         })
     }
 
     #[trace]
     fn reconnect(&self) -> Result<Connection> {
-        Connection::new(self.addr, self.cookie_getter.clone(), self.signal.clone())
+        Connection::new(
+            self.addr,
+            self.fallback,
+            self.cookie_getter.clone(),
+            self.signal.clone(),
+        )
     }
 
     #[trace]
@@ -353,6 +385,7 @@ impl Daemon {
         daemon_dir: &PathBuf,
         blocks_dir: &PathBuf,
         daemon_rpc_addr: SocketAddr,
+        daemon_rpc_fallback_addr: Option<SocketAddr>,
         daemon_parallelism: usize,
         cookie_getter: Arc<dyn CookieGetter>,
         network: Network,
@@ -365,6 +398,7 @@ impl Daemon {
             network,
             conn: Mutex::new(Connection::new(
                 daemon_rpc_addr,
+                daemon_rpc_fallback_addr,
                 cookie_getter,
                 signal.clone(),
             )?),

tests/common.rs

@@ -96,6 +96,7 @@ impl TestRunner {
             daemon_parallelism: 3,
             blocks_dir: daemon_subdir.join("blocks"),
             daemon_rpc_addr: params.rpc_socket.into(),
+            daemon_rpc_fallback_addr: None,
             cookie: None,
             electrum_rpc_addr: rand_available_addr(),
             http_addr: rand_available_addr(),
@@ -138,6 +139,7 @@ impl TestRunner {
             &config.daemon_dir,
             &config.blocks_dir,
             config.daemon_rpc_addr,
+            config.daemon_rpc_fallback_addr,
             config.daemon_parallelism,
             config.cookie_getter(),
             config.network_type,