breez_sdk_spark/chain/

rest_client.rs

use bitcoin::{Address, address::NetworkUnchecked};
use platform_utils::tokio;
use platform_utils::{
    ContentType, HttpClient, HttpError, HttpResponse, REQUEST_TIMEOUT, add_basic_auth_header,
    add_content_type_header,
};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use tracing::{debug, info};

use crate::chain::RecommendedFees;
use crate::{
    Network,
    chain::{ChainServiceError, Outspend, Utxo},
};

use super::BitcoinChainService;

pub const RETRYABLE_ERROR_CODES: [u16; 3] = [
    429, // TOO_MANY_REQUESTS
    500, // INTERNAL_SERVER_ERROR
    503, // SERVICE_UNAVAILABLE
];

/// Base backoff in milliseconds.
const BASE_BACKOFF_MILLIS: Duration = Duration::from_millis(256);

/// Ceiling on a whole retrying call, so retries cost attempts rather than
/// multiplying how long a caller can be blocked for.
const TOTAL_TIMEOUT: Duration = Duration::from_secs(REQUEST_TIMEOUT);

#[derive(Serialize, Deserialize, Clone)]
struct TxInfo {
    txid: String,
    status: super::TxStatus,
}

/// Minimal esplora `/address/:address/txs` entry: the txid, its confirmation
/// status, and the outputs (scanned to find the ones paying the queried address).
#[derive(Deserialize)]
struct AddressTx {
    txid: String,
    status: super::TxStatus,
    vout: Vec<AddressTxVout>,
}

#[derive(Deserialize)]
struct AddressTxVout {
    #[serde(default)]
    scriptpubkey_address: Option<String>,
    value: u64,
}

pub struct BasicAuth {
    username: String,
    password: String,
}

impl BasicAuth {
    pub fn new(username: String, password: String) -> Self {
        Self { username, password }
    }
}

struct RestClientChainServiceInner {
    base_url: String,
    network: Network,
    client: Arc<dyn HttpClient>,
    max_retries: usize,
    basic_auth: Option<BasicAuth>,
    api_type: ChainApiType,
    total_timeout: Duration,
}

/// REST-backed [`BitcoinChainService`].
///
/// The trait is exported through `UniFFI` with `with_foreign`, which makes
/// `UniFFI` re-wrap every `Arc<dyn BitcoinChainService>` that round-trips
/// across the FFI boundary in a foreign-callback proxy — even when both
/// sides are Rust in the same process. That proxy routes calls back into
/// Rust via `UniFFI`'s `RustFuture`, which is polled outside the surrounding
/// tokio runtime context, so `reqwest`'s `tokio::time::sleep` panics with
/// "no reactor running".
///
/// To stay correct under round-tripping (e.g. shared-chain-service in a
/// server-side harness that builds the service in Rust, hands it to a
/// foreign-language integration, and passes it back into multiple SDK
/// instances), we capture a [`tokio::runtime::Handle`] at construction
/// time and dispatch each trait-method body onto it via
/// [`tokio::runtime::Handle::spawn`]. The outer future we return to
/// `UniFFI` is just a `JoinHandle` await — a channel-wakeup poll that
/// needs no tokio context.
pub struct RestClientChainService {
    inner: Arc<RestClientChainServiceInner>,
    #[cfg(not(all(target_family = "wasm", target_os = "unknown")))]
    runtime_handle: tokio::runtime::Handle,
}

