feat: add configurable RPC middleware to RpcAddOns (#17024)

mattsse · web-flow · commit 265700cf2fc9 · 2025-06-24T08:35:19.000Z
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/crates/ethereum/node/src/node.rs b/crates/ethereum/node/src/node.rs
@@ -25,7 +25,7 @@ use reth_node_builder::{
     node::{FullNodeTypes, NodeTypes},
     rpc::{
         BasicEngineApiBuilder, EngineApiBuilder, EngineValidatorAddOn, EngineValidatorBuilder,
-        EthApiBuilder, EthApiCtx, RethRpcAddOns, RpcAddOns, RpcHandle,
+        EthApiBuilder, EthApiCtx, RethRpcAddOns, RpcAddOns, RpcHandle, RpcServiceBuilder,
     },
     BuilderContext, DebugNode, Node, NodeAdapter, NodeComponentsBuilder, PayloadBuilderConfig,
     PayloadTypes,
@@ -174,6 +174,7 @@ where
                 EthereumEthApiBuilder,
                 EthereumEngineValidatorBuilder::default(),
                 BasicEngineApiBuilder::default(),
+                RpcServiceBuilder::new(),
             ),
         }
     }
diff --git a/crates/node/builder/Cargo.toml b/crates/node/builder/Cargo.toml
@@ -82,6 +82,9 @@ serde_json.workspace = true
 # tracing
 tracing.workspace = true
 
+# tower
+tower.workspace = true
+
 [dev-dependencies]
 tempfile.workspace = true
 
diff --git a/crates/node/builder/src/rpc.rs b/crates/node/builder/src/rpc.rs
@@ -1,5 +1,8 @@
 //! Builder support for rpc components.
 
+pub use jsonrpsee::server::middleware::rpc::{RpcService, RpcServiceBuilder};
+pub use reth_rpc_builder::{Identity, RpcRequestMetricsService};
+
 use crate::{BeaconConsensusEngineEvent, BeaconConsensusEngineHandle};
 use alloy_rpc_types::engine::ClientVersionV1;
 use alloy_rpc_types_engine::ExecutionData;
@@ -31,6 +34,7 @@ use std::{
     future::Future,
     ops::{Deref, DerefMut},
 };
+use tower::Layer;
 
 /// Contains the handles to the spawned RPC servers.
 ///
@@ -417,6 +421,7 @@ pub struct RpcAddOns<
     EthB: EthApiBuilder<Node>,
     EV,
     EB = BasicEngineApiBuilder<EV>,
+    RpcMiddleware = Identity,
 > {
     /// Additional RPC add-ons.
     pub hooks: RpcHooks<Node, EthB::EthApi>,
@@ -426,9 +431,14 @@ pub struct RpcAddOns<
     engine_validator_builder: EV,
     /// Builder for `EngineApi`
     engine_api_builder: EB,
+    /// Configurable RPC middleware stack.
+    ///
+    /// This middleware is applied to all RPC requests across all transports (HTTP, WS, IPC).
+    /// See [`RpcAddOns::with_rpc_middleware`] for more details.
+    rpc_middleware: RpcServiceBuilder<RpcMiddleware>,
 }
 
-impl<Node, EthB, EV, EB> Debug for RpcAddOns<Node, EthB, EV, EB>
+impl<Node, EthB, EV, EB, RpcMiddleware> Debug for RpcAddOns<Node, EthB, EV, EB, RpcMiddleware>
 where
     Node: FullNodeComponents,
     EthB: EthApiBuilder<Node>,
@@ -441,11 +451,12 @@ where
             .field("eth_api_builder", &"...")
             .field("engine_validator_builder", &self.engine_validator_builder)
             .field("engine_api_builder", &self.engine_api_builder)
+            .field("rpc_middleware", &"...")
             .finish()
     }
 }
 
-impl<Node, EthB, EV, EB> RpcAddOns<Node, EthB, EV, EB>
+impl<Node, EthB, EV, EB, RpcMiddleware> RpcAddOns<Node, EthB, EV, EB, RpcMiddleware>
 where
     Node: FullNodeComponents,
     EthB: EthApiBuilder<Node>,
