add(mempool): Verify orphaned mempool transactions

zebra-consensus/src/block/tests.rs

@@ -137,7 +137,7 @@ async fn check_transcripts() -> Result<(), Report> {
     let network = Network::Mainnet;
     let state_service = zebra_state::init_test(&network);
 
-    let transaction = transaction::Verifier::new(&network, state_service.clone());
+    let transaction = transaction::Verifier::new_for_tests(&network, state_service.clone());
     let transaction = Buffer::new(BoxService::new(transaction), 1);
     let block_verifier = Buffer::new(
         SemanticBlockVerifier::new(&network, state_service.clone(), transaction),

zebra-consensus/src/router.rs

@@ -21,7 +21,7 @@ use std::{
 
 use futures::{FutureExt, TryFutureExt};
 use thiserror::Error;
-use tokio::task::JoinHandle;
+use tokio::{sync::oneshot, task::JoinHandle};
 use tower::{buffer::Buffer, util::BoxService, Service, ServiceExt};
 use tracing::{instrument, Instrument, Span};
 
@@ -30,6 +30,7 @@ use zebra_chain::{
     parameters::Network,
 };
 
+use zebra_node_services::mempool;
 use zebra_state as zs;
 
 use crate::{
@@ -230,11 +231,12 @@ where
 /// Block and transaction verification requests should be wrapped in a timeout,
 /// so that out-of-order and invalid requests do not hang indefinitely.
 /// See the [`router`](`crate::router`) module documentation for details.
-#[instrument(skip(state_service))]
-pub async fn init<S>(
+#[instrument(skip(state_service, mempool))]
+pub async fn init<S, Mempool>(
     config: Config,
     network: &Network,
     mut state_service: S,
+    mempool: oneshot::Receiver<Mempool>,
 ) -> (
     Buffer<BoxService<Request, block::Hash, RouterError>, Request>,
     Buffer<
@@ -247,6 +249,11 @@ pub async fn init<S>(
 where
     S: Service<zs::Request, Response = zs::Response, Error = BoxError> + Send + Clone + 'static,
     S::Future: Send + 'static,
+    Mempool: Service<mempool::Request, Response = mempool::Response, Error = BoxError>
+        + Send
+        + Clone
+        + 'static,
+    Mempool::Future: Send + 'static,
 {
     // Give other tasks priority before spawning the checkpoint task.
     tokio::task::yield_now().await;
@@ -333,7 +340,7 @@ where
 
     // transaction verification
 
-    let transaction = transaction::Verifier::new(network, state_service.clone());
+    let transaction = transaction::Verifier::new(network, state_service.clone(), mempool);
     let transaction = Buffer::new(BoxService::new(transaction), VERIFIER_BUFFER_BOUND);
 
     // block verification
@@ -397,3 +404,36 @@ pub struct BackgroundTaskHandles {
     /// Finishes when all the checkpoints are verified, or when the state tip is reached.
     pub state_checkpoint_verify_handle: JoinHandle<()>,
 }
+
+/// Calls [`init`] with a closed mempool setup channel for conciseness in tests.
+///
+/// See [`init`] for more details.
+#[cfg(any(test, feature = "proptest-impl"))]
+pub async fn init_test<S>(
+    config: Config,
+    network: &Network,
+    state_service: S,
+) -> (
+    Buffer<BoxService<Request, block::Hash, RouterError>, Request>,
+    Buffer<
+        BoxService<transaction::Request, transaction::Response, TransactionError>,
+        transaction::Request,
+    >,
+    BackgroundTaskHandles,
+    Height,
+)
+where
+    S: Service<zs::Request, Response = zs::Response, Error = BoxError> + Send + Clone + 'static,
+    S::Future: Send + 'static,
+{
+    init(
+        config.clone(),
+        network,
+        state_service.clone(),
+        oneshot::channel::<
+            Buffer<BoxService<mempool::Request, mempool::Response, BoxError>, mempool::Request>,
+        >()
+        .1,
+    )
+    .await
+}

zebra-consensus/src/router/tests.rs

@@ -68,7 +68,7 @@ async fn verifiers_from_network(
         _transaction_verifier,
         _groth16_download_handle,
         _max_checkpoint_height,
-    ) = crate::router::init(Config::default(), &network, state_service.clone()).await;
+    ) = crate::router::init_test(Config::default(), &network, state_service.clone()).await;
 
     // We can drop the download task handle here, because:
     // - if the download task fails, the tests will panic, and
@@ -169,7 +169,7 @@ async fn verify_checkpoint(config: Config) -> Result<(), Report> {
         _transaction_verifier,
         _groth16_download_handle,
         _max_checkpoint_height,
-    ) = super::init(config.clone(), &network, zs::init_test(&network)).await;
+    ) = super::init_test(config.clone(), &network, zs::init_test(&network)).await;
 
     // Add a timeout layer
     let block_verifier_router =