Implement txpool interop support for optimism

crates/optimism/node/src/args.rs

@@ -2,6 +2,9 @@
 
 //! clap [Args](clap::Args) for optimism rollup configuration
 
+use op_alloy_consensus::interop::SafetyLevel;
+use reth_optimism_txpool::supervisor::DEFAULT_SUPERVISOR_URL;
+
 /// Parameters for rollup configuration
 #[derive(Debug, Clone, PartialEq, Eq, clap::Args)]
 #[command(next_help_heading = "Rollup")]
@@ -37,9 +40,23 @@ pub struct RollupArgs {
     /// Enable transaction conditional support on sequencer
     #[arg(long = "rollup.enable-tx-conditional", default_value = "false")]
     pub enable_tx_conditional: bool,
+
+    /// HTTP endpoint for the supervisor
+    #[arg(
+        long = "rollup.supervisor-http",
+        value_name = "SUPERVISOR_HTTP_URL",
+        default_value = DEFAULT_SUPERVISOR_URL
+    )]
+    pub supervisor_http: String,
+
+    /// Safety level for the supervisor
+    #[arg(
+        long = "rollup.supervisor-safety-level",
+        default_value_t = SafetyLevel::CrossUnsafe,
+    )]
+    pub supervisor_safety_level: SafetyLevel,
 }
 
