Merge pull request #1558 from Oscar-Pepper/zingo_sync_transparent_bun…

…dle_scanning

Zingo sync transparent bundle scanning

libtonode-tests/tests/sync.rs

@@ -55,16 +55,29 @@ async fn sync_test() {
     let (regtest_manager, _cph, faucet, mut recipient, _txid) =
         scenarios::orchard_funded_recipient(5_000_000).await;
     from_inputs::quick_send(
-        &recipient,
+        &faucet,
         vec![(
-            &get_base_address_macro!(&faucet, "unified"),
+            &get_base_address_macro!(&recipient, "transparent"),
             100_000,
-            Some("Outgoing decrypt test"),
+            None,
         )],
     )
     .await
     .unwrap();
+    // from_inputs::quick_send(
+    //     &recipient,
+    //     vec![(
+    //         &get_base_address_macro!(&faucet, "unified"),
+    //         100_000,
+    //         Some("Outgoing decrypt test"),
+    //     )],
+    // )
+    // .await
+    // .unwrap();
 
+    increase_server_height(&regtest_manager, 1).await;
+    recipient.do_sync(false).await.unwrap();
+    recipient.quick_shield().await.unwrap();
     increase_server_height(&regtest_manager, 1).await;
 
     let uri = recipient.config().lightwalletd_uri.read().unwrap().clone();
@@ -77,8 +90,9 @@ async fn sync_test() {
     .await
     .unwrap();
 
-    // dbg!(recipient.wallet.wallet_transactions());
+    dbg!(&recipient.wallet.wallet_transactions);
     // dbg!(recipient.wallet.wallet_blocks());
     // dbg!(recipient.wallet.nullifier_map());
+    // dbg!(recipient.wallet.outpoint_map());
     // dbg!(recipient.wallet.sync_state());
 }

zingo-sync/Cargo.toml

@@ -17,6 +17,7 @@ sapling-crypto.workspace = true
 orchard.workspace = true
 incrementalmerkletree.workspace = true
 shardtree.workspace = true
+zip32.workspace = true
 
 # Async
 futures.workspace = true

zingo-sync/src/client.rs

@@ -8,7 +8,7 @@ use zcash_client_backend::{
     data_api::chain::ChainState,
     proto::{
         compact_formats::CompactBlock,
-        service::{BlockId, TreeState},
+        service::{BlockId, GetAddressUtxosReply, TreeState},
     },
 };
 use zcash_primitives::{
@@ -27,10 +27,20 @@ pub enum FetchRequest {
     ChainTip(oneshot::Sender<BlockId>),
     /// Gets the specified range of compact blocks from the server (end exclusive).
     CompactBlockRange(oneshot::Sender<Vec<CompactBlock>>, Range<BlockHeight>),
-    /// Gets the tree states for a specified block height..
+    /// Gets the tree states for a specified block height.
     TreeState(oneshot::Sender<TreeState>, BlockHeight),
     /// Get a full transaction by txid.
     Transaction(oneshot::Sender<(Transaction, BlockHeight)>, TxId),
+    /// Get a list of unspent transparent output metadata for a given list of transparent addresses and start height.
+    UtxoMetadata(
+        oneshot::Sender<Vec<GetAddressUtxosReply>>,
+        (Vec<String>, BlockHeight),
+    ),
+    /// Get a list of transactions for a given transparent address and block range.
+    TransparentAddressTxs(
+        oneshot::Sender<Vec<(BlockHeight, Transaction)>>,
+        (String, Range<BlockHeight>),
+    ),
 }
 
 /// Gets the height of the blockchain from the server.
@@ -39,57 +49,104 @@ pub enum FetchRequest {
 pub async fn get_chain_height(
     fetch_request_sender: UnboundedSender<FetchRequest>,
 ) -> Result<BlockHeight, ()> {
-    let (sender, receiver) = oneshot::channel();
+    let (reply_sender, reply_receiver) = oneshot::channel();
     fetch_request_sender
-        .send(FetchRequest::ChainTip(sender))
+        .send(FetchRequest::ChainTip(reply_sender))
         .unwrap();
-    let chain_tip = receiver.await.unwrap();
+    let chain_tip = reply_receiver.await.unwrap();
 
     Ok(BlockHeight::from_u32(chain_tip.height as u32))
 }
+
 /// Gets the specified range of compact blocks from the server (end exclusive).
 ///
 /// Requires [`crate::client::fetch::fetch`] to be running concurrently, connected via the `fetch_request` channel.
 pub async fn get_compact_block_range(
     fetch_request_sender: UnboundedSender<FetchRequest>,
     block_range: Range<BlockHeight>,
 ) -> Result<Vec<CompactBlock>, ()> {
-    let (sender, receiver) = oneshot::channel();
+    let (reply_sender, reply_receiver) = oneshot::channel();
     fetch_request_sender
-        .send(FetchRequest::CompactBlockRange(sender, block_range))
+        .send(FetchRequest::CompactBlockRange(reply_sender, block_range))
         .unwrap();
-    let compact_blocks = receiver.await.unwrap();
+    let compact_blocks = reply_receiver.await.unwrap();
 
     Ok(compact_blocks)
 }
+
 /// Gets the frontiers for a specified block height.
 ///
 /// Requires [`crate::client::fetch::fetch`] to be running concurrently, connected via the `fetch_request` channel.
 pub async fn get_frontiers(
     fetch_request_sender: UnboundedSender<FetchRequest>,
     block_height: BlockHeight,
 ) -> Result<ChainState, ()> {
-    let (sender, receiver) = oneshot::channel();
+    let (reply_sender, reply_receiver) = oneshot::channel();
     fetch_request_sender
-        .send(FetchRequest::TreeState(sender, block_height))
+        .send(FetchRequest::TreeState(reply_sender, block_height))
         .unwrap();
-    let tree_state = receiver.await.unwrap();
+    let tree_state = reply_receiver.await.unwrap();
     let frontiers = tree_state.to_chain_state().unwrap();
 
     Ok(frontiers)
 }
+
 /// Gets a full transaction for a specified txid.
 ///
 /// Requires [`crate::client::fetch::fetch`] to be running concurrently, connected via the `fetch_request` channel.
 pub async fn get_transaction_and_block_height(
     fetch_request_sender: UnboundedSender<FetchRequest>,
     txid: TxId,
 ) -> Result<(Transaction, BlockHeight), ()> {
-    let (sender, receiver) = oneshot::channel();
+    let (reply_sender, reply_receiver) = oneshot::channel();
     fetch_request_sender
-        .send(FetchRequest::Transaction(sender, txid))
+        .send(FetchRequest::Transaction(reply_sender, txid))
         .unwrap();
-    let transaction_and_block_height = receiver.await.unwrap();
+    let transaction_and_block_height = reply_receiver.await.unwrap();
 
     Ok(transaction_and_block_height)
 }
+
+/// Gets unspent transparent output metadata for a list of `transparent addresses` from the specified `start_height`.
+///
+/// Requires [`crate::client::fetch::fetch`] to be running concurrently, connected via the `fetch_request` channel.
+pub async fn get_utxo_metadata(
+    fetch_request_sender: UnboundedSender<FetchRequest>,
+    transparent_addresses: Vec<String>,
+    start_height: BlockHeight,
+) -> Result<Vec<GetAddressUtxosReply>, ()> {
+    if transparent_addresses.is_empty() {
+        panic!("addresses must be non-empty!");
+    }
+
+    let (reply_sender, reply_receiver) = oneshot::channel();
+    fetch_request_sender
+        .send(FetchRequest::UtxoMetadata(
+            reply_sender,
+            (transparent_addresses, start_height),
+        ))
+        .unwrap();
+    let transparent_output_metadata = reply_receiver.await.unwrap();
+
+    Ok(transparent_output_metadata)
+}
+
+/// Gets transactions relevant to a given `transparent address` in the specified `block_range`.
+///
+/// Requires [`crate::client::fetch::fetch`] to be running concurrently, connected via the `fetch_request` channel.
+pub async fn get_transparent_address_transactions(
+    fetch_request_sender: UnboundedSender<FetchRequest>,
+    transparent_address: String,
+    block_range: Range<BlockHeight>,
+) -> Result<Vec<(BlockHeight, Transaction)>, ()> {
+    let (reply_sender, reply_receiver) = oneshot::channel();
+    fetch_request_sender
+        .send(FetchRequest::TransparentAddressTxs(
+            reply_sender,
+            (transparent_address, block_range),
+        ))
+        .unwrap();
+    let transactions = reply_receiver.await.unwrap();
+
+    Ok(transactions)
+}

zingo-sync/src/client/fetch.rs

@@ -8,6 +8,7 @@ use zcash_client_backend::proto::{
     compact_formats::CompactBlock,
     service::{
         compact_tx_streamer_client::CompactTxStreamerClient, BlockId, BlockRange, ChainSpec,
+        GetAddressUtxosArg, GetAddressUtxosReply, RawTransaction, TransparentAddressBlockFilter,
         TreeState, TxFilter,
     },
 };
@@ -97,7 +98,7 @@ fn select_fetch_request(fetch_request_queue: &mut Vec<FetchRequest>) -> Option<F
 //
 async fn fetch_from_server(
     client: &mut CompactTxStreamerClient<zingo_netutils::UnderlyingService>,
-    parameters: &impl consensus::Parameters,
+    consensus_parameters: &impl consensus::Parameters,
     fetch_request: FetchRequest,
 ) -> Result<(), ()> {
     match fetch_request {
@@ -118,9 +119,33 @@ async fn fetch_from_server(
         }
         FetchRequest::Transaction(sender, txid) => {
             tracing::info!("Fetching transaction. {:?}", txid);
-            let transaction = get_transaction(client, parameters, txid).await.unwrap();
+            let transaction = get_transaction(client, consensus_parameters, txid)
+                .await
+                .unwrap();
             sender.send(transaction).unwrap();
         }
+        FetchRequest::UtxoMetadata(sender, (addresses, start_height)) => {
+            tracing::info!(
+                "Fetching unspent transparent output metadata from {:?} for addresses:\n{:?}",
+                &start_height,
+                &addresses
+            );
+            let utxo_metadata = get_address_utxos(client, addresses, start_height, 0)
+                .await
+                .unwrap();
+            sender.send(utxo_metadata).unwrap();
+        }
+        FetchRequest::TransparentAddressTxs(sender, (address, block_range)) => {
+            tracing::info!(
+                "Fetching raw transactions in block range {:?} for address {:?}",
+                &block_range,
+                &address
+            );
+            let transactions = get_taddress_txs(client, consensus_parameters, address, block_range)
+                .await
+                .unwrap();
+            sender.send(transactions).unwrap();
+        }
     }
 
     Ok(())
@@ -172,7 +197,7 @@ async fn get_tree_state(
 
 async fn get_transaction(
     client: &mut CompactTxStreamerClient<zingo_netutils::UnderlyingService>,
-    parameters: &impl consensus::Parameters,
+    consensus_parameters: &impl consensus::Parameters,
     txid: TxId,
 ) -> Result<(Transaction, BlockHeight), ()> {
     let request = tonic::Request::new(TxFilter {
@@ -182,13 +207,84 @@ async fn get_transaction(
     });
 
     let raw_transaction = client.get_transaction(request).await.unwrap().into_inner();
-    let block_height = BlockHeight::from_u32(raw_transaction.height as u32);
+    let block_height = BlockHeight::from_u32(u32::try_from(raw_transaction.height).unwrap());
 
     let transaction = Transaction::read(
         &raw_transaction.data[..],
-        BranchId::for_height(parameters, block_height),
+        BranchId::for_height(consensus_parameters, block_height),
     )
     .unwrap();
 
     Ok((transaction, block_height))
 }
+
+async fn get_address_utxos(
+    client: &mut CompactTxStreamerClient<zingo_netutils::UnderlyingService>,
+    addresses: Vec<String>,
+    start_height: BlockHeight,
+    max_entries: u32,
+) -> Result<Vec<GetAddressUtxosReply>, ()> {
+    let start_height: u64 = start_height.into();
+    let request = tonic::Request::new(GetAddressUtxosArg {
+        addresses,
+        start_height,
+        max_entries,
+    });
+
+    Ok(client
+        .get_address_utxos(request)
+        .await
+        .unwrap()
+        .into_inner()
+        .address_utxos)
+}
+
+async fn get_taddress_txs(
+    client: &mut CompactTxStreamerClient<zingo_netutils::UnderlyingService>,
+    consensus_parameters: &impl consensus::Parameters,
+    address: String,
+    block_range: Range<BlockHeight>,
+) -> Result<Vec<(BlockHeight, Transaction)>, ()> {
+    let mut raw_transactions: Vec<RawTransaction> = Vec::new();
+
+    let range = Some(BlockRange {
+        start: Some(BlockId {
+            height: block_range.start.into(),
+            hash: vec![],
+        }),
+        end: Some(BlockId {
+            height: u64::from(block_range.end) - 1,
+            hash: vec![],
+        }),
+    });
+
+    let request = tonic::Request::new(TransparentAddressBlockFilter { address, range });
+
+    let mut raw_tx_stream = client
+        .get_taddress_txids(request)
+        .await
+        .unwrap()
+        .into_inner();
+
+    while let Some(raw_tx) = raw_tx_stream.message().await.unwrap() {
+        raw_transactions.push(raw_tx);
+    }
+
+    let transactions: Vec<(BlockHeight, Transaction)> = raw_transactions
+        .into_iter()
+        .map(|raw_transaction| {
+            let block_height =
+                BlockHeight::from_u32(u32::try_from(raw_transaction.height).unwrap());
+
+            let transaction = Transaction::read(
+                &raw_transaction.data[..],
+                BranchId::for_height(consensus_parameters, block_height),
+            )
+            .unwrap();
+
+            (block_height, transaction)
+        })
+        .collect();
+
+    Ok(transactions)
+}

zingo-sync/src/keys.rs

@@ -5,23 +5,30 @@ use std::collections::HashMap;
 use getset::Getters;
 use incrementalmerkletree::Position;
 use orchard::{
-    keys::{FullViewingKey, IncomingViewingKey, Scope},
+    keys::{FullViewingKey, IncomingViewingKey},
     note_encryption::OrchardDomain,
 };
 use sapling_crypto::{
     self as sapling, note_encryption::SaplingDomain, NullifierDerivingKey, SaplingIvk,
 };
 use zcash_keys::keys::UnifiedFullViewingKey;
 use zcash_note_encryption::Domain;
+use zip32::Scope;
 
+/// Child index for the `address_index` path level in the BIP44 hierarchy.
+pub type AddressIndex = u32;
+
+/// Unique ID for shielded keys.
 #[derive(Debug, Clone, Copy, Hash, PartialEq, Eq)]
 pub struct KeyId {
     account_id: zcash_primitives::zip32::AccountId,
     scope: Scope,
 }
 
+pub mod transparent;
+
 impl KeyId {
-    pub fn from_parts(account_id: zcash_primitives::zip32::AccountId, scope: Scope) -> Self {
+    pub(crate) fn from_parts(account_id: zcash_primitives::zip32::AccountId, scope: Scope) -> Self {
         Self { account_id, scope }
     }
 }
@@ -37,7 +44,7 @@ impl memuse::DynamicUsage for KeyId {
 }
 
 /// A key that can be used to perform trial decryption and nullifier
-/// computation for a [`CompactSaplingOutput`] or [`CompactOrchardAction`].
+/// computation for a CompactSaplingOutput or CompactOrchardAction.
 pub trait ScanningKeyOps<D: Domain, Nf> {
     /// Prepare the key for use in batch trial decryption.
     fn prepare(&self) -> D::IncomingViewingKey;