Merge branch 'check-existing-txs' into verify-push-transactions

zebra-consensus/Cargo.toml

@@ -20,7 +20,7 @@ serde = { version = "1", features = ["serde_derive"] }
 futures = "0.3.17"
 futures-util = "0.3.17"
 metrics = "0.13.0-alpha.8"
-thiserror = "1.0.28"
+thiserror = "1.0.29"
 tokio = { version = "0.3.6", features = ["time", "sync", "stream", "tracing"] }
 tower = { version = "0.4", features = ["timeout", "util", "buffer"] }
 tracing = "0.1.26"

zebra-script/Cargo.toml

@@ -10,7 +10,7 @@ edition = "2018"
 [dependencies]
 zcash_script = "0.1.6-alpha.0"
 zebra-chain = { path = "../zebra-chain" }
-thiserror = "1.0.28"
+thiserror = "1.0.29"
 displaydoc = "0.2.2"
 
 [dev-dependencies]

zebra-state/Cargo.toml

@@ -22,7 +22,7 @@ futures = "0.3.17"
 metrics = "0.13.0-alpha.8"
 tower = { version = "0.4", features = ["buffer", "util"] }
 tracing = "0.1"
-thiserror = "1.0.28"
+thiserror = "1.0.29"
 tokio = { version = "0.3.6", features = ["sync"] }
 displaydoc = "0.2.2"
 rocksdb = "0.16.0"

zebra-test/Cargo.toml

@@ -23,7 +23,7 @@ color-eyre = "0.5.11"
 owo-colors = "2.0.0"
 pretty_assertions = "0.7.2"
 spandoc = "0.2.0"
-thiserror = "1.0.28"
+thiserror = "1.0.29"
 tracing = "0.1.26"
 tracing-subscriber = "0.2.19"
 tracing-error = "0.1.2"

zebrad/Cargo.toml

@@ -40,7 +40,7 @@ metrics = "0.13.0-alpha.8"
 metrics-exporter-prometheus = "0.1.0-alpha.7"
 
 dirs = "3.0.2"
-inferno = { version = "0.10.6", default-features = false }
+inferno = { version = "0.10.7", default-features = false }
 atty = "0.2.14"
 
 sentry = { version = "0.21.0", default-features = false, features = ["backtrace", "contexts", "reqwest", "rustls"] }

zebrad/src/components/inbound.rs

@@ -323,16 +323,14 @@ impl Service<zn::Request> for Inbound {
                     .map_ok(zn::Response::Blocks)
                     .boxed()
             }
