16621916274

Committed 30 Jul 2025 12:00PM UTC coverage: 54.226% (+0.003%) from 54.223%

Build # 16621916274

Build Type

push

github

Committed by

web-flow

Commit Message

feat: enhance wallet connectivity monitoring (#7372)

Description
---
Closes #7360 

Enhance wallet connectivity check to proactively check it during wallet
scanning.

Previously grpc call `check_connectivity` falsely returned `Offline`
status while syncing. This was caused because it relied on `is_online`
method in `https.rs` which check if last http call was done in less than
a minute ago but initially no http calls were made and returned
incorrect value.
Now this grpc method relies on pro-actively updated value checked every
5 minutes which verifies if tip of the scanned block has changed since
the previous check.

How Has This Been Tested?
---

Replaced wallet in TU and check `get_state` and `check_connectivity`
methods calls results while scanning wallet. In both cases I received
`Online` status.

```
❯ grpcurl -plaintext -import-path tari/applications/minotari_app_grpc/proto -proto wallet.proto -d '{}' 127.0.0.1:42027 tari.rpc.Wallet/CheckConnectivity
{
  "status": "Online"
}
```

```
14:07:05 INFO  Wallet status: GetStateResponse { scanned_height: 54930, balance: Some(GetBalanceResponse { available_balance: 3107794619, pending_incoming_balance: 190000, pending_outgoing_balance: 2111889, timelocked_balance: 0 }), network: Some(NetworkStatusResponse { status: Online, avg_latency_ms: 18, num_node_connections: 11 }) }
14:07:05 INFO  Initial wallet scanning: 90.81590476977763% (54930/60485)
```

<!-- Checklist -->
<!-- 1. Is the title of your PR in the form that would make nice release
notes? The title, excluding the conventional commit
tag, will be included exactly as is in the CHANGELOG, so please think
about it carefully. -->


Breaking Changes
---

- [x] None
- [ ] Requires data directory on base node to be deleted
- [ ] Requires hard fork
- [ ] Other - Please specify

<!-- Does this include a breaking change? If so, include this line as a
footer -->
<!-- BREAKING CHANGE: Description what the user should do, e.g. delete a
database, resync the ch... (continued)

Coverage Stats

0 of 3 new or added lines in 2 files covered. (0.0%)

9 existing lines in 3 files now uncovered.

75493 of 139218 relevant lines covered (54.23%)

195785.14 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/clients/rust/base_node_wallet_client/src/client/http.rs

// Copyright 2025 The Tari Project
// SPDX-License-Identifier: BSD-3-Clause
use std::time::Instant;

use anyhow::anyhow;
use async_trait::async_trait;
use log::{debug, error, info, warn};
use reqwest::StatusCode;
use tari_core::{
    base_node::rpc::models::{
        self,
        BlockHeader,
        GetUtxosDeletedInfoResponse,
        GetUtxosMinedInfoResponse,
        SyncUtxosByBlockResponse,
        TipInfoResponse,
        TxQueryResponse,
        TxSubmissionResponse,
    },
    mempool::FeePerGramStat,
    transactions::{tari_amount::MicroMinotari, transaction_components::TransactionOutput},
};
use tari_shutdown::ShutdownSignal;
use tari_utilities::hex::Hex;
use tokio::sync::{mpsc, RwLock};
use url::Url;

use crate::{BaseNodeWalletClient, JsonRpcResponse};

const LOG_TARGET: &str = "tari::wallet::client::http";

/// HTTP client for the Base Node wallet service.
pub struct Client {
    local_api_address: Url,
    default_seed_address: Url,
    http_client: reqwest::Client,
    last_latency: RwLock<Option<(std::time::Duration, Instant)>>,
    use_local_api_address: RwLock<Option<bool>>,
}

impl Client {
    pub fn new(local_api_address: Url, default_seed_address: Url) -> Self {
        Self {
            local_api_address,
            default_seed_address,
            http_client: reqwest::Client::new(),
            last_latency: RwLock::new(None),
            use_local_api_address: RwLock::new(None),
        }
    }
}

impl Clone for Client {
    fn clone(&self) -> Self {
        Self {
            local_api_address: self.local_api_address.clone(),
            default_seed_address: self.default_seed_address.clone(),
            http_client: self.http_client.clone(),
            last_latency: RwLock::new(None),
            use_local_api_address: RwLock::new(None),
        }
    }
}
impl Client {
    async fn set_last_latency(&self, duration: std::time::Duration) {
        let mut last_latency = self.last_latency.write().await;
        *last_latency = Some((duration, Instant::now()));
    }

    /// returns the Url of the https server to use
    async fn http_server_address(&self) -> Result<&Url, anyhow::Error> {
        if let Some(use_local) = self.use_local_api_address.read().await.as_ref() {
            if *use_local {
                return Ok(&self.local_api_address);
            } else {
                return Ok(&self.default_seed_address);
            }
        }
        debug!(
            target: LOG_TARGET, "There is no last connected server set, trying local API address: {}",
            self.local_api_address
        );
        // Try to reach the local API address
        let res = match self
            .http_client
            .get(self.local_api_address.join("/get_tip_info")?)
            .send()
            .await
        {
            Ok(response) => response,
            Err(e) => {
                debug!(target: LOG_TARGET, "Failed to reach local API address {}: {}", self.local_api_address, e);
                *self.use_local_api_address.write().await = Some(false);
                return Ok(&self.default_seed_address);
            },
        };
        if res.status().is_client_error() || res.status().is_server_error() {
            debug!(
                target: LOG_TARGET, "Local API address {} is not reachable, falling back to default seed address: {}",
                self.local_api_address, self.default_seed_address
            );
            // we cant use the local, use the default seed address
            *self.use_local_api_address.write().await = Some(false);
            Ok(&self.default_seed_address)
        } else {
            debug!(target: LOG_TARGET, "Using local API address: {}", self.local_api_address);
            // if we can reach the local api, then use it
            *self.use_local_api_address.write().await = Some(true);
            Ok(&self.local_api_address)
        }
    }
}

#[async_trait]
impl BaseNodeWalletClient for Client {
    async fn get_address(&self) -> String {
        match self.http_server_address().await {
            Ok(v) => v.to_string(),
            _ => "".to_string(),
        }
    }

    async fn is_online(&self) -> bool {
        self.get_tip_info().await.is_ok()
    }

    async fn get_tip_info(&self) -> Result<TipInfoResponse, anyhow::Error> {
        let server_address = self.http_server_address().await?;
        debug!(target: LOG_TARGET, "Requesting tip info from Base Node wallet service at {}", server_address);
        let timer = Instant::now();
        let res = self
            .http_client
            .get(server_address.join("/get_tip_info")?)
            .send()
            .await?;
        self.set_last_latency(timer.elapsed()).await;

        if res.status().is_client_error() || res.status().is_server_error() {
            let status = res.status();
            let body = res.text().await.unwrap_or_else(|_| "No response body".to_string());
            warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);
            Err(anyhow!(
                "Received error response from Base Node wallet service: {}. {}",
                status,
                body
            ))
        } else {
            Ok(res.json::<TipInfoResponse>().await?)
        }
    }

    async fn get_header_by_height(&self, height: u64) -> Result<Option<BlockHeader>, anyhow::Error> {
        let server_address = self.http_server_address().await?;
        debug!(
            target: LOG_TARGET,
            "Requesting block header at height {} from Base Node wallet service at {}",
            height, server_address
        );
        let mut target_url = server_address.join("/get_header_by_height")?;
        target_url.set_query(Some(format!("height={}", height).as_str()));
        let timer = Instant::now();
        let res = self.http_client.get(target_url).send().await?;
        self.set_last_latency(timer.elapsed()).await;
        if res.status() == StatusCode::NOT_FOUND {
            debug!(
                target: LOG_TARGET,
                "No block header found at height {} from Base Node wallet service at {}",
                height, server_address
            );
            return Ok(None);
        }
        if res.status().is_client_error() || res.status().is_server_error() {
            let status = res.status();
            let body = res.text().await.unwrap_or_else(|_| "No response body".to_string());
            warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);
            return Err(anyhow!(
                "Received error response from Base Node wallet service: {}. {}",
                status,
                body
            ));
        } else {
            let text = res.text().await?;
            match serde_json::from_str::<BlockHeader>(&text) {
                Ok(header) => Ok(Some(header)),
                Err(e) => {
                    warn!(target: LOG_TARGET, "Error decoding block header at height {}: {}, Received:{}", height, e, text);
                    Err(anyhow!("Error decoding block header at height {}: {}", height, e))
                },
            }
        }
    }

    async fn get_height_at_time(&self, epoch_time: u64) -> Result<u64, anyhow::Error> {
        let server_address = self.http_server_address().await?;
        debug!(
            target: LOG_TARGET, "Requesting block height at epoch time {} from Base Node wallet service at {}",
            epoch_time, server_address
        );
        let mut target_url = server_address.join("/get_height_at_time")?;
        target_url.set_query(Some(format!("time={}", epoch_time).as_str()));
        let timer = Instant::now();
        let res = self.http_client.get(target_url).send().await?;
        self.set_last_latency(timer.elapsed()).await;
        if res.status().is_client_error() || res.status().is_server_error() {
            let status = res.status();
            let body = res.text().await.unwrap_or_else(|_| "No response body".to_string());
            warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);
            Err(anyhow!(
                "Received error response from Base Node wallet service: {}. {}",
                status,
                body
            ))
        } else {
            Ok(res.json::<u64>().await?)
        }
    }

    async fn get_utxos_by_block(&self, header_hash: Vec<u8>) -> Result<models::GetUtxosByBlockResponse, anyhow::Error> {
        let server_address = self.http_server_address().await?;
        debug!(
            target: LOG_TARGET,
            "Requesting UTXOs for block with header hash {} from Base Node wallet service at {}",
            header_hash.to_hex(), server_address
        );
        let mut target_url = server_address.join("/get_utxos_by_block")?;
        target_url.set_query(Some(&format!("header_hash={}", header_hash.to_hex())));
        let timer = Instant::now();
        let res = self
            .http_client
            .get(target_url)
            .json(&models::GetUtxosByBlockRequest { header_hash })
            .send()
            .await?;
        self.set_last_latency(timer.elapsed()).await;
        if res.status().is_client_error() || res.status().is_server_error() {
            let status = res.status();
            let body = res.text().await.unwrap_or_else(|_| "No response body".to_string());
            warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);
            return Err(anyhow!(
                "Received error response from Base Node wallet service: {}. {}",
                status,
                body,
            ));
        }
        Ok(res.json::<models::GetUtxosByBlockResponse>().await?)
    }

    async fn sync_utxos_by_block(
        &self,
        start_header_hash: Vec<u8>,
        shutdown: ShutdownSignal,
    ) -> Result<mpsc::Receiver<Result<SyncUtxosByBlockResponse, anyhow::Error>>, anyhow::Error> {
        debug!(
            target: LOG_TARGET,
            "Starting UTXO sync from {}",
            start_header_hash.to_hex()
        );
        let mut target_url = self.http_server_address().await?.join("/sync_utxos_by_block")?;
        let (resp_tx, resp_rx) = mpsc::channel(1000);
        let start_header_hash_hex = start_header_hash.to_hex();
        let client = self.http_client.clone();

        let limit = 10;
        tokio::spawn(async move {
            let mut page = 0;
            let mut has_next_page = true;
            while has_next_page {
                if shutdown.is_triggered() {
                    info!(target: LOG_TARGET, "UTXO sync task shutdown triggered, exiting loop");
                    break;
                }
                target_url.set_query(Some(
                    format!(
                        "start_header_hash={}&limit={}&page={}",
                        &start_header_hash_hex, limit, page
                    )
                    .as_str(),
                ));
                debug!(target: LOG_TARGET, "Requesting UTXOs by block from Base Node wallet service at {}", target_url);
                match client.get(target_url.clone()).send().await {
                    Ok(response) => match response.json::<SyncUtxosByBlockResponse>().await {
                        Ok(response) => {
                            has_next_page = response.has_next_page;
                            debug!(target: LOG_TARGET, "Received UTXOs for page {}", page);
                            if let Err(send_error) = resp_tx.send(Ok(response)).await {
                                error!(target: LOG_TARGET, "Error sending utxo response: {:?}", send_error);
                            }
                        },
                        Err(error) => {
                            if let Err(send_error) = resp_tx.send(Err(error.into())).await {
                                error!(target: LOG_TARGET, "Error sending error result: {:?}", send_error);
                            }
                            break;
                        },
                    },
                    Err(error) => {
                        if let Err(send_error) = resp_tx.send(Err(error.into())).await {
                            error!(target: LOG_TARGET, "Error sending error result: {:?}", send_error);
                        }
                        break;
                    },
                }

                if has_next_page {
                    page += 1;
                }
            }
        });

        Ok(resp_rx)
    }

    async fn get_last_request_latency(&self) -> Option<std::time::Duration> {
        self.last_latency.read().await.map(|(duration, _)| duration)
    }

    async fn get_utxos_mined_info(&self, hashes: Vec<Vec<u8>>) -> Result<GetUtxosMinedInfoResponse, anyhow::Error> {
        let server_address = self.http_server_address().await?;
        debug!(
            target: LOG_TARGET,
            "Requesting matching UTXOs for {} hashes from Base Node wallet service at {}",
            hashes.len(), server_address
        );
        let mut target_url = server_address.join("/get_utxos_mined_info")?;
        target_url.set_query(Some(&format!(
            "hashes={}",
            hashes.iter().map(|h| h.to_hex()).collect::<Vec<_>>().join(",")
        )));
        let timer = Instant::now();
        let res = self.http_client.get(target_url).send().await?;
        self.set_last_latency(timer.elapsed()).await;
        if res.status().is_client_error() || res.status().is_server_error() {
            let status = res.status();
            let body = res.text().await.unwrap_or_else(|_| "No response body".to_string());
            warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);
            return Err(anyhow!(
                "Received error response from Base Node wallet service: {}. {}",
                status,
                body
            ));
        }
        debug!(
            target: LOG_TARGET,
            "Received UTXOs mined info for {} hashes from Base Node wallet service at {}",
            hashes.len(), server_address
        );

        let res_text = res.text().await?;
        debug!(target: LOG_TARGET, "Response text: {}", res_text);
        let json = serde_json::from_str::<GetUtxosMinedInfoResponse>(&res_text)
            .map_err(|e| anyhow!("Failed to parse response JSON: {}", e))?;
        Ok(json)
    }

    async fn query_deleted_utxos(
        &self,
        hashes: Vec<Vec<u8>>,
        must_include_header: Vec<u8>,
    ) -> Result<GetUtxosDeletedInfoResponse, anyhow::Error> {
        let server_address = self.http_server_address().await?;
        debug!(
            target: LOG_TARGET,
            "Requesting deleted UTXOs for {} hashes, must include header {} from Base Node wallet",
            hashes.len(), &must_include_header.to_hex()
        );
        let mut target_url = server_address.join("/get_utxos_deleted_info")?;
        target_url.set_query(Some(&format!(
            "hashes={}&must_include_header={}",
            hashes.iter().map(|h| h.to_hex()).collect::<Vec<_>>().join(","),
            must_include_header.to_hex()
        )));
        let timer = Instant::now();
        let res = self.http_client.get(target_url).send().await?;
        self.set_last_latency(timer.elapsed()).await;
        if res.status().is_client_error() || res.status().is_server_error() {
            let status = res.status();
            let body = res.text().await.unwrap_or_else(|_| "No response body".to_string());
            warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);
            return Err(anyhow!(
                "Received error response from Base Node wallet service: {}. {}",
                status,
                body
            ));
        }
        debug!(
            target: LOG_TARGET,
            "Received deleted UTXOs for {} hashes from Base Node wallet service at {}",
            hashes.len(), server_address
        );
        let res_text = res.text().await?;
        let json = serde_json::from_str::<GetUtxosDeletedInfoResponse>(&res_text)
            .map_err(|e| anyhow!("Failed to parse response JSON: {}", e))?;
        Ok(json)
    }

    async fn fetch_utxo(&self, utxo: Vec<u8>) -> Result<Option<TransactionOutput>, anyhow::Error> {
        let server_address = self.http_server_address().await?;
        debug!(
            target: LOG_TARGET,
            "Requesting UTXO with hash {} from Base Node wallet service at {}",
            utxo.to_hex(), server_address
        );
        let mut target_url = server_address.join("/fetch_utxo")?;
        target_url.set_query(Some(&format!("utxo={}", utxo.to_hex())));
        let timer = Instant::now();
        let res = self.http_client.get(target_url).send().await?;
        self.set_last_latency(timer.elapsed()).await;
        if res.status().is_client_error() || res.status().is_server_error() {
            let status = res.status();
            let body = res.text().await.unwrap_or_else(|_| "No response body".to_string());
            warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);
            return Err(anyhow!(
                "Received error response from Base Node wallet service: {}. {}",
                status,
                body
            ));
        }
        Ok(res.json::<Option<TransactionOutput>>().await?)
    }

    async fn submit_transaction(
        &self,
        transaction: tari_core::transactions::transaction_components::Transaction,
    ) -> Result<TxSubmissionResponse, anyhow::Error> {
        let server_address = self.http_server_address().await?;
        debug!(target: LOG_TARGET, "Submitting transaction to Base Node wallet service at {}", server_address);
        let target_url = server_address.join("/json_rpc")?;
        let request_body = serde_json::json!({
            "jsonrpc": "2.0",
            "id": "1",
            "method": "submit_transaction",
            "params": {
                "transaction": transaction,
            }
        });

        let res = self.http_client.post(target_url).json(&request_body).send().await?;
        if res.status().is_client_error() || res.status().is_server_error() {
            let status = res.status();
            let body = res.text().await.unwrap_or_else(|_| "No response body".to_string());
            warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);
            return Err(anyhow!(
                "Received error response from Base Node wallet service: {}. {}",
                status,
                body
            ));
        }
        info!(target: LOG_TARGET, "Transaction submitted successfully to Base Node wallet service at {}", server_address);
        let response = res.json::<JsonRpcResponse<TxSubmissionResponse>>().await?;
        match response.result {
            Some(result) => {
                debug!(target: LOG_TARGET, "Transaction submission response: {:?}", result);
                Ok(result)
            },
            None => {
                let error_message = response.error.unwrap_or_else(|| "Unknown error".to_string());
                warn!(target: LOG_TARGET, "Transaction submission failed: {}", error_message);
                Err(anyhow!("Transaction submission failed: {}", error_message))
            },
        }
    }

    async fn transaction_query(
        &self,
        excess_sig_nonce: Vec<u8>,
        excess_sig_sig: Vec<u8>,
    ) -> Result<TxQueryResponse, anyhow::Error> {
        let server_address = self.http_server_address().await?;
        debug!(
            target: LOG_TARGET,
            "Querying transaction with excess signature nonce {} and signature {}",
            excess_sig_nonce.to_hex(), excess_sig_sig.to_hex()
        );
        let mut target_url = server_address.join("/transactions")?;
        target_url.set_query(Some(&format!(
            "excess_sig_nonce={}&excess_sig_sig={}",
            excess_sig_nonce.to_hex(),
            excess_sig_sig.to_hex()
        )));
        let timer = Instant::now();
        let res = self.http_client.get(target_url).send().await?;
        self.set_last_latency(timer.elapsed()).await;
        if res.status().is_client_error() || res.status().is_server_error() {
            let status = res.status();
            let body = res.text().await.unwrap_or_else(|_| "No response body".to_string());
            warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);
            return Err(anyhow!(
                "Received error response from Base Node wallet service: {}. {}",
                status,
                body
            ));
        }
        debug!(
            target: LOG_TARGET,
            "Transaction query successful for excess signature nonce {} and signature {}",
            excess_sig_nonce.to_hex(), excess_sig_sig.to_hex()
        );
        let response = res.json::<TxQueryResponse>().await?;
        Ok(response)
    }

    async fn get_mempool_fee_per_gram_stats(&self, _count: u64) -> Result<FeePerGramStat, anyhow::Error> {
        Ok(FeePerGramStat {
            order: 1,
            min_fee_per_gram: MicroMinotari::from(1),
            avg_fee_per_gram: MicroMinotari::from(1),
            max_fee_per_gram: MicroMinotari::from(1),
        }) // Placeholder implementation
    }
}

