From 0cc9e1cdea59220531f0180246e537ea04655bf4 Mon Sep 17 00:00:00 2001
From: Alekos Filini <alekos.filini@gmail.com>
Date: Thu, 7 May 2020 17:36:45 +0200
Subject: [PATCH] [blockchain] Use async I/O in the various blockchain impls

---
 Cargo.toml                 |  10 ++--
 examples/repl.rs           |  24 ++++++---
 src/blockchain/electrum.rs |  38 +++++++------
 src/blockchain/esplora.rs  | 108 +++++++++++++++++++++++--------------
 src/blockchain/mod.rs      |  15 +++---
 src/blockchain/utils.rs    |  40 ++++++++------
 src/lib.rs                 |   3 ++
 src/wallet/mod.rs          |  23 ++++----
 8 files changed, 163 insertions(+), 98 deletions(-)

diff --git a/Cargo.toml b/Cargo.toml
index 2ef3a529..7bb030e5 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,6 +1,7 @@
 [package]
 name = "magical-bitcoin-wallet"
 version = "0.1.0"
+edition = "2018"
 authors = ["Riccardo Casatta <riccardo@casatta.it>", "Alekos Filini <alekos.filini@gmail.com>"]
 
 [dependencies]
@@ -10,21 +11,24 @@ miniscript = { version = "0.12" }
 serde = { version = "^1.0", features = ["derive"] }
 serde_json = { version = "^1.0" }
 base64 = "^0.11"
+async-trait = "0.1"
 
 # Optional dependencies
 sled = { version = "0.31.0", optional = true }
-electrum-client = { version = "0.1.0-beta.5", optional = true }
-reqwest = { version = "0.10", optional = true, features = ["blocking", "json"] }
+electrum-client = { git = "https://github.com/MagicalBitcoin/rust-electrum-client.git", optional = true }
+reqwest = { version = "0.10", optional = true, features = ["json"] }
+futures = { version = "0.3", optional = true }
 
 [features]
 minimal = []
 compiler = ["miniscript/compiler"]
 default = ["key-value-db", "electrum"]
 electrum = ["electrum-client"]
-esplora = ["reqwest"]
+esplora = ["reqwest", "futures"]
 key-value-db = ["sled"]
 
 [dev-dependencies]
+tokio = { version = "0.2", features = ["macros"] }
 lazy_static = "1.4"
 rustyline = "5.0" # newer version requires 2018 edition
 clap = "2.33"
diff --git a/examples/repl.rs b/examples/repl.rs
index 13ee0951..74d366eb 100644
--- a/examples/repl.rs
+++ b/examples/repl.rs
@@ -9,6 +9,7 @@ extern crate rustyline;
 use std::fs;
 use std::path::PathBuf;
 use std::str::FromStr;
+use std::sync::Arc;
 
 use clap::{App, AppSettings, Arg, ArgMatches, SubCommand};
 
@@ -72,7 +73,8 @@ fn outpoint_validator(s: String) -> Result<(), String> {
     parse_outpoint(&s).map(|_| ())
 }
 