#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))]
#[derive(Clone, Copy, Debug)]
pub enum ChainApiType {
    Esplora,
    MempoolSpace,
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct MempoolSpaceRecommendedFeesResponse {
    fastest_fee: f64,
    half_hour_fee: f64,
    hour_fee: f64,
    economy_fee: f64,
    minimum_fee: f64,
}

impl From<MempoolSpaceRecommendedFeesResponse> for RecommendedFees {
    #[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
    fn from(response: MempoolSpaceRecommendedFeesResponse) -> Self {
        Self {
            fastest_fee: response.fastest_fee.ceil() as u64,
            half_hour_fee: response.half_hour_fee.ceil() as u64,
            hour_fee: response.hour_fee.ceil() as u64,
            economy_fee: response.economy_fee.ceil() as u64,
            minimum_fee: response.minimum_fee.ceil() as u64,
        }
    }
}

impl RestClientChainService {
    pub fn new(
        base_url: String,
        network: Network,
        max_retries: usize,
        http_client: Arc<dyn HttpClient>,
        basic_auth: Option<BasicAuth>,
        api_type: ChainApiType,
    ) -> Self {
        Self {
            inner: Arc::new(RestClientChainServiceInner {
                base_url,
                network,
                client: http_client,
                max_retries,
                basic_auth,
                api_type,
                total_timeout: TOTAL_TIMEOUT,
            }),
            // Captured here so each trait-method body can re-enter the
            // surrounding runtime even when invoked from a `UniFFI`
            // foreign-callback proxy that polls outside any tokio context.
            // Callers reach this constructor from within an async path
            // (`new_rest_chain_service`, `SdkBuilder::build`, etc.), so
            // `Handle::current()` is guaranteed to find a runtime.
            #[cfg(not(all(target_family = "wasm", target_os = "unknown")))]
            runtime_handle: tokio::runtime::Handle::current(),
        }
    }

    /// Caps each whole retrying call at `total_timeout` instead of the default,
    /// so a caller with a fallback backend gives up on this one sooner.
    #[must_use]
    pub(crate) fn with_total_timeout(mut self, total_timeout: Duration) -> Self {
        if let Some(inner) = Arc::get_mut(&mut self.inner) {
            inner.total_timeout = total_timeout;
        }
        self
    }

    /// Runs `work` on the captured tokio runtime (non-WASM) or inline
    /// (WASM, where there's no separate runtime to dispatch onto).
    #[cfg(not(all(target_family = "wasm", target_os = "unknown")))]
    async fn run_on_runtime<F, Fut, T>(&self, work: F) -> Result<T, ChainServiceError>
    where
        F: FnOnce(Arc<RestClientChainServiceInner>) -> Fut + Send + 'static,
        Fut: std::future::Future<Output = Result<T, ChainServiceError>> + Send,
        T: Send + 'static,
    {
        use tracing::Instrument;

        let inner = self.inner.clone();
        self.runtime_handle
            .spawn(async move { work(inner).await }.in_current_span())
            .await
            .map_err(|e| ChainServiceError::Generic(format!("join error: {e}")))?
    }

    #[cfg(all(target_family = "wasm", target_os = "unknown"))]
    async fn run_on_runtime<F, Fut, T>(&self, work: F) -> Result<T, ChainServiceError>
    where
        F: FnOnce(Arc<RestClientChainServiceInner>) -> Fut,
        Fut: std::future::Future<Output = Result<T, ChainServiceError>>,
    {
        work(self.inner.clone()).await
    }
}

impl RestClientChainServiceInner {
    async fn get_response_json<T: serde::de::DeserializeOwned>(
        &self,
        path: &str,
    ) -> Result<T, ChainServiceError> {
        let url = format!("{}{}", self.base_url, path);
        debug!("Fetching response json from {}", url);
        let (response, _) = self.get_with_retry(&url, self.client.as_ref()).await?;

        let response: T = serde_json::from_str(&response)
            .map_err(|e| ChainServiceError::Generic(e.to_string()))?;

        Ok(response)
    }

    async fn get_response_text(&self, path: &str) -> Result<String, ChainServiceError> {
        let url = format!("{}{}", self.base_url, path);
        debug!("Fetching response text from {}", url);
        let (response, _) = self.get_with_retry(&url, self.client.as_ref()).await?;
        Ok(response)
    }

    /// Sleeps out the current backoff and charges an attempt, or reports that
    /// the retry budget is spent so the caller surfaces the error it is holding.
    async fn backoff(&self, attempts: &mut usize, delay: &mut Duration) -> bool {
        if *attempts >= self.max_retries {
            return false;
        }
        tokio::time::sleep(*delay).await;
        *attempts = attempts.saturating_add(1);
        *delay = delay.saturating_mul(2);
        true
    }

