feat(client): add async and blocking clients to submit txs package

Author: acidbunny21

Cargo.toml

@@ -18,6 +18,7 @@ path = "src/lib.rs"
 
 [dependencies]
 serde = { version = "1.0", features = ["derive"] }
+serde_json = { version = "1.0", default-features = false }
 bitcoin = { version = "0.32", features = ["serde", "std"], default-features = false }
 hex = { version = "0.2", package = "hex-conservative" }
 log = "^0.4"
@@ -28,7 +29,6 @@ reqwest = { version = "0.11",  features = ["json"], default-features = false, op
 tokio = { version = "1", features = ["time"], optional = true }
 
 [dev-dependencies]
-serde_json = "1.0"
 tokio = { version = "1.20.1", features = ["full"] }
 electrsd = { version = "0.28.0", features = ["legacy", "esplora_a33e97e1", "bitcoind_25_0"] }
 lazy_static = "1.4.0"

src/api.rs

@@ -2,13 +2,14 @@
 //!
 //! See: <https://github.com/Blockstream/esplora/blob/master/API.md>
 
+use std::collections::HashMap;
+
 pub use bitcoin::consensus::{deserialize, serialize};
 pub use bitcoin::hex::FromHex;
-use bitcoin::Weight;
 pub use bitcoin::{
     transaction, Amount, BlockHash, OutPoint, ScriptBuf, Transaction, TxIn, TxOut, Txid, Witness,
 };
-
+use bitcoin::{FeeRate, Weight, Wtxid};
 use serde::Deserialize;
 
 #[derive(Deserialize, Clone, Debug, PartialEq, Eq)]
@@ -123,6 +124,54 @@ pub struct AddressTxsSummary {
     pub tx_count: u32,
 }
 
+#[derive(Deserialize, Debug)]
+pub struct SubmitPackageResult {
+    /// The transaction package result message. "success" indicates all transactions were accepted
+    /// into or are already in the mempool.
+    pub package_msg: String,
+    /// Transaction results keyed by [`Wtxid`].
+    #[serde(rename = "tx-results")]
+    pub tx_results: HashMap<Wtxid, TxResult>,
+    /// List of txids of replaced transactions.
+    #[serde(rename = "replaced-transactions")]
+    pub replaced_transactions: Option<Vec<Txid>>,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct TxResult {
+    /// The transaction id.
+    pub txid: Txid,
+    /// The [`Wtxid`] of a different transaction with the same [`Txid`] but different witness found
+    /// in the mempool.
+    ///
+    /// If set, this means the submitted transaction was ignored.
+    #[serde(rename = "other-wtxid")]
+    pub other_wtxid: Option<Wtxid>,
+    /// Sigops-adjusted virtual transaction size.
+    pub vsize: Option<u32>,
+    /// Transaction fees.
+    pub fees: Option<MempoolFeesSubmitPackage>,
+    /// The transaction error string, if it was rejected by the mempool
+    pub error: Option<String>,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct MempoolFeesSubmitPackage {
+    /// Transaction fee.
+    #[serde(with = "bitcoin::amount::serde::as_btc")]
+    pub base: Amount,
+    /// The effective feerate.
+    ///
+    /// Will be `None` if the transaction was already in the mempool. For example, the package
+    /// feerate and/or feerate with modified fees from the `prioritisetransaction` JSON-RPC method.
+    #[serde(rename = "effective-feerate", default, deserialize_with = "deserialize_feerate")]
+    pub effective_feerate: Option<FeeRate>,
+    /// If [`Self::effective_fee_rate`] is provided, this holds the [`Wtxid`]s of the transactions
+    /// whose fees and vsizes are included in effective-feerate.
+    #[serde(rename = "effective-includes")]
+    pub effective_includes: Option<Vec<Wtxid>>,
+}
+
 impl Tx {
     pub fn to_tx(&self) -> Transaction {
         Transaction {
@@ -198,3 +247,20 @@ where
         .collect::<Result<Vec<Vec<u8>>, _>>()
         .map_err(serde::de::Error::custom)
 }
+
+fn deserialize_feerate<'de, D>(d: D) -> Result<Option<FeeRate>, D::Error>
+where
+    D: serde::de::Deserializer<'de>,
+{
+       use serde::de::Error;
+
+       let btc_per_kvb = match Option::<f64>::deserialize(d)? {
+               Some(v) => v,
+               None => return Ok(None),
+       };
+       let sat_per_kwu = btc_per_kvb * 25_000_000.0;
+       if sat_per_kwu.is_infinite() {
+               return Err(D::Error::custom("feerate overflow"));
+       }
+       Ok(Some(FeeRate::from_sat_per_kwu(sat_per_kwu as u64)))
+}

src/async.rs

@@ -11,11 +11,12 @@
 
 //! Esplora by way of `reqwest` HTTP client.
 
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::marker::PhantomData;
 use std::str::FromStr;
 
-use bitcoin::consensus::{deserialize, serialize, Decodable, Encodable};
+use bitcoin::consensus::encode::serialize_hex;
+use bitcoin::consensus::{deserialize, serialize, Decodable};
 use bitcoin::hashes::{sha256, Hash};
 use bitcoin::hex::{DisplayHex, FromHex};
 use bitcoin::Address;
@@ -26,12 +27,12 @@ use bitcoin::{
 #[allow(unused_imports)]
 use log::{debug, error, info, trace};
 
-use reqwest::{header, Client, Response};
+use reqwest::{header, Body, Client, Response};
 
 use crate::api::AddressStats;
 use crate::{
-    BlockStatus, BlockSummary, Builder, Error, MerkleProof, OutputStatus, Tx, TxStatus,
-    BASE_BACKOFF_MILLIS, RETRYABLE_ERROR_CODES,
+    BlockStatus, BlockSummary, Builder, Error, MerkleProof, OutputStatus, SubmitPackageResult, Tx,
+    TxStatus, BASE_BACKOFF_MILLIS, RETRYABLE_ERROR_CODES,
 };
 
 #[derive(Debug, Clone)]
@@ -249,21 +250,27 @@ impl<S: Sleeper> AsyncClient<S> {
         }
     }
 
-    /// Make an HTTP POST request to given URL, serializing from any `T` that
-    /// implement [`bitcoin::consensus::Encodable`].
-    ///
-    /// It should be used when requesting Esplora endpoints that expected a
-    /// native bitcoin type serialized with [`bitcoin::consensus::Encodable`].
+    /// Make an HTTP POST request to given URL, converting any `T` that
+    /// implement [`Into<Body>`] and setting query parameters, if any.
     ///
     /// # Errors
     ///
     /// This function will return an error either from the HTTP client, or the
-    /// [`bitcoin::consensus::Encodable`] serialization.
-    async fn post_request_hex<T: Encodable>(&self, path: &str, body: T) -> Result<(), Error> {
-        let url = format!("{}{}", self.url, path);
-        let body = serialize::<T>(&body).to_lower_hex_string();
+    /// response's [`serde_json`] deserialization.
+    async fn post_request_bytes<T: Into<Body>>(
+        &self,
+        path: &str,
+        body: T,
+        query_params: Option<HashSet<(&str, String)>>,
+    ) -> Result<Response, Error> {
+        let url: String = format!("{}{}", self.url, path);
+        let mut request = self.client.post(url).body(body);
+
+        for param in query_params.unwrap_or_default() {
+            request = request.query(&param);
+        }
 
-        let response = self.client.post(url).body(body).send().await?;
+        let response = request.send().await?;
 
         if !response.status().is_success() {
             return Err(Error::HttpResponse {
@@ -272,7 +279,7 @@ impl<S: Sleeper> AsyncClient<S> {
             });
         }
 
-        Ok(())
+        Ok(response)
     }
 
     /// Get a [`Transaction`] option given its [`Txid`]
@@ -359,8 +366,49 @@ impl<S: Sleeper> AsyncClient<S> {
     }
 
     /// Broadcast a [`Transaction`] to Esplora
-    pub async fn broadcast(&self, transaction: &Transaction) -> Result<(), Error> {
-        self.post_request_hex("/tx", transaction).await
+    pub async fn broadcast(&self, transaction: &Transaction) -> Result<Txid, Error> {
+        let body = serialize::<Transaction>(transaction).to_lower_hex_string();
+        let response = self.post_request_bytes("/tx", body, None).await?;
+        let txid = Txid::from_str(&response.text().await?).map_err(|_| Error::InvalidResponse)?;
+        Ok(txid)
+    }
+
+    /// Broadcast a package of [`Transaction`] to Esplora
+    ///
+    /// if `maxfeerate` is provided, any transaction whose
+    /// fee is higher will be rejected
+    ///
+    /// if  `maxburnamount` is provided, any transaction
+    /// with higher provably unspendable outputs amount
+    /// will be rejected
+    pub async fn submit_package(
+        &self,
+        transactions: &[Transaction],
+        maxfeerate: Option<f64>,
+        maxburnamount: Option<f64>,
+    ) -> Result<SubmitPackageResult, Error> {
+        let mut queryparams = HashSet::<(&str, String)>::new();
+        if let Some(maxfeerate) = maxfeerate {
+            queryparams.insert(("maxfeerate", maxfeerate.to_string()));
+        }
+        if let Some(maxburnamount) = maxburnamount {
+            queryparams.insert(("maxburnamount", maxburnamount.to_string()));
+        }
+
+        let serialized_txs = transactions
+            .iter()
+            .map(|tx| serialize_hex(&tx))
+            .collect::<Vec<_>>();
+
+        let response = self
+            .post_request_bytes(
+                "/txs/package",
+                serde_json::to_string(&serialized_txs).unwrap(),
+                Some(queryparams),
+            )
+            .await?;
+
+        Ok(response.json::<SubmitPackageResult>().await?)
     }
 
     /// Get the current height of the blockchain tip

src/blocking.rs

@@ -16,6 +16,7 @@ use std::convert::TryFrom;
 use std::str::FromStr;
 use std::thread;
 
+use bitcoin::consensus::encode::serialize_hex;
 #[allow(unused_imports)]
 use log::{debug, error, info, trace};
 
@@ -31,8 +32,8 @@ use bitcoin::{
 
 use crate::api::AddressStats;
 use crate::{
-    BlockStatus, BlockSummary, Builder, Error, MerkleProof, OutputStatus, Tx, TxStatus,
-    BASE_BACKOFF_MILLIS, RETRYABLE_ERROR_CODES,
+    BlockStatus, BlockSummary, Builder, Error, MerkleProof, OutputStatus, SubmitPackageResult, Tx,
+    TxStatus, BASE_BACKOFF_MILLIS, RETRYABLE_ERROR_CODES,
 };
 
 #[derive(Debug, Clone)]
@@ -88,6 +89,24 @@ impl BlockingClient {
         Ok(request)
     }
 
+    fn post_request<T>(&self, path: &str, body: T) -> Result<Request, Error>
+    where
+        T: Into<Vec<u8>>,
+    {
+        let mut request = minreq::post(format!("{}/{}", self.url, path)).with_body(body);
+
+        if let Some(proxy) = &self.proxy {
+            let proxy = Proxy::new(proxy.as_str())?;
+            request = request.with_proxy(proxy);
+        }
+
+        if let Some(timeout) = &self.timeout {
+            request = request.with_timeout(*timeout);
+        }
+
+        Ok(request)
+    }
+
     fn get_opt_response<T: Decodable>(&self, path: &str) -> Result<Option<T>, Error> {
         match self.get_with_retry(path) {
             Ok(resp) if is_status_not_found(resp.status_code) => Ok(None),
@@ -268,20 +287,58 @@ impl BlockingClient {
 
     /// Broadcast a [`Transaction`] to Esplora
     pub fn broadcast(&self, transaction: &Transaction) -> Result<(), Error> {
-        let mut request = minreq::post(format!("{}/tx", self.url)).with_body(
+        let request = self.post_request(
+            "tx",
             serialize(transaction)
                 .to_lower_hex_string()
                 .as_bytes()
                 .to_vec(),
-        );
+        )?;
 
-        if let Some(proxy) = &self.proxy {
-            let proxy = Proxy::new(proxy.as_str())?;
-            request = request.with_proxy(proxy);
+        match request.send() {
+            Ok(resp) if !is_status_ok(resp.status_code) => {
+                let status = u16::try_from(resp.status_code).map_err(Error::StatusCode)?;
+                let message = resp.as_str().unwrap_or_default().to_string();
+                Err(Error::HttpResponse { status, message })
+            }
+            Ok(_resp) => Ok(()),
+            Err(e) => Err(Error::Minreq(e)),
         }
+    }
 
-        if let Some(timeout) = &self.timeout {
-            request = request.with_timeout(*timeout);
+    /// Broadcast a package of [`Transaction`] to Esplora
+    ///
+    /// if `maxfeerate` is provided, any transaction whose
+    /// fee is higher will be rejected
+    ///
+    /// if  `maxburnamount` is provided, any transaction
+    /// with higher provably unspendable outputs amount
+    /// will be rejected
+    pub fn submit_package(
+        &self,
+        transactions: &[Transaction],
+        maxfeerate: Option<f64>,
+        maxburnamount: Option<f64>,
+    ) -> Result<SubmitPackageResult, Error> {
+        let serialized_txs = transactions
+            .iter()
+            .map(|tx| serialize_hex(&tx))
+            .collect::<Vec<_>>();
+
+        let mut request = self.post_request(
+            "txs/package",
+            serde_json::to_string(&serialized_txs)
+                .unwrap()
+                .as_bytes()
+                .to_vec(),
+        )?;
+
+        if let Some(maxfeerate) = maxfeerate {
+            request = request.with_param("maxfeerate", maxfeerate.to_string())
+        }
+
+        if let Some(maxburnamount) = maxburnamount {
+            request = request.with_param("maxburnamount", maxburnamount.to_string())
         }
 
         match request.send() {
@@ -290,7 +347,7 @@ impl BlockingClient {
                 let message = resp.as_str().unwrap_or_default().to_string();
                 Err(Error::HttpResponse { status, message })
             }
-            Ok(_resp) => Ok(()),
+            Ok(resp) => Ok(resp.json::<SubmitPackageResult>().map_err(Error::Minreq)?),
             Err(e) => Err(Error::Minreq(e)),
         }
     }