feat(rpc): implement the get_address_tx_ids RPC method query

zebra-rpc/src/methods.rs

@@ -647,7 +647,6 @@ where
         end: u32,
     ) -> BoxFuture<Result<Vec<String>>> {
         let mut state = self.state.clone();
-        let mut response_transactions = vec![];
         let start = Height(start);
         let end = Height(end);
 
@@ -661,7 +660,7 @@ where
             // height range checks
             check_height_range(start, end, chain_height?)?;
 
-            let valid_addresses: Result<Vec<Address>> = addresses
+            let valid_addresses: Result<HashSet<Address>> = addresses
                 .iter()
                 .map(|address| {
                     address.parse().map_err(|_| {
@@ -670,8 +669,10 @@ where
                 })
                 .collect();
 
-            let request =
-                zebra_state::ReadRequest::TransactionsByAddresses(valid_addresses?, start, end);
+            let request = zebra_state::ReadRequest::TransactionIdsByAddresses {
+                addresses: valid_addresses?,
+                height_range: start..=end,
+            };
             let response = state
                 .ready()
                 .and_then(|service| service.call(request))
@@ -682,13 +683,14 @@ where
                     data: None,
                 })?;
 
-            match response {
-                zebra_state::ReadResponse::TransactionIds(hashes) => response_transactions
-                    .append(&mut hashes.iter().map(|h| h.to_string()).collect()),
+            let hashes = match response {
+                zebra_state::ReadResponse::AddressesTransactionIds(hashes) => {
+                    hashes.values().map(|tx_id| tx_id.to_string()).collect()
+                }
                 _ => unreachable!("unmatched response to a TransactionsByAddresses request"),
-            }
+            };
 
-            Ok(response_transactions)
+            Ok(hashes)
         }
         .boxed()
     }

zebra-rpc/src/methods/tests/vectors.rs

@@ -1,6 +1,6 @@
 //! Fixed test vectors for RPC methods.
 
-use std::sync::Arc;
+use std::{ops::RangeInclusive, sync::Arc};
 
 use jsonrpc_core::ErrorCode;
 use tower::buffer::Buffer;
@@ -11,6 +11,7 @@ use zebra_chain::{
     parameters::Network::*,
     serialization::{ZcashDeserializeInto, ZcashSerialize},
     transaction::{UnminedTx, UnminedTxId},
+    transparent,
 };
 use zebra_network::constants::USER_AGENT;
 use zebra_node_services::BoxError;