1	// Copyright 2025 The Tari Project
2	// SPDX-License-Identifier: BSD-3-Clause
3	use std::time::Instant;
4
5	use anyhow::anyhow;
6	use async_trait::async_trait;
7	use log::{debug, error, info, warn};
8	use reqwest::StatusCode;
9	use tari_core::{
10	base_node::rpc::models::{
11	self,
12	BlockHeader,
13	GetUtxosDeletedInfoResponse,
14	GetUtxosMinedInfoResponse,
15	SyncUtxosByBlockResponse,
16	TipInfoResponse,
17	TxQueryResponse,
18	TxSubmissionResponse,
19	},
20	mempool::FeePerGramStat,
21	transactions::{tari_amount::MicroMinotari, transaction_components::TransactionOutput},
22	};
23	use tari_shutdown::ShutdownSignal;
24	use tari_utilities::hex::Hex;
25	use tokio::sync::{mpsc, RwLock};
26	use url::Url;
27
28	use crate::{BaseNodeWalletClient, JsonRpcResponse};
29
30	const LOG_TARGET: &str = "tari::wallet::client::http";
31
32	/// HTTP client for the Base Node wallet service.
33	pub struct Client {
34	local_api_address: Url,
35	default_seed_address: Url,
36	http_client: reqwest::Client,
37	last_latency: RwLock<Option<(std::time::Duration, Instant)>>,
38	use_local_api_address: RwLock<Option<bool>>,
39	}
40
41	impl Client {
42	pub fn new(local_api_address: Url, default_seed_address: Url) -> Self {	×
43	Self {	×
44	local_api_address,	×
45	default_seed_address,	×
46	http_client: reqwest::Client::new(),	×
47	last_latency: RwLock::new(None),	×
48	use_local_api_address: RwLock::new(None),	×
49	}	×
50	}	×
51	}
52
53	impl Clone for Client {
54	fn clone(&self) -> Self {	×
55	Self {	×
56	local_api_address: self.local_api_address.clone(),	×
57	default_seed_address: self.default_seed_address.clone(),	×
58	http_client: self.http_client.clone(),	×
59	last_latency: RwLock::new(None),	×
60	use_local_api_address: RwLock::new(None),	×
61	}	×
62	}	×
63	}
64	impl Client {
65	async fn set_last_latency(&self, duration: std::time::Duration) {	×
66	let mut last_latency = self.last_latency.write().await;	×
67	*last_latency = Some((duration, Instant::now()));	×
68	}	×
69
70	/// returns the Url of the https server to use
71	async fn http_server_address(&self) -> Result<&Url, anyhow::Error> {	×
72	if let Some(use_local) = self.use_local_api_address.read().await.as_ref() {	×
73	if *use_local {	×
74	return Ok(&self.local_api_address);	×
75	} else {
76	return Ok(&self.default_seed_address);	×
77	}
78	}	×
79	debug!(	×
80	target: LOG_TARGET, "There is no last connected server set, trying local API address: {}",	×
81	self.local_api_address
82	);
83	// Try to reach the local API address
84	let res = match self	×
85	.http_client	×
86	.get(self.local_api_address.join("/get_tip_info")?)	×
87	.send()	×
88	.await	×
89	{
90	Ok(response) => response,	×
91	Err(e) => {	×
92	debug!(target: LOG_TARGET, "Failed to reach local API address {}: {}", self.local_api_address, e);	×
93	*self.use_local_api_address.write().await = Some(false);	×
94	return Ok(&self.default_seed_address);	×
95	},
96	};
97	if res.status().is_client_error() \|\| res.status().is_server_error() {	×
98	debug!(	×
99	target: LOG_TARGET, "Local API address {} is not reachable, falling back to default seed address: {}",	×
100	self.local_api_address, self.default_seed_address
101	);
102	// we cant use the local, use the default seed address
103	*self.use_local_api_address.write().await = Some(false);	×
104	Ok(&self.default_seed_address)	×
105	} else {
106	debug!(target: LOG_TARGET, "Using local API address: {}", self.local_api_address);	×
107	// if we can reach the local api, then use it
108	*self.use_local_api_address.write().await = Some(true);	×
109	Ok(&self.local_api_address)	×
110	}
111	}	×
112	}
113
114	#[async_trait]
115	impl BaseNodeWalletClient for Client {
116	async fn get_address(&self) -> String {	×
117	match self.http_server_address().await {	×
118	Ok(v) => v.to_string(),	×
119	_ => "".to_string(),	×
120	}
121	}	×
122
123	async fn is_online(&self) -> bool {	×
NEW 124	self.get_tip_info().await.is_ok()	×
125	}	×
126
127	async fn get_tip_info(&self) -> Result<TipInfoResponse, anyhow::Error> {	×
128	let server_address = self.http_server_address().await?;	×
129	debug!(target: LOG_TARGET, "Requesting tip info from Base Node wallet service at {}", server_address);	×
130	let timer = Instant::now();	×
131	let res = self	×
132	.http_client	×
133	.get(server_address.join("/get_tip_info")?)	×
134	.send()	×
135	.await?;	×
136	self.set_last_latency(timer.elapsed()).await;	×
137
138	if res.status().is_client_error() \|\| res.status().is_server_error() {	×
139	let status = res.status();	×
140	let body = res.text().await.unwrap_or_else(\|_\| "No response body".to_string());	×
141	warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);	×
142	Err(anyhow!(	×
143	"Received error response from Base Node wallet service: {}. {}",	×
144	status,	×
145	body	×
146	))	×
147	} else {
148	Ok(res.json::<TipInfoResponse>().await?)	×
149	}
150	}	×
151
152	async fn get_header_by_height(&self, height: u64) -> Result<Option<BlockHeader>, anyhow::Error> {	×
153	let server_address = self.http_server_address().await?;	×
154	debug!(	×
155	target: LOG_TARGET,	×
156	"Requesting block header at height {} from Base Node wallet service at {}",	×
157	height, server_address
158	);
159	let mut target_url = server_address.join("/get_header_by_height")?;	×
160	target_url.set_query(Some(format!("height={}", height).as_str()));	×
161	let timer = Instant::now();	×
162	let res = self.http_client.get(target_url).send().await?;	×
163	self.set_last_latency(timer.elapsed()).await;	×
164	if res.status() == StatusCode::NOT_FOUND {	×
165	debug!(	×
166	target: LOG_TARGET,	×
167	"No block header found at height {} from Base Node wallet service at {}",	×
168	height, server_address
169	);
170	return Ok(None);	×
171	}	×
172	if res.status().is_client_error() \|\| res.status().is_server_error() {	×
173	let status = res.status();	×
174	let body = res.text().await.unwrap_or_else(\|_\| "No response body".to_string());	×
175	warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);	×
176	return Err(anyhow!(	×
177	"Received error response from Base Node wallet service: {}. {}",	×
178	status,	×
179	body	×
180	));	×
181	} else {
182	let text = res.text().await?;	×
183	match serde_json::from_str::<BlockHeader>(&text) {	×
184	Ok(header) => Ok(Some(header)),	×
185	Err(e) => {	×
186	warn!(target: LOG_TARGET, "Error decoding block header at height {}: {}, Received:{}", height, e, text);	×
187	Err(anyhow!("Error decoding block header at height {}: {}", height, e))	×
188	},
189	}
190	}
191	}	×
192
193	async fn get_height_at_time(&self, epoch_time: u64) -> Result<u64, anyhow::Error> {	×
194	let server_address = self.http_server_address().await?;	×
195	debug!(	×
196	target: LOG_TARGET, "Requesting block height at epoch time {} from Base Node wallet service at {}",	×
197	epoch_time, server_address
198	);
199	let mut target_url = server_address.join("/get_height_at_time")?;	×
200	target_url.set_query(Some(format!("time={}", epoch_time).as_str()));	×
201	let timer = Instant::now();	×
202	let res = self.http_client.get(target_url).send().await?;	×
203	self.set_last_latency(timer.elapsed()).await;	×
204	if res.status().is_client_error() \|\| res.status().is_server_error() {	×
205	let status = res.status();	×
206	let body = res.text().await.unwrap_or_else(\|_\| "No response body".to_string());	×
207	warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);	×
208	Err(anyhow!(	×
209	"Received error response from Base Node wallet service: {}. {}",	×
210	status,	×
211	body	×
212	))	×
213	} else {
214	Ok(res.json::<u64>().await?)	×
215	}
216	}	×
217
218	async fn get_utxos_by_block(&self, header_hash: Vec<u8>) -> Result<models::GetUtxosByBlockResponse, anyhow::Error> {	×
219	let server_address = self.http_server_address().await?;	×
220	debug!(	×
221	target: LOG_TARGET,	×
222	"Requesting UTXOs for block with header hash {} from Base Node wallet service at {}",	×
223	header_hash.to_hex(), server_address	×
224	);
225	let mut target_url = server_address.join("/get_utxos_by_block")?;	×
226	target_url.set_query(Some(&format!("header_hash={}", header_hash.to_hex())));	×
227	let timer = Instant::now();	×
228	let res = self	×
229	.http_client	×
230	.get(target_url)	×
231	.json(&models::GetUtxosByBlockRequest { header_hash })	×
232	.send()	×
233	.await?;	×
234	self.set_last_latency(timer.elapsed()).await;	×
235	if res.status().is_client_error() \|\| res.status().is_server_error() {	×
236	let status = res.status();	×
237	let body = res.text().await.unwrap_or_else(\|_\| "No response body".to_string());	×
238	warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);	×
239	return Err(anyhow!(	×
240	"Received error response from Base Node wallet service: {}. {}",	×
241	status,	×
242	body,	×
243	));	×
244	}	×
245	Ok(res.json::<models::GetUtxosByBlockResponse>().await?)	×
246	}	×
247
248	async fn sync_utxos_by_block(
249	&self,
250	start_header_hash: Vec<u8>,
251	shutdown: ShutdownSignal,
252	) -> Result<mpsc::Receiver<Result<SyncUtxosByBlockResponse, anyhow::Error>>, anyhow::Error> {	×
253	debug!(	×
254	target: LOG_TARGET,	×
255	"Starting UTXO sync from {}",	×
256	start_header_hash.to_hex()	×
257	);
258	let mut target_url = self.http_server_address().await?.join("/sync_utxos_by_block")?;	×
259	let (resp_tx, resp_rx) = mpsc::channel(1000);	×
260	let start_header_hash_hex = start_header_hash.to_hex();	×
261	let client = self.http_client.clone();	×
262		×
263	let limit = 10;	×
264	tokio::spawn(async move {	×
265	let mut page = 0;	×
266	let mut has_next_page = true;	×
267	while has_next_page {	×
268	if shutdown.is_triggered() {	×
269	info!(target: LOG_TARGET, "UTXO sync task shutdown triggered, exiting loop");	×
270	break;	×
271	}	×
272	target_url.set_query(Some(	×
273	format!(	×
274	"start_header_hash={}&limit={}&page={}",	×
275	&start_header_hash_hex, limit, page	×
276	)	×
277	.as_str(),	×
278	));	×
279	debug!(target: LOG_TARGET, "Requesting UTXOs by block from Base Node wallet service at {}", target_url);	×
280	match client.get(target_url.clone()).send().await {	×
281	Ok(response) => match response.json::<SyncUtxosByBlockResponse>().await {	×
282	Ok(response) => {	×
283	has_next_page = response.has_next_page;	×
284	debug!(target: LOG_TARGET, "Received UTXOs for page {}", page);	×
285	if let Err(send_error) = resp_tx.send(Ok(response)).await {	×
286	error!(target: LOG_TARGET, "Error sending utxo response: {:?}", send_error);	×
287	}	×
288	},
289	Err(error) => {	×
290	if let Err(send_error) = resp_tx.send(Err(error.into())).await {	×
291	error!(target: LOG_TARGET, "Error sending error result: {:?}", send_error);	×
292	}	×
293	break;	×
294	},
295	},
296	Err(error) => {	×
297	if let Err(send_error) = resp_tx.send(Err(error.into())).await {	×
298	error!(target: LOG_TARGET, "Error sending error result: {:?}", send_error);	×
299	}	×
300	break;	×
301	},
302	}
303
304	if has_next_page {	×
305	page += 1;	×
306	}	×
307	}
308	});	×
309		×
310	Ok(resp_rx)	×
311	}	×
312
313	async fn get_last_request_latency(&self) -> Option<std::time::Duration> {	×
314	self.last_latency.read().await.map(\|(duration, _)\| duration)	×
315	}	×
316
317	async fn get_utxos_mined_info(&self, hashes: Vec<Vec<u8>>) -> Result<GetUtxosMinedInfoResponse, anyhow::Error> {	×
318	let server_address = self.http_server_address().await?;	×
319	debug!(	×
320	target: LOG_TARGET,	×
321	"Requesting matching UTXOs for {} hashes from Base Node wallet service at {}",	×
322	hashes.len(), server_address	×
323	);
324	let mut target_url = server_address.join("/get_utxos_mined_info")?;	×
325	target_url.set_query(Some(&format!(	×
326	"hashes={}",	×
327	hashes.iter().map(\|h\| h.to_hex()).collect::<Vec<_>>().join(",")	×
328	)));	×
329	let timer = Instant::now();	×
330	let res = self.http_client.get(target_url).send().await?;	×
331	self.set_last_latency(timer.elapsed()).await;	×
332	if res.status().is_client_error() \|\| res.status().is_server_error() {	×
333	let status = res.status();	×
334	let body = res.text().await.unwrap_or_else(\|_\| "No response body".to_string());	×
335	warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);	×
336	return Err(anyhow!(	×
337	"Received error response from Base Node wallet service: {}. {}",	×
338	status,	×
339	body	×
340	));	×
341	}	×
342	debug!(	×
343	target: LOG_TARGET,	×
344	"Received UTXOs mined info for {} hashes from Base Node wallet service at {}",	×
345	hashes.len(), server_address	×
346	);
347
348	let res_text = res.text().await?;	×
349	debug!(target: LOG_TARGET, "Response text: {}", res_text);	×
350	let json = serde_json::from_str::<GetUtxosMinedInfoResponse>(&res_text)	×
351	.map_err(\|e\| anyhow!("Failed to parse response JSON: {}", e))?;	×
352	Ok(json)	×
353	}	×
354
355	async fn query_deleted_utxos(
356	&self,
357	hashes: Vec<Vec<u8>>,
358	must_include_header: Vec<u8>,
359	) -> Result<GetUtxosDeletedInfoResponse, anyhow::Error> {	×
360	let server_address = self.http_server_address().await?;	×
361	debug!(	×
362	target: LOG_TARGET,	×
363	"Requesting deleted UTXOs for {} hashes, must include header {} from Base Node wallet",	×
364	hashes.len(), &must_include_header.to_hex()	×
365	);
366	let mut target_url = server_address.join("/get_utxos_deleted_info")?;	×
367	target_url.set_query(Some(&format!(	×
368	"hashes={}&must_include_header={}",	×
369	hashes.iter().map(\|h\| h.to_hex()).collect::<Vec<_>>().join(","),	×
370	must_include_header.to_hex()	×
371	)));	×
372	let timer = Instant::now();	×
373	let res = self.http_client.get(target_url).send().await?;	×
374	self.set_last_latency(timer.elapsed()).await;	×
375	if res.status().is_client_error() \|\| res.status().is_server_error() {	×
376	let status = res.status();	×
377	let body = res.text().await.unwrap_or_else(\|_\| "No response body".to_string());	×
378	warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);	×
379	return Err(anyhow!(	×
380	"Received error response from Base Node wallet service: {}. {}",	×
381	status,	×
382	body	×
383	));	×
384	}	×
385	debug!(	×
386	target: LOG_TARGET,	×
387	"Received deleted UTXOs for {} hashes from Base Node wallet service at {}",	×
388	hashes.len(), server_address	×
389	);
390	let res_text = res.text().await?;	×
391	let json = serde_json::from_str::<GetUtxosDeletedInfoResponse>(&res_text)	×
392	.map_err(\|e\| anyhow!("Failed to parse response JSON: {}", e))?;	×
393	Ok(json)	×
394	}	×
395
396	async fn fetch_utxo(&self, utxo: Vec<u8>) -> Result<Option<TransactionOutput>, anyhow::Error> {	×
397	let server_address = self.http_server_address().await?;	×
398	debug!(	×
399	target: LOG_TARGET,	×
400	"Requesting UTXO with hash {} from Base Node wallet service at {}",	×
401	utxo.to_hex(), server_address	×
402	);
403	let mut target_url = server_address.join("/fetch_utxo")?;	×
404	target_url.set_query(Some(&format!("utxo={}", utxo.to_hex())));	×
405	let timer = Instant::now();	×
406	let res = self.http_client.get(target_url).send().await?;	×
407	self.set_last_latency(timer.elapsed()).await;	×
408	if res.status().is_client_error() \|\| res.status().is_server_error() {	×
409	let status = res.status();	×
410	let body = res.text().await.unwrap_or_else(\|_\| "No response body".to_string());	×
411	warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);	×
412	return Err(anyhow!(	×
413	"Received error response from Base Node wallet service: {}. {}",	×
414	status,	×
415	body	×
416	));	×
417	}	×
418	Ok(res.json::<Option<TransactionOutput>>().await?)	×
419	}	×
420
421	async fn submit_transaction(
422	&self,
423	transaction: tari_core::transactions::transaction_components::Transaction,
424	) -> Result<TxSubmissionResponse, anyhow::Error> {	×
425	let server_address = self.http_server_address().await?;	×
426	debug!(target: LOG_TARGET, "Submitting transaction to Base Node wallet service at {}", server_address);	×
427	let target_url = server_address.join("/json_rpc")?;	×
428	let request_body = serde_json::json!({	×
429	"jsonrpc": "2.0",	×
430	"id": "1",	×
431	"method": "submit_transaction",	×
432	"params": {	×
433	"transaction": transaction,	×
434	}	×
435	});	×
436
437	let res = self.http_client.post(target_url).json(&request_body).send().await?;	×
438	if res.status().is_client_error() \|\| res.status().is_server_error() {	×
439	let status = res.status();	×
440	let body = res.text().await.unwrap_or_else(\|_\| "No response body".to_string());	×
441	warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);	×
442	return Err(anyhow!(	×
443	"Received error response from Base Node wallet service: {}. {}",	×
444	status,	×
445	body	×
446	));	×
447	}	×
448	info!(target: LOG_TARGET, "Transaction submitted successfully to Base Node wallet service at {}", server_address);	×
449	let response = res.json::<JsonRpcResponse<TxSubmissionResponse>>().await?;	×
450	match response.result {	×
451	Some(result) => {	×
452	debug!(target: LOG_TARGET, "Transaction submission response: {:?}", result);	×
453	Ok(result)	×
454	},
455	None => {
456	let error_message = response.error.unwrap_or_else(\|\| "Unknown error".to_string());	×
457	warn!(target: LOG_TARGET, "Transaction submission failed: {}", error_message);	×
458	Err(anyhow!("Transaction submission failed: {}", error_message))	×
459	},
460	}
461	}	×
462
463	async fn transaction_query(
464	&self,
465	excess_sig_nonce: Vec<u8>,
466	excess_sig_sig: Vec<u8>,
467	) -> Result<TxQueryResponse, anyhow::Error> {	×
468	let server_address = self.http_server_address().await?;	×
469	debug!(	×
470	target: LOG_TARGET,	×
471	"Querying transaction with excess signature nonce {} and signature {}",	×
472	excess_sig_nonce.to_hex(), excess_sig_sig.to_hex()	×
473	);
474	let mut target_url = server_address.join("/transactions")?;	×
475	target_url.set_query(Some(&format!(	×
476	"excess_sig_nonce={}&excess_sig_sig={}",	×
477	excess_sig_nonce.to_hex(),	×
478	excess_sig_sig.to_hex()	×
479	)));	×
480	let timer = Instant::now();	×
481	let res = self.http_client.get(target_url).send().await?;	×
482	self.set_last_latency(timer.elapsed()).await;	×
483	if res.status().is_client_error() \|\| res.status().is_server_error() {	×
484	let status = res.status();	×
485	let body = res.text().await.unwrap_or_else(\|_\| "No response body".to_string());	×
486	warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);	×
487	return Err(anyhow!(	×
488	"Received error response from Base Node wallet service: {}. {}",	×
489	status,	×
490	body	×
491	));	×
492	}	×
493	debug!(	×
494	target: LOG_TARGET,	×
495	"Transaction query successful for excess signature nonce {} and signature {}",	×
496	excess_sig_nonce.to_hex(), excess_sig_sig.to_hex()	×
497	);
498	let response = res.json::<TxQueryResponse>().await?;	×
499	Ok(response)	×
500	}	×
501
502	async fn get_mempool_fee_per_gram_stats(&self, _count: u64) -> Result<FeePerGramStat, anyhow::Error> {	×
503	Ok(FeePerGramStat {	×
504	order: 1,	×
505	min_fee_per_gram: MicroMinotari::from(1),	×
506	avg_fee_per_gram: MicroMinotari::from(1),	×
507	max_fee_per_gram: MicroMinotari::from(1),	×
508	}) // Placeholder implementation	×
509	}	×
510	}

tari-project / tari / 16621916274

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous