16313842248

Committed 16 Jul 2025 08:00AM UTC coverage: 57.812% (-0.02%) from 57.833%

Build # 16313842248

Build Type

push

github

Committed by

web-flow

Commit Message

chore: reduce http wallet spam logs (#7321)

Description
---
Reduced HTTP wallet spam logs.

Motivation and Context
---
No meaningful logs for larger wallets - lots of byte vectors filling up
the logs.

How Has This Been Tested?
---
Spam was observed with system-level testing.

What process can a PR reviewer use to test or verify this change?
---
Code review

<!-- 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 chain -->


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **Style**
  * Improved readability of logging statements in the client.  
* **Refactor**
* Reduced log verbosity by changing some informational logs to debug
level.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: SW van Heerden <swvheerden@gmail.com>

Coverage Stats

0 of 49 new or added lines in 1 file covered. (0.0%)

21 existing lines in 4 files now uncovered.

69794 of 120725 relevant lines covered (57.81%)

225684.19 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.last_latency
            .read()
            .await
            .map(|latency| latency.1.elapsed() < std::time::Duration::from_secs(60))
            .unwrap_or(false)
    }

    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>,
        end_header_hash: Vec<u8>,
        shutdown: ShutdownSignal,
    ) -> Result<mpsc::Receiver<Result<SyncUtxosByBlockResponse, anyhow::Error>>, anyhow::Error> {
        debug!(
            target: LOG_TARGET,
            "Starting UTXO sync from {} to {}",
            start_header_hash.to_hex(), end_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 end_header_hash_hex = end_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={}&end_header_hash={}&limit={}&page={}",
                        &start_header_hash_hex, &end_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	}	×
NEW 79	debug!(	×
NEW 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() {	×
NEW 98	debug!(	×
NEW 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 {	×
124	self.last_latency	×
125	.read()	×
126	.await	×
127	.map(\|latency\| latency.1.elapsed() < std::time::Duration::from_secs(60))	×
128	.unwrap_or(false)	×
129	}	×
130
131	async fn get_tip_info(&self) -> Result<TipInfoResponse, anyhow::Error> {	×
132	let server_address = self.http_server_address().await?;	×
133	debug!(target: LOG_TARGET, "Requesting tip info from Base Node wallet service at {}", server_address);	×
134	let timer = Instant::now();	×
135	let res = self	×
136	.http_client	×
137	.get(server_address.join("/get_tip_info")?)	×
138	.send()	×
139	.await?;	×
140	self.set_last_latency(timer.elapsed()).await;	×
141
142	if res.status().is_client_error() \|\| res.status().is_server_error() {	×
143	let status = res.status();	×
144	let body = res.text().await.unwrap_or_else(\|_\| "No response body".to_string());	×
145	warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);	×
146	Err(anyhow!(	×
147	"Received error response from Base Node wallet service: {}. {}",	×
148	status,	×
149	body	×
150	))	×
151	} else {
152	Ok(res.json::<TipInfoResponse>().await?)	×
153	}
154	}	×
155
156	async fn get_header_by_height(&self, height: u64) -> Result<Option<BlockHeader>, anyhow::Error> {	×
157	let server_address = self.http_server_address().await?;	×
NEW 158	debug!(	×
NEW 159	target: LOG_TARGET,	×
NEW 160	"Requesting block header at height {} from Base Node wallet service at {}",	×
161	height, server_address
162	);
163	let mut target_url = server_address.join("/get_header_by_height")?;	×
164	target_url.set_query(Some(format!("height={}", height).as_str()));	×
165	let timer = Instant::now();	×
166	let res = self.http_client.get(target_url).send().await?;	×
167	self.set_last_latency(timer.elapsed()).await;	×
168	if res.status() == StatusCode::NOT_FOUND {	×
NEW 169	debug!(	×
NEW 170	target: LOG_TARGET,	×
NEW 171	"No block header found at height {} from Base Node wallet service at {}",	×
172	height, server_address
173	);
174	return Ok(None);	×
175	}	×
176	if res.status().is_client_error() \|\| res.status().is_server_error() {	×
177	let status = res.status();	×
178	let body = res.text().await.unwrap_or_else(\|_\| "No response body".to_string());	×
179	warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);	×
180	return Err(anyhow!(	×
181	"Received error response from Base Node wallet service: {}. {}",	×
182	status,	×
183	body	×
184	));	×
185	} else {
186	let text = res.text().await?;	×
187	match serde_json::from_str::<BlockHeader>(&text) {	×
NEW 188	Ok(header) => Ok(Some(header)),	×
189	Err(e) => {	×
190	warn!(target: LOG_TARGET, "Error decoding block header at height {}: {}, Received:{}", height, e, text);	×
191	Err(anyhow!("Error decoding block header at height {}: {}", height, e))	×
192	},
193	}
194	}
195	}	×
196
197	async fn get_height_at_time(&self, epoch_time: u64) -> Result<u64, anyhow::Error> {	×
198	let server_address = self.http_server_address().await?;	×
NEW 199	debug!(	×
NEW 200	target: LOG_TARGET, "Requesting block height at epoch time {} from Base Node wallet service at {}",	×
201	epoch_time, server_address
202	);
203	let mut target_url = server_address.join("/get_height_at_time")?;	×
204	target_url.set_query(Some(format!("time={}", epoch_time).as_str()));	×
205	let timer = Instant::now();	×
206	let res = self.http_client.get(target_url).send().await?;	×
207	self.set_last_latency(timer.elapsed()).await;	×
208	if res.status().is_client_error() \|\| res.status().is_server_error() {	×
209	let status = res.status();	×
210	let body = res.text().await.unwrap_or_else(\|_\| "No response body".to_string());	×
211	warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);	×
212	Err(anyhow!(	×
213	"Received error response from Base Node wallet service: {}. {}",	×
214	status,	×
215	body	×
216	))	×
217	} else {
218	Ok(res.json::<u64>().await?)	×
219	}
220	}	×
221
222	async fn get_utxos_by_block(&self, header_hash: Vec<u8>) -> Result<models::GetUtxosByBlockResponse, anyhow::Error> {	×
223	let server_address = self.http_server_address().await?;	×
NEW 224	debug!(	×
NEW 225	target: LOG_TARGET,	×
NEW 226	"Requesting UTXOs for block with header hash {} from Base Node wallet service at {}",	×
NEW 227	header_hash.to_hex(), server_address	×
228	);
229	let mut target_url = server_address.join("/get_utxos_by_block")?;	×
230	target_url.set_query(Some(&format!("header_hash={}", header_hash.to_hex())));	×
231	let timer = Instant::now();	×
232	let res = self	×
233	.http_client	×
234	.get(target_url)	×
235	.json(&models::GetUtxosByBlockRequest { header_hash })	×
236	.send()	×
237	.await?;	×
238	self.set_last_latency(timer.elapsed()).await;	×
239	if res.status().is_client_error() \|\| res.status().is_server_error() {	×
240	let status = res.status();	×
241	let body = res.text().await.unwrap_or_else(\|_\| "No response body".to_string());	×
242	warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);	×
243	return Err(anyhow!(	×
244	"Received error response from Base Node wallet service: {}. {}",	×
245	status,	×
246	body,	×
247	));	×
248	}	×
249	Ok(res.json::<models::GetUtxosByBlockResponse>().await?)	×
250	}	×
251
252	async fn sync_utxos_by_block(
253	&self,
254	start_header_hash: Vec<u8>,
255	end_header_hash: Vec<u8>,
256	shutdown: ShutdownSignal,
257	) -> Result<mpsc::Receiver<Result<SyncUtxosByBlockResponse, anyhow::Error>>, anyhow::Error> {	×
NEW 258	debug!(	×
NEW 259	target: LOG_TARGET,	×
NEW 260	"Starting UTXO sync from {} to {}",	×
NEW 261	start_header_hash.to_hex(), end_header_hash.to_hex()	×
262	);
263	let mut target_url = self.http_server_address().await?.join("/sync_utxos_by_block")?;	×
264	let (resp_tx, resp_rx) = mpsc::channel(1000);	×
265	let start_header_hash_hex = start_header_hash.to_hex();	×
266	let end_header_hash_hex = end_header_hash.to_hex();	×
267	let client = self.http_client.clone();	×
268		×
269	let limit = 10;	×
270	tokio::spawn(async move {	×
271	let mut page = 0;	×
272	let mut has_next_page = true;	×
273	while has_next_page {	×
274	if shutdown.is_triggered() {	×
275	info!(target: LOG_TARGET, "UTXO sync task shutdown triggered, exiting loop");	×
276	break;	×
277	}	×
278	target_url.set_query(Some(	×
279	format!(	×
280	"start_header_hash={}&end_header_hash={}&limit={}&page={}",	×
281	&start_header_hash_hex, &end_header_hash_hex, limit, page	×
282	)	×
283	.as_str(),	×
284	));	×
285	debug!(target: LOG_TARGET, "Requesting UTXOs by block from Base Node wallet service at {}", target_url);	×
286	match client.get(target_url.clone()).send().await {	×
287	Ok(response) => match response.json::<SyncUtxosByBlockResponse>().await {	×
288	Ok(response) => {	×
289	has_next_page = response.has_next_page;	×
290	debug!(target: LOG_TARGET, "Received UTXOs for page {}", page);	×
291	if let Err(send_error) = resp_tx.send(Ok(response)).await {	×
292	error!(target: LOG_TARGET, "Error sending utxo response: {:?}", send_error);	×
293	}	×
294	},
295	Err(error) => {	×
296	if let Err(send_error) = resp_tx.send(Err(error.into())).await {	×
297	error!(target: LOG_TARGET, "Error sending error result: {:?}", send_error);	×
298	}	×
299	break;	×
300	},
301	},
302	Err(error) => {	×
303	if let Err(send_error) = resp_tx.send(Err(error.into())).await {	×
304	error!(target: LOG_TARGET, "Error sending error result: {:?}", send_error);	×
305	}	×
306	break;	×
307	},
308	}
309
310	if has_next_page {	×
311	page += 1;	×
312	}	×
313	}
314	});	×
315		×
316	Ok(resp_rx)	×
317	}	×
318
319	async fn get_last_request_latency(&self) -> Option<std::time::Duration> {	×
320	self.last_latency.read().await.map(\|(duration, _)\| duration)	×
321	}	×
322
323	async fn get_utxos_mined_info(&self, hashes: Vec<Vec<u8>>) -> Result<GetUtxosMinedInfoResponse, anyhow::Error> {	×
324	let server_address = self.http_server_address().await?;	×
NEW 325	debug!(	×
NEW 326	target: LOG_TARGET,	×
NEW 327	"Requesting matching UTXOs for {} hashes from Base Node wallet service at {}",	×
NEW 328	hashes.len(), server_address	×
329	);
330	let mut target_url = server_address.join("/get_utxos_mined_info")?;	×
331	target_url.set_query(Some(&format!(	×
332	"hashes={}",	×
333	hashes.iter().map(\|h\| h.to_hex()).collect::<Vec<_>>().join(",")	×
334	)));	×
335	let timer = Instant::now();	×
336	let res = self.http_client.get(target_url).send().await?;	×
337	self.set_last_latency(timer.elapsed()).await;	×
338	if res.status().is_client_error() \|\| res.status().is_server_error() {	×
339	let status = res.status();	×
340	let body = res.text().await.unwrap_or_else(\|_\| "No response body".to_string());	×
341	warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);	×
342	return Err(anyhow!(	×
343	"Received error response from Base Node wallet service: {}. {}",	×
344	status,	×
345	body	×
346	));	×
347	}	×
NEW 348	debug!(	×
NEW 349	target: LOG_TARGET,	×
NEW 350	"Received UTXOs mined info for {} hashes from Base Node wallet service at {}",	×
NEW 351	hashes.len(), server_address	×
352	);
353
354	let res_text = res.text().await?;	×
355	debug!(target: LOG_TARGET, "Response text: {}", res_text);	×
356	let json = serde_json::from_str::<GetUtxosMinedInfoResponse>(&res_text)	×
357	.map_err(\|e\| anyhow!("Failed to parse response JSON: {}", e))?;	×
358	Ok(json)	×
359	}	×
360
361	async fn query_deleted_utxos(
362	&self,
363	hashes: Vec<Vec<u8>>,
364	must_include_header: Vec<u8>,
365	) -> Result<GetUtxosDeletedInfoResponse, anyhow::Error> {	×
366	let server_address = self.http_server_address().await?;	×
NEW 367	debug!(	×
NEW 368	target: LOG_TARGET,	×
NEW 369	"Requesting deleted UTXOs for {} hashes, must include header {} from Base Node wallet",	×
NEW 370	hashes.len(), &must_include_header.to_hex()	×
371	);
372	let mut target_url = server_address.join("/get_utxos_deleted_info")?;	×
373	target_url.set_query(Some(&format!(	×
374	"hashes={}&must_include_header={}",	×
375	hashes.iter().map(\|h\| h.to_hex()).collect::<Vec<_>>().join(","),	×
376	must_include_header.to_hex()	×
377	)));	×
378	let timer = Instant::now();	×
379	let res = self.http_client.get(target_url).send().await?;	×
380	self.set_last_latency(timer.elapsed()).await;	×
381	if res.status().is_client_error() \|\| res.status().is_server_error() {	×
382	let status = res.status();	×
383	let body = res.text().await.unwrap_or_else(\|_\| "No response body".to_string());	×
384	warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);	×
385	return Err(anyhow!(	×
386	"Received error response from Base Node wallet service: {}. {}",	×
387	status,	×
388	body	×
389	));	×
390	}	×
NEW 391	debug!(	×
NEW 392	target: LOG_TARGET,	×
NEW 393	"Received deleted UTXOs for {} hashes from Base Node wallet service at {}",	×
NEW 394	hashes.len(), server_address	×
395	);
396	let res_text = res.text().await?;	×
397	let json = serde_json::from_str::<GetUtxosDeletedInfoResponse>(&res_text)	×
398	.map_err(\|e\| anyhow!("Failed to parse response JSON: {}", e))?;	×
399	Ok(json)	×
400	}	×
401
402	async fn fetch_utxo(&self, utxo: Vec<u8>) -> Result<Option<TransactionOutput>, anyhow::Error> {	×
403	let server_address = self.http_server_address().await?;	×
NEW 404	debug!(	×
NEW 405	target: LOG_TARGET,	×
NEW 406	"Requesting UTXO with hash {} from Base Node wallet service at {}",	×
NEW 407	utxo.to_hex(), server_address	×
408	);
409	let mut target_url = server_address.join("/fetch_utxo")?;	×
410	target_url.set_query(Some(&format!("utxo={}", utxo.to_hex())));	×
411	let timer = Instant::now();	×
412	let res = self.http_client.get(target_url).send().await?;	×
413	self.set_last_latency(timer.elapsed()).await;	×
414	if res.status().is_client_error() \|\| res.status().is_server_error() {	×
415	let status = res.status();	×
416	let body = res.text().await.unwrap_or_else(\|_\| "No response body".to_string());	×
417	warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);	×
418	return Err(anyhow!(	×
419	"Received error response from Base Node wallet service: {}. {}",	×
420	status,	×
421	body	×
422	));	×
423	}	×
424	Ok(res.json::<Option<TransactionOutput>>().await?)	×
425	}	×
426
427	async fn submit_transaction(
428	&self,
429	transaction: tari_core::transactions::transaction_components::Transaction,
430	) -> Result<TxSubmissionResponse, anyhow::Error> {	×
431	let server_address = self.http_server_address().await?;	×
432	debug!(target: LOG_TARGET, "Submitting transaction to Base Node wallet service at {}", server_address);	×
433	let target_url = server_address.join("/json_rpc")?;	×
434	let request_body = serde_json::json!({	×
435	"jsonrpc": "2.0",	×
436	"id": "1",	×
437	"method": "submit_transaction",	×
438	"params": {	×
439	"transaction": transaction,	×
440	}	×
441	});	×
442
443	let res = self.http_client.post(target_url).json(&request_body).send().await?;	×
444	if res.status().is_client_error() \|\| res.status().is_server_error() {	×
445	let status = res.status();	×
446	let body = res.text().await.unwrap_or_else(\|_\| "No response body".to_string());	×
447	warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);	×
448	return Err(anyhow!(	×
449	"Received error response from Base Node wallet service: {}. {}",	×
450	status,	×
451	body	×
452	));	×
453	}	×
454	info!(target: LOG_TARGET, "Transaction submitted successfully to Base Node wallet service at {}", server_address);	×
455	let response = res.json::<JsonRpcResponse<TxSubmissionResponse>>().await?;	×
456	match response.result {	×
457	Some(result) => {	×
458	debug!(target: LOG_TARGET, "Transaction submission response: {:?}", result);	×
459	Ok(result)	×
460	},
461	None => {
462	let error_message = response.error.unwrap_or_else(\|\| "Unknown error".to_string());	×
463	warn!(target: LOG_TARGET, "Transaction submission failed: {}", error_message);	×
464	Err(anyhow!("Transaction submission failed: {}", error_message))	×
465	},
466	}
467	}	×
468
469	async fn transaction_query(
470	&self,
471	excess_sig_nonce: Vec<u8>,
472	excess_sig_sig: Vec<u8>,
473	) -> Result<TxQueryResponse, anyhow::Error> {	×
474	let server_address = self.http_server_address().await?;	×
NEW 475	debug!(	×
NEW 476	target: LOG_TARGET,	×
NEW 477	"Querying transaction with excess signature nonce {} and signature {}",	×
NEW 478	excess_sig_nonce.to_hex(), excess_sig_sig.to_hex()	×
479	);
480	let mut target_url = server_address.join("/transactions")?;	×
481	target_url.set_query(Some(&format!(	×
482	"excess_sig_nonce={}&excess_sig_sig={}",	×
483	excess_sig_nonce.to_hex(),	×
484	excess_sig_sig.to_hex()	×
485	)));	×
486	let timer = Instant::now();	×
487	let res = self.http_client.get(target_url).send().await?;	×
488	self.set_last_latency(timer.elapsed()).await;	×
489	if res.status().is_client_error() \|\| res.status().is_server_error() {	×
490	let status = res.status();	×
491	let body = res.text().await.unwrap_or_else(\|_\| "No response body".to_string());	×
492	warn!(target: LOG_TARGET, "Received error response from Base Node wallet service: {}. {}", status, body);	×
493	return Err(anyhow!(	×
494	"Received error response from Base Node wallet service: {}. {}",	×
495	status,	×
496	body	×
497	));	×
498	}	×
NEW 499	debug!(	×
NEW 500	target: LOG_TARGET,	×
NEW 501	"Transaction query successful for excess signature nonce {} and signature {}",	×
NEW 502	excess_sig_nonce.to_hex(), excess_sig_sig.to_hex()	×
503	);
504	let response = res.json::<TxQueryResponse>().await?;	×
505	Ok(response)	×
506	}	×
507
508	async fn get_mempool_fee_per_gram_stats(&self, _count: u64) -> Result<FeePerGramStat, anyhow::Error> {	×
509	Ok(FeePerGramStat {	×
510	order: 1,	×
511	min_fee_per_gram: MicroMinotari::from(1),	×
512	avg_fee_per_gram: MicroMinotari::from(1),	×
513	max_fee_per_gram: MicroMinotari::from(1),	×
514	}) // Placeholder implementation	×
515	}	×
516	}

tari-project / tari / 16313842248

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