19257987480

Committed 11 Nov 2025 07:12AM UTC coverage: 50.496% (-1.1%) from 51.608%

Build # 19257987480

Build Type

push

github

Committed by

web-flow

Commit Message

fix(sidechain)!: adds shard group accumulated data to checkpoint (#7577)

Description
---
fix(sidechain)!: adds shard group accumulated data to checkpoint

Motivation and Context
---
This PR adds accumulated data typ to the sidechain header used for epoch
checkpoints.

Related https://github.com/tari-project/tari-ootle/pull/1639

How Has This Been Tested?
---
Updated fixtures for tests

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

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

- [ ] None
- [ ] Requires data directory on base node to be deleted
- [ ] Requires hard fork
- [x] 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 -->

BREAKING CHANGE: Previous commit proofs (epoch checkpoints, etc.) are
not compatible. These are not enabled on mainnet.


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

## Summary by CodeRabbit

## Release Notes

* **New Features**
* Added accumulated exhaust burn tracking to sidechain block headers,
enabling storage and retrieval of total exhaust burn metrics within
block validation data.

* **Chores**
* Updated protocol definitions and conversion logic to support the new
accumulated data structure.
* Refreshed test fixtures and block header schemas to accommodate
expanded metadata requirements.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Run Details

1 of 20 new or added lines in 2 files covered. (5.0%)

1479 existing lines in 36 files now uncovered.

57906 of 114674 relevant lines covered (50.5%)

8050.74 hits per line

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

5.22

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

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

use std::cmp;

use log::trace;
use serde_valid::{validation, Validate};
use tari_common_types::{
    types,
    types::{FixedHash, FixedHashSizeError},
};
use tari_transaction_components::{
    rpc::{
        models,
        models::{
            BlockUtxoInfo,
            GenerateKernelMerkleProofResponse,
            GetUtxosByBlockRequest,
            GetUtxosByBlockResponse,
            MinimalUtxoSyncInfo,
            SyncUtxosByBlockRequest,
            SyncUtxosByBlockResponse,
            TipInfoResponse,
            TxLocation,
            TxQueryResponse,
        },
    },
    transaction_components::TransactionOutput,
};
use tari_utilities::{hex::Hex, ByteArray, ByteArrayError};
use thiserror::Error;

use crate::{
    base_node::{rpc::BaseNodeWalletQueryService, state_machine_service::states::StateInfo, StateMachineHandle},
    chain_storage::{async_db::AsyncBlockchainDb, BlockchainBackend, ChainStorageError},
    mempool::{service::MempoolHandle, MempoolServiceError, TxStorageResponse},
};

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

#[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 },
    #[error("A general error occurred: {0}")]
    General(anyhow::Error),
}

