Applies suggestions from code review

zebra-scan/src/init.rs

@@ -1,4 +1,4 @@
-//! Initializing the scanner and RPC server.
+//! Initializing the scanner and gRPC server.
 
 use color_eyre::Report;
 use tower::ServiceBuilder;

zebra-scan/src/service.rs

@@ -18,6 +18,9 @@ pub mod scan_task;
 
 pub use scan_task::{ScanTask, ScanTaskCommand};
 
+#[cfg(any(test, feature = "proptest-impl"))]
+use std::sync::mpsc::Receiver;
+
 /// Zebra-scan [`tower::Service`]
 #[derive(Debug)]
 pub struct ScanService {
@@ -46,10 +49,9 @@ impl ScanService {
     }
 
     /// Create a new [`ScanService`] with a mock `ScanTask`
+    // TODO: Move this to tests behind `cfg(any(test, feature = "proptest-impl"))`
     #[cfg(any(test, feature = "proptest-impl"))]
-    pub fn new_with_mock_scanner(
-        db: Storage,
-    ) -> (Self, std::sync::mpsc::Receiver<ScanTaskCommand>) {
+    pub fn new_with_mock_scanner(db: Storage) -> (Self, Receiver<ScanTaskCommand>) {
         let (scan_task, cmd_receiver) = ScanTask::mock();
         (Self { db, scan_task }, cmd_receiver)
     }

zebra-scan/src/service/scan_task.rs

@@ -37,6 +37,7 @@ impl ScanTask {
         state: scan::State,
         chain_tip_change: ChainTipChange,
     ) -> Self {
+        // TODO: Use a bounded channel or move this logic to the scan service or another service.
         let (cmd_sender, cmd_receiver) = mpsc::channel();
 
         Self {

zebra-scan/src/service/scan_task/commands.rs

@@ -1,4 +1,4 @@
-//! Types and method implementations for [`ScanTask`]
+//! Types and method implementations for [`ScanTaskCommand`]
 
 use std::{
     collections::HashMap,
@@ -39,12 +39,13 @@ pub enum ScanTaskCommand {
     },
 
     /// Start sending results for key hashes to `result_sender`
+    // TODO: Implement this command (#8206)
     SubscribeResults {
         /// Sender for results
         result_sender: mpsc::Sender<Arc<Transaction>>,
 
         /// Key hashes to send the results of to result channel
-        key_hashes: Vec<()>,
+        keys: Vec<String>,
     },
 }
 

zebra-scan/src/service/scan_task/executor.rs

@@ -2,27 +2,31 @@
 
 use futures::{stream::FuturesUnordered, FutureExt, StreamExt};
 use tokio::{
-    sync::mpsc::{UnboundedReceiver, UnboundedSender},
+    sync::mpsc::{Receiver, Sender},
     task::JoinHandle,
 };
 use tracing::Instrument;
 use zebra_chain::BoxError;
 
 use super::scan::ScanRangeTaskBuilder;
 
+const EXECUTOR_BUFFER_SIZE: usize = 100;
+
 pub fn spawn_init() -> (
-    UnboundedSender<ScanRangeTaskBuilder>,
+    Sender<ScanRangeTaskBuilder>,
     JoinHandle<Result<(), BoxError>>,
 ) {
-    let (scan_task_sender, scan_task_receiver) = tokio::sync::mpsc::unbounded_channel();
+    // TODO: Use a bounded channel.
+    let (scan_task_sender, scan_task_receiver) = tokio::sync::mpsc::channel(EXECUTOR_BUFFER_SIZE);
+
     (
         scan_task_sender,
         tokio::spawn(scan_task_executor(scan_task_receiver).in_current_span()),
     )
 }
 
 pub async fn scan_task_executor(
-    mut scan_task_receiver: UnboundedReceiver<ScanRangeTaskBuilder>,
+    mut scan_task_receiver: Receiver<ScanRangeTaskBuilder>,
 ) -> Result<(), BoxError> {
     let mut scan_range_tasks = FuturesUnordered::new();
 

zebra-scan/src/service/scan_task/scan.rs

@@ -135,7 +135,8 @@ pub async fn start(
 
             scan_task_sender
                 .send(ScanRangeTaskBuilder::new(height, new_keys, state, storage))
-                .expect("scan_until_task channel should not be full");
+                .await
+                .expect("scan_until_task channel should not be closed");
         }
 
         let scanned_height = scan_height_and_store_results(

zebra-scan/src/service/scan_task/tests/vectors.rs

@@ -44,7 +44,7 @@ async fn scan_task_processes_messages_correctly() -> Result<(), Report> {
             .collect(),
     )?;
 
-    // Check that it updated parsed_keys correctly and returned the right new keys when starting with an empty state
+    // Check that no key should be added if they are all already known and the heights are the same
 
     let new_keys = ScanTask::process_messages(&cmd_receiver, &mut parsed_keys)?;
 
@@ -104,7 +104,7 @@ async fn scan_task_processes_messages_correctly() -> Result<(), Report> {
 
     let new_keys = ScanTask::process_messages(&cmd_receiver, &mut parsed_keys)?;
 
-    // Check that it sends the done notification successfully before returning and dropping `done_tx``
+    // Check that it sends the done notification successfully before returning and dropping `done_tx`
     done_rx.await?;
 
     assert!(

zebrad/tests/acceptance.rs

@@ -2965,6 +2965,7 @@ fn scan_start_where_left() -> Result<()> {
 /// Test successful registration of a new key in the scan task.
 ///
 /// See [`common::shielded_scan::register_key`] for more information.
+// TODO: Add this test to CI (#8236)
 #[tokio::test]
 #[ignore]
 #[cfg(feature = "shielded-scan")]