fix: answer comments

Author: greged93

Cargo.lock

@@ -10,7 +10,6 @@ use reth_rpc_builder::config::RethRpcServerConfig;
 use reth_scroll_chainspec::ScrollChainSpec;
 use reth_scroll_primitives::ScrollPrimitives;
 use reth_transaction_pool::{PoolTransaction, TransactionPool};
-use rollup_node_indexer::Indexer;
 use rollup_node_manager::{PoAConsensus, RollupNodeManager};
 use rollup_node_providers::{beacon_provider, DatabaseL1MessageProvider, OnlineL1Provider};
 use rollup_node_watcher::L1Watcher;
@@ -120,9 +119,6 @@ where
         // Wrap the database in an Arc
         let db = Arc::new(db);
 
-        // Spawn the indexer
-        let indexer = Indexer::new(db.clone());
-
         // Spawn the L1Watcher
         let l1_provider_args = self.config.l1_provider_args;
         let l1_notification_rx = if let Some(l1_rpc_url) = l1_provider_args.l1_rpc_url {
@@ -153,8 +149,8 @@ where
             scroll_network_manager,
             engine,
             l1_provider,
+            db,
             l1_notification_rx,
-            indexer,
             ForkchoiceState::genesis(
                 ctx.config().chain.chain.try_into().expect("must be a named chain"),
             ),

bin/rollup/src/network.rs

@@ -86,12 +86,7 @@ where
 
     /// Handles the next batch index in the batch index queue, pushing the future in the pipeline
     /// futures.
-    fn handle_next_batch<
-        F: FnMut(&mut FuturesOrdered<DerivationPipelineFuture>, DerivationPipelineFuture),
-    >(
-        &mut self,
-        mut queue_fut: F,
-    ) {
+    fn handle_next_batch(&mut self) -> Option<DerivationPipelineFuture> {
         let database = self.database.clone();
         let provider = self.l1_provider.clone();
 
@@ -105,8 +100,9 @@ where
 
                 derive(batch, provider).await.map_err(|err| (index, err))
             });
-            queue_fut(&mut self.pipeline_futures, fut);
+            return Some(fut);
         }
+        None
     }
 }
 
@@ -133,7 +129,9 @@ where
 
         // if the futures can still grow, handle the next batch.
         if this.pipeline_futures.len() < MAX_CONCURRENT_DERIVATION_PIPELINE_FUTS {
-            this.handle_next_batch(|queue, fut| queue.push_back(fut));
+            if let Some(fut) = this.handle_next_batch() {
+                this.pipeline_futures.push_back(fut)
+            }
         }
 
         // poll the futures and handle result.
@@ -147,7 +145,9 @@ where
                     tracing::error!(target: "scroll::node::derivation_pipeline", ?index, ?err, "failed to derive payload attributes for batch");
                     // retry polling the same batch index.
                     this.batch_index_queue.push_front(index);
-                    this.handle_next_batch(|queue, fut| queue.push_front(fut));
+                    if let Some(fut) = this.handle_next_batch() {
+                        this.pipeline_futures.push_front(fut)
+                    }
                 }
             }
         }

crates/derivation-pipeline/src/lib.rs

@@ -35,11 +35,6 @@ impl Indexer {
         Self { database, pending_futures: Default::default() }
     }
 
-    /// Returns a reference to the database used by the indexer.
-    pub fn database(&self) -> Arc<Database> {
-        self.database.clone()
-    }
-
     /// Handles an event from the L1.
     pub fn handle_l1_notification(&mut self, event: L1Notification) {
         let fut =

crates/indexer/src/lib.rs

@@ -23,6 +23,7 @@ reth-scroll-primitives.workspace = true
 
 # scroll
 scroll-alloy-rpc-types-engine.workspace = true
+scroll-db.workspace = true
 scroll-derivation-pipeline.workspace = true
 scroll-engine.workspace = true
 scroll-network.workspace = true

crates/node/Cargo.toml

@@ -39,6 +39,7 @@ mod consensus;
 use consensus::Consensus;
 use rollup_node_primitives::BlockInfo;
 use rollup_node_providers::{ExecutionPayloadProvider, L1Provider};
+use scroll_db::Database;
 use scroll_derivation_pipeline::DerivationPipeline;
 
 /// The size of the event channel.
@@ -121,13 +122,13 @@ where
         network: NetworkManager,
         engine: EngineDriver<EC, P>,
         l1_provider: L1P,
+        database: Arc<Database>,
         l1_notification_rx: Option<Receiver<Arc<L1Notification>>>,
-        indexer: Indexer,
         forkchoice_state: ForkchoiceState,
         consensus: C,
         new_block_rx: Option<UnboundedReceiver<NewBlockWithPeer>>,
     ) -> Self {
-        let database = indexer.database();
+        let indexer = Indexer::new(database.clone());
         let derivation_pipeline = DerivationPipeline::new(l1_provider, database);
         Self {
             network,

crates/node/src/lib.rs

@@ -51,7 +51,7 @@ impl<L1P, BP> OnlineL1Provider<L1P, BP>
 where
     BP: BeaconProvider,
 {
-    /// Returns a new [`OnlineBeaconClient`] from the provided [`OnlineBeaconClient`], blob capacity
+    /// Returns a new [`OnlineL1Provider`] from the provided [`BeaconProvider`], blob capacity
     /// and [`L1MessageProvider`].
     pub async fn new(beacon_provider: BP, blob_capacity: usize, l1_message_provider: L1P) -> Self {
         let cache = Arc::new(Mutex::new(LruCache::new(