feat: optimize building stage waiting

rust/main/lander/src/dispatcher/core.rs

@@ -6,7 +6,10 @@ use std::{collections::VecDeque, path::PathBuf, sync::Arc};
 use derive_new::new;
 use eyre::Result;
 use futures_util::future::join_all;
-use tokio::{sync::Mutex, task::JoinHandle};
+use tokio::{
+    sync::{mpsc, Mutex},
+    task::JoinHandle,
+};
 
 use hyperlane_base::{
     db::{HyperlaneRocksDB, DB},
@@ -78,14 +81,16 @@ impl Dispatcher {
     #[instrument(skip(self), fields(domain = %self.domain))]
     pub async fn spawn(self) -> JoinHandle<()> {
         let mut tasks = vec![];
+        let (building_stage_sender, building_stage_receiver) = mpsc::channel(1);
         let building_stage_queue = BuildingStageQueue::new();
         let (inclusion_stage_sender, inclusion_stage_receiver) =
             tokio::sync::mpsc::channel::<Transaction>(SUBMITTER_CHANNEL_SIZE);
         let (finality_stage_sender, finality_stage_receiver) =
             tokio::sync::mpsc::channel::<Transaction>(SUBMITTER_CHANNEL_SIZE);
 
-        let building_stage = BuildingStage::new(
+        let mut building_stage = BuildingStage::new(
             building_stage_queue.clone(),
+            building_stage_receiver,
             inclusion_stage_sender.clone(),
             self.inner.clone(),
             self.domain.clone(),
@@ -137,6 +142,7 @@ impl Dispatcher {
 
         let payload_db_loader = PayloadDbLoader::new(
             self.inner.payload_db.clone(),
+            building_stage_sender,
             building_stage_queue.clone(),
             self.domain.clone(),
         );

rust/main/lander/src/dispatcher/db/loader.rs

@@ -143,7 +143,7 @@ impl<T: LoadableFromDb + Debug> DbIterator<T> {
                 }
 
                 // sleep to wait for new items to be added
-                sleep(Duration::from_millis(10)).await;
+                sleep(Duration::from_millis(100)).await;
                 iteration_count = iteration_count.saturating_add(1);
             } else {
                 debug!(?self, "Loaded item");

rust/main/lander/src/dispatcher/db/payload/loader.rs

@@ -5,6 +5,7 @@ use std::{
 
 use async_trait::async_trait;
 use derive_new::new;
+use tokio::sync::mpsc;
 use tracing::{debug, trace};
 
 use crate::{
@@ -18,6 +19,7 @@ use super::PayloadDb;
 #[derive(new)]
 pub struct PayloadDbLoader {
     db: Arc<dyn PayloadDb>,
+    building_stage_sender: mpsc::Sender<()>,
     building_stage_queue: BuildingStageQueue,
     domain: String,
 }
@@ -27,6 +29,16 @@ impl PayloadDbLoader {
         let domain = self.domain.clone();
         DbIterator::new(Arc::new(self), "Payload".to_string(), false, domain).await
     }
+
+    pub async fn push_back(&self, item: FullPayload) {
+        self.building_stage_queue.push_back(item).await;
+        // Notify building stage of new payloads.
+        // If capacity is zero, then that means building stage
+        // is already notified. Don't need to notify again.
+        if self.building_stage_sender.capacity() > 0 {
+            let _ = self.building_stage_sender.send(()).await;
+        }
+    }
 }
 
 impl Debug for PayloadDbLoader {
@@ -50,7 +62,7 @@ impl LoadableFromDb for PayloadDbLoader {
     async fn load(&self, item: FullPayload) -> Result<LoadingOutcome, LanderError> {
         match item.status {
             PayloadStatus::ReadyToSubmit | PayloadStatus::Retry(_) => {
-                self.building_stage_queue.push_back(item).await;
+                self.push_back(item).await;
                 Ok(LoadingOutcome::Loaded)
             }
             PayloadStatus::Dropped(_) | PayloadStatus::InTransaction(_) => {
@@ -60,3 +72,6 @@ impl LoadableFromDb for PayloadDbLoader {
         }
     }
 }
+
+#[cfg(test)]
+mod loader_tests;