@@ -395,47 +396,89 @@ async fn rpc_getaddresstxids_invalid_arguments() {
 async fn rpc_getaddresstxids_response() {
     zebra_test::init();
 
-    let blocks: Vec<Arc<Block>> = zebra_test::vectors::CONTINUOUS_MAINNET_BLOCKS
+    for network in [Mainnet, Testnet] {
+        let blocks: Vec<Arc<Block>> = match network {
+            Mainnet => &*zebra_test::vectors::CONTINUOUS_MAINNET_BLOCKS,
+            Testnet => &*zebra_test::vectors::CONTINUOUS_TESTNET_BLOCKS,
+        }
         .iter()
         .map(|(_height, block_bytes)| block_bytes.zcash_deserialize_into().unwrap())
         .collect();
 
-    // get the first transaction of the first block
-    let first_block_first_transaction = &blocks[1].transactions[0];
-    // get the address, this is always `t3Vz22vK5z2LcKEdg16Yv4FFneEL1zg9ojd`
-    let address = &first_block_first_transaction.outputs()[1]
-        .address(Mainnet)
-        .unwrap();
+        // The first few blocks after genesis send funds to the same founders reward address,
+        // in one output per coinbase transaction.
+        //
+        // Get the coinbase transaction of the first block
+        // (the genesis block coinbase transaction is ignored by the consensus rules).
+        let first_block_first_transaction = &blocks[1].transactions[0];
+
+        // Get the address.
+        let address = first_block_first_transaction.outputs()[1]
+            .address(network)
+            .unwrap();
+
+        if network == Mainnet {
+            // Exhaustively test possible block ranges for mainnet.
+            //
+            // TODO: if it takes too long on slower machines, turn this into a proptest with 10-20 cases
+            for start in 1..=10 {
+                for end in start..=10 {
+                    rpc_getaddresstxids_response_with(network, start..=end, &blocks, &address)
+                        .await;
+                }
+            }
+        } else {
+            // Just test the full range for testnet.
+            rpc_getaddresstxids_response_with(network, 1..=10, &blocks, &address).await;
+        }
+    }
+}
 
+async fn rpc_getaddresstxids_response_with(
+    network: Network,
+    range: RangeInclusive<u32>,
+    blocks: &[Arc<Block>],
+    address: &transparent::Address,
+) {
     let mut mempool: MockService<_, _, _, BoxError> = MockService::build().for_unit_tests();
     // Create a populated state service
     let (_state, read_state, latest_chain_tip, _chain_tip_change) =
-        zebra_state::populated_state(blocks.clone(), Mainnet).await;
+        zebra_state::populated_state(blocks.to_owned(), network).await;
 
     let (rpc, rpc_tx_queue_task_handle) = RpcImpl::new(
         "RPC test",
         Buffer::new(mempool.clone(), 1),
         Buffer::new(read_state.clone(), 1),
         latest_chain_tip,
-        Mainnet,
+        network,
     );
 
     // call the method with valid arguments
     let addresses = vec![address.to_string()];
-    let start: u32 = 1;
-    let end: u32 = 1;
     let response = rpc
-        .get_address_tx_ids(addresses, start, end)
+        .get_address_tx_ids(addresses, *range.start(), *range.end())
         .await
         .expect("arguments are valid so no error can happen here");
 
-    // TODO: The length of the response should be 1
-    // Fix in the context of #3147
-    assert_eq!(response.len(), 0);
+    // One founders reward output per coinbase transactions, no other transactions.
+    assert_eq!(response.len(), range.count());
 
     mempool.expect_no_requests().await;
 
-    // The queue task should continue without errors or panics
+    // Shut down the queue task, to close the state's file descriptors.
+    // (If we don't, opening ~100 simultaneous states causes process file descriptor limit errors.)
+    //
+    // TODO: abort all the join handles in all the tests, except one?
+    rpc_tx_queue_task_handle.abort();
+
+    // The queue task should not have panicked or exited by itself.
+    // It can still be running, or it can have exited due to the abort.
     let rpc_tx_queue_task_result = rpc_tx_queue_task_handle.now_or_never();
-    assert!(matches!(rpc_tx_queue_task_result, None));
+    assert!(
+        rpc_tx_queue_task_result.is_none()
+            || rpc_tx_queue_task_result
+                .unwrap()
+                .unwrap_err()
+                .is_cancelled()
+    );
 }

zebra-state/src/request.rs

@@ -2,6 +2,7 @@
 
 use std::{
     collections::{HashMap, HashSet},
+    ops::RangeInclusive,
     sync::Arc,
 };
 
@@ -438,19 +439,27 @@ pub enum ReadRequest {
     /// * [`Response::Transaction(None)`](Response::Transaction) otherwise.
     Transaction(transaction::Hash),
 
-    /// Looks up transactions hashes that were made by provided addresses in a blockchain height range.
+    /// Looks up the balance of a set of transparent addresses.
+    ///
+    /// Returns an [`Amount`] with the total balance of the set of addresses.
+    AddressBalance(HashSet<transparent::Address>),
+
+    /// Looks up transaction hashes that sent or received from addresses,
+    /// in an inclusive blockchain height range.
     ///
     /// Returns
     ///
-    /// * A vector of transaction hashes.
+    /// * A set of transaction hashes.
     /// * An empty vector if no transactions were found for the given arguments.
     ///
-    /// Returned txids are in the order they appear in blocks, which ensures that they are topologically sorted
+    /// Returned txids are in the order they appear in blocks,
+    /// which ensures that they are topologically sorted
     /// (i.e. parent txids will appear before child txids).
-    TransactionsByAddresses(Vec<transparent::Address>, block::Height, block::Height),
+    TransactionIdsByAddresses {
+        /// The requested addresses.
+        addresses: HashSet<transparent::Address>,
 
-    /// Looks up the balance of a set of transparent addresses.
-    ///
-    /// Returns an [`Amount`] with the total balance of the set of addresses.
-    AddressBalance(HashSet<transparent::Address>),
+        /// The blocks to be queried for transactions.
+        height_range: RangeInclusive<block::Height>,
+    },
 }

zebra-state/src/response.rs

@@ -1,18 +1,19 @@
 //! State [`tower::Service`] response types.
 
-use std::sync::Arc;
+use std::{collections::BTreeMap, sync::Arc};
 
 use zebra_chain::{
     amount::{Amount, NonNegative},
     block::{self, Block},
-    transaction::{Hash, Transaction},
+    transaction::{self, Transaction},
     transparent,
 };
 
 // Allow *only* this unused import, so that rustdoc link resolution
 // will work with inline links.
 #[allow(unused_imports)]
 use crate::Request;
+use crate::TransactionLocation;
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 /// A response to a [`StateService`] [`Request`].
@@ -55,10 +56,10 @@ pub enum ReadResponse {
     /// Response to [`ReadRequest::Transaction`] with the specified transaction.
     Transaction(Option<(Arc<Transaction>, block::Height)>),
 
-    /// Response to [`ReadRequest::TransactionsByAddresses`] with the obtained transaction ids,
-    /// in the order they appear in blocks.
-    TransactionIds(Vec<Hash>),
-
     /// Response to [`ReadRequest::AddressBalance`] with the total balance of the addresses.
     AddressBalance(Amount<NonNegative>),
+
+    /// Response to [`ReadRequest::TransactionIdsByAddresses`] with the obtained transaction ids,
+    /// in the order they appear in blocks.
+    AddressesTransactionIds(BTreeMap<TransactionLocation, transaction::Hash>),
 }

zebra-state/src/service.rs

@@ -993,24 +993,25 @@ impl Service<ReadRequest> for ReadStateService {
             }
 
             // For the get_address_tx_ids RPC.
-            ReadRequest::TransactionsByAddresses(_addresses, _start, _end) => {
+            ReadRequest::TransactionIdsByAddresses {
+                addresses,
+                height_range,
+            } => {
                 metrics::counter!(
                     "state.requests",
                     1,
                     "service" => "read_state",
-                    "type" => "transactions_by_addresses",
+                    "type" => "transaction_ids_by_addresses",
                 );
 
-                let _state = self.clone();
+                let state = self.clone();
 
                 async move {
-                    // TODO: Respond with found transactions
-                    // At least the following pull requests should be merged:
-                    // - #4022
-                    // - #4038
-                    // Do the corresponding update in the context of #3147
-                    let transaction_ids = vec![];
-                    Ok(ReadResponse::TransactionIds(transaction_ids))
+                    let tx_ids = state.best_chain_receiver.with_watch_data(|best_chain| {
+                        read::transparent_tx_ids(best_chain, &state.db, addresses, height_range)
+                    });
+
+                    tx_ids.map(ReadResponse::AddressesTransactionIds)
                 }
                 .boxed()
             }

zebra-state/src/service/finalized_state/disk_format/transparent.rs

@@ -5,7 +5,7 @@
 //! The [`crate::constants::DATABASE_FORMAT_VERSION`] constant must
 //! be incremented each time the database format (column, serialization, etc) changes.
 
-use std::fmt::Debug;
+use std::{cmp::max, fmt::Debug};
 
 use zebra_chain::{
     amount::{self, Amount, NonNegative},
@@ -396,18 +396,29 @@ impl AddressTransaction {
     }
 
     /// Create an [`AddressTransaction`] which starts iteration for the supplied address.
+    /// Starts at the first UTXO, or at the `query_start` height, whichever is greater.
+    ///
     /// Used to look up the first transaction with [`ReadDisk::zs_next_key_value_from`].
     ///
-    /// The transaction location is before all unspent output locations in the index.
-    /// It is always invalid, due to the genesis consensus rules. But this is not an issue
-    /// since [`ReadDisk::zs_next_key_value_from`] will fetch the next existing (valid) value.
-    pub fn address_iterator_start(address_location: AddressLocation) -> AddressTransaction {
+    /// The transaction location might be invalid, if it is based on the `query_start` height.
+    /// But this is not an issue, since [`ReadDisk::zs_next_key_value_from`]
+    /// will fetch the next existing (valid) value.
+    pub fn address_iterator_start(
+        address_location: AddressLocation,
+        query_start: Height,
+    ) -> AddressTransaction {
         // Iterating from the lowest possible transaction location gets us the first transaction.
-        let zero_transaction_location = TransactionLocation::from_usize(Height(0), 0);
+        //
+        // The address location is the output location of the first UTXO sent to the address,
+        // and addresses can not spend funds until they receive their first UTXO.
+        let first_utxo_location = address_location.transaction_location();
+
+        // Iterating from the start height filters out transactions that aren't needed.
+        let query_start_location = TransactionLocation::from_usize(query_start, 0);
 
         AddressTransaction {
             address_location,
-            transaction_location: zero_transaction_location,
+            transaction_location: max(first_utxo_location, query_start_location),
         }
     }
 

zebra-state/src/service/finalized_state/zebra_db/block/tests/snapshot.rs

@@ -45,11 +45,14 @@ use zebra_chain::{
 };
 
 use crate::{
-    service::finalized_state::{
-        disk_format::{
-            block::TransactionIndex, transparent::OutputLocation, FromDisk, TransactionLocation,
+    service::{
+        finalized_state::{
+            disk_format::{
+                block::TransactionIndex, transparent::OutputLocation, FromDisk, TransactionLocation,
+            },
+            FinalizedState,
         },
-        FinalizedState,
+        read::ADDRESS_HEIGHTS_FULL_RANGE,
     },
     Config,
 };
@@ -495,7 +498,9 @@ fn snapshot_transparent_address_data(state: &FinalizedState, height: u32) {
         }
 
         let mut stored_transaction_locations = Vec::new();
-        for transaction_location in state.address_transaction_locations(stored_address_location) {
+        for transaction_location in
+            state.address_transaction_locations(stored_address_location, ADDRESS_HEIGHTS_FULL_RANGE)
+        {
             assert_eq!(
                 transaction_location.address_location(),
                 stored_address_location