    async fn get_with_retry(
        &self,
        url: &str,
        client: &dyn HttpClient,
    ) -> Result<(String, u16), ChainServiceError> {
        within_timeout(self.total_timeout, self.get_attempts(url, client)).await
    }

    async fn get_attempts(
        &self,
        url: &str,
        client: &dyn HttpClient,
    ) -> Result<(String, u16), ChainServiceError> {
        let mut delay = BASE_BACKOFF_MILLIS;
        let mut attempts = 0;

        loop {
            let mut headers = HashMap::new();
            if let Some(basic_auth) = &self.basic_auth {
                add_basic_auth_header(&mut headers, &basic_auth.username, &basic_auth.password);
            }

            let response = match client.get(url.to_string(), Some(headers)).await {
                Ok(response) => response,
                Err(e) => {
                    if is_transport_retryable(&e) && self.backoff(&mut attempts, &mut delay).await {
                        continue;
                    }
                    return Err(e.into());
                }
            };

            let HttpResponse { body, status, .. } = response;
            if is_status_retryable(status) && self.backoff(&mut attempts, &mut delay).await {
                continue;
            }
            if !(200..300).contains(&status) {
                return Err(HttpError::Status { status, body }.into());
            }
            return Ok((body, status));
        }
    }

    async fn post(&self, url: &str, body: Option<String>) -> Result<String, ChainServiceError> {
        within_timeout(self.total_timeout, self.post_attempts(url, body)).await
    }

    async fn post_attempts(
        &self,
        url: &str,
        body: Option<String>,
    ) -> Result<String, ChainServiceError> {
        info!("Posting to {}", url);
        debug!(
            "Posting to {} with body {}",
            url,
            body.clone().unwrap_or_default()
        );

        let mut delay = BASE_BACKOFF_MILLIS;
        let mut attempts = 0;

        loop {
            let mut headers: HashMap<String, String> = HashMap::new();
            add_content_type_header(&mut headers, ContentType::TextPlain);
            if let Some(basic_auth) = &self.basic_auth {
                add_basic_auth_header(&mut headers, &basic_auth.username, &basic_auth.password);
            }

            let response = match self
                .client
                .post(url.to_string(), Some(headers), body.clone())
                .await
            {
                Ok(response) => response,
                Err(e) => {
                    if is_transport_retryable(&e) && self.backoff(&mut attempts, &mut delay).await {
                        continue;
                    }
                    return Err(e.into());
                }
            };

            let HttpResponse { body, status, .. } = response;
            if is_status_retryable(status) && self.backoff(&mut attempts, &mut delay).await {
                continue;
            }
            if !(200..300).contains(&status) {
                return Err(HttpError::Status { status, body }.into());
            }
            return Ok(body);
        }
    }

    async fn recommended_fees_esplora(&self) -> Result<RecommendedFees, ChainServiceError> {
        let fee_map = self
            .get_response_json::<HashMap<u16, f64>>("/fee-estimates")
            .await?;
        #[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
        let get_fees = |block: &u16| fee_map.get(block).map_or(0, |fee| fee.ceil() as u64);

        Ok(RecommendedFees {
            fastest_fee: get_fees(&1),
            half_hour_fee: get_fees(&3),
            hour_fee: get_fees(&6),
            economy_fee: get_fees(&25),
            minimum_fee: get_fees(&1008),
        })
    }

    async fn recommended_fees_mempool_space(&self) -> Result<RecommendedFees, ChainServiceError> {
        let response = self
            .get_response_json::<MempoolSpaceRecommendedFeesResponse>("/v1/fees/recommended")
            .await?;
        Ok(response.into())
    }

    // ---- BitcoinChainService method bodies (run on the captured runtime
    //      via the outer struct's `run_on_runtime` helper) ---------------

    async fn do_get_address_utxos(&self, address: String) -> Result<Vec<Utxo>, ChainServiceError> {
        let address = address
            .parse::<Address<NetworkUnchecked>>()?
            .require_network(self.network.into())?;

        let utxos = self
            .get_response_json::<Vec<Utxo>>(format!("/address/{address}/utxo").as_str())
            .await?;

        Ok(utxos)
    }