@@ -455,28 +466,98 @@ where
         eth_api_builder: EthB,
         engine_validator_builder: EV,
         engine_api_builder: EB,
+        rpc_middleware: RpcServiceBuilder<RpcMiddleware>,
     ) -> Self {
         Self {
             hooks: RpcHooks::default(),
             eth_api_builder,
             engine_validator_builder,
             engine_api_builder,
+            rpc_middleware,
         }
     }
 
     /// Maps the [`EngineApiBuilder`] builder type.
-    pub fn with_engine_api<T>(self, engine_api_builder: T) -> RpcAddOns<Node, EthB, EV, T> {
-        let Self { hooks, eth_api_builder, engine_validator_builder, .. } = self;
-        RpcAddOns { hooks, eth_api_builder, engine_validator_builder, engine_api_builder }
+    pub fn with_engine_api<T>(
+        self,
+        engine_api_builder: T,
+    ) -> RpcAddOns<Node, EthB, EV, T, RpcMiddleware> {
+        let Self { hooks, eth_api_builder, engine_validator_builder, rpc_middleware, .. } = self;
+        RpcAddOns {
+            hooks,
+            eth_api_builder,
+            engine_validator_builder,
+            engine_api_builder,
+            rpc_middleware,
+        }
     }
 
     /// Maps the [`EngineValidatorBuilder`] builder type.
     pub fn with_engine_validator<T>(
         self,
         engine_validator_builder: T,
-    ) -> RpcAddOns<Node, EthB, T, EB> {
-        let Self { hooks, eth_api_builder, engine_api_builder, .. } = self;
-        RpcAddOns { hooks, eth_api_builder, engine_validator_builder, engine_api_builder }
+    ) -> RpcAddOns<Node, EthB, T, EB, RpcMiddleware> {
+        let Self { hooks, eth_api_builder, engine_api_builder, rpc_middleware, .. } = self;
+        RpcAddOns {
+            hooks,
+            eth_api_builder,
+            engine_validator_builder,
+            engine_api_builder,
+            rpc_middleware,
+        }
+    }
+
+    /// Sets the RPC middleware stack for processing RPC requests.
+    ///
+    /// This method configures a custom middleware stack that will be applied to all RPC requests
+    /// across HTTP, `WebSocket`, and IPC transports. The middleware is applied to the RPC service
+    /// layer, allowing you to intercept, modify, or enhance RPC request processing.
+    ///
+    ///
+    /// # How It Works
+    ///
+    /// The middleware uses the Tower ecosystem's `Layer` pattern. When an RPC server is started,
+    /// the configured middleware stack is applied to create a layered service that processes
+    /// requests in the order the layers were added.
+    ///
+    /// # Examples
+    ///
+    /// ```ignore
+    /// use reth_rpc_builder::{RpcServiceBuilder, RpcRequestMetrics};
+    /// use tower::Layer;
+    ///
+    /// // Simple example with metrics
+    /// let metrics_layer = RpcRequestMetrics::new(metrics_recorder);
+    /// let with_metrics = rpc_addons.with_rpc_middleware(
+    ///     RpcServiceBuilder::new().layer(metrics_layer)
+    /// );
+    ///
+    /// // Composing multiple middleware layers
+    /// let middleware_stack = RpcServiceBuilder::new()
+    ///     .layer(rate_limit_layer)
+    ///     .layer(logging_layer)
+    ///     .layer(metrics_layer);
+    /// let with_full_stack = rpc_addons.with_rpc_middleware(middleware_stack);
+    /// ```
+    ///
+    /// # Notes
+    ///
+    /// - Middleware is applied to the RPC service layer, not the HTTP transport layer
+    /// - The default middleware is `Identity` (no-op), which passes through requests unchanged
+    /// - Middleware layers are applied in the order they are added via `.layer()`
+    pub fn with_rpc_middleware<T>(
+        self,
+        rpc_middleware: RpcServiceBuilder<T>,
+    ) -> RpcAddOns<Node, EthB, EV, EB, T> {
+        let Self { hooks, eth_api_builder, engine_validator_builder, engine_api_builder, .. } =
+            self;
+        RpcAddOns {
+            hooks,
+            eth_api_builder,
+            engine_validator_builder,
+            engine_api_builder,
+            rpc_middleware,
+        }
     }
 
     /// Sets the hook that is run once the rpc server is started.
@@ -500,25 +581,35 @@ where
     }
 }
 
