feat: extra blob providers (#203)

Author: greged93

Cargo.lock

@@ -187,6 +187,7 @@ scroll-migration = { path = "crates/database/migration" }
 arbitrary = { version = "1.4", default-features = false }
 async-trait = "0.1"
 auto_impl = "1.2"
+clap = { version = "4", features = ["derive", "env"] }
 derive_more = { version = "2.0", default-features = false }
 eyre = "0.6"
 futures = { version = "0.3", default-features = false }

Cargo.toml

@@ -304,7 +304,7 @@ mod tests {
     use futures::StreamExt;
     use rollup_node_primitives::L1MessageEnvelope;
     use rollup_node_providers::{
-        test_utils::NoBlobProvider, DatabaseL1MessageProvider, L1BlobProvider, L1MessageProvider,
+        test_utils::NoBlobProvider, BlobProvider, DatabaseL1MessageProvider, L1MessageProvider,
         L1ProviderError,
     };
     use scroll_alloy_consensus::TxL1Message;
@@ -325,7 +325,7 @@ mod tests {
     }
 
     #[async_trait::async_trait]
-    impl L1BlobProvider for MockL1MessageProvider {
+    impl BlobProvider for MockL1MessageProvider {
         async fn blob(
             &self,
             _block_timestamp: u64,

crates/derivation-pipeline/src/lib.rs

@@ -86,7 +86,7 @@ pub struct RollupNodeManager<
     /// The engine driver used to communicate with the engine.
     engine: EngineDriver<EC, CS, P>,
     /// The derivation pipeline, used to derive payload attributes from batches.
-    derivation_pipeline: Option<DerivationPipeline<L1P>>,
+    derivation_pipeline: DerivationPipeline<L1P>,
     /// A receiver for [`L1Notification`]s from the [`rollup_node_watcher::L1Watcher`].
     l1_notification_rx: Option<ReceiverStream<Arc<L1Notification>>>,
     /// An indexer used to index data for the rollup node.
@@ -153,7 +153,7 @@ where
     pub fn new(
         network: ScrollNetworkManager<N>,
         engine: EngineDriver<EC, CS, P>,
-        l1_provider: Option<L1P>,
+        l1_provider: L1P,
         database: Arc<Database>,
         l1_notification_rx: Option<Receiver<Arc<L1Notification>>>,
         consensus: Box<dyn Consensus>,
@@ -165,8 +165,7 @@ where
     ) -> (Self, RollupManagerHandle) {
         let (handle_tx, handle_rx) = mpsc::channel(EVENT_CHANNEL_SIZE);
         let indexer = Indexer::new(database.clone(), chain_spec.clone());
-        let derivation_pipeline =
-            l1_provider.map(|provider| DerivationPipeline::new(provider, database));
+        let derivation_pipeline = DerivationPipeline::new(l1_provider, database);
         let rnm = Self {
             handle_rx,
             chain_spec,
@@ -240,17 +239,13 @@ where
             IndexerEvent::BatchCommitIndexed { batch_info, safe_head, l1_block_number } => {
                 // if we detected a batch revert event, we reset the pipeline and the engine driver.
                 if let Some(new_safe_head) = safe_head {
-                    if let Some(pipeline) = self.derivation_pipeline.as_mut() {
-                        pipeline.flush()
-                    }
+                    self.derivation_pipeline.flush();
                     self.engine.clear_l1_payload_attributes();
                     self.engine.set_head_block_info(new_safe_head);
                     self.engine.set_safe_block_info(new_safe_head);
                 }
                 // push the batch info into the derivation pipeline.
-                if let Some(pipeline) = &mut self.derivation_pipeline {
-                    pipeline.handle_batch_commit(batch_info, l1_block_number);
-                }
+                self.derivation_pipeline.handle_batch_commit(batch_info, l1_block_number);
             }
             IndexerEvent::BatchFinalizationIndexed(_, Some(finalized_block)) => {
                 // update the fcs on new finalized block.
@@ -285,9 +280,7 @@ where
                 }
 
                 // Handle the reorg in the derivation pipeline.
-                if let Some(pipeline) = self.derivation_pipeline.as_mut() {
-                    pipeline.handle_reorg(l1_block_number);
-                }
+                self.derivation_pipeline.handle_reorg(l1_block_number);
             }
             IndexerEvent::L1MessageIndexed(index) => {
                 if let Some(event_sender) = self.event_sender.as_ref() {
@@ -520,9 +513,7 @@ where
         );
 
         // Poll Derivation Pipeline and push attribute in queue if any.
-        while let Some(Poll::Ready(Some(attributes))) =
-            this.derivation_pipeline.as_mut().map(|f| f.poll_next_unpin(cx))
-        {
+        while let Poll::Ready(Some(attributes)) = this.derivation_pipeline.poll_next_unpin(cx) {
             this.engine.handle_l1_consolidation(attributes)
         }
 

crates/manager/src/manager/mod.rs

@@ -83,7 +83,7 @@ scroll-wire.workspace = true
 scroll-network.workspace = true
 
 auto_impl.workspace = true
-clap = { version = "4", features = ["derive", "env"] }
+clap = { workspace = true, features = ["derive", "env"] }
 eyre.workspace = true
 reqwest.workspace = true
 tokio.workspace = true

crates/node/Cargo.toml

@@ -1,7 +1,9 @@
 use crate::{
     add_ons::IsDevChain,
-    constants::{self, PROVIDER_BLOB_CACHE_SIZE},
+    constants::{self},
 };
+use std::{fs, path::PathBuf, sync::Arc, time::Duration};
+
 use alloy_primitives::{hex, Address};
 use alloy_provider::{Provider, ProviderBuilder};
 use alloy_rpc_client::RpcClient;
@@ -22,7 +24,7 @@ use rollup_node_manager::{
 };
 use rollup_node_primitives::{BlockInfo, NodeConfig};
 use rollup_node_providers::{
-    beacon_provider, DatabaseL1MessageProvider, L1MessageProvider, L1Provider, OnlineL1Provider,
+    BlobSource, DatabaseL1MessageProvider, FullL1Provider, L1MessageProvider, L1Provider,
     SystemContractProvider,
 };
 use rollup_node_sequencer::{L1MessageInclusionMode, Sequencer};
@@ -35,7 +37,6 @@ use scroll_engine::{genesis_hash_from_chain_spec, EngineDriver, ForkchoiceState}
 use scroll_migration::traits::ScrollMigrator;
 use scroll_network::ScrollNetworkManager;
 use scroll_wire::ScrollWireEvent;
-use std::{fs, path::PathBuf, sync::Arc, time::Duration};
 use tokio::sync::mpsc::{Sender, UnboundedReceiver};
 
 /// A struct that represents the arguments for the rollup node.
@@ -228,18 +229,13 @@ impl ScrollRollupNodeConfig {
 
         // Construct the l1 provider.
         let l1_messages_provider = DatabaseL1MessageProvider::new(db.clone(), 0);
-        let l1_provider = if let Some(url) = self.beacon_provider_args.url {
-            let beacon_provider = beacon_provider(url.to_string());
-            let l1_provider = OnlineL1Provider::new(
-                beacon_provider,
-                PROVIDER_BLOB_CACHE_SIZE,
-                l1_messages_provider.clone(),
-            )
-            .await;
-            Some(l1_provider)
-        } else {
-            None
-        };
+        let blob_provider = self
+            .beacon_provider_args
+            .blob_source
+            .provider(self.beacon_provider_args.url)
+            .await
+            .expect("failed to construct L1 blob provider");
+        let l1_provider = FullL1Provider::new(blob_provider, l1_messages_provider.clone()).await;
 
         // Construct the Sequencer.
         let (sequencer, block_time) = if self.sequencer_args.sequencer_enabled {
@@ -345,6 +341,13 @@ pub struct BeaconProviderArgs {
     /// The URL for the Beacon chain.
     #[arg(long = "beacon.url", id = "beacon_url", value_name = "BEACON_URL")]
     pub url: Option<reqwest::Url>,
+    /// The blob source for the provider.
+    #[arg(
+        long = "beacon.blob-source",
+        id = "beacon_blob_source",
+        value_name = "BEACON_BLOB_SOURCE"
+    )]
+    pub blob_source: BlobSource,
     /// The compute units per second for the provider.
     #[arg(long = "beacon.cups", id = "beacon_compute_units_per_second", value_name = "BEACON_COMPUTE_UNITS_PER_SECOND", default_value_t = constants::PROVIDER_COMPUTE_UNITS_PER_SECOND)]
     pub compute_units_per_second: u64,

crates/node/src/args.rs

@@ -1,6 +1,3 @@
-/// The size of the blob cache for the provider.
-pub(crate) const PROVIDER_BLOB_CACHE_SIZE: usize = 100;
-
 /// The max retries for the L1 provider.
 pub(crate) const PROVIDER_MAX_RETRIES: u32 = 10;
 

crates/node/src/constants.rs

@@ -19,6 +19,7 @@ use reth_node_core::args::{DiscoveryArgs, NetworkArgs, RpcServerArgs};
 use reth_provider::providers::BlockchainProvider;
 use reth_rpc_server_types::RpcModuleSelection;
 use reth_tasks::TaskManager;
+use rollup_node_providers::BlobSource;
 use rollup_node_sequencer::L1MessageInclusionMode;
 use std::{path::PathBuf, sync::Arc};
 use tokio::sync::Mutex;
@@ -138,7 +139,10 @@ pub fn default_test_scroll_rollup_node_config() -> ScrollRollupNodeConfig {
         l1_provider_args: L1ProviderArgs::default(),
         engine_driver_args: EngineDriverArgs { en_sync_trigger: 100, sync_at_startup: true },
         sequencer_args: SequencerArgs { payload_building_duration: 1000, ..Default::default() },
-        beacon_provider_args: BeaconProviderArgs::default(),
+        beacon_provider_args: BeaconProviderArgs {
+            blob_source: BlobSource::Mock,
+            ..Default::default()
+        },
         signer_args: Default::default(),
     }
 }
@@ -162,7 +166,10 @@ pub fn default_sequencer_test_scroll_rollup_node_config() -> ScrollRollupNodeCon
             fee_recipient: Default::default(),
             l1_message_inclusion_mode: L1MessageInclusionMode::BlockDepth(0),
         },
-        beacon_provider_args: BeaconProviderArgs::default(),
+        beacon_provider_args: BeaconProviderArgs {
+            blob_source: BlobSource::Mock,
+            ..Default::default()
+        },
         signer_args: Default::default(),
     }
 }

crates/node/src/test_utils.rs

@@ -14,6 +14,7 @@ use rollup_node::{
 };
 use rollup_node_manager::{RollupManagerEvent, RollupManagerHandle};
 use rollup_node_primitives::BatchCommitData;
+use rollup_node_providers::BlobSource;
 use rollup_node_sequencer::L1MessageInclusionMode;
 use rollup_node_watcher::L1Notification;
 use scroll_alloy_consensus::TxL1Message;
@@ -49,7 +50,10 @@ async fn can_bridge_l1_messages() -> eyre::Result<()> {
             l1_message_inclusion_mode: L1MessageInclusionMode::BlockDepth(0),
             ..SequencerArgs::default()
         },
-        beacon_provider_args: BeaconProviderArgs::default(),
+        beacon_provider_args: BeaconProviderArgs {
+            blob_source: BlobSource::Mock,
+            ..Default::default()
+        },
         signer_args: Default::default(),
     };
     let (mut nodes, _tasks, _wallet) = setup_engine(node_args, 1, chain_spec, false).await?;
@@ -114,7 +118,10 @@ async fn can_sequence_and_gossip_blocks() {
             payload_building_duration: 1000,
             ..SequencerArgs::default()
         },
-        beacon_provider_args: BeaconProviderArgs::default(),
+        beacon_provider_args: BeaconProviderArgs {
+            blob_source: BlobSource::Mock,
+            ..Default::default()
+        },
         signer_args: Default::default(),
     };
 

crates/node/tests/e2e.rs

@@ -16,6 +16,7 @@ use rollup_node::{
     ScrollRollupNode, ScrollRollupNodeConfig, SequencerArgs,
 };
 use rollup_node_manager::{RollupManagerCommand, RollupManagerEvent};
+use rollup_node_providers::BlobSource;
 use tokio::sync::oneshot;
 
 #[tokio::test]
@@ -47,6 +48,7 @@ async fn test_should_consolidate_to_block_15k() -> eyre::Result<()> {
             compute_units_per_second: 100,
             max_retries: 10,
             initial_backoff: 100,
+            blob_source: BlobSource::Beacon,
         },
         signer_args: Default::default(),
     };