Merge pull request #20 from interlay/greg/feat/bitcoin-connect-sync

feat: connect to bitcoin with retry

bitcoin/Cargo.toml

@@ -18,6 +18,7 @@ num-derive = "0.3"
 futures = "0.3.5"
 serde_json = "1"
 log = "0.4.0"
+hyper = "0.10"
 
 # Substrate dependencies
 sp-core = { git = "https://github.com/paritytech/substrate", branch = "rococo-v1" }

bitcoin/src/error.rs

@@ -10,6 +10,7 @@ use bitcoincore_rpc::{
 use hex::FromHexError;
 use serde_json::Error as SerdeJsonError;
 use thiserror::Error;
+use tokio::time::Elapsed;
 
 #[derive(Error, Debug)]
 pub enum Error {
@@ -27,6 +28,8 @@ pub enum Error {
     Secp256k1Error(#[from] Secp256k1Error),
     #[error("KeyError: {0}")]
     KeyError(#[from] KeyError),
+    #[error("Timeout: {0}")]
+    TimeElapsed(#[from] Elapsed),
 
     #[error("Could not confirm transaction")]
     ConfirmationError,
@@ -57,7 +60,7 @@ pub enum ConversionError {
 }
 
 // https://github.com/bitcoin/bitcoin/blob/be3af4f31089726267ce2dbdd6c9c153bb5aeae1/src/rpc/protocol.h#L43
-#[derive(Debug, FromPrimitive)]
+#[derive(Debug, FromPrimitive, PartialEq, Eq)]
 pub enum BitcoinRpcError {
     /// Standard JSON-RPC 2.0 errors
     RpcInvalidRequest = -32600,

bitcoin/src/iter.rs

@@ -254,7 +254,6 @@ mod tests {
             async fn wallet_has_public_key<P>(&self, public_key: P) -> Result<bool, Error>
                 where
                     P: Into<[u8; PUBLIC_KEY_SIZE]> + From<[u8; PUBLIC_KEY_SIZE]> + Clone + PartialEq + Send + Sync + 'static;
-            async fn wait_for_block_sync(&self, timeout: Duration) -> Result<(), Error>;
         }
     }
 

bitcoin/src/lib.rs

@@ -30,18 +30,22 @@ pub use bitcoincore_rpc::{
     Auth, Client, Error as BitcoinError, RpcApi,
 };
 pub use error::{BitcoinRpcError, ConversionError, Error};
+use hyper::Error as HyperError;
 pub use iter::{get_transactions, stream_blocks, stream_in_chain_transactions};
-use log::trace;
+use log::{info, trace};
 use sp_core::H256;
+use std::io::ErrorKind as IoErrorKind;
 use std::{sync::Arc, time::Duration};
 use tokio::sync::{Mutex, OwnedMutexGuard};
-use tokio::time::delay_for;
+use tokio::time::{delay_for, timeout};
 
 #[macro_use]
 extern crate num_derive;
 
 const NOT_IN_MEMPOOL_ERROR_CODE: i32 = BitcoinRpcError::RpcInvalidAddressOrKey as i32;
 
+const RETRY_DURATION_MS: Duration = Duration::from_millis(1000);
+
 #[derive(Debug, Clone)]
 pub struct TransactionMetadata {
     pub txid: Txid,
@@ -135,8 +139,6 @@ pub trait BitcoinCoreApi {
             + Send
             + Sync
             + 'static;
-
-    async fn wait_for_block_sync(&self, timeout: Duration) -> Result<(), Error>;
 }
 
 pub struct LockedTransaction {
@@ -166,6 +168,67 @@ impl BitcoinCore {
         }
     }
 
+    pub async fn new_with_retry(
+        rpc: Client,
+        network: Network,
+        timeout_duration: Duration,
+    ) -> Result<Self, Error> {
+        let core = Self::new(rpc, network);
+        core.connect(timeout_duration).await?;
+        core.sync().await?;
+        Ok(core)
+    }
+
+    /// Connect to a bitcoin-core full node or timeout.
+    ///
+    /// # Arguments
+    /// * `timeout_duration` - maximum duration before elapsing
+    async fn connect(&self, timeout_duration: Duration) -> Result<(), Error> {
+        info!("Connecting to bitcoin-core...");
+        timeout(timeout_duration, async move {
+            loop {
+                match self.rpc.get_blockchain_info() {
+                    Err(BitcoinError::JsonRpc(JsonRpcError::Hyper(HyperError::Io(err))))
+                        if err.kind() == IoErrorKind::ConnectionRefused =>
+                    {
+                        trace!("could not connect to bitcoin-core");
+                        delay_for(RETRY_DURATION_MS).await;
+                        continue;
+                    }
+                    Err(BitcoinError::JsonRpc(JsonRpcError::Rpc(err)))
+                        if BitcoinRpcError::from(err.clone()) == BitcoinRpcError::RpcInWarmup =>
+                    {
+                        // may be loading block index or verifying wallet
+                        trace!("bitcoin-core still in warm up");
+                        delay_for(RETRY_DURATION_MS).await;
+                        continue;
+                    }
+                    Ok(_) => {
+                        info!("Connected!");
+                        return Ok(());
+                    }
+                    Err(err) => return Err(err.into()),
+                }
+            }
+        })
+        .await?
+    }
+
+    /// Wait indefinitely for the node to sync.
+    async fn sync(&self) -> Result<(), Error> {
+        info!("Waiting for bitcoin-core to sync...");
+        loop {
+            let info = self.rpc.get_blockchain_info()?;
+            // NOTE: initial_block_download is always true on regtest
+            if !info.initial_block_download || info.verification_progress == 1.0 {
+                info!("Synced!");
+                return Ok(());
+            }
+            trace!("bitcoin-core not synced");
+            delay_for(RETRY_DURATION_MS).await;
+        }
+    }
+
     /// Wrapper of rust_bitcoincore_rpc::create_raw_transaction_hex that accepts an optional op_return
     fn create_raw_transaction_hex(
         &self,
@@ -596,18 +659,6 @@ impl BitcoinCoreApi for BitcoinCore {
         let wallet_pubkey = address_info.pubkey.ok_or(Error::MissingPublicKey)?;
         Ok(P::from(wallet_pubkey.key.serialize()) == public_key)
     }
-
-    async fn wait_for_block_sync(&self, timeout: Duration) -> Result<(), Error> {
-        loop {
-            let info = self.rpc.get_blockchain_info()?;
-            // NOTE: initial_block_download is always true on regtest
-            if !info.initial_block_download || info.verification_progress == 1.0 {
-                return Ok(());
-            }
-            trace!("Bitcoin not synced, sleeping for {:?}", timeout);
-            delay_for(timeout).await;
-        }
-    }
 }
 
 /// Extension trait for transaction, adding methods to help to match the Transaction to Replace/Redeem requests

runtime/src/integration/bitcoin_simulator.rs

@@ -451,7 +451,4 @@ impl BitcoinCoreApi for MockBitcoinCore {
     {
         Ok(true)
     }
-    async fn wait_for_block_sync(&self, _timeout: Duration) -> Result<(), BitcoinError> {
-        Ok(())
-    }
 }

staked-relayer/src/main.rs

@@ -93,15 +93,14 @@ async fn main() -> Result<(), Error> {
     let oracle_timeout_ms = opts.oracle_timeout_ms;
 
     let dummy_network = bitcoin::Network::Regtest; // we don't make any transaction so this is not used
-    let btc_rpc = Arc::new(BitcoinCore::new(
-        opts.bitcoin.new_client(None)?,
-        dummy_network,
-    ));
-
-    info!("Waiting for bitcoin core to sync");
-    btc_rpc
-        .wait_for_block_sync(Duration::from_millis(bitcoin_timeout_ms))
-        .await?;
+    let btc_rpc = Arc::new(
+        BitcoinCore::new_with_retry(
+            opts.bitcoin.new_client(None)?,
+            dummy_network,
+            Duration::from_millis(opts.connection_timeout_ms),
+        )
+        .await?,
+    );
 
     let (key_pair, _) = opts.account_info.get_key_pair()?;
     let signer = PairSigner::<PolkaBtcRuntime, _>::new(key_pair);

staked-relayer/src/status.rs

@@ -334,7 +334,6 @@ mod tests {
             async fn wallet_has_public_key<P>(&self, public_key: P) -> Result<bool, BitcoinError>
                 where
                     P: Into<[u8; PUBLIC_KEY_SIZE]> + From<[u8; PUBLIC_KEY_SIZE]> + Clone + PartialEq + Send + Sync + 'static;
-            async fn wait_for_block_sync(&self, timeout: Duration) -> Result<(), BitcoinError>;
         }
     }
 

staked-relayer/src/vault.rs

@@ -365,7 +365,6 @@ mod tests {
             async fn wallet_has_public_key<P>(&self, public_key: P) -> Result<bool, BitcoinError>
                 where
                     P: Into<[u8; PUBLIC_KEY_SIZE]> + From<[u8; PUBLIC_KEY_SIZE]> + Clone + PartialEq + Send + Sync + 'static;
-            async fn wait_for_block_sync(&self, timeout: Duration) -> Result<(), BitcoinError>;
         }
     }
 

vault/src/main.rs

@@ -17,15 +17,14 @@ async fn main() -> Result<(), Error> {
 
     let (pair, wallet) = opts.account_info.get_key_pair()?;
 
-    let btc_rpc = Arc::new(BitcoinCore::new(
-        opts.bitcoin.new_client(Some(&wallet))?,
-        opts.network.0,
-    ));
-
-    info!("Waiting for bitcoin core to sync");
-    btc_rpc
-        .wait_for_block_sync(Duration::from_millis(opts.bitcoin_timeout_ms))
-        .await?;
+    let btc_rpc = Arc::new(
+        BitcoinCore::new_with_retry(
+            opts.bitcoin.new_client(Some(&wallet))?,
+            opts.network.0,
+            Duration::from_millis(opts.connection_timeout_ms),
+        )
+        .await?,
+    );
 
     // load wallet. Exit on failure, since without wallet we can't do a lot
     btc_rpc

vault/src/replace.rs

@@ -413,7 +413,6 @@ mod tests {
             async fn wallet_has_public_key<P>(&self, public_key: P) -> Result<bool, BitcoinError>
                 where
                     P: Into<[u8; PUBLIC_KEY_SIZE]> + From<[u8; PUBLIC_KEY_SIZE]> + Clone + PartialEq + Send + Sync + 'static;
-            async fn wait_for_block_sync(&self, timeout: Duration) -> Result<(), BitcoinError>;
         }
     }