impl Error {
    fn general(err: impl Into<anyhow::Error>) -> Self {
        Error::General(err.into())
    }
}

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::CompressedSignature) -> 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)
    }

    #[allow(clippy::too_many_lines)]
    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 tip_header = self.db.fetch_tip_header().await?;
        // we only allow wallets to ask for a max of 100 blocks at a time and we want to cache the queries to ensure
        // they are in batch of 100 and we want to ensure they request goes to the nearest 100 block height so
        // we can cache all wallet's queries
        let increase = ((start_header.height + 100) / 100) * 100;
        let end_height = cmp::min(tip_header.header().height, increase);

        // pagination
        let start_header_height = start_header.height + (request.page * request.limit);
        if start_header_height > tip_header.header().height {
            return Err(Error::HeaderHeightMismatch {
                start_height: start_header.height,
                end_height: tip_header.header().height,
            });
        }
        let start_header = self
            .db
            .fetch_header(start_header_height)
            .await?
            .ok_or_else(|| Error::HeaderNotFound {
                height: start_header_height,
            })?;
        // fetch utxos
        let mut utxos = vec![];
        let mut current_header = start_header;
        let mut fetched_utxos = 0;
        let spending_end_header_hash = self
            .db
            .fetch_header(
                tip_header
                    .header()
                    .height
                    .saturating_sub(SYNC_UTXOS_SPEND_TIP_SAFETY_LIMIT),
            )
            .await?
            .ok_or_else(|| Error::HeaderNotFound {
                height: tip_header
                    .header()
                    .height
                    .saturating_sub(SYNC_UTXOS_SPEND_TIP_SAFETY_LIMIT),
            })?
            .hash();
        let next_header_to_request;
        let mut has_next_page = false;
        loop {
            let current_header_hash = current_header.hash();
            trace!(
                target: LOG_TARGET,
                "current header = {} ({})",
                current_header.height,
                current_header_hash.to_hex()
            );
            let outputs = if request.exclude_spent {
                self.db
                    .fetch_outputs_in_block_with_spend_state(current_header_hash, Some(spending_end_header_hash))
                    .await?
                    .into_iter()
                    .filter(|(_, spent)| !spent)
                    .map(|(output, _spent)| output)
                    .collect::<Vec<TransactionOutput>>()
            } else {
                self.db
                    .fetch_outputs_in_block_with_spend_state(current_header_hash, None)
                    .await?
                    .into_iter()
                    .map(|(output, _spent)| output)
                    .collect::<Vec<TransactionOutput>>()
            };
            let mut inputs = self
                .db
                .fetch_inputs_in_block(current_header_hash)
                .await?
                .into_iter()
                .map(|input| input.output_hash())
                .collect::<Vec<FixedHash>>();
            for output_chunk in outputs.chunks(2000) {
                let inputs_to_send = if inputs.is_empty() {
                    Vec::new()
                } else {
                    let num_to_drain = inputs.len().min(2000);
                    inputs.drain(..num_to_drain).map(|h| h.to_vec()).collect()
                };

                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(),
                    inputs: inputs_to_send,
                    height: current_header.height,
                    header_hash: current_header_hash.to_vec(),
                    mined_timestamp: current_header.timestamp.as_u64(),
                };
                utxos.push(output_block_response);
            }
            // We might still have inputs left to send if they are more than the outputs
            for input_chunk in inputs.chunks(2000) {
                let output_block_response = BlockUtxoInfo {
                    outputs: Vec::new(),
                    inputs: input_chunk.iter().map(|h| h.to_vec()).collect::<Vec<_>>().to_vec(),
                    height: current_header.height,
                    header_hash: current_header_hash.to_vec(),
                    mined_timestamp: current_header.timestamp.as_u64(),
                };
                utxos.push(output_block_response);
            }

            fetched_utxos += 1;

            if current_header.height >= tip_header.header().height {
                next_header_to_request = vec![];
                has_next_page = (end_height.saturating_sub(current_header.height)) > 0;
                break;
            }
            if fetched_utxos >= request.limit {
                next_header_to_request = current_header_hash.to_vec();
                // This is a special edge case, our request has reached the page limit, but we are also not done with
                // the block. We also dont want to split up the block over two requests. So we need to ensure that we
                // remove the partial block we added so that it can be requested fully in the next request. We also dont
                // want to get in a loop where the block cannot fit into the page limit, so if the block is the same as
                // the first one, we just send it as is, partial. If not we remove it and let it be sent in the next
                // request.
                if utxos.first().ok_or(Error::General(anyhow::anyhow!("No utxos founds")))? // should never happen as we always add at least one block
                    .header_hash ==
                    current_header_hash.to_vec()
                {
                    // special edge case where the first block is also the last block we can send, so we just send it as
                    // is, partial
                    break;
                }
                while !utxos.is_empty() &&
                    utxos.last().ok_or(Error::General(anyhow::anyhow!("No utxos found")))? // should never happen as we always add at least one block
                    .header_hash ==
                        current_header_hash.to_vec()
                {
                    utxos.pop();
                }
                break;
            }

            current_header =
                self.db
                    .fetch_header(current_header.height + 1)
                    .await?
                    .ok_or_else(|| Error::HeaderNotFound {
                        height: current_header.height + 1,
                    })?;
            if current_header.height == end_height {
                next_header_to_request = current_header.hash().to_vec();
                has_next_page = (end_height.saturating_sub(current_header.height)) > 0;
                break; // Stop if we reach the end height
            }
        }

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

#[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: {epoch_time}, selected height: {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(),
        })
    }

    async fn generate_kernel_merkle_proof(
        &self,
        excess_sig: types::CompressedSignature,
    ) -> Result<GenerateKernelMerkleProofResponse, Self::Error> {
        let proof = self.db().generate_kernel_merkle_proof(excess_sig).await?;
        Ok(GenerateKernelMerkleProofResponse {
            encoded_merkle_proof: bincode::serialize(&proof.merkle_proof).map_err(Error::general)?,
            block_hash: proof.block_hash,
            leaf_index: proof.leaf_index.value() as u64,
        })
    }
}

#[cfg(test)]
mod tests {
    use tari_common::configuration::Network;
    use tari_shutdown::Shutdown;

    use super::*;
    use crate::test_helpers::blockchain::create_new_blockchain_with_network;
    fn make_state_machine_handle() -> StateMachineHandle {
        use tokio::sync::{broadcast, watch};
        let (state_tx, _state_rx) = broadcast::channel(10);
        let (_status_tx, status_rx) =
            watch::channel(crate::base_node::state_machine_service::states::StatusInfo::new());
        let shutdown = Shutdown::new();
        StateMachineHandle::new(state_tx, status_rx, shutdown.to_signal())
    }

