Merge pull request #1317 from Oscar-Pepper/zingo_sync_scanner

Zingo sync scanner module

Cargo.toml

@@ -21,7 +21,7 @@ resolver = "2"
 zcash_address = { git = "https://github.com/zingolabs/librustzcash.git", tag = "always_require_change" }
 zcash_client_backend = { git = "https://github.com/zingolabs/librustzcash.git", tag = "always_require_change", features = ["lightwalletd-tonic", "orchard", "transparent-inputs"] }
 zcash_encoding = { git = "https://github.com/zingolabs/librustzcash.git", tag = "always_require_change" }
-zcash_keys = { git = "https://github.com/zingolabs/librustzcash.git", tag = "always_require_change", features = ["orchard"] }
+zcash_keys = { git = "https://github.com/zingolabs/librustzcash.git", tag = "always_require_change", features = ["transparent-inputs", "sapling", "orchard" ] }
 zcash_note_encryption = "0.4"
 zcash_primitives = { git = "https://github.com/zingolabs/librustzcash.git", tag = "always_require_change" }
 zcash_proofs = { git = "https://github.com/zingolabs/librustzcash.git", tag = "always_require_change" }
@@ -89,9 +89,15 @@ tempfile = "3.3.0"
 test-case = "3.3.1"
 tokio = "1.28.2"
 tonic-build = "0.10"
+tracing = "0.1.40"
 tracing-subscriber = "0.3.15"
+memuse = "0.2.1"
 enum_dispatch = "0.3.13"
 
+# Parallel processing
+crossbeam-channel = "0.5"
+rayon = "1.5"
+
 [profile.release]
 debug = false
 

libtonode-tests/Cargo.toml

@@ -8,11 +8,13 @@ chain_generic_tests = []
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 [dependencies]
-zingolib = { path = "../zingolib", features = ["deprecations", "test-elevation"] }
+zingolib = { path = "../zingolib", features = ["deprecations", "test-elevation", "sync"] }
 zingo-status = { path = "../zingo-status" }
 zingo-testutils = { path = "../zingo-testutils" }
 zingo-testvectors = { path = "../zingo-testvectors" }
 zingoconfig = { path = "../zingoconfig" }
+zingo-netutils = { path = "../zingo-netutils" }
+zingo-sync = { path = "../zingo-sync" }
 
 zcash_primitives = { workspace = true }
 orchard = { workspace = true }
@@ -29,3 +31,4 @@ itertools = { workspace = true }
 serde_json = { workspace = true }
 http.workspace = true
 tempfile.workspace = true
+tracing-subscriber.workspace = true

libtonode-tests/tests/sync.rs

@@ -1,16 +1,26 @@
 use tempfile::TempDir;
+use zingo_netutils::GrpcConnector;
+use zingo_sync::sync::sync;
+use zingo_testutils::scenarios;
 use zingo_testvectors::seeds::HOSPITAL_MUSEUM_SEED;
 use zingoconfig::{construct_lightwalletd_uri, load_clientconfig, DEFAULT_LIGHTWALLETD_SERVER};
 use zingolib::{lightclient::LightClient, wallet::WalletBase};
 
 #[tokio::test]