-#[expect(clippy::derivable_impls)]
 impl Default for RollupArgs {
     fn default() -> Self {
         Self {
@@ -49,6 +66,8 @@ impl Default for RollupArgs {
             compute_pending_block: false,
             discovery_v4: false,
             enable_tx_conditional: false,
+            supervisor_http: DEFAULT_SUPERVISOR_URL.to_string(),
+            supervisor_safety_level: SafetyLevel::CrossUnsafe,
         }
     }
 }

crates/optimism/node/src/node.rs

@@ -6,7 +6,7 @@ use crate::{
     txpool::{OpTransactionPool, OpTransactionValidator},
     OpEngineApiBuilder, OpEngineTypes,
 };
-use op_alloy_consensus::OpPooledTransaction;
+use op_alloy_consensus::{interop::SafetyLevel, OpPooledTransaction};
 use reth_chainspec::{EthChainSpec, Hardforks};
 use reth_evm::{execute::BasicBlockExecutorProvider, ConfigureEvm, EvmFactory, EvmFactoryFor};
 use reth_network::{NetworkConfig, NetworkHandle, NetworkManager, NetworkPrimitives, PeersInfo};
@@ -41,7 +41,10 @@ use reth_optimism_rpc::{
     OpEthApi, OpEthApiError, SequencerClient,
 };
 use reth_optimism_txpool::{
-    conditional::MaybeConditionalTransaction, interop::MaybeInteropTransaction, OpPooledTx,
+    conditional::MaybeConditionalTransaction,
+    interop::MaybeInteropTransaction,
+    supervisor::{SupervisorClient, DEFAULT_SUPERVISOR_URL},
+    OpPooledTx,
 };
 use reth_provider::{providers::ProviderFactoryBuilder, CanonStateSubscriptions, EthStorage};
 use reth_rpc_api::DebugApiServer;
@@ -113,7 +116,11 @@ impl OpNode {
             .node_types::<Node>()
             .pool(
                 OpPoolBuilder::default()
-                    .with_enable_tx_conditional(self.args.enable_tx_conditional),
+                    .with_enable_tx_conditional(self.args.enable_tx_conditional)
+                    .with_supervisor(
+                        self.args.supervisor_http.clone(),
+                        self.args.supervisor_safety_level,
+                    ),
             )
             .payload(BasicPayloadServiceBuilder::new(
                 OpPayloadBuilder::new(compute_pending_block).with_da_config(self.da_config.clone()),
@@ -479,6 +486,10 @@ pub struct OpPoolBuilder<T = crate::txpool::OpPooledTransaction> {
     pub pool_config_overrides: PoolBuilderConfigOverrides,
     /// Enable transaction conditionals.
     pub enable_tx_conditional: bool,
+    /// Supervisor client url
+    pub supervisor_http: String,
+    /// Supervisor safety level
+    pub supervisor_safety_level: SafetyLevel,
     /// Marker for the pooled transaction type.
     _pd: core::marker::PhantomData<T>,
 }
@@ -488,6 +499,8 @@ impl<T> Default for OpPoolBuilder<T> {
         Self {
             pool_config_overrides: Default::default(),
             enable_tx_conditional: false,
+            supervisor_http: DEFAULT_SUPERVISOR_URL.to_string(),
+            supervisor_safety_level: SafetyLevel::CrossUnsafe,
             _pd: Default::default(),
         }
     }
@@ -508,6 +521,17 @@ impl<T> OpPoolBuilder<T> {
         self.pool_config_overrides = pool_config_overrides;
         self
     }
+
+    /// Sets the supervisor client
+    pub fn with_supervisor(
+        mut self,
+        supervisor_client: String,
+        supervisor_safety_level: SafetyLevel,
+    ) -> Self {
+        self.supervisor_http = supervisor_client;
+        self.supervisor_safety_level = supervisor_safety_level;
+        self
+    }
 }
 
 impl<Node, T> PoolBuilder<Node> for OpPoolBuilder<T>
@@ -523,6 +547,17 @@ where
         let Self { pool_config_overrides, .. } = self;
         let data_dir = ctx.config().datadir();
         let blob_store = DiskFileBlobStore::open(data_dir.blobstore(), Default::default())?;
+        // supervisor used for interop
+        if ctx.chain_spec().is_interop_active_at_timestamp(ctx.head().timestamp) &&
+            self.supervisor_http == DEFAULT_SUPERVISOR_URL
+        {
+            info!(target: "reth::cli",
+                url=%DEFAULT_SUPERVISOR_URL,
+                "Default supervisor url is used, consider changing --rollup.supervisor-http."
+            );
+        }
+        let supervisor_client =
+            SupervisorClient::new(self.supervisor_http.clone(), self.supervisor_safety_level).await;
 
         let validator = TransactionValidationTaskExecutor::eth_builder(ctx.provider().clone())
             .no_eip4844()
@@ -539,6 +574,7 @@ where
                     // In --dev mode we can't require gas fees because we're unable to decode
                     // the L1 block info
                     .require_l1_data_gas_fee(!ctx.config().dev.dev)
+                    .with_supervisor(supervisor_client.clone())
             });
 
         let transaction_pool = reth_transaction_pool::Pool::new(

crates/optimism/txpool/Cargo.toml

@@ -17,6 +17,7 @@ alloy-consensus.workspace = true
 alloy-eips.workspace = true
 alloy-primitives.workspace = true
 alloy-rpc-types-eth.workspace = true
+alloy-rpc-client = { workspace = true, features = ["reqwest", "default"] }
 
 # reth
 reth-chainspec.workspace = true
@@ -44,6 +45,10 @@ c-kzg.workspace = true
 derive_more.workspace = true
 futures-util.workspace = true
 parking_lot.workspace = true
+serde.workspace = true
+tracing.workspace = true
+thiserror.workspace = true
+tokio = { workspace = true, features = ["time"] }
 
 [dev-dependencies]
 reth-optimism-chainspec.workspace = true

crates/optimism/txpool/src/error.rs

@@ -0,0 +1,47 @@
+use crate::supervisor::{InteropTxValidatorError, InvalidInboxEntry};
+use op_alloy_consensus::interop::SafetyLevel;
+use reth_transaction_pool::error::PoolTransactionError;
+use std::any::Any;
+
+/// Wrapper for [`InteropTxValidatorError`] to implement [`PoolTransactionError`] for it.
+#[derive(thiserror::Error, Debug)]
+pub enum InvalidCrossTx {
+    /// Errors produced by supervisor validation
+    #[error(transparent)]
+    ValidationError(#[from] InteropTxValidatorError),
+    /// Error cause by cross chain tx during not active interop hardfork
+    #[error("cross chain tx is invalid before interop")]
+    CrossChainTxPreInterop,
+}
+
+impl PoolTransactionError for InvalidCrossTx {
+    fn is_bad_transaction(&self) -> bool {
+        match self {
+            Self::ValidationError(err) => {
+                match err {
+                    InteropTxValidatorError::InvalidInboxEntry(err) => match err {
+                        // This transaction could become valid after a while
+                        InvalidInboxEntry::MinimumSafety { got, .. } => match got {
+                            // This transaction will never become valid
+                            SafetyLevel::Invalid => true,
+                            // This transaction will become valid when origin chain progress
+                            _ => false,
+                        },
+                        // This tx will not become valid unless supervisor is reconfigured
+                        InvalidInboxEntry::UnknownChain(_) => true,
+                    },
+                    // Rpc error or supervisor haven't responded in time
+                    InteropTxValidatorError::RpcClientError(_) |
+                    InteropTxValidatorError::ValidationTimeout(_) => false,
+                    // Transaction caused unknown (for parsing) error in supervisor
+                    InteropTxValidatorError::SupervisorServerError(_) => true,
+                }
+            }
+            Self::CrossChainTxPreInterop => true,
+        }
+    }
+
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+}

crates/optimism/txpool/src/lib.rs

@@ -12,10 +12,13 @@ mod validator;
 pub use validator::{OpL1BlockInfo, OpTransactionValidator};
 
 pub mod conditional;
+pub mod supervisor;
 mod transaction;
 pub use transaction::{OpPooledTransaction, OpPooledTx};
+mod error;
 pub mod interop;
 pub mod maintain;
+pub use error::InvalidCrossTx;
 
 use reth_transaction_pool::{CoinbaseTipOrdering, Pool, TransactionValidationTaskExecutor};
 

crates/optimism/txpool/src/supervisor/access_list.rs

@@ -0,0 +1,41 @@
+// Source: https://github.com/op-rs/kona
+// Copyright © 2023 kona contributors Copyright © 2024 Optimism
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy of this software and
+// associated documentation files (the “Software”), to deal in the Software without restriction,
+// including without limitation the rights to use, copy, modify, merge, publish, distribute,
+// sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all copies or
+// substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT
+// NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+// NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
+// DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+use crate::supervisor::CROSS_L2_INBOX_ADDRESS;
+use alloy_eips::eip2930::AccessListItem;
+use alloy_primitives::B256;
+
+/// Parses [`AccessListItem`]s to inbox entries.
+///
+/// Return flattened iterator with all inbox entries.
+pub fn parse_access_list_items_to_inbox_entries<'a>(
+    access_list_items: impl Iterator<Item = &'a AccessListItem>,
+) -> impl Iterator<Item = &'a B256> {
+    access_list_items.filter_map(parse_access_list_item_to_inbox_entries).flatten()
+}
+
+/// Parse [`AccessListItem`] to inbox entries, if any.
+/// Max 3 inbox entries can exist per [`AccessListItem`] that points to [`CROSS_L2_INBOX_ADDRESS`].
+///
+/// Returns `Vec::new()` if [`AccessListItem`] address doesn't point to [`CROSS_L2_INBOX_ADDRESS`].
+// TODO: add url to spec once [pr](https://github.com/ethereum-optimism/specs/pull/612) is merged
+fn parse_access_list_item_to_inbox_entries(
+    access_list_item: &AccessListItem,
+) -> Option<impl Iterator<Item = &B256>> {
+    (access_list_item.address == CROSS_L2_INBOX_ADDRESS)
+        .then(|| access_list_item.storage_keys.iter())
+}

crates/optimism/txpool/src/supervisor/client.rs

@@ -0,0 +1,107 @@
+//! This is our custom implementation of validator struct
+
+use crate::supervisor::{ExecutingDescriptor, InteropTxValidatorError};
+use alloy_primitives::B256;
+use alloy_rpc_client::ReqwestClient;
+use futures_util::future::BoxFuture;
+use op_alloy_consensus::interop::SafetyLevel;
+use std::{borrow::Cow, future::IntoFuture, time::Duration};
+
+/// Supervisor hosted by op-labs
+// TODO: This should be changes to actual supervisor url
+pub const DEFAULT_SUPERVISOR_URL: &str = "http://localhost:1337/";
+
+/// The default request timeout to use
+const DEFAULT_REQUEST_TIMOUT: Duration = Duration::from_millis(100);
+
+/// Implementation of the supervisor trait for the interop.
+#[derive(Debug, Clone)]
+pub struct SupervisorClient {
+    client: ReqwestClient,
+    /// The default
+    safety: SafetyLevel,
+    /// The default request timeout
+    timeout: Duration,
+}
+
+impl SupervisorClient {
+    /// Creates a new supervisor validator.
+    pub async fn new(supervisor_endpoint: impl Into<String>, safety: SafetyLevel) -> Self {
+        let client = ReqwestClient::builder()
+            .connect(supervisor_endpoint.into().as_str())
+            .await
+            .expect("building supervisor client");
+        Self { client, safety, timeout: DEFAULT_REQUEST_TIMOUT }
+    }
+
+    /// Configures a custom timeout
+    pub fn with_timeout(mut self, timeout: Duration) -> Self {
+        self.timeout = timeout;
+        self
+    }
+
+    /// Returns safely level
+    pub fn safety(&self) -> SafetyLevel {
+        self.safety
+    }
+
+    /// Executes a `supervisor_checkAccessList` with the configured safety level.
+    pub fn check_access_list<'a>(
+        &self,
+        inbox_entries: &'a [B256],
+        executing_descriptor: ExecutingDescriptor,
+    ) -> CheckAccessListRequest<'a> {
+        CheckAccessListRequest {
+            client: self.client.clone(),
+            inbox_entries: Cow::Borrowed(inbox_entries),
+            executing_descriptor,
+            timeout: self.timeout,
+            safety: self.safety,
+        }
+    }
+}
+
+/// A Request future that issues a `supervisor_checkAccessList` request.
+#[derive(Debug, Clone)]
+pub struct CheckAccessListRequest<'a> {
+    client: ReqwestClient,
+    inbox_entries: Cow<'a, [B256]>,
+    executing_descriptor: ExecutingDescriptor,
+    timeout: Duration,
+    safety: SafetyLevel,
+}
+
+impl CheckAccessListRequest<'_> {
+    /// Configures the timeout to use for the request if any.
+    pub fn with_timeout(mut self, timeout: Duration) -> Self {
+        self.timeout = timeout;
+        self
+    }
+
+    /// Configures the [`SafetyLevel`] for this request
+    pub fn with_safety(mut self, safety: SafetyLevel) -> Self {
+        self.safety = safety;
+        self
+    }
+}
+
+impl<'a> IntoFuture for CheckAccessListRequest<'a> {
+    type Output = Result<(), InteropTxValidatorError>;
+    type IntoFuture = BoxFuture<'a, Self::Output>;
+
+    fn into_future(self) -> Self::IntoFuture {
+        let Self { client, inbox_entries, executing_descriptor, timeout, safety } = self;
+        Box::pin(async move {
+            tokio::time::timeout(
+                timeout,
+                client.request(
+                    "supervisor_checkAccessList",
+                    (inbox_entries, safety, executing_descriptor),
+                ),
+            )
+            .await
+            .map_err(|_| InteropTxValidatorError::ValidationTimeout(timeout.as_secs()))?
+            .map_err(InteropTxValidatorError::client)
+        })
+    }
+}