    fn make_mempool_handle() -> MempoolHandle {
        use crate::mempool::test_utils::mock::create_mempool_service_mock;
        let (handle, _state) = create_mempool_service_mock();
        handle
    }

    async fn make_service() -> Service<crate::test_helpers::blockchain::TempDatabase> {
        let db = create_new_blockchain_with_network(Network::default());
        let adb = AsyncBlockchainDb::from(db);
        let state_machine = make_state_machine_handle();
        let mempool = make_mempool_handle();
        Service::new(adb, state_machine, mempool)
    }

    #[tokio::test]
    async fn fetch_utxos_start_header_not_found() {
        let service = make_service().await;
        let req = SyncUtxosByBlockRequest {
            start_header_hash: vec![0xAB; 32],
            limit: 4,
            page: 0,
            exclude_spent: false,
        };
        let err = service.fetch_utxos(req).await.unwrap_err();
        match err {
            Error::StartHeaderHashNotFound => {},
            other => panic!("unexpected error: {other:?}"),
        }
    }

    #[tokio::test]
    async fn fetch_utxos_header_height_mismatch() {
        let service = make_service().await;
        let genesis = service.db().fetch_header(0).await.unwrap().unwrap();
        // page * limit moves start height beyond tip (0)
        let req = SyncUtxosByBlockRequest {
            start_header_hash: genesis.hash().to_vec(),
            limit: 1,
            page: 1,
            exclude_spent: false,
        };
        let err = service.fetch_utxos(req).await.unwrap_err();
        match err {
            Error::HeaderHeightMismatch { .. } => {},
            other => panic!("unexpected error: {other:?}"),
        }
    }
}

