Send crawled transaction IDs to downloader

zebra-chain/src/parameters/network_upgrade.rs

@@ -10,11 +10,15 @@ use std::ops::Bound::*;
 
 use chrono::{DateTime, Duration, Utc};
 
+#[cfg(any(test, feature = "proptest-impl"))]
+use proptest_derive::Arbitrary;
+
 /// A Zcash network upgrade.
 ///
 /// Network upgrades can change the Zcash network protocol or consensus rules in
 /// incompatible ways.
 #[derive(Copy, Clone, Debug, Eq, Hash, PartialEq, Serialize, Deserialize)]
+#[cfg_attr(any(test, feature = "proptest-impl"), derive(Arbitrary))]
 pub enum NetworkUpgrade {
     /// The Zcash protocol for a Genesis block.
     ///

zebra-consensus/Cargo.toml

@@ -5,6 +5,10 @@ authors = ["Zcash Foundation <[email protected]>"]
 license = "MIT OR Apache-2.0"
 edition = "2018"
 
+[features]
+default = []
+proptest-impl = ["proptest", "proptest-derive", "zebra-chain/proptest-impl"]
+
 [dependencies]
 blake2b_simd = "0.5.11"
 bellman = "0.10.0"
@@ -33,8 +37,13 @@ zebra-state = { path = "../zebra-state" }
 zebra-script = { path = "../zebra-script" }
 wagyu-zcash-parameters = "0.2.0"
 
+proptest = { version = "0.10", optional = true }
+proptest-derive = { version = "0.3.0", optional = true }
+
 [dev-dependencies]
 color-eyre = "0.5.11"
+proptest = "0.10"
+proptest-derive = "0.3.0"
 rand07 = { package = "rand", version = "0.7" }
 spandoc = "0.2"
 tokio = { version = "0.3.6", features = ["full"] }

zebra-consensus/src/error.rs

@@ -9,6 +9,9 @@ use thiserror::Error;
 
 use crate::BoxError;
 
+#[cfg(any(test, feature = "proptest-impl"))]
+use proptest_derive::Arbitrary;
+
 #[derive(Error, Copy, Clone, Debug, PartialEq)]
 pub enum SubsidyError {
     #[error("no coinbase transaction in block")]
@@ -19,6 +22,7 @@ pub enum SubsidyError {
 }
 
 #[derive(Error, Clone, Debug, PartialEq)]
+#[cfg_attr(any(test, feature = "proptest-impl"), derive(Arbitrary))]
 pub enum TransactionError {
     #[error("first transaction must be coinbase")]
     CoinbasePosition,
@@ -45,6 +49,7 @@ pub enum TransactionError {
     CoinbaseInMempool,
 
     #[error("coinbase transaction failed subsidy validation")]
+    #[cfg_attr(any(test, feature = "proptest-impl"), proptest(skip))]
     Subsidy(#[from] SubsidyError),
 
     #[error("transaction version number MUST be >= 4")]
@@ -63,6 +68,7 @@ pub enum TransactionError {
     BadBalance,
 
     #[error("could not verify a transparent script")]
+    #[cfg_attr(any(test, feature = "proptest-impl"), proptest(skip))]
     Script(#[from] zebra_script::Error),
 
     #[error("spend description cv and rk MUST NOT be of small order")]
@@ -76,12 +82,15 @@ pub enum TransactionError {
     #[error(
         "Sprout joinSplitSig MUST represent a valid signature under joinSplitPubKey of dataToBeSigned"
     )]
+    #[cfg_attr(any(test, feature = "proptest-impl"), proptest(skip))]
     Ed25519(#[from] zebra_chain::primitives::ed25519::Error),
 
     #[error("Sapling bindingSig MUST represent a valid signature under the transaction binding validating key bvk of SigHash")]
+    #[cfg_attr(any(test, feature = "proptest-impl"), proptest(skip))]
     RedJubjub(zebra_chain::primitives::redjubjub::Error),
 
     #[error("Orchard bindingSig MUST represent a valid signature under the transaction binding validating key bvk of SigHash")]
+    #[cfg_attr(any(test, feature = "proptest-impl"), proptest(skip))]
     RedPallas(zebra_chain::primitives::redpallas::Error),
 
     // temporary error type until #1186 is fixed

zebrad/Cargo.toml

@@ -61,6 +61,7 @@ proptest = "0.10"
 proptest-derive = "0.3"
 
 zebra-chain = { path = "../zebra-chain", features = ["proptest-impl"] }
+zebra-consensus = { path = "../zebra-consensus/", features = ["proptest-impl"] }
 zebra-state = { path = "../zebra-state", features = ["proptest-impl"] }
 zebra-test = { path = "../zebra-test" }
 

zebrad/src/commands/start.rs

@@ -102,12 +102,12 @@ impl StartCmd {
         let mempool = ServiceBuilder::new().buffer(20).service(mempool_service);
 
         setup_tx
-            .send((peer_set.clone(), address_book, mempool))
+            .send((peer_set.clone(), address_book, mempool.clone()))
             .map_err(|_| eyre!("could not send setup data to inbound service"))?;
 
         select! {
             result = syncer.sync().fuse() => result,
-            _ = mempool::Crawler::spawn(peer_set, sync_status).fuse() => {
+            _ = mempool::Crawler::spawn(peer_set, mempool, sync_status).fuse() => {
                 unreachable!("The mempool crawler only stops if it panics");
             }
         }

zebrad/src/components/mempool.rs

@@ -50,7 +50,7 @@ type TxVerifier = Buffer<
 >;
 type InboundTxDownloads = TxDownloads<Timeout<Outbound>, Timeout<TxVerifier>, State>;
 
-#[derive(Debug)]
+#[derive(Debug, Eq, PartialEq)]
 #[allow(dead_code)]
 pub enum Request {
     TransactionIds,

zebrad/src/components/mempool/crawler.rs

@@ -8,9 +8,12 @@ use futures::{stream::FuturesUnordered, StreamExt};
 use tokio::{task::JoinHandle, time::sleep};
 use tower::{timeout::Timeout, BoxError, Service, ServiceExt};
 
-use zebra_network::{Request, Response};
+use zebra_network as zn;
 
-use super::super::sync::SyncStatus;
+use super::{
+    super::{mempool, sync::SyncStatus},
+    downloads::Gossip,
+};
 
 #[cfg(test)]
 mod tests;
@@ -31,20 +34,30 @@ const RATE_LIMIT_DELAY: Duration = Duration::from_secs(75);
 const PEER_RESPONSE_TIMEOUT: Duration = Duration::from_secs(6);
 
 /// The mempool transaction crawler.
-pub struct Crawler<S> {
-    peer_set: Timeout<S>,
+pub struct Crawler<PeerSet, Mempool> {
+    peer_set: Timeout<PeerSet>,
+    mempool: Mempool,
     status: SyncStatus,
 }
 
-impl<S> Crawler<S>
+impl<PeerSet, Mempool> Crawler<PeerSet, Mempool>
 where
-    S: Service<Request, Response = Response, Error = BoxError> + Clone + Send + 'static,
-    S::Future: Send,
+    PeerSet:
+        Service<zn::Request, Response = zn::Response, Error = BoxError> + Clone + Send + 'static,
+    PeerSet::Future: Send,
+    Mempool:
+        Service<mempool::Request, Response = mempool::Response, Error = BoxError> + Send + 'static,
+    Mempool::Future: Send,
 {
     /// Spawn an asynchronous task to run the mempool crawler.
-    pub fn spawn(peer_set: S, status: SyncStatus) -> JoinHandle<Result<(), BoxError>> {
+    pub fn spawn(
+        peer_set: PeerSet,
+        mempool: Mempool,
+        status: SyncStatus,
+    ) -> JoinHandle<Result<(), BoxError>> {
         let crawler = Crawler {
             peer_set: Timeout::new(peer_set, PEER_RESPONSE_TIMEOUT),
+            mempool,
             status,
         };
 
@@ -79,13 +92,13 @@ where
             // end the task on permanent peer set errors
             let peer_set = peer_set.ready_and().await?;
 
-            requests.push(peer_set.call(Request::MempoolTransactionIds));
+            requests.push(peer_set.call(zn::Request::MempoolTransactionIds));
         }
 
         while let Some(result) = requests.next().await {
             // log individual response errors
             match result {
-                Ok(response) => self.handle_response(response).await,
+                Ok(response) => self.handle_response(response).await?,
                 // TODO: Reduce the log level of the errors (#2655).
                 Err(error) => info!("Failed to crawl peer for mempool transactions: {}", error),
             }
@@ -95,9 +108,9 @@ where
     }
 
     /// Handle a peer's response to the crawler's request for transactions.
-    async fn handle_response(&mut self, response: Response) {
-        let transaction_ids = match response {
-            Response::TransactionIds(ids) => ids,
+    async fn handle_response(&mut self, response: zn::Response) -> Result<(), BoxError> {
+        let transaction_ids: Vec<_> = match response {
+            zn::Response::TransactionIds(ids) => ids.into_iter().map(Gossip::Id).collect(),
             _ => unreachable!("Peer set did not respond with transaction IDs to mempool crawler"),
         };
 
@@ -106,6 +119,37 @@ where
             transaction_ids.len()
         );
 
-        // TODO: Send transaction IDs to the download and verify stream (#2650)
+        if !transaction_ids.is_empty() {
+            self.queue_transactions(transaction_ids).await?;
+        }
+
+        Ok(())
+    }
+
+    /// Forward the crawled transactions IDs to the mempool transaction downloader.
+    async fn queue_transactions(&mut self, transaction_ids: Vec<Gossip>) -> Result<(), BoxError> {
+        let call_result = self
+            .mempool
+            .ready_and()
+            .await?
+            .call(mempool::Request::Queue(transaction_ids))
+            .await;
+
+        let queue_errors = match call_result {
+            Ok(mempool::Response::Queued(queue_results)) => {
+                queue_results.into_iter().filter_map(Result::err)
+            }
+            Ok(_) => unreachable!("Mempool did not respond with queue results to mempool crawler"),
+            Err(call_error) => {
+                debug!("Ignoring unexpected peer behavior: {}", call_error);
+                return Ok(());
+            }
+        };
+
+        for error in queue_errors {
+            debug!("Failed to download a crawled transaction: {}", error);
+        }
+
+        Ok(())
     }
 }