introduce watcher back pressure in the rollup node manager (#278)

Co-authored-by: greg <[email protected]>

Author: frisitano

crates/chain-orchestrator/src/lib.rs

@@ -125,6 +125,11 @@ impl<
         })
     }
 
+    /// Returns the number of pending futures.
+    pub fn pending_futures_len(&self) -> usize {
+        self.pending_futures.len()
+    }
+
     /// Wraps a pending chain orchestrator future, metering the completion of it.
     pub fn handle_metered(
         &mut self,

crates/engine/src/driver.rs

@@ -88,6 +88,13 @@ where
         }
     }
 
+    /// Returns the number of pending futures in the queue.
+    ///
+    /// This only considers the length of the L1 payload attributes and chain import queues.
+    pub fn pending_futures_len(&self) -> usize {
+        self.l1_payload_attributes.len() + self.chain_imports.len()
+    }
+
     /// Sets the finalized block info.
     pub fn set_finalized_block_info(&mut self, block_info: BlockInfo) {
         self.fcs.update_finalized_block_info(block_info);

crates/manager/src/manager/mod.rs

@@ -59,6 +59,14 @@ pub use handle::RollupManagerHandle;
 /// The size of the event channel.
 const EVENT_CHANNEL_SIZE: usize = 100;
 
+/// The maximum capacity of the pending futures queue in the chain orchestrator for acceptance of
+/// new events from the L1 notification channel.
+const CHAIN_ORCHESTRATOR_MAX_PENDING_FUTURES: usize = 5000;
+
+/// The maximum number of pending futures in the engine driver for acceptance of new events from the
+/// L1 notification channel.
+const ENGINE_MAX_PENDING_FUTURES: usize = 5000;
+
 /// The main manager for the rollup node.
 ///
 /// This is an endless [`Future`] that drives the state of the entire network forward and includes
@@ -444,6 +452,14 @@ where
 
         drop(graceful_guard);
     }
+
+    /// Returns true if the manager has capacity to accept new L1 notifications.
+    pub fn has_capacity_for_l1_notifications(&self) -> bool {
+        let chain_orchestrator_has_capacity = self.chain.pending_futures_len() <
+            CHAIN_ORCHESTRATOR_MAX_PENDING_FUTURES - L1_NOTIFICATION_CHANNEL_BUDGET as usize;
+        let engine_has_capacity = self.engine.pending_futures_len() < ENGINE_MAX_PENDING_FUTURES;
+        chain_orchestrator_has_capacity && engine_has_capacity
+    }
 }
 
 impl<N, EC, P, L1P, L1MP, CS> Future for RollupNodeManager<N, EC, P, L1P, L1MP, CS>
@@ -516,7 +532,7 @@ where
 
         let mut maybe_more_l1_rx_events = false;
         proceed_if!(
-            en_synced,
+            en_synced && this.has_capacity_for_l1_notifications(),
             maybe_more_l1_rx_events = poll_nested_stream_with_budget!(
                 "l1_notification_rx",
                 "L1Notification channel",