feat: add budget to L1 notification channel (#265)

* add budget to L1 notification channel

* add budget.rs

Author: frisitano

crates/manager/src/manager/budget.rs

@@ -0,0 +1,28 @@
+/// The budget for the L1 notification channel.
+pub(crate) const L1_NOTIFICATION_CHANNEL_BUDGET: u32 = 10;
+
+/// Polls the given stream. Breaks with `true` if there maybe is more work.
+#[macro_export]
+macro_rules! poll_nested_stream_with_budget {
+    ($target:literal, $label:literal, $budget:ident, $poll_stream:expr, $on_ready_some:expr $(, $on_ready_none:expr;)? $(,)?) => {{
+        let mut budget: u32 = $budget;
+
+            loop {
+                match $poll_stream {
+                    Poll::Ready(Some(item)) => {
+                        $on_ready_some(item);
+
+                        budget -= 1;
+                        if budget == 0 {
+                            break true
+                        }
+                    }
+                    Poll::Ready(None) => {
+                        $($on_ready_none;)? // todo: handle error case with $target and $label
+                        break false
+                    }
+                    Poll::Pending => break false,
+                }
+            }
+    }};
+}

crates/manager/src/manager/mod.rs

@@ -3,6 +3,7 @@
 //! responsible for handling events from these components and coordinating their actions.
 
 use super::Consensus;
+use crate::poll_nested_stream_with_budget;
 use alloy_primitives::Signature;
 use alloy_provider::Provider;
 use futures::StreamExt;
@@ -41,6 +42,9 @@ use rollup_node_providers::{L1MessageProvider, L1Provider};
 use scroll_db::Database;
 use scroll_derivation_pipeline::DerivationPipeline;
 
+mod budget;
+use budget::L1_NOTIFICATION_CHANNEL_BUDGET;
+
 mod command;
 pub use command::RollupManagerCommand;
 
@@ -473,14 +477,19 @@ where
             }
         );
 
+        let mut maybe_more_l1_rx_events = false;
         proceed_if!(
             en_synced,
-            // Drain all L1 notifications.
-            while let Some(Poll::Ready(Some(event))) =
-                this.l1_notification_rx.as_mut().map(|rx| rx.poll_next_unpin(cx))
-            {
-                this.handle_l1_notification((*event).clone());
-            }
+            maybe_more_l1_rx_events = poll_nested_stream_with_budget!(
+                "l1_notification_rx",
+                "L1Notification channel",
+                L1_NOTIFICATION_CHANNEL_BUDGET,
+                this.l1_notification_rx
+                    .as_mut()
+                    .map(|rx| rx.poll_next_unpin(cx))
+                    .unwrap_or(Poll::Ready(None)),
+                |event: Arc<L1Notification>| this.handle_l1_notification((*event).clone()),
+            )
         );
 
         // Drain all Indexer events.
@@ -551,6 +560,10 @@ where
             this.handle_network_manager_event(event);
         }
 
+        if maybe_more_l1_rx_events {
+            cx.waker().wake_by_ref();
+        }
+
         Poll::Pending
     }
 }