    async fn do_get_address_txos(&self, address: String) -> Result<Vec<Utxo>, ChainServiceError> {
        let address = address
            .parse::<Address<NetworkUnchecked>>()?
            .require_network(self.network.into())?;
        let address_str = address.to_string();

        // A refund address only ever holds its one refund, a child paying that
        // refund's fee and the sweep, so a single (un-paginated) page of history
        // covers it.
        let txs = self
            .get_response_json::<Vec<AddressTx>>(format!("/address/{address}/txs").as_str())
            .await?;
        let mut txos = Vec::new();
        for tx in txs {
            for (vout, out) in tx.vout.iter().enumerate() {
                if out.scriptpubkey_address.as_deref() != Some(address_str.as_str()) {
                    continue;
                }
                txos.push(Utxo {
                    txid: tx.txid.clone(),
                    vout: u32::try_from(vout)
                        .map_err(|_| ChainServiceError::Generic("vout overflow".to_string()))?,
                    value: out.value,
                    status: tx.status.clone(),
                });
            }
        }
        Ok(txos)
    }

    async fn do_get_transaction_status(
        &self,
        txid: String,
    ) -> Result<super::TxStatus, ChainServiceError> {
        let tx_info = self
            .get_response_json::<TxInfo>(format!("/tx/{txid}").as_str())
            .await?;
        Ok(tx_info.status)
    }

    async fn do_tip_height(&self) -> Result<u32, ChainServiceError> {
        // Plain-text height, served by both esplora and mempool.space.
        let height = self.get_response_text("/blocks/tip/height").await?;
        height
            .trim()
            .parse()
            .map_err(|_| ChainServiceError::Generic(format!("invalid tip height: {height}")))
    }

    async fn do_get_transaction_hex(&self, txid: String) -> Result<String, ChainServiceError> {
        let tx = self
            .get_response_text(format!("/tx/{txid}/hex").as_str())
            .await?;
        Ok(tx)
    }

    async fn do_get_outspend(
        &self,
        txid: String,
        vout: u32,
    ) -> Result<Outspend, ChainServiceError> {
        let outspend = self
            .get_response_json::<Outspend>(format!("/tx/{txid}/outspend/{vout}").as_str())
            .await?;
        Ok(outspend)
    }

    /// Safe to retry: a re-broadcast of a transaction the server did accept is
    /// answered "already known" rather than acted on twice.
    async fn do_broadcast_transaction(&self, tx: String) -> Result<(), ChainServiceError> {
        let url = format!("{}{}", self.base_url, "/tx");
        self.post(&url, Some(tx)).await?;
        Ok(())
    }

    async fn do_recommended_fees(&self) -> Result<RecommendedFees, ChainServiceError> {
        match self.api_type {
            ChainApiType::Esplora => self.recommended_fees_esplora().await,
            ChainApiType::MempoolSpace => self.recommended_fees_mempool_space().await,
        }
    }
}

#[macros::async_trait]
impl BitcoinChainService for RestClientChainService {
    async fn get_address_utxos(&self, address: String) -> Result<Vec<Utxo>, ChainServiceError> {
        self.run_on_runtime(|inner| async move { inner.do_get_address_utxos(address).await })
            .await
    }

    async fn get_address_txos(&self, address: String) -> Result<Vec<Utxo>, ChainServiceError> {
        self.run_on_runtime(|inner| async move { inner.do_get_address_txos(address).await })
            .await
    }

    async fn get_transaction_status(
        &self,
        txid: String,
    ) -> Result<super::TxStatus, ChainServiceError> {
        self.run_on_runtime(|inner| async move { inner.do_get_transaction_status(txid).await })
            .await
    }

    async fn tip_height(&self) -> Result<u32, ChainServiceError> {
        self.run_on_runtime(|inner| async move { inner.do_tip_height().await })
            .await
    }

    async fn get_transaction_hex(&self, txid: String) -> Result<String, ChainServiceError> {
        self.run_on_runtime(|inner| async move { inner.do_get_transaction_hex(txid).await })
            .await
    }