-async fn sync_test() {
+async fn sync_mainnet_test() {
+    tracing_subscriber::fmt().init();
+
     let uri = construct_lightwalletd_uri(Some(DEFAULT_LIGHTWALLETD_SERVER.to_string()));
     let temp_dir = TempDir::new().unwrap();
     let temp_path = temp_dir.path().to_path_buf();
-    let config =
-        load_clientconfig(uri, Some(temp_path), zingoconfig::ChainType::Mainnet, true).unwrap();
-    let lightclient = LightClient::create_from_wallet_base_async(
+    let config = load_clientconfig(
+        uri.clone(),
+        Some(temp_path),
+        zingoconfig::ChainType::Mainnet,
+        true,
+    )
+    .unwrap();
+    let mut lightclient = LightClient::create_from_wallet_base_async(
         WalletBase::from_string(HOSPITAL_MUSEUM_SEED.to_string()),
         &config,
         2_590_000,
@@ -19,5 +29,27 @@ async fn sync_test() {
     .await
     .unwrap();
 
-    lightclient.do_sync(true).await.unwrap();
+    let client = GrpcConnector::new(uri).get_client().await.unwrap();
+
+    sync(client, &config.chain, &mut lightclient.wallet)
+        .await
+        .unwrap();
+}
+#[tokio::test]
+async fn sync_test() {
+    tracing_subscriber::fmt().init();
+
+    let (_regtest_manager, _cph, _faucet, mut recipient, _txid) =
+        scenarios::orchard_funded_recipient(5_000_000).await;
+    let uri = recipient.config().lightwalletd_uri.read().unwrap().clone();
+
+    let client = GrpcConnector::new(uri).get_client().await.unwrap();
+
+    sync(
+        client,
+        &recipient.config().chain.clone(),
+        &mut recipient.wallet,
+    )
+    .await
+    .unwrap();
 }

zingo-sync/Cargo.toml

@@ -4,12 +4,32 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
+# Zingo
 zingo-netutils = { path = "../zingo-netutils" }
 
+# Zcash
 zcash_client_backend.workspace = true
 zcash_primitives.workspace = true
+zcash_note_encryption.workspace = true
+sapling-crypto.workspace = true
+orchard.workspace = true
 
+# Async
 futures.workspace = true
 tokio.workspace = true
+
+# Client
 tonic.workspace = true
-tracing = "0.1.40"
+
+# Logging
+tracing.workspace = true
+
+# Metrics
+memuse.workspace = true
+
+# Parallel processing
+crossbeam-channel.workspace = true
+rayon.workspace = true
+
+# Minimise boilerplate
+getset.workspace = true

zingo-sync/src/client.rs

@@ -2,7 +2,13 @@
 
 use std::ops::Range;
 
-use zcash_client_backend::proto::{compact_formats::CompactBlock, service::BlockId};
+use zcash_client_backend::{
+    data_api::chain::ChainState,
+    proto::{
+        compact_formats::CompactBlock,
+        service::{BlockId, TreeState},
+    },
+};
 use zcash_primitives::consensus::BlockHeight;
 
 use tokio::sync::{mpsc::UnboundedSender, oneshot};
@@ -12,11 +18,14 @@ pub mod fetcher;
 /// Fetch requests are created and sent to the [`crate::client::fetcher::fetcher`] task when a connection to the server is required.
 ///
 /// Each variant includes a [`tokio::sync::oneshot::Sender`] for returning the fetched data to the requester.
+#[derive(Debug)]
 pub enum FetchRequest {
     /// Gets the height of the blockchain from the server.
     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..
+    TreeState(oneshot::Sender<TreeState>, BlockHeight),
 }
 
 /// Gets the height of the blockchain from the server.
@@ -48,3 +57,19 @@ pub async fn get_compact_block_range(
 
     Ok(compact_blocks)
 }
+/// Gets the frontiers for a specified block height..
+///
+/// Requires [`crate::client::fetcher::fetcher`] 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::<TreeState>();
+    fetch_request_sender
+        .send(FetchRequest::TreeState(sender, block_height))
+        .unwrap();
+    let tree_state = receiver.await.unwrap();
+    let frontiers = tree_state.to_chain_state().unwrap();
+
+    Ok(frontiers)
+}

zingo-sync/src/client/fetcher.rs

@@ -8,6 +8,7 @@ use zcash_client_backend::proto::{
     compact_formats::CompactBlock,
     service::{
         compact_tx_streamer_client::CompactTxStreamerClient, BlockId, BlockRange, ChainSpec,
+        TreeState,
     },
 };
 use zcash_primitives::consensus::BlockHeight;
@@ -52,7 +53,7 @@ async fn receive_fetch_requests(
     // if there are no fetch requests to process, sleep until the next fetch request is received
     // or channel is closed
     if fetch_request_queue.is_empty() {
-        while let Some(fetch_request) = receiver.recv().await {
+        if let Some(fetch_request) = receiver.recv().await {
             fetch_request_queue.push(fetch_request);
         }
     }
@@ -79,14 +80,12 @@ async fn receive_fetch_requests(
 // TODO: placeholder for algorythm that selects the next fetch request to be processed
 // return `None` if a fetch request could not be selected
 fn select_fetch_request(fetch_request_queue: &mut Vec<FetchRequest>) -> Option<FetchRequest> {
-    // TODO: add other fetch requests with priorities
-    let fetch_request_index = fetch_request_queue
-        .iter()
-        .enumerate()
-        .find(|(_, request)| matches!(request, FetchRequest::ChainTip(_)))
-        .map(|(index, _)| index);
-
-    fetch_request_index.map(|index| fetch_request_queue.remove(index))
+    // TODO: improve priority logic
+    if fetch_request_queue.first().is_some() {
+        Some(fetch_request_queue.remove(0))
+    } else {
+        None
+    }
 }
 
 //
@@ -96,13 +95,20 @@ async fn fetch_from_server(
 ) -> Result<(), ()> {
     match fetch_request {
         FetchRequest::ChainTip(sender) => {
+            tracing::info!("Fetching chain tip.");
             let block_id = get_latest_block(client).await;
             sender.send(block_id).unwrap();
         }
         FetchRequest::CompactBlockRange(sender, block_range) => {
+            tracing::info!("Fetching compact blocks. {:?}", &block_range);
             let compact_blocks = get_block_range(client, block_range).await;
             sender.send(compact_blocks).unwrap();
         }
+        FetchRequest::TreeState(sender, block_height) => {
+            tracing::info!("Fetching tree state. {:?}", &block_height);
+            let tree_state = get_tree_state(client, block_height).await;
+            sender.send(tree_state).unwrap();
+        }
     }
 
     Ok(())
@@ -115,7 +121,6 @@ async fn get_latest_block(
 
     client.get_latest_block(request).await.unwrap().into_inner()
 }
-
 async fn get_block_range(
     client: &mut CompactTxStreamerClient<zingo_netutils::UnderlyingService>,
     block_range: Range<BlockHeight>,
@@ -126,11 +131,11 @@ async fn get_block_range(
     let request = tonic::Request::new(BlockRange {
         start: Some(BlockId {
             height: u64::from(block_range.start),
-            hash: Vec::new(),
+            hash: vec![],
         }),
         end: Some(BlockId {
             height: u64::from(block_range.end) - 1,
-            hash: Vec::new(),
+            hash: vec![],
         }),
     });
     let mut block_stream = client.get_block_range(request).await.unwrap().into_inner();
@@ -141,3 +146,14 @@ async fn get_block_range(
 
     compact_blocks
 }
+async fn get_tree_state(
+    client: &mut CompactTxStreamerClient<zingo_netutils::UnderlyingService>,
+    block_height: BlockHeight,
+) -> TreeState {
+    let request = tonic::Request::new(BlockId {
+        height: block_height.into(),
+        hash: vec![],
+    });
+
+    client.get_tree_state(request).await.unwrap().into_inner()
+}

zingo-sync/src/interface.rs

@@ -1,12 +1,18 @@
 //! Traits for interfacing a wallet with the sync engine
 
-use crate::SyncState;
+use std::collections::HashMap;
+use std::fmt::Debug;
+
+use zcash_client_backend::keys::UnifiedFullViewingKey;
+use zcash_primitives::zip32::AccountId;
 
 /// Temporary dump for all neccessary wallet functionality for PoC
 pub trait SyncWallet {
-    /// Errors associated with interacting with the wallet data
-    type Error: std::fmt::Debug;
+    /// Errors associated with interfacing the sync engine with wallet data
+    type Error: Debug;
 
-    /// Mutable reference to the wallet sync state
-    fn set_sync_state(&mut self) -> Result<&mut SyncState, Self::Error>;
+    /// Returns all unified full viewing keys known to this wallet.
+    fn get_unified_full_viewing_keys(
+        &self,
+    ) -> Result<HashMap<AccountId, UnifiedFullViewingKey>, Self::Error>;
 }

zingo-sync/src/lib.rs

@@ -3,33 +3,9 @@
 //!
 //! Entrypoint: [`crate::sync::sync`]
 
-use zcash_client_backend::data_api::scanning::ScanRange;
-
 pub mod client;
 pub mod interface;
+#[allow(missing_docs)]
+pub mod primitives;
+pub(crate) mod scanner;
 pub mod sync;
-
-/// Encapsulates the current state of sync
-pub struct SyncState {
-    scan_ranges: Vec<ScanRange>,
-}
-
-impl SyncState {
-    /// Create new SyncState
-    pub fn new() -> Self {
-        SyncState {
-            scan_ranges: Vec::new(),
-        }
-    }
-
-    /// TODO: doc comment
-    pub fn set_scan_ranges(&mut self) -> &mut Vec<ScanRange> {
-        &mut self.scan_ranges
-    }
-}
-
-impl Default for SyncState {
-    fn default() -> Self {
-        Self::new()
-    }
-}