16123384529

Committed 07 Jul 2025 05:11PM UTC coverage: 64.327% (-7.6%) from 71.89%

Build # 16123384529

Build Type

push

github

Committed by

web-flow

Commit Message

chore: new release v4.9.0-pre.0 (#7289)

Description
---
new release esmeralda

Run Details

77151 of 119935 relevant lines covered (64.33%)

227108.34 hits per line

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

0.0

/base_layer/core/src/base_node/rpc/query_service.rs

// Copyright 2025 The Tari Project
// SPDX-License-Identifier: BSD-3-Clause

use log::trace;
use serde_valid::{validation, Validate};
use tari_common_types::{types, types::FixedHashSizeError};
use tari_utilities::{hex::Hex, ByteArray, ByteArrayError};
use thiserror::Error;

use crate::{
    base_node::{
        rpc::{
            models::{
                self,
                BlockUtxoInfo,
                GetUtxosByBlockRequest,
                GetUtxosByBlockResponse,
                MinimalUtxoSyncInfo,
                SyncUtxosByBlockRequest,
                SyncUtxosByBlockResponse,
                TipInfoResponse,
                TxLocation,
                TxQueryResponse,
            },
            BaseNodeWalletQueryService,
        },
        state_machine_service::states::StateInfo,
        StateMachineHandle,
    },
    chain_storage::{async_db::AsyncBlockchainDb, BlockchainBackend, ChainStorageError},
    mempool::{service::MempoolHandle, MempoolServiceError, TxStorageResponse},
    transactions::transaction_components::TransactionOutput,
};

const LOG_TARGET: &str = "c::bn::rpc::query_service";

#[derive(Debug, Error)]
pub enum Error {
    #[error("Failed to get chain metadata: {0}")]
    FailedToGetChainMetadata(#[from] ChainStorageError),
    #[error("Header not found at height: {height}")]
    HeaderNotFound { height: u64 },
    #[error("Signature conversion error: {0}")]
    SignatureConversion(ByteArrayError),
    #[error("Mempool service error: {0}")]
    MempoolService(#[from] MempoolServiceError),
    #[error("Serde validation error: {0}")]
    SerdeValidation(#[from] validation::Errors),
    #[error("Hash conversion error: {0}")]
    HashConversion(#[from] FixedHashSizeError),
    #[error("Start header hash not found")]
    StartHeaderHashNotFound,
    #[error("End header hash not found")]
    EndHeaderHashNotFound,
    #[error("Header hash not found")]
    HeaderHashNotFound,
    #[error("Start header height {start_height} cannot be greater than the end header height {end_height}")]
    HeaderHeightMismatch { start_height: u64, end_height: u64 },
}

pub struct Service<B> {
    db: AsyncBlockchainDb<B>,
    state_machine: StateMachineHandle,
    mempool: MempoolHandle,
}

impl<B: BlockchainBackend + 'static> Service<B> {
    pub fn new(db: AsyncBlockchainDb<B>, state_machine: StateMachineHandle, mempool: MempoolHandle) -> Self {
        Self {
            db,
            state_machine,
            mempool,
        }
    }

    fn state_machine(&self) -> StateMachineHandle {
        self.state_machine.clone()
    }

    fn db(&self) -> &AsyncBlockchainDb<B> {
        &self.db
    }

    fn mempool(&self) -> MempoolHandle {
        self.mempool.clone()
    }

    async fn fetch_kernel(&self, signature: types::Signature) -> Result<TxQueryResponse, Error> {
        let db = self.db();

        match db.fetch_kernel_by_excess_sig(signature.clone()).await? {
            None => (),
            Some((_, block_hash)) => match db.fetch_header_by_block_hash(block_hash).await? {
                None => (),
                Some(header) => {
                    let response = TxQueryResponse {
                        location: TxLocation::Mined,
                        mined_header_hash: Some(block_hash.to_vec()),
                        mined_height: Some(header.height),
                        mined_timestamp: Some(header.timestamp.as_u64()),
                    };
                    return Ok(response);
                },
            },
        };

        // If not in a block then check the mempool
        let mut mempool = self.mempool();
        let mempool_response = match mempool.get_tx_state_by_excess_sig(signature.clone()).await? {
            TxStorageResponse::UnconfirmedPool => TxQueryResponse {
                location: TxLocation::InMempool,
                mined_header_hash: None,
                mined_height: None,
                mined_timestamp: None,
            },
            TxStorageResponse::ReorgPool |
            TxStorageResponse::NotStoredOrphan |
            TxStorageResponse::NotStoredTimeLocked |
            TxStorageResponse::NotStoredAlreadySpent |
            TxStorageResponse::NotStoredConsensus |
            TxStorageResponse::NotStored |
            TxStorageResponse::NotStoredFeeTooLow |
            TxStorageResponse::NotStoredAlreadyMined => TxQueryResponse {
                location: TxLocation::NotStored,
                mined_timestamp: None,
                mined_height: None,
                mined_header_hash: None,
            },
        };

        Ok(mempool_response)
    }

    async fn fetch_utxos_by_block(&self, request: GetUtxosByBlockRequest) -> Result<GetUtxosByBlockResponse, Error> {
        request.validate()?;

        let hash = request.header_hash.clone().try_into()?;

        let header = self
            .db()
            .fetch_header_by_block_hash(hash)
            .await?
            .ok_or_else(|| Error::HeaderHashNotFound)?;

        // fetch utxos
        let outputs_with_statuses = self.db.fetch_outputs_in_block_with_spend_state(hash, None).await?;

        let outputs = outputs_with_statuses
            .into_iter()
            .map(|(output, _spent)| output)
            .collect::<Vec<TransactionOutput>>();

        // if its empty, we need to send an empty vec of outputs.
        let utxo_block_response = GetUtxosByBlockResponse {
            outputs,
            height: header.height,
            header_hash: hash.to_vec(),
            mined_timestamp: header.timestamp.as_u64(),
        };

        Ok(utxo_block_response)
    }

    async fn fetch_utxos(&self, request: SyncUtxosByBlockRequest) -> Result<SyncUtxosByBlockResponse, Error> {
        // validate and fetch inputs
        request.validate()?;

        let hash = request.start_header_hash.clone().try_into()?;

        let start_header = self
            .db()
            .fetch_header_by_block_hash(hash)
            .await?
            .ok_or_else(|| Error::StartHeaderHashNotFound)?;

        let hash = request.end_header_hash.clone().try_into()?;

        let end_header = self
            .db
            .fetch_header_by_block_hash(hash)
            .await?
            .ok_or_else(|| Error::EndHeaderHashNotFound)?;

        // pagination
        let start_header_height = start_header.height + (request.page * request.limit);
        let start_header = self
            .db
            .fetch_header(start_header_height)
            .await?
            .ok_or_else(|| Error::HeaderNotFound {
                height: start_header_height,
            })?;

        if start_header.height > end_header.height {
            return Err(Error::HeaderHeightMismatch {
                start_height: start_header.height,
                end_height: end_header.height,
            });
        }

        // fetch utxos
        let mut utxos = vec![];
        let mut current_header = start_header;
        let mut fetched_utxos = 0;
        loop {
            let current_header_hash = current_header.hash();

            trace!(
                target: LOG_TARGET,
                "current header = {} ({})",
                current_header.height,
                current_header_hash.to_hex()
            );

            let outputs_with_statuses = self
                .db
                .fetch_outputs_in_block_with_spend_state(current_header.hash(), None)
                .await?;

            let outputs = outputs_with_statuses
                .into_iter()
                .map(|(output, _spent)| output)
                .collect::<Vec<TransactionOutput>>();

            for output_chunk in outputs.chunks(2000) {
                let output_block_response = BlockUtxoInfo {
                    outputs: output_chunk
                        .iter()
                        .map(|output| MinimalUtxoSyncInfo {
                            output_hash: output.hash().to_vec(),
                            commitment: output.commitment().to_vec(),
                            encrypted_data: output.encrypted_data().as_bytes().to_vec(),
                            sender_offset_public_key: output.sender_offset_public_key.to_vec(),
                        })
                        .collect(),
                    height: current_header.height,
                    header_hash: current_header_hash.to_vec(),
                    mined_timestamp: current_header.timestamp.as_u64(),
                };
                utxos.push(output_block_response);
            }
            if outputs.is_empty() {
                // if its empty, we need to send an empty vec of outputs.
                let utxo_block_response = BlockUtxoInfo {
                    outputs: Vec::new(),
                    height: current_header.height,
                    header_hash: current_header_hash.to_vec(),
                    mined_timestamp: current_header.timestamp.as_u64(),
                };
                utxos.push(utxo_block_response);
            }

            fetched_utxos += 1;

            if current_header.height >= end_header.height || fetched_utxos >= request.limit {
                break;
            }

            current_header =
                self.db
                    .fetch_header(current_header.height + 1)
                    .await?
                    .ok_or_else(|| Error::HeaderNotFound {
                        height: current_header.height + 1,
                    })?;
        }

        let has_next_page = (end_header.height - current_header.height) > 0;

        Ok(SyncUtxosByBlockResponse {
            blocks: utxos,
            has_next_page,
        })
    }
}

#[async_trait::async_trait]
impl<B: BlockchainBackend + 'static> BaseNodeWalletQueryService for Service<B> {
    type Error = Error;

    async fn get_tip_info(&self) -> Result<TipInfoResponse, Self::Error> {
        let state_machine = self.state_machine();
        let status_watch = state_machine.get_status_info_watch();
        let is_synced = match status_watch.borrow().state_info {
            StateInfo::Listening(li) => li.is_synced(),
            _ => false,
        };

        let metadata = self.db.get_chain_metadata().await?;

        Ok(TipInfoResponse {
            metadata: Some(metadata),
            is_synced,
        })
    }

    async fn get_header_by_height(&self, height: u64) -> Result<models::BlockHeader, Self::Error> {
        let result = self
            .db
            .fetch_header(height)
            .await?
            .ok_or(Error::HeaderNotFound { height })?
            .into();
        Ok(result)
    }

    async fn get_height_at_time(&self, epoch_time: u64) -> Result<u64, Self::Error> {
        trace!(target: LOG_TARGET, "requested_epoch_time: {}", epoch_time);
        let tip_header = self.db.fetch_tip_header().await?;

        let mut left_height = 0u64;
        let mut right_height = tip_header.height();

        while left_height <= right_height {
            let mut mid_height = (left_height + right_height) / 2;

            if mid_height == 0 {
                return Ok(0u64);
            }
            // If the two bounds are adjacent then perform the test between the right and left sides
            if left_height == mid_height {
                mid_height = right_height;
            }

            let mid_header = self
                .db
                .fetch_header(mid_height)
                .await?
                .ok_or_else(|| Error::HeaderNotFound { height: mid_height })?;
            let before_mid_header = self
                .db
                .fetch_header(mid_height - 1)
                .await?
                .ok_or_else(|| Error::HeaderNotFound { height: mid_height - 1 })?;
            trace!(
                target: LOG_TARGET,
                "requested_epoch_time: {}, left: {}, mid: {}/{} ({}/{}), right: {}",
                epoch_time,
                left_height,
                mid_height,
                mid_height-1,
                mid_header.timestamp.as_u64(),
                before_mid_header.timestamp.as_u64(),
                right_height
            );
            if epoch_time < mid_header.timestamp.as_u64() && epoch_time >= before_mid_header.timestamp.as_u64() {
                trace!(
                    target: LOG_TARGET,
                    "requested_epoch_time: {}, selected height: {}",
                    epoch_time, before_mid_header.height
                );
                return Ok(before_mid_header.height);
            } else if mid_height == right_height {
                trace!(
                    target: LOG_TARGET,
                    "requested_epoch_time: {}, selected height: {}",
                    epoch_time, right_height
                );
                return Ok(right_height);
            } else if epoch_time <= mid_header.timestamp.as_u64() {
                right_height = mid_height;
            } else {
                left_height = mid_height;
            }
        }

        Ok(0u64)
    }

    async fn transaction_query(
        &self,
        signature: crate::base_node::rpc::models::Signature,
    ) -> Result<TxQueryResponse, Self::Error> {
        let signature = signature.try_into().map_err(Error::SignatureConversion)?;

        let response = self.fetch_kernel(signature).await?;

        Ok(response)
    }

    async fn sync_utxos_by_block(
        &self,
        request: SyncUtxosByBlockRequest,
    ) -> Result<SyncUtxosByBlockResponse, Self::Error> {
        self.fetch_utxos(request).await
    }

    async fn get_utxos_by_block(
        &self,
        request: GetUtxosByBlockRequest,
    ) -> Result<GetUtxosByBlockResponse, Self::Error> {
        self.fetch_utxos_by_block(request).await
    }

    async fn get_utxos_mined_info(
        &self,
        request: models::GetUtxosMinedInfoRequest,
    ) -> Result<models::GetUtxosMinedInfoResponse, Self::Error> {
        request.validate()?;

        let mut utxos = vec![];

        let tip_header = self.db().fetch_tip_header().await?;
        for hash in request.hashes {
            let hash = hash.try_into()?;
            let output = self.db().fetch_output(hash).await?;
            if let Some(output) = output {
                utxos.push(models::MinedUtxoInfo {
                    utxo_hash: hash.to_vec(),
                    mined_in_hash: output.header_hash.to_vec(),
                    mined_in_height: output.mined_height,
                    mined_in_timestamp: output.mined_timestamp,
                });
            }
        }

        Ok(models::GetUtxosMinedInfoResponse {
            utxos,
            best_block_hash: tip_header.hash().to_vec(),
            best_block_height: tip_header.height(),
        })
    }

    async fn get_utxos_deleted_info(
        &self,
        request: models::GetUtxosDeletedInfoRequest,
    ) -> Result<models::GetUtxosDeletedInfoResponse, Self::Error> {
        request.validate()?;

        let mut utxos = vec![];

        let must_include_header = request.must_include_header.clone().try_into()?;
        if self
            .db()
            .fetch_header_by_block_hash(must_include_header)
            .await?
            .is_none()
        {
            return Err(Error::HeaderHashNotFound);
        }

        let tip_header = self.db().fetch_tip_header().await?;
        for hash in request.hashes {
            let hash = hash.try_into()?;
            let output = self.db().fetch_output(hash).await?;

            if let Some(output) = output {
                // is it still unspent?
                let input = self.db().fetch_input(hash).await?;
                if let Some(i) = input {
                    utxos.push(models::DeletedUtxoInfo {
                        utxo_hash: hash.to_vec(),
                        found_in_header: Some((output.mined_height, output.header_hash.to_vec())),
                        spent_in_header: Some((i.spent_height, i.header_hash.to_vec())),
                    });
                } else {
                    utxos.push(models::DeletedUtxoInfo {
                        utxo_hash: hash.to_vec(),
                        found_in_header: Some((output.mined_height, output.header_hash.to_vec())),
                        spent_in_header: None,
                    });
                }
            } else {
                utxos.push(models::DeletedUtxoInfo {
                    utxo_hash: hash.to_vec(),
                    found_in_header: None,
                    spent_in_header: None,
                });
            }
        }

        Ok(models::GetUtxosDeletedInfoResponse {
            utxos,
            best_block_hash: tip_header.hash().to_vec(),
            best_block_height: tip_header.height(),
        })
    }
}

1	// Copyright 2025 The Tari Project
2	// SPDX-License-Identifier: BSD-3-Clause
3
4	use log::trace;
5	use serde_valid::{validation, Validate};
6	use tari_common_types::{types, types::FixedHashSizeError};
7	use tari_utilities::{hex::Hex, ByteArray, ByteArrayError};
8	use thiserror::Error;
9
10	use crate::{
11	base_node::{
12	rpc::{
13	models::{
14	self,
15	BlockUtxoInfo,
16	GetUtxosByBlockRequest,
17	GetUtxosByBlockResponse,
18	MinimalUtxoSyncInfo,
19	SyncUtxosByBlockRequest,
20	SyncUtxosByBlockResponse,
21	TipInfoResponse,
22	TxLocation,
23	TxQueryResponse,
24	},
25	BaseNodeWalletQueryService,
26	},
27	state_machine_service::states::StateInfo,
28	StateMachineHandle,
29	},
30	chain_storage::{async_db::AsyncBlockchainDb, BlockchainBackend, ChainStorageError},
31	mempool::{service::MempoolHandle, MempoolServiceError, TxStorageResponse},
32	transactions::transaction_components::TransactionOutput,
33	};
34
35	const LOG_TARGET: &str = "c::bn::rpc::query_service";
36
37	#[derive(Debug, Error)]
38	pub enum Error {
39	#[error("Failed to get chain metadata: {0}")]
40	FailedToGetChainMetadata(#[from] ChainStorageError),
41	#[error("Header not found at height: {height}")]
42	HeaderNotFound { height: u64 },
43	#[error("Signature conversion error: {0}")]
44	SignatureConversion(ByteArrayError),
45	#[error("Mempool service error: {0}")]
46	MempoolService(#[from] MempoolServiceError),
47	#[error("Serde validation error: {0}")]
48	SerdeValidation(#[from] validation::Errors),
49	#[error("Hash conversion error: {0}")]
50	HashConversion(#[from] FixedHashSizeError),
51	#[error("Start header hash not found")]
52	StartHeaderHashNotFound,
53	#[error("End header hash not found")]
54	EndHeaderHashNotFound,
55	#[error("Header hash not found")]
56	HeaderHashNotFound,
57	#[error("Start header height {start_height} cannot be greater than the end header height {end_height}")]
58	HeaderHeightMismatch { start_height: u64, end_height: u64 },
59	}
60
61	pub struct Service<B> {
62	db: AsyncBlockchainDb<B>,
63	state_machine: StateMachineHandle,
64	mempool: MempoolHandle,
65	}
66
67	impl<B: BlockchainBackend + 'static> Service<B> {
68	pub fn new(db: AsyncBlockchainDb<B>, state_machine: StateMachineHandle, mempool: MempoolHandle) -> Self {	×
69	Self {	×
70	db,	×
71	state_machine,	×
72	mempool,	×
73	}	×
74	}	×
75
76	fn state_machine(&self) -> StateMachineHandle {	×
77	self.state_machine.clone()	×
78	}	×
79
80	fn db(&self) -> &AsyncBlockchainDb<B> {	×
81	&self.db	×
82	}	×
83
84	fn mempool(&self) -> MempoolHandle {	×
85	self.mempool.clone()	×
86	}	×
87
88	async fn fetch_kernel(&self, signature: types::Signature) -> Result<TxQueryResponse, Error> {	×
89	let db = self.db();	×
90		×
91	match db.fetch_kernel_by_excess_sig(signature.clone()).await? {	×
92	None => (),	×
93	Some((_, block_hash)) => match db.fetch_header_by_block_hash(block_hash).await? {	×
94	None => (),	×
95	Some(header) => {	×
96	let response = TxQueryResponse {	×
97	location: TxLocation::Mined,	×
98	mined_header_hash: Some(block_hash.to_vec()),	×
99	mined_height: Some(header.height),	×
100	mined_timestamp: Some(header.timestamp.as_u64()),	×
101	};	×
102	return Ok(response);	×
103	},
104	},
105	};
106
107	// If not in a block then check the mempool
108	let mut mempool = self.mempool();	×
109	let mempool_response = match mempool.get_tx_state_by_excess_sig(signature.clone()).await? {	×
110	TxStorageResponse::UnconfirmedPool => TxQueryResponse {	×
111	location: TxLocation::InMempool,	×
112	mined_header_hash: None,	×
113	mined_height: None,	×
114	mined_timestamp: None,	×
115	},	×
116	TxStorageResponse::ReorgPool \|
117	TxStorageResponse::NotStoredOrphan \|
118	TxStorageResponse::NotStoredTimeLocked \|
119	TxStorageResponse::NotStoredAlreadySpent \|
120	TxStorageResponse::NotStoredConsensus \|
121	TxStorageResponse::NotStored \|
122	TxStorageResponse::NotStoredFeeTooLow \|
123	TxStorageResponse::NotStoredAlreadyMined => TxQueryResponse {	×
124	location: TxLocation::NotStored,	×
125	mined_timestamp: None,	×
126	mined_height: None,	×
127	mined_header_hash: None,	×
128	},	×
129	};
130
131	Ok(mempool_response)	×
132	}	×
133
134	async fn fetch_utxos_by_block(&self, request: GetUtxosByBlockRequest) -> Result<GetUtxosByBlockResponse, Error> {	×
135	request.validate()?;	×
136
137	let hash = request.header_hash.clone().try_into()?;	×
138
139	let header = self	×
140	.db()	×
141	.fetch_header_by_block_hash(hash)	×
142	.await?	×
143	.ok_or_else(\|\| Error::HeaderHashNotFound)?;	×
144
145	// fetch utxos
146	let outputs_with_statuses = self.db.fetch_outputs_in_block_with_spend_state(hash, None).await?;	×
147
148	let outputs = outputs_with_statuses	×
149	.into_iter()	×
150	.map(\|(output, _spent)\| output)	×
151	.collect::<Vec<TransactionOutput>>();	×
152		×
153	// if its empty, we need to send an empty vec of outputs.	×
154	let utxo_block_response = GetUtxosByBlockResponse {	×
155	outputs,	×
156	height: header.height,	×
157	header_hash: hash.to_vec(),	×
158	mined_timestamp: header.timestamp.as_u64(),	×
159	};	×
160		×
161	Ok(utxo_block_response)	×
162	}	×
163
164	async fn fetch_utxos(&self, request: SyncUtxosByBlockRequest) -> Result<SyncUtxosByBlockResponse, Error> {	×
165	// validate and fetch inputs	×
166	request.validate()?;	×
167
168	let hash = request.start_header_hash.clone().try_into()?;	×
169
170	let start_header = self	×
171	.db()	×
172	.fetch_header_by_block_hash(hash)	×
173	.await?	×
174	.ok_or_else(\|\| Error::StartHeaderHashNotFound)?;	×
175
176	let hash = request.end_header_hash.clone().try_into()?;	×
177
178	let end_header = self	×
179	.db	×
180	.fetch_header_by_block_hash(hash)	×
181	.await?	×
182	.ok_or_else(\|\| Error::EndHeaderHashNotFound)?;	×
183
184	// pagination
185	let start_header_height = start_header.height + (request.page * request.limit);	×
186	let start_header = self	×
187	.db	×
188	.fetch_header(start_header_height)	×
189	.await?	×
190	.ok_or_else(\|\| Error::HeaderNotFound {	×
191	height: start_header_height,	×
192	})?;	×
193
194	if start_header.height > end_header.height {	×
195	return Err(Error::HeaderHeightMismatch {	×
196	start_height: start_header.height,	×
197	end_height: end_header.height,	×
198	});	×
199	}	×
200		×
201	// fetch utxos	×
202	let mut utxos = vec![];	×
203	let mut current_header = start_header;	×
204	let mut fetched_utxos = 0;	×
205	loop {
206	let current_header_hash = current_header.hash();	×
207		×
208	trace!(	×
209	target: LOG_TARGET,	×
210	"current header = {} ({})",	×
211	current_header.height,	×
212	current_header_hash.to_hex()	×
213	);
214
215	let outputs_with_statuses = self	×
216	.db	×
217	.fetch_outputs_in_block_with_spend_state(current_header.hash(), None)	×
218	.await?;	×
219
220	let outputs = outputs_with_statuses	×
221	.into_iter()	×
222	.map(\|(output, _spent)\| output)	×
223	.collect::<Vec<TransactionOutput>>();	×
224
225	for output_chunk in outputs.chunks(2000) {	×
226	let output_block_response = BlockUtxoInfo {	×
227	outputs: output_chunk	×
228	.iter()	×
229	.map(\|output\| MinimalUtxoSyncInfo {	×
230	output_hash: output.hash().to_vec(),	×
231	commitment: output.commitment().to_vec(),	×
232	encrypted_data: output.encrypted_data().as_bytes().to_vec(),	×
233	sender_offset_public_key: output.sender_offset_public_key.to_vec(),	×
234	})	×
235	.collect(),	×
236	height: current_header.height,	×
237	header_hash: current_header_hash.to_vec(),	×
238	mined_timestamp: current_header.timestamp.as_u64(),	×
239	};	×
240	utxos.push(output_block_response);	×
241	}	×
242	if outputs.is_empty() {	×
243	// if its empty, we need to send an empty vec of outputs.	×
244	let utxo_block_response = BlockUtxoInfo {	×
245	outputs: Vec::new(),	×
246	height: current_header.height,	×
247	header_hash: current_header_hash.to_vec(),	×
248	mined_timestamp: current_header.timestamp.as_u64(),	×
249	};	×
250	utxos.push(utxo_block_response);	×
251	}	×
252
253	fetched_utxos += 1;	×
254		×
255	if current_header.height >= end_header.height \|\| fetched_utxos >= request.limit {	×
256	break;	×
257	}	×
258
259	current_header =	×
260	self.db	×
261	.fetch_header(current_header.height + 1)	×
262	.await?	×
263	.ok_or_else(\|\| Error::HeaderNotFound {	×
264	height: current_header.height + 1,	×
265	})?;	×
266	}
267
268	let has_next_page = (end_header.height - current_header.height) > 0;	×
269		×
270	Ok(SyncUtxosByBlockResponse {	×
271	blocks: utxos,	×
272	has_next_page,	×
273	})	×
274	}	×
275	}
276
277	#[async_trait::async_trait]
278	impl<B: BlockchainBackend + 'static> BaseNodeWalletQueryService for Service<B> {
279	type Error = Error;
280
281	async fn get_tip_info(&self) -> Result<TipInfoResponse, Self::Error> {	×
282	let state_machine = self.state_machine();	×
283	let status_watch = state_machine.get_status_info_watch();	×
284	let is_synced = match status_watch.borrow().state_info {	×
285	StateInfo::Listening(li) => li.is_synced(),	×
286	_ => false,	×
287	};
288
289	let metadata = self.db.get_chain_metadata().await?;	×
290
291	Ok(TipInfoResponse {	×
292	metadata: Some(metadata),	×
293	is_synced,	×
294	})	×
295	}	×
296
297	async fn get_header_by_height(&self, height: u64) -> Result<models::BlockHeader, Self::Error> {	×
298	let result = self	×
299	.db	×
300	.fetch_header(height)	×
301	.await?	×
302	.ok_or(Error::HeaderNotFound { height })?	×
303	.into();	×
304	Ok(result)	×
305	}	×
306
307	async fn get_height_at_time(&self, epoch_time: u64) -> Result<u64, Self::Error> {	×
308	trace!(target: LOG_TARGET, "requested_epoch_time: {}", epoch_time);	×
309	let tip_header = self.db.fetch_tip_header().await?;	×
310
311	let mut left_height = 0u64;	×
312	let mut right_height = tip_header.height();	×
313
314	while left_height <= right_height {	×
315	let mut mid_height = (left_height + right_height) / 2;	×
316		×
317	if mid_height == 0 {	×
318	return Ok(0u64);	×
319	}	×
320	// If the two bounds are adjacent then perform the test between the right and left sides	×
321	if left_height == mid_height {	×
322	mid_height = right_height;	×
323	}	×
324
325	let mid_header = self	×
326	.db	×
327	.fetch_header(mid_height)	×
328	.await?	×
329	.ok_or_else(\|\| Error::HeaderNotFound { height: mid_height })?;	×
330	let before_mid_header = self	×
331	.db	×
332	.fetch_header(mid_height - 1)	×
333	.await?	×
334	.ok_or_else(\|\| Error::HeaderNotFound { height: mid_height - 1 })?;	×
335	trace!(	×
336	target: LOG_TARGET,	×
337	"requested_epoch_time: {}, left: {}, mid: {}/{} ({}/{}), right: {}",	×
338	epoch_time,	×
339	left_height,	×
340	mid_height,	×
341	mid_height-1,	×
342	mid_header.timestamp.as_u64(),	×
343	before_mid_header.timestamp.as_u64(),	×
344	right_height
345	);
346	if epoch_time < mid_header.timestamp.as_u64() && epoch_time >= before_mid_header.timestamp.as_u64() {	×
347	trace!(	×
348	target: LOG_TARGET,	×
349	"requested_epoch_time: {}, selected height: {}",	×
350	epoch_time, before_mid_header.height
351	);
352	return Ok(before_mid_header.height);	×
353	} else if mid_height == right_height {	×
354	trace!(	×
355	target: LOG_TARGET,	×
356	"requested_epoch_time: {}, selected height: {}",	×
357	epoch_time, right_height
358	);
359	return Ok(right_height);	×
360	} else if epoch_time <= mid_header.timestamp.as_u64() {	×
361	right_height = mid_height;	×
362	} else {	×
363	left_height = mid_height;	×
364	}	×
365	}
366
367	Ok(0u64)	×
368	}	×
369
370	async fn transaction_query(
371	&self,
372	signature: crate::base_node::rpc::models::Signature,
373	) -> Result<TxQueryResponse, Self::Error> {	×
374	let signature = signature.try_into().map_err(Error::SignatureConversion)?;	×
375
376	let response = self.fetch_kernel(signature).await?;	×
377
378	Ok(response)	×
379	}	×
380
381	async fn sync_utxos_by_block(
382	&self,
383	request: SyncUtxosByBlockRequest,
384	) -> Result<SyncUtxosByBlockResponse, Self::Error> {	×
385	self.fetch_utxos(request).await	×
386	}	×
387
388	async fn get_utxos_by_block(
389	&self,
390	request: GetUtxosByBlockRequest,
391	) -> Result<GetUtxosByBlockResponse, Self::Error> {	×
392	self.fetch_utxos_by_block(request).await	×
393	}	×
394
395	async fn get_utxos_mined_info(
396	&self,
397	request: models::GetUtxosMinedInfoRequest,
398	) -> Result<models::GetUtxosMinedInfoResponse, Self::Error> {	×
399	request.validate()?;	×
400
401	let mut utxos = vec![];	×
402
403	let tip_header = self.db().fetch_tip_header().await?;	×
404	for hash in request.hashes {	×
405	let hash = hash.try_into()?;	×
406	let output = self.db().fetch_output(hash).await?;	×
407	if let Some(output) = output {	×
408	utxos.push(models::MinedUtxoInfo {	×
409	utxo_hash: hash.to_vec(),	×
410	mined_in_hash: output.header_hash.to_vec(),	×
411	mined_in_height: output.mined_height,	×
412	mined_in_timestamp: output.mined_timestamp,	×
413	});	×
414	}	×
415	}
416
417	Ok(models::GetUtxosMinedInfoResponse {	×
418	utxos,	×
419	best_block_hash: tip_header.hash().to_vec(),	×
420	best_block_height: tip_header.height(),	×
421	})	×
422	}	×
423
424	async fn get_utxos_deleted_info(
425	&self,
426	request: models::GetUtxosDeletedInfoRequest,
427	) -> Result<models::GetUtxosDeletedInfoResponse, Self::Error> {	×
428	request.validate()?;	×
429
430	let mut utxos = vec![];	×
431
432	let must_include_header = request.must_include_header.clone().try_into()?;	×
433	if self	×
434	.db()	×
435	.fetch_header_by_block_hash(must_include_header)	×
436	.await?	×
437	.is_none()	×
438	{
439	return Err(Error::HeaderHashNotFound);	×
440	}	×
441
442	let tip_header = self.db().fetch_tip_header().await?;	×
443	for hash in request.hashes {	×
444	let hash = hash.try_into()?;	×
445	let output = self.db().fetch_output(hash).await?;	×
446
447	if let Some(output) = output {	×
448	// is it still unspent?
449	let input = self.db().fetch_input(hash).await?;	×
450	if let Some(i) = input {	×
451	utxos.push(models::DeletedUtxoInfo {	×
452	utxo_hash: hash.to_vec(),	×
453	found_in_header: Some((output.mined_height, output.header_hash.to_vec())),	×
454	spent_in_header: Some((i.spent_height, i.header_hash.to_vec())),	×
455	});	×
456	} else {	×
457	utxos.push(models::DeletedUtxoInfo {	×
458	utxo_hash: hash.to_vec(),	×
459	found_in_header: Some((output.mined_height, output.header_hash.to_vec())),	×
460	spent_in_header: None,	×
461	});	×
462	}	×
463	} else {	×
464	utxos.push(models::DeletedUtxoInfo {	×
465	utxo_hash: hash.to_vec(),	×
466	found_in_header: None,	×
467	spent_in_header: None,	×
468	});	×
469	}	×
470	}
471
472	Ok(models::GetUtxosDeletedInfoResponse {	×
473	utxos,	×
474	best_block_hash: tip_header.hash().to_vec(),	×
475	best_block_height: tip_header.height(),	×
476	})	×
477	}	×
478	}

tari-project / tari / 16123384529

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