Merge pull request zingolabs#1589 from Oscar-Pepper/move_mempool_to_s…

…eparate_task

Move mempool to separate task

libtonode-tests/tests/sync.rs

@@ -31,7 +31,7 @@ async fn sync_mainnet_test() {
     let mut lightclient = LightClient::create_from_wallet_base_async(
         WalletBase::from_string(HOSPITAL_MUSEUM_SEED.to_string()),
         &config,
-        2_715_150,
+        2_650_318,
         true,
     )
     .await

zingo-sync/src/client.rs

@@ -8,7 +8,10 @@ use zcash_client_backend::{
     data_api::chain::ChainState,
     proto::{
         compact_formats::CompactBlock,
-        service::{BlockId, GetAddressUtxosReply, RawTransaction, TreeState},
+        service::{
+            compact_tx_streamer_client::CompactTxStreamerClient, BlockId, GetAddressUtxosReply,
+            RawTransaction, TreeState,
+        },
     },
 };
 use zcash_primitives::{
@@ -41,8 +44,6 @@ pub enum FetchRequest {
         oneshot::Sender<Vec<(BlockHeight, Transaction)>>,
         (String, Range<BlockHeight>),
     ),
-    /// Get a stream of mempool transactions until a new block is mined.
-    MempoolStream(oneshot::Sender<tonic::Streaming<RawTransaction>>),
 }
 
 /// Gets the height of the blockchain from the server.
@@ -154,16 +155,11 @@ pub async fn get_transparent_address_transactions(
 }
 
 /// Gets stream of mempool transactions until the next block is mined.
-///
-/// Requires [`crate::client::fetch::fetch`] to be running concurrently, connected via the `fetch_request` channel.
 pub async fn get_mempool_transaction_stream(
-    fetch_request_sender: UnboundedSender<FetchRequest>,
+    client: &mut CompactTxStreamerClient<zingo_netutils::UnderlyingService>,
 ) -> Result<tonic::Streaming<RawTransaction>, ()> {
-    let (reply_sender, reply_receiver) = oneshot::channel();
-    fetch_request_sender
-        .send(FetchRequest::MempoolStream(reply_sender))
-        .unwrap();
-    let mempool_stream = reply_receiver.await.unwrap();
+    tracing::info!("Fetching mempool stream");
+    let mempool_stream = fetch::get_mempool_stream(client).await.unwrap();
 
     Ok(mempool_stream)
 }

zingo-sync/src/client/fetch.rs

@@ -146,12 +146,6 @@ async fn fetch_from_server(
                 .unwrap();
             sender.send(transactions).unwrap();
         }
-
-        FetchRequest::MempoolStream(sender) => {
-            tracing::info!("Fetching mempool stream");
-            let mempool_stream = get_mempool_stream(client).await.unwrap();
-            sender.send(mempool_stream).unwrap();
-        }
     }
 
     Ok(())
@@ -295,7 +289,10 @@ async fn get_taddress_txs(
     Ok(transactions)
 }
 
-async fn get_mempool_stream(
+/// Call `GetMempoolStream` client gPRC
+///
+/// This is not called from the fetch request framework and is intended to be called independently.
+pub(super) async fn get_mempool_stream(
     client: &mut CompactTxStreamerClient<zingo_netutils::UnderlyingService>,
 ) -> Result<tonic::Streaming<RawTransaction>, ()> {
     let request = tonic::Request::new(Empty {});

zingo-sync/src/scan/task.rs

@@ -152,7 +152,7 @@ where
     /// If verification is still in progress, do not create scan tasks.
     /// If there is an idle worker, create a new scan task and add to worker.
     /// If there are no more range available to scan, shutdown the idle workers.
-    pub(crate) async fn update<W>(&mut self, wallet: &mut W)
+    pub(crate) async fn update<W>(&mut self, wallet: &mut W, shutdown_mempool: Arc<AtomicBool>)
     where
         W: SyncWallet + SyncBlocks,
     {
@@ -193,6 +193,9 @@ where
                 }
             }
             ScannerState::Shutdown => {
+                // shutdown mempool
+                shutdown_mempool.store(true, atomic::Ordering::Relaxed);
+
                 // shutdown idle workers
                 while let Some(worker) = self.idle_worker() {
                     self.shutdown_worker(worker.id)