feat(rollup): init exex driver and pipeline

bin/op-rs/src/main.rs

@@ -56,7 +56,7 @@ fn main() -> Result<()> {
             };
 
             let node = EthereumNode::default();
-            let hera = move |ctx| async { Ok(Driver::new(ctx, hera_args, cfg).await.start()) };
+            let hera = move |ctx| async { Ok(Driver::exex(ctx, hera_args, cfg).start()) };
             let handle = builder.node(node).install_exex(HERA_EXEX_ID, hera).launch().await?;
             handle.wait_for_node_exit().await
         } else {

crates/kona-providers/src/blob_provider.rs

@@ -18,6 +18,14 @@ use reth::primitives::BlobTransactionSidecar;
 use tracing::warn;
 use url::Url;
 
+/// A blob provider that first attempts to fetch blobs from a primary beacon client and
+/// falls back to a secondary blob archiver if the primary fails.
+///
+/// Any blob archiver just needs to implement the beacon
+/// [`blob_sidecars` API](https://ethereum.github.io/beacon-APIs/#/Beacon/getBlobSidecars)
+pub type DurableBlobProvider =
+    OnlineBlobProviderWithFallback<OnlineBeaconClient, OnlineBeaconClient, SimpleSlotDerivation>;
+
 /// Layered [BlobProvider] for the Kona derivation pipeline.
 ///
 /// This provider wraps different blob sources in an ordered manner:
@@ -35,13 +43,9 @@ pub struct LayeredBlobProvider {
     /// This is used primarily during sync when archived blobs
     /// aren't provided by reth since they'll be too old.
     ///
-    /// The `WithFallback` setup allows to specify two different
+    /// The `Durable` setup allows to specify two different
     /// endpoints for a primary and a fallback blob provider.
-    online: OnlineBlobProviderWithFallback<
-        OnlineBeaconClient,
-        OnlineBeaconClient,
-        SimpleSlotDerivation,
-    >,
+    online: DurableBlobProvider,
 }
 
 /// A blob provider that hold blobs in memory.

crates/rollup/Cargo.toml

@@ -10,6 +10,8 @@ keywords.workspace = true
 categories.workspace = true
 
 [dependencies]
+kona-providers = { path = "../kona-providers" }
+
 # Workspace
 eyre.workspace = true
 tracing.workspace = true

crates/rollup/src/cli.rs

@@ -8,6 +8,9 @@ use url::Url;
 /// The default L2 chain ID to use. This corresponds to OP Mainnet.
 pub const DEFAULT_L2_CHAIN_ID: u64 = 10;
 
+/// The default L1 RPC URL to use.
+pub const DEFAULT_L1_RPC_URL: &str = "https://eth.llamarpc.com/";
+
 /// The default L2 RPC URL to use.
 pub const DEFAULT_L2_RPC_URL: &str = "https://optimism.llamarpc.com/";
 
@@ -30,6 +33,11 @@ pub struct HeraArgsExt {
     #[clap(long = "hera.l2-rpc-url", default_value = DEFAULT_L2_RPC_URL)]
     pub l2_rpc_url: Url,
 
+    /// RPC URL of an L1 execution client
+    /// (This is only needed when running in Standalone mode)
+    #[clap(long = "hera.l1-rpc-url", default_value = DEFAULT_L1_RPC_URL)]
+    pub l1_rpc_url: Url,
+
     /// URL of an L1 beacon client to fetch blobs
     #[clap(long = "hera.l1-beacon-client-url", default_value = DEFAULT_L1_BEACON_CLIENT_URL)]
     pub l1_beacon_client_url: Url,

crates/rollup/src/driver.rs

@@ -1,16 +1,24 @@
 //! Rollup Node Driver
 
-use std::sync::Arc;
+use std::{fmt::Debug, sync::Arc};
 
 use async_trait::async_trait;
 use eyre::{bail, Result};
+use kona_derive::{
+    online::{AlloyChainProvider, AlloyL2ChainProvider, OnlineBlobProviderBuilder},
+    traits::{BlobProvider, ChainProvider, L2ChainProvider},
+};
+use kona_primitives::BlockInfo;
+use kona_providers::{
+    blob_provider::DurableBlobProvider, InMemoryChainProvider, LayeredBlobProvider,
+};
 use reth_exex::{ExExContext, ExExEvent, ExExNotification};
 use reth_node_api::FullNodeComponents;
 use superchain_registry::RollupConfig;
 use tokio::sync::mpsc::error::SendError;
 use tracing::{debug, info};
 
-use crate::cli::HeraArgsExt;
+use crate::{new_rollup_pipeline, HeraArgsExt, RollupPipeline};
 
 #[async_trait]
 pub trait DriverContext {
@@ -30,22 +38,76 @@ impl<N: FullNodeComponents> DriverContext for ExExContext<N> {
     }
 }
 
+#[derive(Debug)]
+pub struct StandaloneContext;
+
+#[async_trait]
+impl DriverContext for StandaloneContext {
+    async fn recv_notification(&mut self) -> Option<ExExNotification> {
+        // TODO: we will need a background task to listen for new blocks
+        // (either polling or via websocket), parse them into ExExNotifications
+        // and handle reorgs for the driver to react to.
+        todo!()
+    }
+
+    fn send_event(&mut self, _event: ExExEvent) -> Result<(), SendError<ExExEvent>> {
+        // TODO: When we have a better background notifier abstraction, sending
+        // FinishedHeight events will be useful to make the pipeline advance
+        // safely through reorgs (as it is currently done for ExExContext).
+        Ok(())
+    }
+}
+
 /// The Rollup Driver entrypoint.
 #[derive(Debug)]
-pub struct Driver<DC: DriverContext> {
+pub struct Driver<DC, CP, BP, L2CP> {
     /// The rollup configuration
     cfg: Arc<RollupConfig>,
     /// The context of the node
     ctx: DC,
+    /// The L1 chain provider
+    chain_provider: CP,
+    /// The L1 blob provider
+    blob_provider: BP,
+    /// The L2 chain provider
+    l2_chain_provider: L2CP,
 }
 
-#[allow(unused)]
-impl<DC: DriverContext> Driver<DC> {
-    /// Creates a new instance of the Hera Execution Extension.
-    pub async fn new(ctx: DC, args: HeraArgsExt, cfg: Arc<RollupConfig>) -> Self {
-        Self { ctx, cfg }
+impl<N> Driver<ExExContext<N>, InMemoryChainProvider, LayeredBlobProvider, AlloyL2ChainProvider>
+where
+    N: FullNodeComponents,
+{
+    /// Create a new Hera Execution Extension Driver
+    pub fn exex(ctx: ExExContext<N>, args: HeraArgsExt, cfg: Arc<RollupConfig>) -> Self {
+        let cp = InMemoryChainProvider::with_capacity(1024);
+        let bp = LayeredBlobProvider::new(args.l1_beacon_client_url, args.l1_blob_archiver_url);
+        let l2_cp = AlloyL2ChainProvider::new_http(args.l2_rpc_url, cfg.clone());
+
+        Self { cfg, ctx, chain_provider: cp, blob_provider: bp, l2_chain_provider: l2_cp }
     }
+}
+
+impl Driver<StandaloneContext, AlloyChainProvider, DurableBlobProvider, AlloyL2ChainProvider> {
+    /// Create a new standalone Hera Driver
+    pub fn standalone(ctx: StandaloneContext, args: HeraArgsExt, cfg: Arc<RollupConfig>) -> Self {
+        let cp = AlloyChainProvider::new_http(args.l1_rpc_url);
+        let l2_cp = AlloyL2ChainProvider::new_http(args.l2_rpc_url, cfg.clone());
+        let bp = OnlineBlobProviderBuilder::new()
+            .with_primary(args.l1_beacon_client_url.to_string())
+            .with_fallback(args.l1_blob_archiver_url.map(|url| url.to_string()))
+            .build();
 
+        Self { cfg, ctx, chain_provider: cp, blob_provider: bp, l2_chain_provider: l2_cp }
+    }
+}
+
+impl<DC, CP, BP, L2CP> Driver<DC, CP, BP, L2CP>
+where
+    DC: DriverContext,
+    CP: ChainProvider + Clone + Send + Sync + Debug + 'static,
+    BP: BlobProvider + Clone + Send + Sync + Debug + 'static,
+    L2CP: L2ChainProvider + Clone + Send + Sync + Debug + 'static,
+{
     /// Wait for the L2 genesis L1 block (aka "origin block") to be available in the L1 chain.
     async fn wait_for_l2_genesis_l1_block(&mut self) -> Result<()> {
         loop {
@@ -69,12 +131,30 @@ impl<DC: DriverContext> Driver<DC> {
         }
     }
 
+    /// Initialize the rollup pipeline from the driver's components.
+    fn init_pipeline(&mut self) -> RollupPipeline<CP, BP, L2CP> {
+        new_rollup_pipeline(
+            self.cfg.clone(),
+            self.chain_provider.clone(),
+            self.blob_provider.clone(),
+            self.l2_chain_provider.clone(),
+            // TODO: use a dynamic "tip" block instead of genesis
+            BlockInfo {
+                hash: self.cfg.genesis.l2.hash,
+                number: self.cfg.genesis.l2.number,
+                ..Default::default()
+            },
+        )
+    }
+
     /// Starts the Hera Execution Extension loop.
     pub async fn start(mut self) -> Result<()> {
         // Step 1: Wait for the L2 origin block to be available
         self.wait_for_l2_genesis_l1_block().await?;
         info!("Chain synced to rollup genesis");
 
-        todo!("init pipeline and start processing events");
+        let _pipeline = self.init_pipeline();
+
+        todo!("start processing events");
     }
 }