-impl<Node, EthB, EV, EB> Default for RpcAddOns<Node, EthB, EV, EB>
+impl<Node, EthB, EV, EB> Default for RpcAddOns<Node, EthB, EV, EB, Identity>
 where
     Node: FullNodeComponents,
     EthB: EthApiBuilder<Node>,
     EV: Default,
     EB: Default,
 {
     fn default() -> Self {
-        Self::new(EthB::default(), EV::default(), EB::default())
+        Self::new(EthB::default(), EV::default(), EB::default(), RpcServiceBuilder::new())
     }
 }
 
-impl<N, EthB, EV, EB> RpcAddOns<N, EthB, EV, EB>
+impl<N, EthB, EV, EB, RpcMiddleware> RpcAddOns<N, EthB, EV, EB, RpcMiddleware>
 where
     N: FullNodeComponents,
     N::Provider: ChainSpecProvider<ChainSpec: EthereumHardforks>,
     EthB: EthApiBuilder<N>,
     EV: EngineValidatorBuilder<N>,
     EB: EngineApiBuilder<N>,
+    RpcMiddleware: Layer<RpcRequestMetricsService<RpcService>> + Clone + Send + 'static,
+    <RpcMiddleware as Layer<RpcRequestMetricsService<RpcService>>>::Service:
+        Send
+            + Sync
+            + 'static
+            + jsonrpsee::server::middleware::rpc::RpcServiceT<
+                MethodResponse = jsonrpsee::MethodResponse,
+                BatchResponse = jsonrpsee::MethodResponse,
+                NotificationResponse = jsonrpsee::MethodResponse,
+            >,
 {
     /// Launches only the regular RPC server (HTTP/WS/IPC), without the authenticated Engine API
     /// server.
@@ -533,6 +624,7 @@ where
     where
         F: FnOnce(RpcModuleContainer<'_, N, EthB::EthApi>) -> eyre::Result<()>,
     {
+        let rpc_middleware = self.rpc_middleware.clone();
         let setup_ctx = self.setup_rpc_components(ctx, ext).await?;
         let RpcSetupContext {
             node,
@@ -546,7 +638,7 @@ where
             engine_handle,
         } = setup_ctx;
 
-        let server_config = config.rpc.rpc_server_config();
+        let server_config = config.rpc.rpc_server_config().set_rpc_middleware(rpc_middleware);
         let rpc_server_handle = Self::launch_rpc_server_internal(server_config, &modules).await?;
 
         let handles =
@@ -579,6 +671,7 @@ where
     where
         F: FnOnce(RpcModuleContainer<'_, N, EthB::EthApi>) -> eyre::Result<()>,
     {
+        let rpc_middleware = self.rpc_middleware.clone();
         let setup_ctx = self.setup_rpc_components(ctx, ext).await?;
         let RpcSetupContext {
             node,
@@ -592,7 +685,7 @@ where
             engine_handle,
         } = setup_ctx;
 
-        let server_config = config.rpc.rpc_server_config();
+        let server_config = config.rpc.rpc_server_config().set_rpc_middleware(rpc_middleware);
         let auth_module_clone = auth_module.clone();
 
         // launch servers concurrently
@@ -706,10 +799,22 @@ where
     }
 
     /// Helper to launch the RPC server
-    async fn launch_rpc_server_internal(
-        server_config: RpcServerConfig,
+    async fn launch_rpc_server_internal<M>(
+        server_config: RpcServerConfig<M>,
         modules: &TransportRpcModules,
-    ) -> eyre::Result<RpcServerHandle> {
+    ) -> eyre::Result<RpcServerHandle>
+    where
+        M: Layer<RpcRequestMetricsService<RpcService>> + Clone + Send + 'static,
+        for<'a> <M as Layer<RpcRequestMetricsService<RpcService>>>::Service:
+            Send
+                + Sync
+                + 'static
+                + jsonrpsee::server::middleware::rpc::RpcServiceT<
+                    MethodResponse = jsonrpsee::MethodResponse,
+                    BatchResponse = jsonrpsee::MethodResponse,
+                    NotificationResponse = jsonrpsee::MethodResponse,
+                >,
+    {
         let handle = server_config.start(modules).await?;
 
         if let Some(path) = handle.ipc_endpoint() {
@@ -760,13 +865,23 @@ where
     }
 }
 
-impl<N, EthB, EV, EB> NodeAddOns<N> for RpcAddOns<N, EthB, EV, EB>
+impl<N, EthB, EV, EB, RpcMiddleware> NodeAddOns<N> for RpcAddOns<N, EthB, EV, EB, RpcMiddleware>
 where
     N: FullNodeComponents,
     <N as FullNodeTypes>::Provider: ChainSpecProvider<ChainSpec: EthereumHardforks>,
     EthB: EthApiBuilder<N>,
     EV: EngineValidatorBuilder<N>,
     EB: EngineApiBuilder<N>,
+    RpcMiddleware: Layer<RpcRequestMetricsService<RpcService>> + Clone + Send + 'static,
+    <RpcMiddleware as Layer<RpcRequestMetricsService<RpcService>>>::Service:
+        Send
+            + Sync
+            + 'static
+            + jsonrpsee::server::middleware::rpc::RpcServiceT<
+                MethodResponse = jsonrpsee::MethodResponse,
+                BatchResponse = jsonrpsee::MethodResponse,
+                NotificationResponse = jsonrpsee::MethodResponse,
+            >,
 {
     type Handle = RpcHandle<N, EthB::EthApi>;
 
@@ -787,7 +902,8 @@ pub trait RethRpcAddOns<N: FullNodeComponents>:
     fn hooks_mut(&mut self) -> &mut RpcHooks<N, Self::EthApi>;
 }
 
-impl<N: FullNodeComponents, EthB, EV, EB> RethRpcAddOns<N> for RpcAddOns<N, EthB, EV, EB>
+impl<N: FullNodeComponents, EthB, EV, EB, RpcMiddleware> RethRpcAddOns<N>
+    for RpcAddOns<N, EthB, EV, EB, RpcMiddleware>
 where
     Self: NodeAddOns<N, Handle = RpcHandle<N, EthB::EthApi>>,
     EthB: EthApiBuilder<N>,
diff --git a/crates/optimism/node/src/node.rs b/crates/optimism/node/src/node.rs
@@ -27,7 +27,7 @@ use reth_node_builder::{
     node::{FullNodeTypes, NodeTypes},
     rpc::{
         EngineApiBuilder, EngineValidatorAddOn, EngineValidatorBuilder, EthApiBuilder,
-        RethRpcAddOns, RethRpcServerHandles, RpcAddOns, RpcContext, RpcHandle,
+        RethRpcAddOns, RethRpcServerHandles, RpcAddOns, RpcContext, RpcHandle, RpcServiceBuilder,
     },
     BuilderContext, DebugNode, Node, NodeAdapter, NodeComponentsBuilder,
 };
@@ -255,6 +255,9 @@ impl NodeTypes for OpNode {
 }
 
 /// Add-ons w.r.t. optimism.
+///
+/// This type provides optimism-specific addons to the node and exposes the RPC server and engine
+/// API.
 #[derive(Debug)]
 pub struct OpAddOns<N: FullNodeComponents, EthB: EthApiBuilder<N>, EV, EB> {
     /// Rpc add-ons responsible for launching the RPC servers and instantiating the RPC handlers
@@ -591,6 +594,7 @@ impl<NetworkT> OpAddOnsBuilder<NetworkT> {
                     .with_min_suggested_priority_fee(min_suggested_priority_fee),
                 EV::default(),
                 EB::default(),
+                RpcServiceBuilder::new(),
             ),
             da_config: da_config.unwrap_or_default(),
             sequencer_url,
