[wip] Kona ExEx

Cargo.toml

@@ -7,6 +7,7 @@ members = [
   "op-bridge",
   "remote",
   "rollup",
+  "kona"
 ]
 resolver = "2"
 

kona/Cargo.toml

@@ -0,0 +1,39 @@
+[package]
+name = "kona"
+version = "0.1.0"
+publish = false
+edition.workspace = true
+license.workspace = true
+
+[dependencies]
+# kona
+kona-derive = { git = "https://github.com/ethereum-optimism/kona", rev = "e069eb4", features = ["online", "serde"] }
+superchain-registry = { version = "0.2.2", default-features = false }
+
+# needed for compatibility with kona's ChainProvider trait
+anyhow = { version = "1.0.86", default-features = false }
+
+# reth
+reth.workspace = true
+reth-exex.workspace = true
+reth-node-api.workspace = true
+reth-node-ethereum.workspace = true
+
+# alloy
+alloy = { version = "0.2", features = ["full"] }
+alloy-rlp = "0.3.4"
+
+# other
+clap = "4" 
+tracing = "0.1.40"
+async-trait = "0.1.81"
+reqwest = "0.12.5"
+serde_json = "1.0.120"
+parking_lot = "0.12.3"
+eyre.workspace = true
+
+[dev-dependencies]
+reth-exex-test-utils.workspace = true
+reth-testing-utils.workspace = true
+
+tokio.workspace = true

kona/src/blobs.rs

@@ -0,0 +1,137 @@
+//! Blob Provider
+
+use std::{
+    boxed::Box,
+    collections::{HashMap, VecDeque},
+    sync::Arc,
+};
+
+use async_trait::async_trait;
+use kona_derive::{
+    online::{
+        OnlineBeaconClient, OnlineBlobProviderBuilder, OnlineBlobProviderWithFallback,
+        SimpleSlotDerivation,
+    },
+    traits::BlobProvider,
+    types::{alloy_primitives::B256, Blob, BlobProviderError, BlockInfo, IndexedBlobHash},
+};
+use parking_lot::Mutex;
+use reqwest::Url;
+use tracing::warn;
+
+/// [BlobProvider] for the Kona derivation pipeline.
+#[derive(Debug, Clone)]
+pub struct ExExBlobProvider {
+    /// In-memory blob provider, used for locally caching blobs as
+    /// they come during live sync (when following the chain tip).
+    memory: Arc<Mutex<InMemoryBlobProvider>>,
+
+    /// Fallback online blob provider.
+    /// 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
+    /// endpoints for a primary and a fallback blob provider.
+    online: OnlineBlobProviderWithFallback<
+        OnlineBeaconClient,
+        OnlineBeaconClient,
+        SimpleSlotDerivation,
+    >,
+}
+
+/// A blob provider that hold blobs in memory.
+#[derive(Debug)]
+pub struct InMemoryBlobProvider {
+    /// Maximum number of blobs to keep in memory.
+    capacity: usize,
+    /// Order of key insertion for oldest entry eviction.
+    key_order: VecDeque<B256>,
+    /// Maps block hashes to blobs.
+    blocks_to_blob: HashMap<B256, Vec<Blob>>,
+}
+
+impl InMemoryBlobProvider {
+    /// Creates a new [InMemoryBlobProvider].
+    pub fn with_capacity(cap: usize) -> Self {
+        Self {
+            capacity: cap,
+            blocks_to_blob: HashMap::with_capacity(cap),
+            key_order: VecDeque::with_capacity(cap),
+        }
+    }
+
+    /// Inserts multiple blobs into the provider.
+    #[allow(unused)]
+    pub fn insert_blobs(&mut self, block_hash: B256, blobs: Vec<Blob>) {
+        if let Some(existing_blobs) = self.blocks_to_blob.get_mut(&block_hash) {
+            existing_blobs.extend(blobs);
+        } else {
+            if self.blocks_to_blob.len() >= self.capacity {
+                if let Some(oldest) = self.key_order.pop_front() {
+                    self.blocks_to_blob.remove(&oldest);
+                }
+            }
+            self.blocks_to_blob.insert(block_hash, blobs);
+        }
+    }
+}
+
+impl ExExBlobProvider {
+    /// Creates a new [ExExBlobProvider] with a local blob store, an online primary beacon
+    /// client and an optional fallback blob archiver for fetching blobs.
+    pub fn new(beacon_client_url: Url, blob_archiver_url: Option<Url>) -> Self {
+        let memory = Arc::new(Mutex::new(InMemoryBlobProvider::with_capacity(256)));
+
+        let online = OnlineBlobProviderBuilder::new()
+            .with_primary(beacon_client_url.to_string())
+            .with_fallback(blob_archiver_url.map(|url| url.to_string()))
+            .build();
+
+        Self { memory, online }
+    }
+
+    /// Inserts multiple blobs into the in-memory provider.
+    pub fn insert_blobs(&mut self, block_hash: B256, blobs: Vec<Blob>) {
+        self.memory.lock().insert_blobs(block_hash, blobs);
+    }
+
+    /// Attempts to fetch blobs using the in-memory blob store.
+    async fn memory_blob_load(
+        &mut self,
+        block_ref: &BlockInfo,
+        hashes: &[IndexedBlobHash],
+    ) -> eyre::Result<Vec<Blob>> {
+        let locked = self.memory.lock();
+
+        let blobs_for_block = locked
+            .blocks_to_blob
+            .get(&block_ref.hash)
+            .ok_or_else(|| eyre::eyre!("Blob not found for block ref: {:?}", block_ref))?;
+
+        let mut blobs = Vec::new();
+        for _ in hashes {
+            for blob in blobs_for_block {
+                blobs.push(*blob);
+            }
+        }
+
+        Ok(blobs)
+    }
+}
+
+#[async_trait]
+impl BlobProvider for ExExBlobProvider {
+    /// Fetches blobs for a given block ref and the blob hashes.
+    async fn get_blobs(
+        &mut self,
+        block_ref: &BlockInfo,
+        blob_hashes: &[IndexedBlobHash],
+    ) -> Result<Vec<Blob>, BlobProviderError> {
+        if let Ok(b) = self.memory_blob_load(block_ref, blob_hashes).await {
+            return Ok(b);
+        } else {
+            warn!(target: "blob-provider", "Blob provider falling back to online provider");
+            self.online.get_blobs(block_ref, blob_hashes).await
+        }
+    }
+}