-fn main() {
+#[tokio::main]
+async fn main() {
     env_logger::init();
 
     let app = App::new("Magical Bitcoin Wallet")
@@ -264,7 +266,9 @@ fn main() {
         .unwrap();
     debug!("database opened successfully");
 
-    let client = Client::new(matches.value_of("server").unwrap()).unwrap();
+    let client = Client::new(matches.value_of("server").unwrap())
+        .await
+        .unwrap();
     let wallet = Wallet::new(
         descriptor,
         change_descriptor,
@@ -272,14 +276,20 @@ fn main() {
         tree,
         ElectrumBlockchain::from(client),
     )
+    .await
     .unwrap();
+    let wallet = Arc::new(wallet);
 
     // TODO: print errors in a nice way
-    let handle_matches = |matches: ArgMatches<'_>| {
+    async fn handle_matches<C, D>(wallet: Arc<Wallet<C, D>>, matches: ArgMatches<'_>)
+    where
+        C: magical_bitcoin_wallet::blockchain::OnlineBlockchain,
+        D: magical_bitcoin_wallet::database::BatchDatabase,
+    {
         if let Some(_sub_matches) = matches.subcommand_matches("get_new_address") {
             println!("{}", wallet.get_new_address().unwrap().to_string());
         } else if let Some(_sub_matches) = matches.subcommand_matches("sync") {
-            wallet.sync(None, None).unwrap();
+            wallet.sync(None, None).await.unwrap();
         } else if let Some(_sub_matches) = matches.subcommand_matches("list_unspent") {
             for utxo in wallet.list_unspent().unwrap() {
                 println!("{} value {} SAT", utxo.outpoint, utxo.txout.value);
@@ -344,7 +354,7 @@ fn main() {
         } else if let Some(sub_matches) = matches.subcommand_matches("broadcast") {
             let psbt = base64::decode(sub_matches.value_of("psbt").unwrap()).unwrap();
             let psbt: PartiallySignedTransaction = deserialize(&psbt).unwrap();
-            let (txid, _) = wallet.broadcast(psbt).unwrap();
+            let (txid, _) = wallet.broadcast(psbt).await.unwrap();
 
             println!("TXID: {}", txid);
         }
@@ -372,7 +382,7 @@ fn main() {
                         continue;
                     }
 
-                    handle_matches(matches.unwrap());
+                    handle_matches(Arc::clone(&wallet), matches.unwrap()).await;
                 }
                 Err(ReadlineError::Interrupted) => continue,
                 Err(ReadlineError::Eof) => break,
@@ -385,6 +395,6 @@ fn main() {
 
     // rl.save_history("history.txt").unwrap();
     } else {
-        handle_matches(matches);
+        handle_matches(wallet, matches).await;
     }
 }
diff --git a/src/blockchain/electrum.rs b/src/blockchain/electrum.rs
index df1ee2d5..fe40c78b 100644
--- a/src/blockchain/electrum.rs
+++ b/src/blockchain/electrum.rs
@@ -1,11 +1,11 @@
 use std::collections::HashSet;
-use std::io::{Read, Write};
 
 #[allow(unused_imports)]
 use log::{debug, error, info, trace};
 
 use bitcoin::{Script, Transaction, Txid};
 
+use electrum_client::tokio::io::{AsyncRead, AsyncWrite};
 use electrum_client::Client;
 
 use self::utils::{ELSGetHistoryRes, ELSListUnspentRes, ElectrumLikeSync};
@@ -13,15 +13,15 @@ use super::*;
 use crate::database::{BatchDatabase, DatabaseUtils};
 use crate::error::Error;
 
-pub struct ElectrumBlockchain<T: Read + Write>(Option<Client<T>>);
+pub struct ElectrumBlockchain<T: AsyncRead + AsyncWrite + Send>(Option<Client<T>>);
 
-impl<T: Read + Write> std::convert::From<Client<T>> for ElectrumBlockchain<T> {
+impl<T: AsyncRead + AsyncWrite + Send> std::convert::From<Client<T>> for ElectrumBlockchain<T> {
     fn from(client: Client<T>) -> Self {
         ElectrumBlockchain(Some(client))
     }
 }
 
-impl<T: Read + Write> Blockchain for ElectrumBlockchain<T> {
+impl<T: AsyncRead + AsyncWrite + Send> Blockchain for ElectrumBlockchain<T> {
     fn offline() -> Self {
         ElectrumBlockchain(None)
     }
@@ -31,14 +31,15 @@ impl<T: Read + Write> Blockchain for ElectrumBlockchain<T> {
     }
 }
 
-impl<T: Read + Write> OnlineBlockchain for ElectrumBlockchain<T> {
-    fn get_capabilities(&self) -> HashSet<Capability> {
+#[async_trait(?Send)]
+impl<T: AsyncRead + AsyncWrite + Send> OnlineBlockchain for ElectrumBlockchain<T> {
+    async fn get_capabilities(&self) -> HashSet<Capability> {
         vec![Capability::FullHistory, Capability::GetAnyTx]
             .into_iter()
             .collect()
     }
 
-    fn setup<D: BatchDatabase + DatabaseUtils, P: Progress>(
+    async fn setup<D: BatchDatabase + DatabaseUtils, P: Progress>(
         &mut self,
         stop_gap: Option<usize>,
         database: &mut D,
@@ -48,27 +49,30 @@ impl<T: Read + Write> OnlineBlockchain for ElectrumBlockchain<T> {
             .as_mut()
             .ok_or(Error::OfflineClient)?
             .electrum_like_setup(stop_gap, database, progress_update)
+            .await
     }
 
-    fn get_tx(&mut self, txid: &Txid) -> Result<Option<Transaction>, Error> {
+    async fn get_tx(&mut self, txid: &Txid) -> Result<Option<Transaction>, Error> {
         Ok(self
             .0
             .as_mut()
             .ok_or(Error::OfflineClient)?
             .transaction_get(txid)
+            .await
             .map(Option::Some)?)
     }
 
-    fn broadcast(&mut self, tx: &Transaction) -> Result<(), Error> {
+    async fn broadcast(&mut self, tx: &Transaction) -> Result<(), Error> {
         Ok(self
             .0
             .as_mut()
             .ok_or(Error::OfflineClient)?
             .transaction_broadcast(tx)
+            .await
             .map(|_| ())?)
     }
 
-    fn get_height(&mut self) -> Result<usize, Error> {
+    async fn get_height(&mut self) -> Result<usize, Error> {
         // TODO: unsubscribe when added to the client, or is there a better call to use here?
 
         Ok(self
@@ -76,16 +80,19 @@ impl<T: Read + Write> OnlineBlockchain for ElectrumBlockchain<T> {
             .as_mut()
             .ok_or(Error::OfflineClient)?
             .block_headers_subscribe()
+            .await
             .map(|data| data.height)?)
     }
 }
 
-impl<T: Read + Write> ElectrumLikeSync for Client<T> {
-    fn els_batch_script_get_history<'s, I: IntoIterator<Item = &'s Script>>(
+#[async_trait(?Send)]
+impl<T: AsyncRead + AsyncWrite + Send> ElectrumLikeSync for Client<T> {
+    async fn els_batch_script_get_history<'s, I: IntoIterator<Item = &'s Script>>(
         &mut self,
         scripts: I,
     ) -> Result<Vec<Vec<ELSGetHistoryRes>>, Error> {
         self.batch_script_get_history(scripts)
+            .await
             .map(|v| {
                 v.into_iter()
                     .map(|v| {
@@ -105,11 +112,12 @@ impl<T: Read + Write> ElectrumLikeSync for Client<T> {
             .map_err(Error::Electrum)
     }
 
-    fn els_batch_script_list_unspent<'s, I: IntoIterator<Item = &'s Script>>(
+    async fn els_batch_script_list_unspent<'s, I: IntoIterator<Item = &'s Script>>(
         &mut self,
         scripts: I,
     ) -> Result<Vec<Vec<ELSListUnspentRes>>, Error> {
         self.batch_script_list_unspent(scripts)
+            .await
             .map(|v| {
                 v.into_iter()
                     .map(|v| {
@@ -133,7 +141,7 @@ impl<T: Read + Write> ElectrumLikeSync for Client<T> {
             .map_err(Error::Electrum)
     }
 
-    fn els_transaction_get(&mut self, txid: &Txid) -> Result<Transaction, Error> {
-        self.transaction_get(txid).map_err(Error::Electrum)
+    async fn els_transaction_get(&mut self, txid: &Txid) -> Result<Transaction, Error> {
+        self.transaction_get(txid).await.map_err(Error::Electrum)
     }
 }
diff --git a/src/blockchain/esplora.rs b/src/blockchain/esplora.rs
index b6574b9f..0e254324 100644
--- a/src/blockchain/esplora.rs
+++ b/src/blockchain/esplora.rs
@@ -1,11 +1,13 @@
 use std::collections::HashSet;
 
+use futures::stream::{self, StreamExt, TryStreamExt};
+
 #[allow(unused_imports)]
 use log::{debug, error, info, trace};
 
 use serde::Deserialize;
 
-use reqwest::blocking::Client;
+use reqwest::Client;
 use reqwest::StatusCode;
 
 use bitcoin::consensus::{deserialize, serialize};
@@ -52,14 +54,15 @@ impl Blockchain for EsploraBlockchain {
     }
 }
 
+#[async_trait(?Send)]
 impl OnlineBlockchain for EsploraBlockchain {
-    fn get_capabilities(&self) -> HashSet<Capability> {
+    async fn get_capabilities(&self) -> HashSet<Capability> {
         vec![Capability::FullHistory, Capability::GetAnyTx]
             .into_iter()
             .collect()
     }
 
-    fn setup<D: BatchDatabase + DatabaseUtils, P: Progress>(
+    async fn setup<D: BatchDatabase + DatabaseUtils, P: Progress>(
         &mut self,
         stop_gap: Option<usize>,
         database: &mut D,
@@ -69,22 +72,34 @@ impl OnlineBlockchain for EsploraBlockchain {
             .as_mut()
             .ok_or(Error::OfflineClient)?
             .electrum_like_setup(stop_gap, database, progress_update)
+            .await
     }
 
-    fn get_tx(&mut self, txid: &Txid) -> Result<Option<Transaction>, Error> {
-        Ok(self.0.as_mut().ok_or(Error::OfflineClient)?._get_tx(txid)?)
-    }
-
-    fn broadcast(&mut self, tx: &Transaction) -> Result<(), Error> {
+    async fn get_tx(&mut self, txid: &Txid) -> Result<Option<Transaction>, Error> {
         Ok(self
             .0
             .as_mut()
             .ok_or(Error::OfflineClient)?
-            ._broadcast(tx)?)
+            ._get_tx(txid)
+            .await?)
     }
 
-    fn get_height(&mut self) -> Result<usize, Error> {
-        Ok(self.0.as_mut().ok_or(Error::OfflineClient)?._get_height()?)
+    async fn broadcast(&mut self, tx: &Transaction) -> Result<(), Error> {
+        Ok(self
+            .0
+            .as_mut()
+            .ok_or(Error::OfflineClient)?
+            ._broadcast(tx)
+            .await?)
+    }
+
+    async fn get_height(&mut self) -> Result<usize, Error> {
+        Ok(self
+            .0
+            .as_mut()
+            .ok_or(Error::OfflineClient)?
+            ._get_height()
+            .await?)
     }
 }
 
@@ -93,40 +108,47 @@ impl UrlClient {
         sha256::Hash::hash(script.as_bytes()).into_inner().to_hex()
     }
 
-    fn _get_tx(&self, txid: &Txid) -> Result<Option<Transaction>, EsploraError> {
+    async fn _get_tx(&self, txid: &Txid) -> Result<Option<Transaction>, EsploraError> {
         let resp = self
             .client
             .get(&format!("{}/api/tx/{}/raw", self.url, txid))
-            .send()?;
+            .send()
+            .await?;
 
         if let StatusCode::NOT_FOUND = resp.status() {
             return Ok(None);
         }
 
-        Ok(Some(deserialize(&resp.error_for_status()?.bytes()?)?))
+        Ok(Some(deserialize(&resp.error_for_status()?.bytes().await?)?))
     }
 
-    fn _broadcast(&self, transaction: &Transaction) -> Result<(), EsploraError> {
+    async fn _broadcast(&self, transaction: &Transaction) -> Result<(), EsploraError> {
         self.client
             .post(&format!("{}/api/tx", self.url))
             .body(serialize(transaction).to_hex())
-            .send()?
+            .send()
+            .await?
             .error_for_status()?;
 
         Ok(())
     }
 
-    fn _get_height(&self) -> Result<usize, EsploraError> {
+    async fn _get_height(&self) -> Result<usize, EsploraError> {
         Ok(self
             .client
             .get(&format!("{}/api/blocks/tip/height", self.url))
-            .send()?
+            .send()
+            .await?
             .error_for_status()?
-            .text()?
+            .text()
+            .await?
             .parse()?)
     }
 
-    fn _script_get_history(&self, script: &Script) -> Result<Vec<ELSGetHistoryRes>, EsploraError> {
+    async fn _script_get_history(
+        &self,
+        script: &Script,
+    ) -> Result<Vec<ELSGetHistoryRes>, EsploraError> {
         let mut result = Vec::new();
         let scripthash = Self::script_to_scripthash(script);
 
@@ -137,9 +159,11 @@ impl UrlClient {
                     "{}/api/scripthash/{}/txs/mempool",
                     self.url, scripthash
                 ))
-                .send()?
+                .send()
+                .await?
                 .error_for_status()?
-                .json::<Vec<EsploraGetHistory>>()?
+                .json::<Vec<EsploraGetHistory>>()
+                .await?
                 .into_iter()
                 .map(|x| ELSGetHistoryRes {
                     tx_hash: x.txid,
@@ -163,9 +187,11 @@ impl UrlClient {
                     "{}/api/scripthash/{}/txs/chain/{}",
                     self.url, scripthash, last_txid
                 ))
-                .send()?
+                .send()
+                .await?
                 .error_for_status()?
-                .json::<Vec<EsploraGetHistory>>()?;
+                .json::<Vec<EsploraGetHistory>>()
+                .await?;
             let len = response.len();
             if let Some(elem) = response.last() {
                 last_txid = elem.txid.to_hex();
@@ -186,7 +212,7 @@ impl UrlClient {
         Ok(result)
     }
 
-    fn _script_list_unspent(
+    async fn _script_list_unspent(
         &self,
         script: &Script,
     ) -> Result<Vec<ELSListUnspentRes>, EsploraError> {
@@ -197,9 +223,11 @@ impl UrlClient {
                 self.url,
                 Self::script_to_scripthash(script)
             ))
-            .send()?
+            .send()
+            .await?
             .error_for_status()?
-            .json::<Vec<EsploraListUnspent>>()?
+            .json::<Vec<EsploraListUnspent>>()
+            .await?
             .into_iter()
             .map(|x| ELSListUnspentRes {
                 tx_hash: x.txid,
@@ -210,30 +238,32 @@ impl UrlClient {
     }
 }
 
+#[async_trait(?Send)]
 impl ElectrumLikeSync for UrlClient {
-    fn els_batch_script_get_history<'s, I: IntoIterator<Item = &'s Script>>(
+    async fn els_batch_script_get_history<'s, I: IntoIterator<Item = &'s Script>>(
         &mut self,
         scripts: I,
     ) -> Result<Vec<Vec<ELSGetHistoryRes>>, Error> {
-        Ok(scripts
-            .into_iter()
-            .map(|script| self._script_get_history(script))
-            .collect::<Result<Vec<_>, _>>()?)
+        Ok(stream::iter(scripts)
+            .then(|script| self._script_get_history(&script))
+            .try_collect()
+            .await?)
     }
 
-    fn els_batch_script_list_unspent<'s, I: IntoIterator<Item = &'s Script>>(
+    async fn els_batch_script_list_unspent<'s, I: IntoIterator<Item = &'s Script>>(
         &mut self,
         scripts: I,
     ) -> Result<Vec<Vec<ELSListUnspentRes>>, Error> {
-        Ok(scripts
-            .into_iter()
-            .map(|script| self._script_list_unspent(script))
-            .collect::<Result<Vec<_>, _>>()?)
+        Ok(stream::iter(scripts)
+            .then(|script| self._script_list_unspent(&script))
+            .try_collect()
+            .await?)
     }
 
-    fn els_transaction_get(&mut self, txid: &Txid) -> Result<Transaction, Error> {
+    async fn els_transaction_get(&mut self, txid: &Txid) -> Result<Transaction, Error> {
         Ok(self
-            ._get_tx(txid)?
+            ._get_tx(txid)
+            .await?
             .ok_or_else(|| EsploraError::TransactionNotFound(*txid))?)
     }
 }
diff --git a/src/blockchain/mod.rs b/src/blockchain/mod.rs
index 3e13d5f9..3418d967 100644
--- a/src/blockchain/mod.rs
+++ b/src/blockchain/mod.rs
@@ -41,28 +41,29 @@ impl Blockchain for OfflineBlockchain {
     }
 }
 
+#[async_trait(?Send)]
 pub trait OnlineBlockchain: Blockchain {
-    fn get_capabilities(&self) -> HashSet<Capability>;
+    async fn get_capabilities(&self) -> HashSet<Capability>;
 
-    fn setup<D: BatchDatabase + DatabaseUtils, P: Progress>(
+    async fn setup<D: BatchDatabase + DatabaseUtils, P: Progress>(
         &mut self,
         stop_gap: Option<usize>,
         database: &mut D,
         progress_update: P,
     ) -> Result<(), Error>;
-    fn sync<D: BatchDatabase + DatabaseUtils, P: Progress>(
+    async fn sync<D: BatchDatabase + DatabaseUtils, P: Progress>(
         &mut self,
         stop_gap: Option<usize>,
         database: &mut D,
         progress_update: P,
     ) -> Result<(), Error> {
-        self.setup(stop_gap, database, progress_update)
+        self.setup(stop_gap, database, progress_update).await
     }
 
-    fn get_tx(&mut self, txid: &Txid) -> Result<Option<Transaction>, Error>;
-    fn broadcast(&mut self, tx: &Transaction) -> Result<(), Error>;
+    async fn get_tx(&mut self, txid: &Txid) -> Result<Option<Transaction>, Error>;
+    async fn broadcast(&mut self, tx: &Transaction) -> Result<(), Error>;
 
-    fn get_height(&mut self) -> Result<usize, Error>;
+    async fn get_height(&mut self) -> Result<usize, Error>;
 }
 
 pub type ProgressData = (f32, Option<String>);
diff --git a/src/blockchain/utils.rs b/src/blockchain/utils.rs
index 9231f0ce..326bd1d9 100644
--- a/src/blockchain/utils.rs
+++ b/src/blockchain/utils.rs
@@ -27,22 +27,23 @@ pub struct ELSListUnspentRes {
 }
 
 /// Implements the synchronization logic for an Electrum-like client.
+#[async_trait(?Send)]
 pub trait ElectrumLikeSync {
-    fn els_batch_script_get_history<'s, I: IntoIterator<Item = &'s Script>>(
+    async fn els_batch_script_get_history<'s, I: IntoIterator<Item = &'s Script>>(
         &mut self,
         scripts: I,
     ) -> Result<Vec<Vec<ELSGetHistoryRes>>, Error>;
 
-    fn els_batch_script_list_unspent<'s, I: IntoIterator<Item = &'s Script>>(
+    async fn els_batch_script_list_unspent<'s, I: IntoIterator<Item = &'s Script>>(
         &mut self,
         scripts: I,
     ) -> Result<Vec<Vec<ELSListUnspentRes>>, Error>;
 
-    fn els_transaction_get(&mut self, txid: &Txid) -> Result<Transaction, Error>;
+    async fn els_transaction_get(&mut self, txid: &Txid) -> Result<Transaction, Error>;
 
     // Provided methods down here...
 
-    fn electrum_like_setup<D: BatchDatabase + DatabaseUtils, P: Progress>(
+    async fn electrum_like_setup<D: BatchDatabase + DatabaseUtils, P: Progress>(
         &mut self,
         stop_gap: Option<usize>,
         database: &mut D,
@@ -85,7 +86,7 @@ pub trait ElectrumLikeSync {
 
             let until = cmp::min(to_check_later.len(), batch_query_size);
             let chunk: Vec<Script> = to_check_later.drain(..until).collect();
-            let call_result = self.els_batch_script_get_history(chunk.iter())?;
+            let call_result = self.els_batch_script_get_history(chunk.iter()).await?;
 
             for (script, history) in chunk.into_iter().zip(call_result.into_iter()) {
                 trace!("received history for {:?}, size {}", script, history.len());
@@ -94,7 +95,8 @@ pub trait ElectrumLikeSync {
                     last_found = index;
 
                     let mut check_later_scripts = self
-                        .check_history(database, script, history, &mut change_max_deriv)?
+                        .check_history(database, script, history, &mut change_max_deriv)
+                        .await?
                         .into_iter()
                         .filter(|x| already_checked.insert(x.clone()))
                         .collect();
@@ -124,7 +126,7 @@ pub trait ElectrumLikeSync {
         let mut batch = database.begin_batch();
         for chunk in ChunksIterator::new(database.iter_utxos()?.into_iter(), batch_query_size) {
             let scripts: Vec<_> = chunk.iter().map(|u| &u.txout.script_pubkey).collect();
-            let call_result = self.els_batch_script_list_unspent(scripts)?;
+            let call_result = self.els_batch_script_list_unspent(scripts).await?;
 
             // check which utxos are actually still unspent
             for (utxo, list_unspent) in chunk.into_iter().zip(call_result.iter()) {
@@ -167,7 +169,7 @@ pub trait ElectrumLikeSync {
         Ok(())
     }
 
-    fn check_tx_and_descendant<D: DatabaseUtils + BatchDatabase>(
+    async fn check_tx_and_descendant<D: DatabaseUtils + BatchDatabase>(
         &mut self,
         database: &mut D,
         txid: &Txid,
@@ -199,7 +201,7 @@ pub trait ElectrumLikeSync {
                 // went wrong
                 saved_tx.transaction.unwrap()
             }
-            None => self.els_transaction_get(&txid)?,
+            None => self.els_transaction_get(&txid).await?,
         };
 
         let mut incoming: u64 = 0;
@@ -264,7 +266,7 @@ pub trait ElectrumLikeSync {
         Ok(to_check_later)
     }
 
-    fn check_history<D: DatabaseUtils + BatchDatabase>(
+    async fn check_history<D: DatabaseUtils + BatchDatabase>(
         &mut self,
         database: &mut D,
         script_pubkey: Script,
@@ -286,13 +288,17 @@ pub trait ElectrumLikeSync {
                 x => u32::try_from(x).ok(),
             };
 
-            to_check_later.extend_from_slice(&self.check_tx_and_descendant(
-                database,
-                &tx.tx_hash,
-                height,
-                &script_pubkey,
-                change_max_deriv,
-            )?);
+            to_check_later.extend_from_slice(
+                &self
+                    .check_tx_and_descendant(
+                        database,
+                        &tx.tx_hash,
+                        height,
+                        &script_pubkey,
+                        change_max_deriv,
+                    )
+                    .await?,
+            );
         }
 
         Ok(to_check_later)
diff --git a/src/lib.rs b/src/lib.rs
index 7273ed50..2b7ceece 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -9,6 +9,9 @@ extern crate serde_json;
 #[macro_use]
 extern crate lazy_static;
 
+#[macro_use]
+extern crate async_trait;
+
 #[cfg(feature = "electrum")]
 pub extern crate electrum_client;
 #[cfg(feature = "electrum")]
diff --git a/src/wallet/mod.rs b/src/wallet/mod.rs
index b0332c13..f1501786 100644
--- a/src/wallet/mod.rs
+++ b/src/wallet/mod.rs
@@ -695,7 +695,7 @@ where
     B: OnlineBlockchain,
     D: BatchDatabase,
 {
-    pub fn new(
+    pub async fn new(
         descriptor: &str,
         change_descriptor: Option<&str>,
         network: Network,
@@ -724,7 +724,7 @@ where
             None => None,
         };
 
-        let current_height = Some(client.get_height()? as u32);
+        let current_height = Some(client.get_height().await? as u32);
 
         Ok(Wallet {
             descriptor,
@@ -738,7 +738,7 @@ where
         })
     }
 
-    pub fn sync(
+    pub async fn sync(
         &self,
         max_address: Option<u32>,
         _batch_query_size: Option<usize>,
@@ -798,16 +798,19 @@ where
             self.database.borrow_mut().commit_batch(address_batch)?;
         }
 
-        self.client.borrow_mut().sync(
-            None,
-            self.database.borrow_mut().deref_mut(),
-            noop_progress(),
-        )
+        self.client
+            .borrow_mut()
+            .sync(
+                None,
+                self.database.borrow_mut().deref_mut(),
+                noop_progress(),
+            )
+            .await
     }
 
-    pub fn broadcast(&self, psbt: PSBT) -> Result<(Txid, Transaction), Error> {
+    pub async fn broadcast(&self, psbt: PSBT) -> Result<(Txid, Transaction), Error> {
         let extracted = psbt.extract_tx();
-        self.client.borrow_mut().broadcast(&extracted)?;
+        self.client.borrow_mut().broadcast(&extracted).await?;
 
         Ok((extracted.txid(), extracted))
     }