diff --git a/packages/zaino-state/src/backends/fetch.rs b/packages/zaino-state/src/backends/fetch.rs index 1e9867fe0..75cbe6799 100644 --- a/packages/zaino-state/src/backends/fetch.rs +++ b/packages/zaino-state/src/backends/fetch.rs @@ -1866,7 +1866,14 @@ impl LightWalletIndexer for FetchServiceSubscriber { request: GetAddressUtxosArg, ) -> Result { let taddrs = GetAddressBalanceRequest::new(request.addresses); - let utxos = self.z_get_address_utxos(taddrs).await?; + let utxos = self + .indexer + .get_address_utxos_bounded( + taddrs, + request.start_height, + (request.max_entries > 0).then_some(request.max_entries), + ) + .await?; let mut address_utxos: Vec = Vec::new(); let mut entries: u32 = 0; for utxo in utxos { @@ -1919,7 +1926,14 @@ impl LightWalletIndexer for FetchServiceSubscriber { request: GetAddressUtxosArg, ) -> Result { let taddrs = GetAddressBalanceRequest::new(request.addresses); - let utxos = self.z_get_address_utxos(taddrs).await?; + let utxos = self + .indexer + .get_address_utxos_bounded( + taddrs, + request.start_height, + (request.max_entries > 0).then_some(request.max_entries), + ) + .await?; let service_timeout = self.config.common.service.timeout; let (channel_tx, channel_rx) = mpsc::channel(self.config.common.service.channel_size as usize); diff --git a/packages/zaino-state/src/backends/state.rs b/packages/zaino-state/src/backends/state.rs index 9bf0ca4f2..0ad4af970 100644 --- a/packages/zaino-state/src/backends/state.rs +++ b/packages/zaino-state/src/backends/state.rs @@ -2512,7 +2512,14 @@ impl LightWalletIndexer for StateServiceSubscriber { request: GetAddressUtxosArg, ) -> Result { let taddrs = GetAddressBalanceRequest::new(request.addresses); - let utxos = self.z_get_address_utxos(taddrs).await?; + let utxos = self + .indexer + .get_address_utxos_bounded( + taddrs, + request.start_height, + (request.max_entries > 0).then_some(request.max_entries), + ) + .await?; let mut address_utxos: Vec = Vec::new(); let mut entries: u32 = 0; for utxo in utxos { @@ -2564,7 +2571,14 @@ impl LightWalletIndexer for StateServiceSubscriber { request: GetAddressUtxosArg, ) -> Result { let taddrs = GetAddressBalanceRequest::new(request.addresses); - let utxos = self.z_get_address_utxos(taddrs).await?; + let utxos = self + .indexer + .get_address_utxos_bounded( + taddrs, + request.start_height, + (request.max_entries > 0).then_some(request.max_entries), + ) + .await?; let service_timeout = self.config.common.service.timeout; let (channel_tx, channel_rx) = mpsc::channel(self.config.common.service.channel_size as usize); diff --git a/packages/zaino-state/src/chain_index.rs b/packages/zaino-state/src/chain_index.rs index 9e4ef163f..6319b2f0f 100644 --- a/packages/zaino-state/src/chain_index.rs +++ b/packages/zaino-state/src/chain_index.rs @@ -31,7 +31,7 @@ use arc_swap::ArcSwapOption; use futures::{FutureExt, Stream}; use hex::FromHex as _; use non_finalised_state::NonfinalizedBlockCacheSnapshot; -use source::{BlockchainSource, ValidatorConnector}; +use source::{AddressUtxosRequest, BlockchainSource, ValidatorConnector}; use tokio_stream::StreamExt; use tokio_util::sync::CancellationToken; use tracing::{info, instrument}; @@ -513,6 +513,14 @@ pub trait ChainIndex { address_strings: GetAddressBalanceRequest, ) -> impl std::future::Future, Self::Error>>; + /// Returns bounded unspent transparent outputs for the given addresses. + fn get_address_utxos_bounded( + &self, + address_strings: GetAddressBalanceRequest, + start_height: u64, + max_entries: Option, + ) -> impl std::future::Future, Self::Error>>; + // ********** Metadata methods ********** /// Returns Information about the mempool state: @@ -2301,9 +2309,24 @@ impl ChainIndex for NodeBackedChainIndexSubscriber Result, Self::Error> { + self.get_address_utxos_bounded(address_strings, 0, None) + .await + } + + /// Returns bounded unspent transparent outputs for the given addresses. + async fn get_address_utxos_bounded( + &self, + address_strings: GetAddressBalanceRequest, + start_height: u64, + max_entries: Option, ) -> Result, Self::Error> { self.source() - .get_address_utxos(address_strings) + .get_address_utxos(AddressUtxosRequest::new( + address_strings, + start_height, + max_entries, + )) .await .map_err(ChainIndexError::backing_validator) } diff --git a/packages/zaino-state/src/chain_index/source.rs b/packages/zaino-state/src/chain_index/source.rs index 4775f6b69..f1d512452 100644 --- a/packages/zaino-state/src/chain_index/source.rs +++ b/packages/zaino-state/src/chain_index/source.rs @@ -34,6 +34,42 @@ pub(crate) mod mockchain_source; pub mod validator_connector; pub use validator_connector::*; +/// Internal request for transparent UTXOs with lightwallet response bounds. +/// +/// The backing `getaddressutxos` RPC only accepts addresses, so fetch-backed +/// sources still receive the full validator response before Zaino can apply +/// these bounds. State-backed sources can apply the bounds while converting the +/// state response into RPC-compatible UTXO rows. +#[derive(Clone, Debug)] +pub struct AddressUtxosRequest { + /// Addresses to query. + pub addresses: GetAddressBalanceRequest, + /// Ignore UTXOs below this block height. + pub start_height: u64, + /// Maximum number of UTXOs to return. `None` means unrestricted. + pub max_entries: Option, +} + +impl AddressUtxosRequest { + /// Create a bounded UTXO query. + pub fn new( + addresses: GetAddressBalanceRequest, + start_height: u64, + max_entries: Option, + ) -> Self { + Self { + addresses, + start_height, + max_entries, + } + } + + /// Create the unbounded query used by the zcash-compatible RPC surface. + pub fn unbounded(addresses: GetAddressBalanceRequest) -> Self { + Self::new(addresses, 0, None) + } +} + /// A trait for accessing blockchain data from different backends. /// /// TODO: Explore whether this should be split into separate capability based traits. @@ -187,7 +223,7 @@ pub trait BlockchainSource: Clone + Send + Sync + 'static { /// async fn get_address_utxos( &self, - address_strings: GetAddressBalanceRequest, + request: AddressUtxosRequest, ) -> BlockchainSourceResult>; // ********** Utility methods ********** diff --git a/packages/zaino-state/src/chain_index/source/mockchain_source.rs b/packages/zaino-state/src/chain_index/source/mockchain_source.rs index da232e39f..21f9a5fcb 100644 --- a/packages/zaino-state/src/chain_index/source/mockchain_source.rs +++ b/packages/zaino-state/src/chain_index/source/mockchain_source.rs @@ -882,9 +882,15 @@ impl BlockchainSource for MockchainSource { async fn get_address_utxos( &self, - address_strings: GetAddressBalanceRequest, + request: AddressUtxosRequest, ) -> BlockchainSourceResult> { - let valid_addresses = address_strings.valid_addresses().map_err(|error| { + let AddressUtxosRequest { + addresses, + start_height, + max_entries, + } = request; + + let valid_addresses = addresses.valid_addresses().map_err(|error| { BlockchainSourceError::Unrecoverable(format!("invalid address: {error}")) })?; @@ -907,6 +913,10 @@ impl BlockchainSource for MockchainSource { let utxos = unspent_outputs .into_iter() + .filter(|(_outpoint, matching_output)| { + (matching_output.height.0 as u64) >= start_height + }) + .take(max_entries.map_or(usize::MAX, |limit| limit as usize)) .map(|(_outpoint, matching_output)| { GetAddressUtxos::new( matching_output.address, diff --git a/packages/zaino-state/src/chain_index/source/validator_connector.rs b/packages/zaino-state/src/chain_index/source/validator_connector.rs index c3d597870..6f9623b87 100644 --- a/packages/zaino-state/src/chain_index/source/validator_connector.rs +++ b/packages/zaino-state/src/chain_index/source/validator_connector.rs @@ -976,8 +976,14 @@ impl BlockchainSource for ValidatorConnector { async fn get_address_utxos( &self, - addresses: GetAddressBalanceRequest, + request: AddressUtxosRequest, ) -> BlockchainSourceResult> { + let AddressUtxosRequest { + addresses, + start_height, + max_entries, + } = request; + match self { ValidatorConnector::State(state) => { let mut state = state.read_state_service.clone(); @@ -997,47 +1003,79 @@ impl BlockchainSource for ValidatorConnector { let mut last_output_location = zebra_state::OutputLocation::from_usize(zebra_chain::block::Height(0), 0, 0); - Ok(utxos - .utxos() - .map( - |( - utxo_address, - utxo_hash, - utxo_output_location, - utxo_transparent_output, - )| { - assert!(utxo_output_location > &last_output_location); - last_output_location = *utxo_output_location; - GetAddressUtxos::new( - utxo_address, - *utxo_hash, - utxo_output_location.output_index(), - utxo_transparent_output.lock_script.clone(), - u64::from(utxo_transparent_output.value()), - utxo_output_location.height(), - ) - }, + let mut address_utxos = Vec::new(); + + for (utxo_address, utxo_hash, utxo_output_location, utxo_transparent_output) in + utxos.utxos() + { + assert!(utxo_output_location > &last_output_location); + last_output_location = *utxo_output_location; + + if (utxo_output_location.height().0 as u64) < start_height { + continue; + } + if let Some(limit) = max_entries { + if address_utxos.len() >= limit as usize { + break; + } + } + + address_utxos.push(GetAddressUtxos::new( + utxo_address, + *utxo_hash, + utxo_output_location.output_index(), + utxo_transparent_output.lock_script.clone(), + u64::from(utxo_transparent_output.value()), + utxo_output_location.height(), + )); + } + + Ok(address_utxos) + } + ValidatorConnector::Fetch(fetch) => { + let rpc_utxos = fetch + .get_address_utxos( + addresses + .valid_addresses() + .map_err(|_error| { + BlockchainSourceError::Unrecoverable( + "Invalid address provided".to_string(), + ) + })? + .into_iter() + .map(|address| address.to_string()) + .collect(), ) - .collect()) + .await + .map_err(|error| BlockchainSourceError::Unrecoverable(error.to_string()))?; + + let mut address_utxos = Vec::new(); + + for utxo in rpc_utxos { + let (address, tx_hash, output_index, script, satoshis, height) = + GetAddressUtxos::from(utxo).into_parts(); + + if (height.0 as u64) < start_height { + continue; + } + if let Some(limit) = max_entries { + if address_utxos.len() >= limit as usize { + break; + } + } + + address_utxos.push(GetAddressUtxos::new( + address, + tx_hash, + output_index, + script, + satoshis, + height, + )); + } + + Ok(address_utxos) } - ValidatorConnector::Fetch(fetch) => Ok(fetch - .get_address_utxos( - addresses - .valid_addresses() - .map_err(|_error| { - BlockchainSourceError::Unrecoverable( - "Invalid address provided".to_string(), - ) - })? - .into_iter() - .map(|address| address.to_string()) - .collect(), - ) - .await - .map_err(|error| BlockchainSourceError::Unrecoverable(error.to_string()))? - .into_iter() - .map(|utxos| utxos.into()) - .collect()), } } diff --git a/packages/zaino-state/src/chain_index/tests/mockchain_tests.rs b/packages/zaino-state/src/chain_index/tests/mockchain_tests.rs index 05c949757..38043d80d 100644 --- a/packages/zaino-state/src/chain_index/tests/mockchain_tests.rs +++ b/packages/zaino-state/src/chain_index/tests/mockchain_tests.rs @@ -1,7 +1,7 @@ use super::{load_test_vectors_and_sync_chain_index, MockchainMode}; use crate::{ chain_index::{ - source::mockchain_source::MockchainSource, + source::{mockchain_source::MockchainSource, AddressUtxosRequest}, tests::{ poll::poll_until, vectors::{load_test_vectors, TestVectorBlockData}, @@ -765,20 +765,75 @@ async fn get_address_utxos() { let transparent_address = faucet_transparent_address(); let expected_utxos = mockchain - .get_address_utxos(GetAddressBalanceRequest::new(vec![ - transparent_address.clone() - ])) + .get_address_utxos(AddressUtxosRequest::unbounded( + GetAddressBalanceRequest::new(vec![transparent_address.clone()]), + )) .await .unwrap(); let indexer_utxos = index_reader - .get_address_utxos(GetAddressBalanceRequest::new(vec![transparent_address])) + .get_address_utxos(GetAddressBalanceRequest::new(vec![ + transparent_address.clone() + ])) .await .unwrap(); assert!(!indexer_utxos.is_empty()); assert_eq!(indexer_utxos, expected_utxos); + let explicitly_unbounded_indexer_utxos = index_reader + .get_address_utxos_bounded( + GetAddressBalanceRequest::new(vec![transparent_address.clone()]), + 0, + None, + ) + .await + .unwrap(); + + assert_eq!(explicitly_unbounded_indexer_utxos, expected_utxos); + + assert!( + expected_utxos.len() > 1, + "test vector must include multiple UTXOs to prove bounded behavior" + ); + + let capped_indexer_utxos = index_reader + .get_address_utxos_bounded( + GetAddressBalanceRequest::new(vec![transparent_address.clone()]), + 0, + Some(1), + ) + .await + .unwrap(); + + assert_eq!( + capped_indexer_utxos, + expected_utxos.iter().take(1).cloned().collect::>() + ); + + let start_height = { + let (_, _, _, _, _, height) = expected_utxos[1].clone().into_parts(); + height.0 as u64 + }; + let expected_from_start_height = expected_utxos + .iter() + .filter(|utxo| { + let (_, _, _, _, _, height) = (*utxo).clone().into_parts(); + (height.0 as u64) >= start_height + }) + .cloned() + .collect::>(); + let height_filtered_indexer_utxos = index_reader + .get_address_utxos_bounded( + GetAddressBalanceRequest::new(vec![transparent_address]), + start_height, + None, + ) + .await + .unwrap(); + + assert_eq!(height_filtered_indexer_utxos, expected_from_start_height); + let invalid_address_result = index_reader .get_address_utxos(GetAddressBalanceRequest::new(vec![ "not_a_valid_transparent_address".to_string(), diff --git a/packages/zaino-state/src/chain_index/tests/proptest_blockgen.rs b/packages/zaino-state/src/chain_index/tests/proptest_blockgen.rs index 51740fe0a..20c3e4d67 100644 --- a/packages/zaino-state/src/chain_index/tests/proptest_blockgen.rs +++ b/packages/zaino-state/src/chain_index/tests/proptest_blockgen.rs @@ -29,7 +29,7 @@ use crate::{ chain_index::{ finalized_height_floor, non_finalised_state::ChainIndexSnapshot, - source::{BlockchainSourceResult, GetTransactionLocation}, + source::{AddressUtxosRequest, BlockchainSourceResult, GetTransactionLocation}, tests::{init_tracing, poll::poll_until, proptest_blockgen::proptest_helpers::add_segment}, types::BestChainLocation, NonFinalizedSnapshot, @@ -792,7 +792,7 @@ impl BlockchainSource for ProptestMockchain { async fn get_address_utxos( &self, - _address_strings: GetAddressBalanceRequest, + _request: AddressUtxosRequest, ) -> BlockchainSourceResult> { // todo!()