[sBTC DR] Bitcoin client retry logic and prevention of race condition…

…s for Stacks contract calls (#102)

* feat: add retry logic to the bitcoin client and prevent racing contract call stacks transactions

* feat: improve how we track created but not broadcasted txs

* feat: make sure we track created transactions so we don't create duplicated

* fix: remove redundant clone

* fix: remove double space

---------

Co-authored-by: Friedger Müffke <[email protected]>

Cargo.toml

@@ -4,6 +4,7 @@ members = ["sbtc-cli", "sbtc-core", "stacks-core", "romeo"]
 [workspace.dependencies]
 anyhow = "1.0"
 array-bytes = "6.1.0"
+backoff = "0.4.0"
 bdk = "0.28.1"
 bitcoin = "0.29.2"
 clap = "4.1.1"

romeo/Cargo.toml

@@ -5,6 +5,7 @@ edition = "2021"
 
 [dependencies]
 anyhow.workspace = true
+backoff = { workspace = true, features = ["tokio"] }
 bdk = { workspace = true, features = ["esplora", "use-esplora-async"]}
 blockstack-core = { git = "https://github.com/stacks-network/stacks-blockchain/", branch = "master" }
 clap = { workspace = true, features = ["derive"] }

romeo/src/bitcoin_client.rs

@@ -1,41 +1,41 @@
 //! Bitcoin client
 
+use std::time::Duration;
+
 use bdk::bitcoin;
-use bdk::bitcoin::schnorr;
-use bdk::bitcoin::secp256k1;
 use bdk::esplora_client;
+use futures::Future;
+use tracing::trace;
 
 use crate::event;
 
+const BLOCK_POLLING_INTERVAL: Duration = Duration::from_secs(5);
+
 /// Facilitates communication with a Bitcoin esplora server
 #[derive(Debug, Clone)]
 pub struct BitcoinClient {
     client: esplora_client::AsyncClient,
-    private_key: bitcoin::PrivateKey,
 }
 
 impl BitcoinClient {
     /// Construct a new bitcoin client
-    pub fn new(esplora_url: &str, private_key: bitcoin::PrivateKey) -> anyhow::Result<Self> {
+    pub fn new(esplora_url: &str) -> anyhow::Result<Self> {
         let client = esplora_client::Builder::new(esplora_url).build_async()?;
 
-        Ok(Self {
-            client,
-            private_key,
-        })
+        Ok(Self { client })
     }
 
     /// Broadcast a bitcoin transaction
     pub async fn broadcast(&self, tx: &bitcoin::Transaction) -> anyhow::Result<()> {
-        Ok(self.client.broadcast(tx).await?)
+        retry(|| self.client.broadcast(tx)).await
     }
 
     /// Get the status of a transaction
     pub async fn get_tx_status(
         &self,
         txid: &bitcoin::Txid,
     ) -> anyhow::Result<event::TransactionStatus> {
-        let status = self.client.get_tx_status(txid).await?;
+        let status = retry(|| self.client.get_tx_status(txid)).await?;
 
         Ok(match status {
             Some(esplora_client::TxStatus {
@@ -53,65 +53,69 @@ impl BitcoinClient {
     /// this function will poll the blockchain until that height is reached.
     #[tracing::instrument(skip(self))]
     pub async fn fetch_block(&self, block_height: u32) -> anyhow::Result<bitcoin::Block> {
-        let mut current_height = self.client.get_height().await?;
+        let mut current_height = retry(|| self.client.get_height()).await?;
 
+        trace!("Looking for block height: {}", current_height + 1);
         while current_height < block_height {
-            tracing::debug!("Polling: {}", current_height);
-            tokio::time::sleep(std::time::Duration::from_secs(5)).await;
-            current_height = self.client.get_height().await?;
+            tokio::time::sleep(BLOCK_POLLING_INTERVAL).await;
+            current_height = retry(|| self.client.get_height()).await?;
         }
 
-        let block_summaries = self.client.get_blocks(Some(block_height)).await?;
+        let block_summaries = retry(|| self.client.get_blocks(Some(block_height))).await?;
         let block_summary = block_summaries
             .first()
             .ok_or_else(|| anyhow::anyhow!("Could not find block at given block height"))?;
 
-        let block = self
-            .client
-            .get_block_by_hash(&block_summary.id)
+        let block = retry(|| self.client.get_block_by_hash(&block_summary.id))
             .await?
             .ok_or_else(|| anyhow::anyhow!("Found no block for the given block hash"))?;
 
-        tracing::debug!("Fetched block");
+        trace!("Fetched block");
+
         Ok(block)
     }
 
     /// Get the current height of the Bitcoin chain
     pub async fn get_height(&self) -> anyhow::Result<u32> {
-        Ok(self.client.get_height().await?)
+        retry(|| self.client.get_height()).await
     }
 
     /// Sign relevant inputs of a bitcoin transaction
     pub async fn sign(&self, _tx: bitcoin::Transaction) -> anyhow::Result<bitcoin::Transaction> {
         // TODO #68
         todo!()
     }
+}
 
-    /// Bitcoin taproot address associated with the private key
-    pub async fn taproot_address(&self) -> bitcoin::Address {
-        let secp = secp256k1::Secp256k1::new();
-        let internal_key: schnorr::UntweakedPublicKey =
-            self.private_key.public_key(&secp).inner.into();
+async fn retry<T, O, Fut>(operation: O) -> anyhow::Result<T>
+where
+    O: Clone + Fn() -> Fut,
+    Fut: Future<Output = Result<T, bdk::esplora_client::Error>>,
+{
+    let operation = || async {
+        operation.clone()().await.map_err(|err| match err {
+            esplora_client::Error::Reqwest(_) => backoff::Error::transient(anyhow::anyhow!(err)),
+            err => backoff::Error::permanent(anyhow::anyhow!(err)),
+        })
+    };
 
-        bitcoin::Address::p2tr(&secp, internal_key, None, self.private_key.network)
-    }
+    let notify = |err, duration| {
+        trace!("Retrying in {:?} after error: {:?}", duration, err);
+    };
+
+    backoff::future::retry_notify(backoff::ExponentialBackoff::default(), operation, notify).await
 }
 
 #[cfg(test)]
 mod tests {
-    use crate::config::Config;
 
     use super::*;
 
     // These integration tests are for exploration/experimentation but should be removed once we have more decent tests
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     #[ignore]
     async fn get_block() {
-        let config =
-            Config::from_path("./testing/config.json").expect("Failed to find config file");
-
-        let bitcoin_client =
-            BitcoinClient::new("https://blockstream.info/testnet/api", config.private_key).unwrap();
+        let bitcoin_client = BitcoinClient::new("https://blockstream.info/testnet/api").unwrap();
 
         let block_height = bitcoin_client.get_height().await.unwrap();
         let block = bitcoin_client.fetch_block(block_height).await.unwrap();

romeo/src/config.rs

@@ -5,7 +5,7 @@ use std::{
     path::{Path, PathBuf},
 };
 
-use bdk::bitcoin::{self, PrivateKey};
+use bdk::bitcoin::{self, schnorr, secp256k1, PrivateKey};
 use blockstack_lib::{
     address::{
         AddressHashMode, C32_ADDRESS_VERSION_MAINNET_SINGLESIG,
@@ -115,6 +115,15 @@ impl Config {
         )
         .unwrap()
     }
+
+    /// Bitcoin taproot address associated with the private key
+    pub async fn taproot_address(&self) -> bitcoin::Address {
+        let secp = secp256k1::Secp256k1::new();
+        let internal_key: schnorr::UntweakedPublicKey =
+            self.private_key.public_key(&secp).inner.into();
+
+        bitcoin::Address::p2tr(&secp, internal_key, None, self.private_key.network)
+    }
 }
 
 fn normalize(root_dir: PathBuf, path: impl AsRef<Path>) -> PathBuf {

romeo/src/event.rs

@@ -12,16 +12,16 @@ use crate::state::WithdrawalInfo;
 #[derivative(Debug)]
 pub enum Event {
     /// A mint transaction has been created and broadcasted
-    MintCreated(DepositInfo, StacksTxId),
+    MintBroadcasted(DepositInfo, StacksTxId),
 
     /// A burn transaction has been created and broadcasted
-    BurnCreated(WithdrawalInfo, StacksTxId),
+    BurnBroadcasted(WithdrawalInfo, StacksTxId),
 
     /// A fulfill transaction has been created and broadcasted
-    FulfillCreated(WithdrawalInfo, BitcoinTxId),
+    FulfillBroadcasted(WithdrawalInfo, BitcoinTxId),
 
     /// The asset contract deploy transaction has been created and broadcasted
-    AssetContractCreated(StacksTxId),
+    AssetContractBroadcasted(StacksTxId),
 
     /// A stacks node has responded with an updated status regarding this txid
     StacksTransactionUpdate(StacksTxId, TransactionStatus),
@@ -36,7 +36,7 @@ pub enum Event {
 /// Status of a broadcasted transaction, useful for implementing retry logic
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
 pub enum TransactionStatus {
-    /// This transaction has been broadcasted to a node
+    /// Broadcasted to a node
     Broadcasted,
     /// This transaction has received `Config::number_of_required_confirmations` confirmations
     Confirmed,

romeo/src/stacks_client.rs

@@ -7,17 +7,17 @@ use blockstack_lib::codec::StacksMessageCodec;
 use blockstack_lib::core::CHAIN_ID_TESTNET;
 use rand::distributions::Alphanumeric;
 use rand::{thread_rng, Rng};
-use reqwest::{Request, StatusCode};
+use reqwest::Request;
 use serde_json::Value;
 use tokio::sync::{Mutex, MutexGuard};
-use tracing::trace;
 
 use serde::de::DeserializeOwned;
 
 use blockstack_lib::burnchains::Txid as StacksTxId;
 use blockstack_lib::chainstate::stacks::{
     StacksTransaction, StacksTransactionSigner, TransactionAnchorMode, TransactionPostConditionMode,
 };
+use tracing::debug;
 
 use crate::config::Config;
 use crate::event::TransactionStatus;
@@ -63,23 +63,32 @@ impl StacksClient {
         T: DeserializeOwned,
     {
         let request_url = req.url().to_string();
-        let res = self.http_client.execute(req).await?;
-
-        if res.status() == StatusCode::OK {
-            Ok(res.json::<T>().await?)
-        } else {
-            let details = res.json::<Value>().await?;
 
-            trace!(
-                "Request failure details: {:?}",
-                serde_json::to_string(&details)?
-            );
-
-            Err(anyhow!(format!(
-                "Request not 200: {}: {}",
-                request_url, details["error"]
-            )))
-        }
+        let res = self.http_client.execute(req).await?;
+        let status = res.status();
+        let body = res.text().await?;
+
+        serde_json::from_str(&body).map_err(|err| {
+            let error_details = serde_json::from_str::<Value>(&body)
+                .ok()
+                .and_then(|details| {
+                    details["error"]
+                        .as_str()
+                        .map(|description| format!(" {}", description))
+                });
+
+            if error_details.is_none() {
+                debug!("Failed request response body: {:?}", body);
+            }
+
+            anyhow!(
+                "Could not parse response JSON, URL is {}, status is {}: {:?}{}",
+                request_url,
+                status,
+                err,
+                error_details.unwrap_or_default()
+            )
+        })
     }
 
     /// Sign and broadcast an unsigned stacks transaction