    async fn get_outspend(&self, txid: String, vout: u32) -> Result<Outspend, ChainServiceError> {
        self.run_on_runtime(move |inner| async move { inner.do_get_outspend(txid, vout).await })
            .await
    }

    async fn broadcast_transaction(&self, tx: String) -> Result<(), ChainServiceError> {
        self.run_on_runtime(|inner| async move { inner.do_broadcast_transaction(tx).await })
            .await
    }

    async fn recommended_fees(&self) -> Result<RecommendedFees, ChainServiceError> {
        self.run_on_runtime(|inner| async move { inner.do_recommended_fees().await })
            .await
    }
}

fn is_status_retryable(status: u16) -> bool {
    RETRYABLE_ERROR_CODES.contains(&status)
}

fn is_transport_retryable(error: &HttpError) -> bool {
    matches!(error, HttpError::Request(_))
}

/// Fails `work` once `timeout` has elapsed, whatever attempt it is on.
async fn within_timeout<T>(
    timeout: Duration,
    work: impl std::future::Future<Output = Result<T, ChainServiceError>>,
) -> Result<T, ChainServiceError> {
    match tokio::time::timeout(timeout, work).await {
        Ok(result) => result,
        Err(_) => Err(HttpError::Timeout(format!(
            "gave up after {}s of attempts",
            timeout.as_secs()
        ))
        .into()),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::Network;
    use std::sync::Mutex;

    use macros::async_test_all;

    #[cfg(feature = "browser-tests")]
    wasm_bindgen_test::wasm_bindgen_test_configure!(run_in_browser);

    #[cfg(test)]
    use breez_sdk_common::test_utils::mock_rest_client::{MockResponse, MockRestClient};

    #[async_test_all]
    async fn test_get_address_utxos() {
        // Mock JSON response from the actual API call
        let mock_response = r#"[
            {
                "txid": "277bbdc3557f163810feea810bf390ed90724ec75de779ab181b865292bb1dc1",
                "vout": 3,
                "status": {
                    "confirmed": true,
                    "block_height": 725850,
                    "block_hash": "00000000000000000002d5aace1354d3f5420fcabf4e931f1c4c7ae9c0b405f8",
                    "block_time": 1646382740
                },
                "value": 24201
            },
            {
                "txid": "3a3774433c15d8c1791806d25043335c2a53e5c0ed19517defa4dba9d0b2019f",
                "vout": 0,
                "status": {
                    "confirmed": true,
                    "block_height": 840719,
                    "block_hash": "0000000000000000000170deaa4ccf2de2f1c94346dfef40318d0a7c5178ffd3",
                    "block_time": 1713994081
                },
                "value": 30236
            },
            {
                "txid": "5f2712d4ab1c9aa09c82c28e881724dc3c8c85cbbe71692e593f3911296d40fd",
                "vout": 74,
                "status": {
                    "confirmed": true,
                    "block_height": 726892,
                    "block_hash": "0000000000000000000841798eb13e9230c11f508121e6e1ba25fff3ad3bc448",
                    "block_time": 1647033214
                },
                "value": 5155
            },
            {
                "txid": "7cb4410874b99055fda468dbca45b20ed910909641b46d9fb86869d560c462de",
                "vout": 0,
                "status": {
                    "confirmed": true,
                    "block_height": 857808,
                    "block_hash": "0000000000000000000286598ae217ea4e5b3c63359f3fe105106556182cb926",
                    "block_time": 1724272387
                },
                "value": 6127
            },
            {
                "txid": "4654a83d953c68ba2c50473a80921bb4e1f01d428b18c65ff0128920865cc314",
                "vout": 126,
                "status": {
                    "confirmed": true,
                    "block_height": 748177,
                    "block_hash": "00000000000000000004a65956b7e99b3fcdfb1c01a9dfe5d6d43618427116be",
                    "block_time": 1659763398
                },
                "value": 22190
            }
        ]"#;

        let mock = MockRestClient::new();
        mock.add_response(MockResponse::new(200, mock_response.to_string()));

        // Create the service with the mock server URL
        let service = RestClientChainService::new(
            "http://localhost:8080".to_string(),
            Network::Mainnet,
            3,
            Arc::new(mock),
            None,
            ChainApiType::Esplora,
        );

        // Call the method under test
        let mut result = service
            .get_address_utxos("1wiz18xYmhRX6xStj2b9t1rwWX4GKUgpv".to_string())
            .await
            .unwrap();

        // Sort results by value for consistent testing
        result.sort_by_key(|a| a.value);

        // Verify we got the expected number of UTXOs
        assert_eq!(result.len(), 5);

        // Verify the UTXOs are correctly parsed and sorted by value
        assert_eq!(result[0].value, 5155); // Smallest value
        assert_eq!(
            result[0].txid,
            "5f2712d4ab1c9aa09c82c28e881724dc3c8c85cbbe71692e593f3911296d40fd"
        );
        assert_eq!(result[0].vout, 74);
        assert!(result[0].status.confirmed);
        assert_eq!(result[0].status.block_height, Some(726_892));

        assert_eq!(result[1].value, 6127);
        assert_eq!(
            result[1].txid,
            "7cb4410874b99055fda468dbca45b20ed910909641b46d9fb86869d560c462de"
        );

        assert_eq!(result[2].value, 22190);
        assert_eq!(result[3].value, 24201);
        assert_eq!(result[4].value, 30236); // Largest value

        // Verify all UTXOs are confirmed
        for utxo in &result {
            assert!(utxo.status.confirmed);
            assert!(utxo.status.block_height.is_some());
            assert!(utxo.status.block_time.is_some());
        }
    }

    #[async_test_all]
    async fn test_get_address_txos_returns_spent_outputs() {
        let address = "1wiz18xYmhRX6xStj2b9t1rwWX4GKUgpv";
        // The refund tx pays the address (vout 0); a later sweep spends it and pays
        // elsewhere. get_address_txos must still surface the refund output.
        let mock_response = format!(
            r#"[
                {{
                    "txid": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
                    "status": {{ "confirmed": false }},
                    "vout": [
                        {{ "scriptpubkey_address": "1DestinationaaaaaaaaaaaaaaaaaaaaZ", "value": 49000 }}
                    ]
                }},
                {{
                    "txid": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
                    "status": {{ "confirmed": true, "block_height": 800000, "block_time": 1700000000 }},
                    "vout": [
                        {{ "scriptpubkey_address": "{address}", "value": 50000 }},
                        {{ "scriptpubkey_address": "1OtheraaaaaaaaaaaaaaaaaaaaaaaaaaZ", "value": 10000 }}
                    ]
                }}
            ]"#
        );

        let mock = MockRestClient::new();
        mock.add_response(MockResponse::new(200, mock_response));
        let service = RestClientChainService::new(
            "http://localhost:8080".to_string(),
            Network::Mainnet,
            3,
            Arc::new(mock),
            None,
            ChainApiType::Esplora,
        );

        let result = service.get_address_txos(address.to_string()).await.unwrap();

        assert_eq!(
            result.len(),
            1,
            "only the output paying the address is kept"
        );
        assert_eq!(
            result[0].txid,
            "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
        );
        assert_eq!(result[0].vout, 0);
        assert_eq!(result[0].value, 50000);
        assert!(result[0].status.confirmed);
    }

