frost/bdk
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[blockchain] Use async I/O in the various blockchain impls

Browse Source
This commit is contained in:
Alekos Filini
2020-05-07 17:36:45 +02:00
parent 95b2cd4c32
commit 0cc9e1cdea
8 changed files with 163 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
src/blockchain/electrum.rs
View File

@@ -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)
}
}

108
src/blockchain/esplora.rs
View File

@@ -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))?)
}
}

15
src/blockchain/mod.rs
View File

@@ -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>);

40
src/blockchain/utils.rs
View File

@@ -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)