-            zn::Request::TransactionsById(_transactions) => {
-                // `zcashd` returns a list of found transactions, followed by a
-                // `NotFound` message if any transactions are missing. `zcashd`
-                // says that Simplified Payment Verification (SPV) clients rely on
-                // this behaviour - are there any of them on the Zcash network?
-                // https://github.com/zcash/zcash/blob/e7b425298f6d9a54810cb7183f00be547e4d9415/src/main.cpp#L5632
-                // We'll implement this request once we have a mempool:
-                // https://en.bitcoin.it/wiki/Protocol_documentation#getdata
-                debug!("ignoring unimplemented request");
-                async { Ok(zn::Response::Nil) }.boxed()
+            zn::Request::TransactionsById(transactions) => {
+                let request = mempool::Request::TransactionsById(transactions);
+                self.mempool.clone().oneshot(request).map_ok(|resp| match resp {
+                    mempool::Response::Transactions(transactions) => zn::Response::Transactions(transactions),
+                    _ => unreachable!("Mempool component should always respond to a `TransactionsById` request with a `Transactions` response"),
+                })
+                .boxed()
+
             }
             zn::Request::FindBlocks { known_blocks, stop } => {
                 let request = zs::Request::FindBlockHashes { known_blocks, stop };

zebrad/src/components/inbound/tests.rs

@@ -1,9 +1,9 @@
-use tower::{util::BoxService, ServiceExt};
+use std::collections::HashSet;
 
 use super::mempool::{unmined_transactions_in_blocks, Mempool};
 
 use tokio::sync::oneshot;
-use tower::builder::ServiceBuilder;
+use tower::{builder::ServiceBuilder, util::BoxService, ServiceExt};
 
 use zebra_chain::{
     parameters::Network,
@@ -14,7 +14,7 @@ use zebra_network::{Request, Response};
 use zebra_state::Config as StateConfig;
 
 #[tokio::test]
-async fn mempool_requests_for_transaction_ids() {
+async fn mempool_requests_for_transactions() {
     let network = Network::Mainnet;
     let consensus_config = ConsensusConfig::default();
     let state_config = StateConfig::ephemeral();
@@ -23,11 +23,8 @@ async fn mempool_requests_for_transaction_ids() {
     let state_service = ServiceBuilder::new().buffer(1).service(state);
     let mut mempool_service = Mempool::new(network);
 
-    let added_transaction_ids: Vec<UnminedTxId> =
-        add_some_stuff_to_mempool(&mut mempool_service, network)
-            .iter()
-            .map(|t| t.id)
-            .collect();
+    let added_transactions = add_some_stuff_to_mempool(&mut mempool_service, network);
+    let added_transaction_ids: Vec<UnminedTxId> = added_transactions.iter().map(|t| t.id).collect();
 
     let mempool_service = BoxService::new(mempool_service);
     let mempool = ServiceBuilder::new().buffer(20).service(mempool_service);
@@ -48,7 +45,9 @@ async fn mempool_requests_for_transaction_ids() {
             mempool,
         ));
 
+    // Test `Request::MempoolTransactionIds`
     let request = inbound_service
+        .clone()
         .oneshot(Request::MempoolTransactionIds)
         .await;
     match request {
@@ -57,6 +56,21 @@ async fn mempool_requests_for_transaction_ids() {
             "`MempoolTransactionIds` requests should always respond `Ok(Vec<UnminedTxId>)`"
         ),
     };
+
+    // Test `Request::TransactionsById`
+    let hash_set = added_transaction_ids
+        .iter()
+        .copied()
+        .collect::<HashSet<_>>();
+
+    let request = inbound_service
+        .oneshot(Request::TransactionsById(hash_set))
+        .await;
+
+    match request {
+        Ok(Response::Transactions(response)) => assert_eq!(response, added_transactions),
+        _ => unreachable!("`TransactionsById` requests should always respond `Ok(Vec<UnminedTx>)`"),
+    };
 }
 
 fn add_some_stuff_to_mempool(mempool_service: &mut Mempool, network: Network) -> Vec<UnminedTx> {

zebrad/src/components/sync/status.rs

@@ -17,6 +17,13 @@ pub struct SyncStatus {
 }
 
 impl SyncStatus {
+    /// The threshold that determines if the synchronization is at the chain
+    /// tip.
+    ///
+    /// This is based on the fact that sync lengths are around 2-20 blocks long
+    /// once Zebra reaches the tip.
+    const MIN_DIST_FROM_TIP: usize = 20;
+
     /// Create an instance of [`SyncStatus`].
     ///
     /// The status is determined based on the latest counts of synchronized blocks, observed
@@ -41,9 +48,25 @@ impl SyncStatus {
 
     /// Check if the synchronization is likely close to the chain tip.
     pub fn is_close_to_tip(&self) -> bool {
-        let _sync_lengths = self.latest_sync_length.borrow();
+        let sync_lengths = self.latest_sync_length.borrow();
+
+        // Return early if sync_lengths is empty.
+        if sync_lengths.is_empty() {
+            return false;
+        }
+
+        // Compute the sum of the `sync_lengths`.
+        // The sum is computed by saturating addition in order to avoid overflowing.
+        let sum = sync_lengths
+            .iter()
+            .fold(0usize, |sum, rhs| sum.saturating_add(*rhs));
+
+        // Compute the average sync length.
+        // This value effectively represents a simple moving average.
+        let avg = sum / sync_lengths.len();
 
-        // TODO: Determine if the synchronization is actually close to the tip (#2592).
-        true
+        // The synchronization process is close to the chain tip once the
+        // average sync length falls below the threshold.
+        avg < Self::MIN_DIST_FROM_TIP
     }
 }

zebrad/src/components/sync/status/tests.rs

@@ -12,7 +12,7 @@ const DEFAULT_ASYNC_SYNCHRONIZED_TASKS_PROPTEST_CASES: u32 = 32;
 /// The maximum time one test instance should run.
 ///
 /// If the test exceeds this time it is considered to have failed.
-const MAX_TEST_EXECUTION: Duration = Duration::from_secs(1);
+const MAX_TEST_EXECUTION: Duration = Duration::from_secs(10);
 
 /// The maximum time to wait for an event to be received.
 ///
@@ -120,7 +120,7 @@ proptest! {
         ///    looping repeatedly while [`SyncStatus`] reports that it is close to the chain tip.
         /// 2. Waits until [`SyncStatus`] reports that it is close to the chain tip.
         /// 3. Notifies the main task that it awoke, i.e., that the [`SyncStatus`] has finished
-        ///    wating until it was close to the chain tip.
+        ///    waiting until it was close to the chain tip.
         async fn wait_task(
             mut status: SyncStatus,
             update_events: Arc<Semaphore>,
@@ -138,3 +138,37 @@ proptest! {
         }
     }
 }
+
+/// Test if totally empty sync lengths array is not near tip.
+#[test]
+fn empty_sync_lengths() {
+    let (status, _recent_sync_lengths) = SyncStatus::new();
+
+    assert!(!status.is_close_to_tip());
+}
+
+/// Test if sync lengths array with all zeroes is near tip.
+#[test]
+fn zero_sync_lengths() {
+    let (status, mut recent_sync_lengths) = SyncStatus::new();
+
+    for _ in 0..RecentSyncLengths::MAX_RECENT_LENGTHS {
+        recent_sync_lengths.push_extend_tips_length(0);
+    }
+
+    assert!(status.is_close_to_tip());
+}
+
+/// Test if sync lengths array with high values is not near tip.
+#[test]
+fn high_sync_lengths() {
+    let (status, mut recent_sync_lengths) = SyncStatus::new();
+
+    // The value 500 is based on the fact that sync lengths are around 500
+    // blocks long when Zebra is syncing.
+    for _ in 0..RecentSyncLengths::MAX_RECENT_LENGTHS {
+        recent_sync_lengths.push_extend_tips_length(500);
+    }
+
+    assert!(!status.is_close_to_tip());
+}