    /// Answers with `failures` transport errors before serving `response`.
    struct FlakyTransport {
        remaining_failures: Mutex<usize>,
        attempts: Mutex<usize>,
        error: HttpError,
        response: String,
    }

    impl FlakyTransport {
        fn new(failures: usize, error: HttpError, response: &str) -> Self {
            Self {
                remaining_failures: Mutex::new(failures),
                attempts: Mutex::new(0),
                error,
                response: response.to_string(),
            }
        }

        fn answer(&self) -> Result<HttpResponse, HttpError> {
            {
                let mut attempts = self.attempts.lock().unwrap();
                *attempts = attempts.saturating_add(1);
            }
            let mut remaining = self.remaining_failures.lock().unwrap();
            if *remaining > 0 {
                *remaining = remaining.saturating_sub(1);
                return Err(self.error.clone());
            }
            Ok(HttpResponse {
                status: 200,
                body: self.response.clone(),
                headers: HashMap::new(),
            })
        }
    }

    fn broken_pipe() -> HttpError {
        HttpError::Request(
            "client error (SendRequest) : connection error : stream closed because of a \
             broken pipe"
                .to_string(),
        )
    }

    #[macros::async_trait]
    impl HttpClient for FlakyTransport {
        async fn get(
            &self,
            _url: String,
            _headers: Option<HashMap<String, String>>,
        ) -> Result<HttpResponse, HttpError> {
            self.answer()
        }

