feat(based-op-reth): based-op changes for reth without fork

based/crates/common/src/p2p.rs

@@ -17,17 +17,17 @@ pub type ExtraData = VariableList<u8, MaxExtraDataSize>;
 #[derive(Debug, Clone, PartialEq, Eq, TreeHash, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct EnvV0 {
-    number: u64,
-    parent_hash: B256,
-    beneficiary: Address,
-    timestamp: u64,
-    gas_limit: u64,
-    basefee: u64,
-    difficulty: U256,
-    prevrandao: B256,
+    pub number: u64,
+    pub parent_hash: B256,
+    pub beneficiary: Address,
+    pub timestamp: u64,
+    pub gas_limit: u64,
+    pub basefee: u64,
+    pub difficulty: U256,
+    pub prevrandao: B256,
     #[serde(with = "ssz_types::serde_utils::hex_var_list")]
-    extra_data: ExtraData,
-    parent_beacon_block_root: B256,
+    pub extra_data: ExtraData,
+    pub parent_beacon_block_root: B256,
 }
 
 impl EnvV0 {
@@ -81,6 +81,10 @@ impl FragV0 {
         let txs = builder_txs.map(|tx| tx.encode().to_vec()).map(Transaction::from).collect::<Vec<_>>();
         Self { block_number, seq, txs: Transactions::from(txs), is_last, blob_gas_used }
     }
+
+    pub fn is_first(&self) -> bool {
+        self.seq == 0
+    }
 }
 
 /// A message sealing a sequence of frags, with fields from the block header