1	// Copyright 2025 The Tari Project
2	// SPDX-License-Identifier: BSD-3-Clause
3
4	use std::cmp;
5
6	use log::trace;
7	use serde_valid::{validation, Validate};
8	use tari_common_types::{
9	types,
10	types::{FixedHash, FixedHashSizeError},
11	};
12	use tari_transaction_components::{
13	rpc::{
14	models,
15	models::{
16	BlockUtxoInfo,
17	GenerateKernelMerkleProofResponse,
18	GetUtxosByBlockRequest,
19	GetUtxosByBlockResponse,
20	MinimalUtxoSyncInfo,
21	SyncUtxosByBlockRequest,
22	SyncUtxosByBlockResponse,
23	TipInfoResponse,
24	TxLocation,
25	TxQueryResponse,
26	},
27	},
28	transaction_components::TransactionOutput,
29	};
30	use tari_utilities::{hex::Hex, ByteArray, ByteArrayError};
31	use thiserror::Error;
32
33	use crate::{
34	base_node::{rpc::BaseNodeWalletQueryService, state_machine_service::states::StateInfo, StateMachineHandle},
35	chain_storage::{async_db::AsyncBlockchainDb, BlockchainBackend, ChainStorageError},
36	mempool::{service::MempoolHandle, MempoolServiceError, TxStorageResponse},
37	};
38
39	const LOG_TARGET: &str = "c::bn::rpc::query_service";
40	const SYNC_UTXOS_SPEND_TIP_SAFETY_LIMIT: u64 = 1000;
41
42	#[derive(Debug, Error)]
43	pub enum Error {
44	#[error("Failed to get chain metadata: {0}")]
45	FailedToGetChainMetadata(#[from] ChainStorageError),
46	#[error("Header not found at height: {height}")]
47	HeaderNotFound { height: u64 },
48	#[error("Signature conversion error: {0}")]
49	SignatureConversion(ByteArrayError),
50	#[error("Mempool service error: {0}")]
51	MempoolService(#[from] MempoolServiceError),
52	#[error("Serde validation error: {0}")]
53	SerdeValidation(#[from] validation::Errors),
54	#[error("Hash conversion error: {0}")]
55	HashConversion(#[from] FixedHashSizeError),
56	#[error("Start header hash not found")]
57	StartHeaderHashNotFound,
58	#[error("End header hash not found")]
59	EndHeaderHashNotFound,
60	#[error("Header hash not found")]
61	HeaderHashNotFound,
62	#[error("Start header height {start_height} cannot be greater than the end header height {end_height}")]
63	HeaderHeightMismatch { start_height: u64, end_height: u64 },
64	#[error("A general error occurred: {0}")]
65	General(anyhow::Error),
66	}
67
68	impl Error {
69	fn general(err: impl Into<anyhow::Error>) -> Self {	×
70	Error::General(err.into())	×
71	}	×
72	}
73
74	pub struct Service<B> {
75	db: AsyncBlockchainDb<B>,
76	state_machine: StateMachineHandle,
77	mempool: MempoolHandle,
78	}
79
80	impl<B: BlockchainBackend + 'static> Service<B> {
UNCOV 81	pub fn new(db: AsyncBlockchainDb<B>, state_machine: StateMachineHandle, mempool: MempoolHandle) -> Self {	×
UNCOV 82	Self {	×
UNCOV 83	db,	×
UNCOV 84	state_machine,	×
UNCOV 85	mempool,	×
UNCOV 86	}	×
UNCOV 87	}	×
88
89	fn state_machine(&self) -> StateMachineHandle {	×
90	self.state_machine.clone()	×
91	}	×
92
UNCOV 93	fn db(&self) -> &AsyncBlockchainDb<B> {	×
UNCOV 94	&self.db	×
UNCOV 95	}	×
96
97	fn mempool(&self) -> MempoolHandle {	×
98	self.mempool.clone()	×
99	}	×
100
101	async fn fetch_kernel(&self, signature: types::CompressedSignature) -> Result<TxQueryResponse, Error> {	×
102	let db = self.db();	×
103
104	match db.fetch_kernel_by_excess_sig(signature.clone()).await? {	×
105	None => (),	×
106	Some((_, block_hash)) => match db.fetch_header_by_block_hash(block_hash).await? {	×
107	None => (),	×
108	Some(header) => {	×
109	let response = TxQueryResponse {	×
110	location: TxLocation::Mined,	×
111	mined_header_hash: Some(block_hash.to_vec()),	×
112	mined_height: Some(header.height),	×
113	mined_timestamp: Some(header.timestamp.as_u64()),	×
114	};	×
115	return Ok(response);	×
116	},
117	},
118	};
119
120	// If not in a block then check the mempool
121	let mut mempool = self.mempool();	×
122	let mempool_response = match mempool.get_tx_state_by_excess_sig(signature.clone()).await? {	×
123	TxStorageResponse::UnconfirmedPool => TxQueryResponse {	×
124	location: TxLocation::InMempool,	×
125	mined_header_hash: None,	×
126	mined_height: None,	×
127	mined_timestamp: None,	×
128	},	×
129	TxStorageResponse::ReorgPool \|
130	TxStorageResponse::NotStoredOrphan \|
131	TxStorageResponse::NotStoredTimeLocked \|
132	TxStorageResponse::NotStoredAlreadySpent \|
133	TxStorageResponse::NotStoredConsensus \|
134	TxStorageResponse::NotStored \|
135	TxStorageResponse::NotStoredFeeTooLow \|
136	TxStorageResponse::NotStoredAlreadyMined => TxQueryResponse {	×
137	location: TxLocation::NotStored,	×
138	mined_timestamp: None,	×
139	mined_height: None,	×
140	mined_header_hash: None,	×
141	},	×
142	};
143
144	Ok(mempool_response)	×
145	}	×
146
147	async fn fetch_utxos_by_block(&self, request: GetUtxosByBlockRequest) -> Result<GetUtxosByBlockResponse, Error> {	×
148	request.validate()?;	×
149
150	let hash = request.header_hash.clone().try_into()?;	×
151
152	let header = self	×
153	.db()	×
154	.fetch_header_by_block_hash(hash)	×
155	.await?	×
156	.ok_or_else(\|\| Error::HeaderHashNotFound)?;	×
157
158	// fetch utxos
159	let outputs_with_statuses = self.db.fetch_outputs_in_block_with_spend_state(hash, None).await?;	×
160
161	let outputs = outputs_with_statuses	×
162	.into_iter()	×
163	.map(\|(output, _spent)\| output)	×
164	.collect::<Vec<TransactionOutput>>();	×
165
166	// if its empty, we need to send an empty vec of outputs.
167	let utxo_block_response = GetUtxosByBlockResponse {	×
168	outputs,	×
169	height: header.height,	×
170	header_hash: hash.to_vec(),	×
171	mined_timestamp: header.timestamp.as_u64(),	×
172	};	×
173
174	Ok(utxo_block_response)	×
175	}	×
176
177	#[allow(clippy::too_many_lines)]
UNCOV 178	async fn fetch_utxos(&self, request: SyncUtxosByBlockRequest) -> Result<SyncUtxosByBlockResponse, Error> {	×
179	// validate and fetch inputs
UNCOV 180	request.validate()?;	×
181
UNCOV 182	let hash = request.start_header_hash.clone().try_into()?;	×
UNCOV 183	let start_header = self	×
UNCOV 184	.db()	×
UNCOV 185	.fetch_header_by_block_hash(hash)	×
UNCOV 186	.await?	×
UNCOV 187	.ok_or_else(\|\| Error::StartHeaderHashNotFound)?;	×
188
UNCOV 189	let tip_header = self.db.fetch_tip_header().await?;	×
190	// we only allow wallets to ask for a max of 100 blocks at a time and we want to cache the queries to ensure
191	// they are in batch of 100 and we want to ensure they request goes to the nearest 100 block height so
192	// we can cache all wallet's queries
UNCOV 193	let increase = ((start_header.height + 100) / 100) * 100;	×
UNCOV 194	let end_height = cmp::min(tip_header.header().height, increase);	×
195
196	// pagination
UNCOV 197	let start_header_height = start_header.height + (request.page * request.limit);	×
UNCOV 198	if start_header_height > tip_header.header().height {	×
UNCOV 199	return Err(Error::HeaderHeightMismatch {	×
UNCOV 200	start_height: start_header.height,	×
UNCOV 201	end_height: tip_header.header().height,	×
UNCOV 202	});	×
203	}	×
204	let start_header = self	×
205	.db	×
206	.fetch_header(start_header_height)	×
207	.await?	×
208	.ok_or_else(\|\| Error::HeaderNotFound {	×
209	height: start_header_height,	×
210	})?;	×
211	// fetch utxos
212	let mut utxos = vec![];	×
213	let mut current_header = start_header;	×
214	let mut fetched_utxos = 0;	×
215	let spending_end_header_hash = self	×
216	.db	×
217	.fetch_header(	×
218	tip_header	×
219	.header()	×
220	.height	×
221	.saturating_sub(SYNC_UTXOS_SPEND_TIP_SAFETY_LIMIT),	×
222	)	×
223	.await?	×
224	.ok_or_else(\|\| Error::HeaderNotFound {	×
225	height: tip_header	×
226	.header()	×
227	.height	×
228	.saturating_sub(SYNC_UTXOS_SPEND_TIP_SAFETY_LIMIT),	×
229	})?	×
230	.hash();	×
231	let next_header_to_request;
232	let mut has_next_page = false;	×
233	loop {
234	let current_header_hash = current_header.hash();	×
235	trace!(	×
236	target: LOG_TARGET,	×
237	"current header = {} ({})",	×
238	current_header.height,
239	current_header_hash.to_hex()	×
240	);
241	let outputs = if request.exclude_spent {	×
242	self.db	×
243	.fetch_outputs_in_block_with_spend_state(current_header_hash, Some(spending_end_header_hash))	×
244	.await?	×
245	.into_iter()	×
246	.filter(\|(_, spent)\| !spent)	×
247	.map(\|(output, _spent)\| output)	×
248	.collect::<Vec<TransactionOutput>>()	×
249	} else {
250	self.db	×
251	.fetch_outputs_in_block_with_spend_state(current_header_hash, None)	×
252	.await?	×
253	.into_iter()	×
254	.map(\|(output, _spent)\| output)	×
255	.collect::<Vec<TransactionOutput>>()	×
256	};
257	let mut inputs = self	×
258	.db	×
259	.fetch_inputs_in_block(current_header_hash)	×
260	.await?	×
261	.into_iter()	×
262	.map(\|input\| input.output_hash())	×
263	.collect::<Vec<FixedHash>>();	×
264	for output_chunk in outputs.chunks(2000) {	×
265	let inputs_to_send = if inputs.is_empty() {	×
266	Vec::new()	×
267	} else {
268	let num_to_drain = inputs.len().min(2000);	×
269	inputs.drain(..num_to_drain).map(\|h\| h.to_vec()).collect()	×
270	};
271
272	let output_block_response = BlockUtxoInfo {	×
273	outputs: output_chunk	×
274	.iter()	×
275	.map(\|output\| MinimalUtxoSyncInfo {	×
276	output_hash: output.hash().to_vec(),	×
277	commitment: output.commitment().to_vec(),	×
278	encrypted_data: output.encrypted_data().as_bytes().to_vec(),	×
279	sender_offset_public_key: output.sender_offset_public_key.to_vec(),	×
280	})	×
281	.collect(),	×
282	inputs: inputs_to_send,	×
283	height: current_header.height,	×
284	header_hash: current_header_hash.to_vec(),	×
285	mined_timestamp: current_header.timestamp.as_u64(),	×
286	};
287	utxos.push(output_block_response);	×
288	}
289	// We might still have inputs left to send if they are more than the outputs
290	for input_chunk in inputs.chunks(2000) {	×
291	let output_block_response = BlockUtxoInfo {	×
292	outputs: Vec::new(),	×
293	inputs: input_chunk.iter().map(\|h\| h.to_vec()).collect::<Vec<_>>().to_vec(),	×
294	height: current_header.height,	×
295	header_hash: current_header_hash.to_vec(),	×
296	mined_timestamp: current_header.timestamp.as_u64(),	×
297	};
298	utxos.push(output_block_response);	×
299	}
300
301	fetched_utxos += 1;	×
302
303	if current_header.height >= tip_header.header().height {	×
304	next_header_to_request = vec![];	×
305	has_next_page = (end_height.saturating_sub(current_header.height)) > 0;	×
306	break;	×
307	}	×
308	if fetched_utxos >= request.limit {	×
309	next_header_to_request = current_header_hash.to_vec();	×
310	// This is a special edge case, our request has reached the page limit, but we are also not done with
311	// the block. We also dont want to split up the block over two requests. So we need to ensure that we
312	// remove the partial block we added so that it can be requested fully in the next request. We also dont
313	// want to get in a loop where the block cannot fit into the page limit, so if the block is the same as
314	// the first one, we just send it as is, partial. If not we remove it and let it be sent in the next
315	// request.
316	if utxos.first().ok_or(Error::General(anyhow::anyhow!("No utxos founds")))? // should never happen as we always add at least one block	×
317	.header_hash ==
318	current_header_hash.to_vec()	×
319	{
320	// special edge case where the first block is also the last block we can send, so we just send it as
321	// is, partial
322	break;	×
323	}	×
324	while !utxos.is_empty() &&	×
325	utxos.last().ok_or(Error::General(anyhow::anyhow!("No utxos found")))? // should never happen as we always add at least one block	×
326	.header_hash ==
327	current_header_hash.to_vec()	×
328	{	×
329	utxos.pop();	×
330	}	×
331	break;	×
332	}	×
333
334	current_header =	×
335	self.db	×
336	.fetch_header(current_header.height + 1)	×
337	.await?	×
338	.ok_or_else(\|\| Error::HeaderNotFound {	×
339	height: current_header.height + 1,	×
340	})?;	×
341	if current_header.height == end_height {	×
342	next_header_to_request = current_header.hash().to_vec();	×
343	has_next_page = (end_height.saturating_sub(current_header.height)) > 0;	×
344	break; // Stop if we reach the end height	×
345	}	×
346	}
347
348	Ok(SyncUtxosByBlockResponse {	×
349	blocks: utxos,	×
350	has_next_page,	×
351	next_header_to_scan: next_header_to_request,	×
352	})	×
UNCOV 353	}	×
354	}
355
356	#[async_trait::async_trait]
357	impl<B: BlockchainBackend + 'static> BaseNodeWalletQueryService for Service<B> {
358	type Error = Error;
359
360	async fn get_tip_info(&self) -> Result<TipInfoResponse, Self::Error> {	×
361	let state_machine = self.state_machine();	×
362	let status_watch = state_machine.get_status_info_watch();	×
363	let is_synced = match status_watch.borrow().state_info {	×
364	StateInfo::Listening(li) => li.is_synced(),	×
365	_ => false,	×
366	};
367
368	let metadata = self.db.get_chain_metadata().await?;	×
369
370	Ok(TipInfoResponse {	×
371	metadata: Some(metadata),	×
372	is_synced,	×
373	})	×
374	}	×
375
376	async fn get_header_by_height(&self, height: u64) -> Result<models::BlockHeader, Self::Error> {	×
377	let result = self	×
378	.db	×
379	.fetch_header(height)	×
380	.await?	×
381	.ok_or(Error::HeaderNotFound { height })?	×
382	.into();	×
383	Ok(result)	×
384	}	×
385
386	async fn get_height_at_time(&self, epoch_time: u64) -> Result<u64, Self::Error> {	×
387	trace!(target: LOG_TARGET, "requested_epoch_time: {}", epoch_time);	×
388	let tip_header = self.db.fetch_tip_header().await?;	×
389
390	let mut left_height = 0u64;	×
391	let mut right_height = tip_header.height();	×
392
393	while left_height <= right_height {	×
394	let mut mid_height = (left_height + right_height) / 2;	×
395
396	if mid_height == 0 {	×
397	return Ok(0u64);	×
398	}	×
399	// If the two bounds are adjacent then perform the test between the right and left sides
400	if left_height == mid_height {	×
401	mid_height = right_height;	×
402	}	×
403
404	let mid_header = self	×
405	.db	×
406	.fetch_header(mid_height)	×
407	.await?	×
408	.ok_or_else(\|\| Error::HeaderNotFound { height: mid_height })?;	×
409	let before_mid_header = self	×
410	.db	×
411	.fetch_header(mid_height - 1)	×
412	.await?	×
413	.ok_or_else(\|\| Error::HeaderNotFound { height: mid_height - 1 })?;	×
414	trace!(	×
415	target: LOG_TARGET,	×
416	"requested_epoch_time: {}, left: {}, mid: {}/{} ({}/{}), right: {}",	×
417	epoch_time,
418	left_height,
419	mid_height,
420	mid_height-1,	×
421	mid_header.timestamp.as_u64(),	×
422	before_mid_header.timestamp.as_u64(),	×
423	right_height
424	);
425	if epoch_time < mid_header.timestamp.as_u64() && epoch_time >= before_mid_header.timestamp.as_u64() {	×
426	trace!(	×
427	target: LOG_TARGET,	×
428	"requested_epoch_time: {}, selected height: {}",	×
429	epoch_time, before_mid_header.height
430	);
431	return Ok(before_mid_header.height);	×
432	} else if mid_height == right_height {	×
433	trace!(	×
434	target: LOG_TARGET,	×
435	"requested_epoch_time: {epoch_time}, selected height: {right_height}"	×
436	);
437	return Ok(right_height);	×
438	} else if epoch_time <= mid_header.timestamp.as_u64() {	×
439	right_height = mid_height;	×
440	} else {	×
441	left_height = mid_height;	×
442	}	×
443	}
444
445	Ok(0u64)	×
446	}	×
447
448	async fn transaction_query(
449	&self,
450	signature: crate::base_node::rpc::models::Signature,
451	) -> Result<TxQueryResponse, Self::Error> {	×
452	let signature = signature.try_into().map_err(Error::SignatureConversion)?;	×
453
454	let response = self.fetch_kernel(signature).await?;	×
455
456	Ok(response)	×
457	}	×
458
459	async fn sync_utxos_by_block(
460	&self,
461	request: SyncUtxosByBlockRequest,
462	) -> Result<SyncUtxosByBlockResponse, Self::Error> {	×
463	self.fetch_utxos(request).await	×
464	}	×
465
466	async fn get_utxos_by_block(
467	&self,
468	request: GetUtxosByBlockRequest,
469	) -> Result<GetUtxosByBlockResponse, Self::Error> {	×
470	self.fetch_utxos_by_block(request).await	×
471	}	×
472
473	async fn get_utxos_mined_info(
474	&self,
475	request: models::GetUtxosMinedInfoRequest,
476	) -> Result<models::GetUtxosMinedInfoResponse, Self::Error> {	×
477	request.validate()?;	×
478
479	let mut utxos = vec![];	×
480
481	let tip_header = self.db().fetch_tip_header().await?;	×
482	for hash in request.hashes {	×
483	let hash = hash.try_into()?;	×
484	let output = self.db().fetch_output(hash).await?;	×
485	if let Some(output) = output {	×
486	utxos.push(models::MinedUtxoInfo {	×
487	utxo_hash: hash.to_vec(),	×
488	mined_in_hash: output.header_hash.to_vec(),	×
489	mined_in_height: output.mined_height,	×
490	mined_in_timestamp: output.mined_timestamp,	×
491	});	×
492	}	×
493	}
494
495	Ok(models::GetUtxosMinedInfoResponse {	×
496	utxos,	×
497	best_block_hash: tip_header.hash().to_vec(),	×
498	best_block_height: tip_header.height(),	×
499	})	×
500	}	×
501
502	async fn get_utxos_deleted_info(
503	&self,
504	request: models::GetUtxosDeletedInfoRequest,
505	) -> Result<models::GetUtxosDeletedInfoResponse, Self::Error> {	×
506	request.validate()?;	×
507
508	let mut utxos = vec![];	×
509
510	let must_include_header = request.must_include_header.clone().try_into()?;	×
511	if self	×
512	.db()	×
513	.fetch_header_by_block_hash(must_include_header)	×
514	.await?	×
515	.is_none()	×
516	{
517	return Err(Error::HeaderHashNotFound);	×
518	}	×
519
520	let tip_header = self.db().fetch_tip_header().await?;	×
521	for hash in request.hashes {	×
522	let hash = hash.try_into()?;	×
523	let output = self.db().fetch_output(hash).await?;	×
524
525	if let Some(output) = output {	×
526	// is it still unspent?
527	let input = self.db().fetch_input(hash).await?;	×
528	if let Some(i) = input {	×
529	utxos.push(models::DeletedUtxoInfo {	×
530	utxo_hash: hash.to_vec(),	×
531	found_in_header: Some((output.mined_height, output.header_hash.to_vec())),	×
532	spent_in_header: Some((i.spent_height, i.header_hash.to_vec())),	×
533	});	×
534	} else {	×
535	utxos.push(models::DeletedUtxoInfo {	×
536	utxo_hash: hash.to_vec(),	×
537	found_in_header: Some((output.mined_height, output.header_hash.to_vec())),	×
538	spent_in_header: None,	×
539	});	×
540	}	×
541	} else {	×
542	utxos.push(models::DeletedUtxoInfo {	×
543	utxo_hash: hash.to_vec(),	×
544	found_in_header: None,	×
545	spent_in_header: None,	×
546	});	×
547	}	×
548	}
549
550	Ok(models::GetUtxosDeletedInfoResponse {	×
551	utxos,	×
552	best_block_hash: tip_header.hash().to_vec(),	×
553	best_block_height: tip_header.height(),	×
554	})	×
555	}	×
556
557	async fn generate_kernel_merkle_proof(
558	&self,
559	excess_sig: types::CompressedSignature,
560	) -> Result<GenerateKernelMerkleProofResponse, Self::Error> {	×
561	let proof = self.db().generate_kernel_merkle_proof(excess_sig).await?;	×
562	Ok(GenerateKernelMerkleProofResponse {
563	encoded_merkle_proof: bincode::serialize(&proof.merkle_proof).map_err(Error::general)?,	×
564	block_hash: proof.block_hash,	×
565	leaf_index: proof.leaf_index.value() as u64,	×
566	})
567	}	×
568	}
569
570	#[cfg(test)]
571	mod tests {
572	use tari_common::configuration::Network;
573	use tari_shutdown::Shutdown;
574
575	use super::*;
576	use crate::test_helpers::blockchain::create_new_blockchain_with_network;
UNCOV 577	fn make_state_machine_handle() -> StateMachineHandle {	×
578	use tokio::sync::{broadcast, watch};
UNCOV 579	let (state_tx, _state_rx) = broadcast::channel(10);	×
UNCOV 580	let (_status_tx, status_rx) =	×
UNCOV 581	watch::channel(crate::base_node::state_machine_service::states::StatusInfo::new());	×
UNCOV 582	let shutdown = Shutdown::new();	×
UNCOV 583	StateMachineHandle::new(state_tx, status_rx, shutdown.to_signal())	×
UNCOV 584	}	×
585
UNCOV 586	fn make_mempool_handle() -> MempoolHandle {	×
587	use crate::mempool::test_utils::mock::create_mempool_service_mock;
UNCOV 588	let (handle, _state) = create_mempool_service_mock();	×
UNCOV 589	handle	×
UNCOV 590	}	×
591
592	async fn make_service() -> Service<crate::test_helpers::blockchain::TempDatabase> {	2✔
593	let db = create_new_blockchain_with_network(Network::default());	2✔
594	let adb = AsyncBlockchainDb::from(db);	2✔
595	let state_machine = make_state_machine_handle();	2✔
596	let mempool = make_mempool_handle();	2✔
597	Service::new(adb, state_machine, mempool)	2✔
598	}	2✔
599
600	#[tokio::test]
601	async fn fetch_utxos_start_header_not_found() {	1✔
602	let service = make_service().await;	1✔
UNCOV 603	let req = SyncUtxosByBlockRequest {	×
UNCOV 604	start_header_hash: vec![0xAB; 32],	×
UNCOV 605	limit: 4,	×
UNCOV 606	page: 0,	×
UNCOV 607	exclude_spent: false,	×
UNCOV 608	};	×
UNCOV 609	let err = service.fetch_utxos(req).await.unwrap_err();	×
610	match err {	1✔
611	Error::StartHeaderHashNotFound => {},	1✔
612	other => panic!("unexpected error: {other:?}"),	1✔
613	}	1✔
614	}	1✔
615
616	#[tokio::test]
617	async fn fetch_utxos_header_height_mismatch() {	1✔
618	let service = make_service().await;	1✔
UNCOV 619	let genesis = service.db().fetch_header(0).await.unwrap().unwrap();	×
620	// page * limit moves start height beyond tip (0)
UNCOV 621	let req = SyncUtxosByBlockRequest {	×
UNCOV 622	start_header_hash: genesis.hash().to_vec(),	×
UNCOV 623	limit: 1,	×
UNCOV 624	page: 1,	×
UNCOV 625	exclude_spent: false,	×
UNCOV 626	};	×
UNCOV 627	let err = service.fetch_utxos(req).await.unwrap_err();	×
628	match err {	1✔
629	Error::HeaderHeightMismatch { .. } => {},	1✔
630	other => panic!("unexpected error: {other:?}"),	1✔
631	}	1✔
632	}	1✔
633	}

tari-project / tari / 19257987480

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