        async fn post(
            &self,
            _url: String,
            _headers: Option<HashMap<String, String>>,
            _body: Option<String>,
        ) -> Result<HttpResponse, HttpError> {
            self.answer()
        }

        async fn delete(
            &self,
            _url: String,
            _headers: Option<HashMap<String, String>>,
            _body: Option<String>,
        ) -> Result<HttpResponse, HttpError> {
            self.answer()
        }
    }

    fn flaky_service(
        failures: usize,
        error: HttpError,
        response: &str,
    ) -> (RestClientChainService, Arc<FlakyTransport>) {
        let transport = Arc::new(FlakyTransport::new(failures, error, response));
        let service = RestClientChainService::new(
            "http://localhost:8080".to_string(),
            Network::Mainnet,
            3,
            transport.clone(),
            None,
            ChainApiType::Esplora,
        );
        (service, transport)
    }

    const TX_STATUS_RESPONSE: &str = r#"{
        "txid": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
        "status": { "confirmed": true, "block_height": 100 }
    }"#;

    #[async_test_all]
    async fn test_get_retries_transport_errors() {
        let (service, _) = flaky_service(3, broken_pipe(), TX_STATUS_RESPONSE);

        let status = service
            .get_transaction_status("aaaa".to_string())
            .await
            .unwrap();

        assert!(status.confirmed);
    }

    #[async_test_all]
    async fn test_get_gives_up_past_the_retry_budget() {
        let (service, _) = flaky_service(4, broken_pipe(), TX_STATUS_RESPONSE);

        let result = service.get_transaction_status("aaaa".to_string()).await;

        assert!(matches!(
            result,
            Err(ChainServiceError::ServiceConnectivity(_))
        ));
    }

    #[async_test_all]
    async fn test_broadcast_retries_transport_errors() {
        let (service, _) = flaky_service(3, broken_pipe(), "aaaa");

        service
            .broadcast_transaction("00".to_string())
            .await
            .unwrap();
    }

    /// A timeout has already cost the full per-request wait, so it is surfaced
    /// rather than spent again.
    #[async_test_all]
    async fn test_get_does_not_retry_timeouts() {
        let (service, transport) = flaky_service(
            1,
            HttpError::Timeout("operation timed out".to_string()),
            TX_STATUS_RESPONSE,
        );

        let result = service.get_transaction_status("aaaa".to_string()).await;

        assert!(matches!(
            result,
            Err(ChainServiceError::ServiceConnectivity(_))
        ));
        assert_eq!(*transport.attempts.lock().unwrap(), 1);
    }

    /// The timeout covers the whole call, so a transport that keeps failing
    /// cannot stretch it by one more attempt.
    #[async_test_all]
    async fn test_timeout_bounds_the_whole_call() {
        let slow = async {
            tokio::time::sleep(Duration::from_secs(30)).await;
            Ok::<_, ChainServiceError>(())
        };

        let result = within_timeout(Duration::from_millis(50), slow).await;

        assert!(matches!(
            result,
            Err(ChainServiceError::ServiceConnectivity(_))
        ));
    }

    /// A body over the size cap is refused identically every attempt, so it is
    /// not worth another one.
    #[async_test_all]
    async fn test_get_does_not_retry_body_errors() {
        let (service, transport) = flaky_service(
            1,
            HttpError::Body("response exceeds the 10485760 byte limit".to_string()),
            TX_STATUS_RESPONSE,
        );

        let result = service.get_transaction_status("aaaa".to_string()).await;

        assert!(matches!(
            result,
            Err(ChainServiceError::ServiceConnectivity(_))
        ));
        assert_eq!(*transport.attempts.lock().unwrap(), 1);
    }
}