@@ -145,6 +149,12 @@ impl From<EnvV0> for VersionedMessage {
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Signed<T: TreeHash> {
+    pub message: T,
+    pub signature: Bytes,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct SignedVersionedMessage {
     pub message: VersionedMessage,

based/crates/reth/Cargo.toml

@@ -0,0 +1,30 @@
+[package]
+edition.workspace = true
+name = "reth"
+repository.workspace = true
+rust-version.workspace = true
+version.workspace = true
+
+[dependencies]
+alloy-consensus.workspace = true
+alloy-eips.workspace = true
+alloy-network.workspace = true
+alloy-primitives.workspace = true
+alloy-rpc-types.workspace = true
+anyhow = "1.0.98"
+bop-common.workspace = true
+clap.workspace = true
+eyre.workspace = true
+futures.workspace = true
+jsonrpsee.workspace = true
+op-alloy-rpc-types.workspace = true
+thiserror.workspace = true
+tokio.workspace = true
+tracing.workspace = true
+
+reth-exex = { git = "https://github.com/paradigmxyz/reth", tag = "v1.9.3" }
+reth-node-builder = { git = "https://github.com/paradigmxyz/reth", tag = "v1.9.3" }
+reth-optimism-chainspec = { git = "https://github.com/paradigmxyz/reth", tag = "v1.9.3" }
+reth-optimism-cli = { git = "https://github.com/paradigmxyz/reth", tag = "v1.9.3" }
+reth-optimism-node = { git = "https://github.com/paradigmxyz/reth", tag = "v1.9.3" }
+reth-rpc = { git = "https://github.com/paradigmxyz/reth", tag = "v1.9.3" }

based/crates/reth/src/api/engine.rs

@@ -0,0 +1,65 @@
+use bop_common::p2p::{EnvV0, FragV0, SealV0, Signed};
+use jsonrpsee::{
+    core::{RpcResult, async_trait},
+    proc_macros::rpc,
+    types::{ErrorCode, ErrorObject},
+};
+
+use crate::{driver::Driver, error::DriverError};
+
+#[rpc(server, namespace = "engine")]
+pub trait BasedEngineApi {
+    #[method(name = "envV0")]
+    async fn env_v0(&self, env: Signed<EnvV0>) -> RpcResult<()>;
+
+    #[method(name = "newFragV0")]
+    async fn new_frag_v0(&self, frag: Signed<FragV0>) -> RpcResult<()>;
+
+    #[method(name = "sealFragV0")]
+    async fn seal_frag_v0(&self, seal: Signed<SealV0>) -> RpcResult<()>;
+}
+
+pub struct BasedEngineApi {
+    driver: Driver,
+}
+
+impl BasedEngineApi {
+    /// Initialize a new based engine API instance.
+    pub fn new(driver: Driver) -> Self {
+        Self { driver }
+    }
+}
+
+#[async_trait]
+impl BasedEngineApiServer for BasedEngineApi {
+    #[tracing::instrument(skip(self))]
+    async fn env_v0(&self, env: Signed<EnvV0>) -> RpcResult<()> {
+        tracing::debug!("handling engine_envV0");
+
+        self.driver.env_v0(env.message).await?;
+        Ok(())
+    }
+
+    #[tracing::instrument(skip(self))]
+    async fn new_frag_v0(&self, frag: Signed<FragV0>) -> RpcResult<()> {
+        tracing::debug!("handling engine_newFragV0");
+
+        self.driver.new_frag_v0(frag.message).await?;
+        Ok(())
+    }
+
+    #[tracing::instrument(skip(self))]
+    async fn seal_frag_v0(&self, seal: Signed<SealV0>) -> RpcResult<()> {
+        tracing::debug!("handling engine_sealFragV0");
+
+        self.driver.seal_frag_v0(seal.message).await?;
+        Ok(())
+    }
+}
+
+impl From<DriverError> for ErrorObject<'static> {
+    // TODO: Better error handling
+    fn from(e: DriverError) -> Self {
+        ErrorObject::owned(ErrorCode::InternalError.code(), e.to_string(), Option::<()>::None)
+    }
+}

based/crates/reth/src/api/mod.rs

@@ -0,0 +1 @@
+pub mod engine;

based/crates/reth/src/cli.rs

@@ -0,0 +1,76 @@
+use clap::Parser;
+use futures::TryStreamExt as _;
+use reth_exex::ExExEvent;
+use reth_node_builder::Node;
+use reth_optimism_cli::{Cli, chainspec::OpChainSpecParser};
+use reth_optimism_node::{OpNode, args::RollupArgs};
+
+use crate::{
+    api::engine::{BasedEngineApi, BasedEngineApiServer as _},
+    driver::Driver,
+};
+
+#[derive(Parser, Debug, Clone)]
+pub struct BasedOpRethArgs {
+    #[command(flatten)]
+    pub rollup: RollupArgs,
+    #[command(flatten)]
+    pub based_op: BasedOpArgs,
+}
+
+#[derive(Parser, Debug, Clone)]
+pub struct BasedOpArgs {
+    /// Whether to use the unsealed block as the "latest" state in RPC calls.
+    #[arg(long)]
+    unsealed_as_latest: bool,
+}
+
+pub fn run() -> eyre::Result<()> {
+    Cli::<OpChainSpecParser, BasedOpRethArgs>::parse().run(|builder, args| async move {
+        let driver = Driver::new(args.based_op.unsealed_as_latest);
+
+        let op_node = OpNode::new(args.rollup.clone());
+
+        let node_handle = builder
+            .with_types::<OpNode>()
+            .with_components(op_node.components())
+            .with_add_ons(op_node.add_ons())
+            // Install the execution extension to handle canonical chain updates
+            .install_exex("based-op", {
+                // Get a clone of the driver handle.
+                let driver = driver.clone();
+                move |mut ctx| async move {
+                    Ok(async move {
+                        while let Some(note) = ctx.notifications.try_next().await? {
+                            if let Some(committed) = note.committed_chain() {
+                                // Handle committed blocks by notifying the driver
+                                for block in committed.blocks_iter() {
+                                    driver.forkchoice_updated(block.clone().into_block()).await?;
+                                }
+                                let _ = ctx.events.send(ExExEvent::FinishedHeight(committed.tip().num_hash()));
+                            }
+                        }
+
+                        Ok(())
+                    })
+                }
+            })
+            .extend_rpc_modules(move |ctx| {
+                // Add based engine API modules to the existing auth module.
+                ctx.auth_module
+                    .merge_auth_methods(BasedEngineApi::new(driver).into_rpc())
+                    .expect("failed to merge modules");
+                // TODO:
+                // - Replace / extend the engine API
+                // - Replace eth API
+                Ok(())
+            })
+            .launch()
+            .await?;
+
+        // Run to completion
+        node_handle.wait_for_node_exit().await?;
+
+        Ok(())
+    })
+}