kona/src/cli_ext.rs

@@ -0,0 +1,76 @@
+//! CLI Extension module for Kona
+
+use std::path::PathBuf;
+
+use clap::Args;
+use reqwest::Url;
+
+pub const DEFAULT_L2_CHAIN_ID: u64 = 10;
+
+pub const DEFAULT_L2_RPC_URL: &str = "https://optimism.llamarpc.com";
+
+pub const DEFAULT_BEACON_CLIENT_URL: &str = "http://localhost:5052";
+
+#[derive(Debug, Clone, Args)]
+pub(crate) struct KonaArgsExt {
+    /// Chain ID of the L2 network
+    #[clap(long = "kona.l2-chain-id", default_value_t = DEFAULT_L2_CHAIN_ID)]
+    pub l2_chain_id: u64,
+
+    /// RPC URL of an L2 node
+    #[clap(long = "kona.l2-rpc-url", default_value = DEFAULT_L2_RPC_URL)]
+    pub l2_rpc_url: Url,
+
+    /// URL of the beacon client to fetch blobs
+    #[clap(long = "kona.beacon-client-url", default_value = DEFAULT_BEACON_CLIENT_URL)]
+    pub beacon_client_url: Url,
+
+    /// URL of the blob archiver to fetch blobs that are expired on
+    /// the beacon client but still needed for processing.
+    ///
+    /// Blob archivers need to implement the `blob_sidecars` API:
+    /// <https://ethereum.github.io/beacon-APIs/#/Beacon/getBlobSidecars>
+    #[clap(long = "kona.blob-archiver-url")]
+    pub blob_archiver_url: Option<Url>,
+
+    /// The payload validation mode to use.
+    ///
+    /// - Trusted: rely on a trusted synced L2 execution client. Validation happens by fetching the
+    ///   same block and comparing the results.
+    /// - Engine API: use a local or remote engine API of an L2 execution client. Validation
+    ///   happens by sending the `new_payload` to the API and expecting a VALID response.
+    #[clap(
+        long = "kona.validation-mode",
+        default_value = "trusted",
+        requires_ifs([("engine-api", "kona.l2-engine-api-url"), ("engine-api", "kona.l2-engine-jwt-secret")]),
+    )]
+    pub validation_mode: ValidationMode,
+
+    /// If the mode is "engine api", we also need an URL for the engine API endpoint of
+    /// the execution client to validate the payload.
+    #[clap(long = "kona.l2-engine-api-url")]
+    pub l2_engine_api_url: Option<Url>,
+
+    /// If the mode is "engine api", we also need a JWT secret for the auth-rpc.
+    /// This MUST be a valid path to a file containing the hex-encoded JWT secret.
+    #[clap(long = "kona.l2-engine-jwt")]
+    pub l2_engine_jwt_secret: Option<PathBuf>,
+}
+
+#[derive(Debug, Clone)]
+pub(crate) enum ValidationMode {
+    Trusted,
+    EngineApi,
+}
+
+impl std::str::FromStr for ValidationMode {
+    type Err = String;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s.to_lowercase().as_str() {
+            "trusted" => Ok(ValidationMode::Trusted),
+            "engine-api" => Ok(ValidationMode::EngineApi),
+            _ => Err(format!("Invalid validation mode: {}", s)),
+        }
+    }
+}