From fe513bb4a86187055eeef6db8b7f0cd8432bfc65 Mon Sep 17 00:00:00 2001 From: Mitko Tschimev Date: Tue, 22 Sep 2026 13:46:49 +0400 Subject: [PATCH] feat: isolate consensus RPC on an optional second IPC socket Public HTTP/WS and arc-consensus share one EthApi eth_call semaphore. Under load, get_signing_validator_set waits out ETH_DEFAULT_TIMEOUT and the CL exits. --consensus-ipcpath serves consensus on a second EthApi with 16 permits. Unset, consensus still uses --ipcpath. --- Cargo.lock | 1 + crates/eth-engine/tests/integration.rs | 1 + crates/evm-node/Cargo.toml | 1 + crates/evm-node/src/node.rs | 149 ++++++++++++++++++++++--- crates/node/src/main.rs | 24 ++++ crates/test/integration/src/runner.rs | 1 + 6 files changed, 160 insertions(+), 17 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 560acdde..496b4ff7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1288,6 +1288,7 @@ dependencies = [ "reth-ethereum-payload-builder", "reth-ethereum-primitives", "reth-evm", + "reth-ipc", "reth-network", "reth-node-api", "reth-node-builder", diff --git a/crates/eth-engine/tests/integration.rs b/crates/eth-engine/tests/integration.rs index 3922ab91..75c87f99 100644 --- a/crates/eth-engine/tests/integration.rs +++ b/crates/eth-engine/tests/integration.rs @@ -186,6 +186,7 @@ async fn test_engine() { std::time::Duration::from_secs(0), // disable rebroadcast in integration tests Vec::new(), DEFAULT_TX_RELAY_TIMEOUT, + None, ); let node_handle = NodeBuilder::new(node_config) .testing_node(executor) diff --git a/crates/evm-node/Cargo.toml b/crates/evm-node/Cargo.toml index e78900dd..b90c0a91 100644 --- a/crates/evm-node/Cargo.toml +++ b/crates/evm-node/Cargo.toml @@ -42,6 +42,7 @@ reth-ethereum-engine-primitives.workspace = true reth-ethereum-payload-builder.workspace = true reth-ethereum-primitives.workspace = true reth-evm.workspace = true +reth-ipc.workspace = true reth-network.workspace = true reth-node-api.workspace = true reth-node-builder.workspace = true diff --git a/crates/evm-node/src/node.rs b/crates/evm-node/src/node.rs index b8ce9f4d..8e5fafe1 100644 --- a/crates/evm-node/src/node.rs +++ b/crates/evm-node/src/node.rs @@ -32,6 +32,7 @@ use reth_ethereum::{node::EthEngineTypes, node::EthEvmConfig}; use reth_ethereum_engine_primitives::{EthBuiltPayload, EthPayloadAttributes}; use reth_ethereum_primitives::EthPrimitives; use reth_evm::{ConfigureEvm, EvmFactory, EvmFactoryFor, NextBlockEnvAttributes}; +use reth_ipc::server::Builder as IpcServerBuilder; use reth_network::{primitives::BasicNetworkPrimitives, NetworkHandle, PeersInfo}; use reth_node_api::{ AddOnsContext, FullNodeComponents, HeaderTy, NodeAddOns, PayloadAttributesBuilder, @@ -56,7 +57,7 @@ use reth_rpc::{ eth::core::{EthApiFor, EthRpcConverterFor}, ValidationApi, }; -use reth_rpc_api::servers::BlockSubmissionValidationApiServer; +use reth_rpc_api::servers::{BlockSubmissionValidationApiServer, EthApiServer, RethApiServer}; use reth_rpc_builder::{ config::RethRpcServerConfig, middleware::RethRpcMiddleware, TransportRpcModules, }; @@ -67,7 +68,7 @@ use reth_rpc_eth_api::{ }, RpcConvert, RpcTypes, SignableTxRequest, }; -use reth_rpc_eth_types::{error::FromEvmError, EthApiError}; +use reth_rpc_eth_types::{error::FromEvmError, EthApiError, EthStateCache}; use reth_rpc_server_types::RethRpcModule; use reth_tracing::tracing::{info, warn}; use reth_transaction_pool::{PoolPooledTx, PoolTransaction, TransactionPool}; @@ -87,6 +88,10 @@ use crate::rpc_middleware::{ }; use crate::ArcEngineValidator; +/// Concurrent `eth_call` permits on the consensus-only IPC socket. Public +/// HTTP/WS keep Reth's `--rpc.max-blocking-io-requests` (default 256). +const CONSENSUS_IPC_MAX_BLOCKING_IO: usize = 16; + /// Bundle RPC methods that Arc never exposes on public RPC transports. const BUNDLE_RPC_METHODS: [&str; 6] = [ "eth_callBundle", @@ -136,6 +141,9 @@ pub struct ArcNode { pub tx_relays: Vec, /// Connection timeout for relays, and request timeout for async relayed submissions. pub tx_relay_timeout: std::time::Duration, + /// Optional second IPC socket reserved for `arc-consensus`. Empty/`None` keeps + /// upstream behaviour (consensus shares the public `--ipcpath` EthApi). + pub consensus_ipcpath: Option, } impl ArcNode { @@ -154,6 +162,7 @@ impl ArcNode { rebroadcast_interval: crate::rebroadcast::DEFAULT_REBROADCAST_INTERVAL, tx_relays: Vec::new(), tx_relay_timeout: DEFAULT_TX_RELAY_TIMEOUT, + consensus_ipcpath: None, } } @@ -172,6 +181,7 @@ impl ArcNode { rebroadcast_interval: std::time::Duration, tx_relays: Vec, tx_relay_timeout: std::time::Duration, + consensus_ipcpath: Option, ) -> Self { Self { rpc_cfg, @@ -186,6 +196,7 @@ impl ArcNode { rebroadcast_interval, tx_relays, tx_relay_timeout, + consensus_ipcpath, } } @@ -315,6 +326,7 @@ pub struct ArcAddOns< > { inner: RpcAddOns, arc_rpc: ArcRpcConfig, + consensus_ipcpath: Option, } impl ArcAddOns @@ -327,7 +339,11 @@ where inner: RpcAddOns, arc_rpc: ArcRpcConfig, ) -> Self { - Self { inner, arc_rpc } + Self { + inner, + arc_rpc, + consensus_ipcpath: None, + } } } @@ -369,8 +385,16 @@ where where T: Send, { - let Self { inner, arc_rpc } = self; - ArcAddOns::new(inner.with_engine_api(engine_api_builder), arc_rpc) + let Self { + inner, + arc_rpc, + consensus_ipcpath, + } = self; + ArcAddOns { + inner: inner.with_engine_api(engine_api_builder), + arc_rpc, + consensus_ipcpath, + } } /// Replace the payload validator builder. @@ -378,11 +402,16 @@ where self, payload_validator_builder: T, ) -> ArcAddOns { - let Self { inner, arc_rpc } = self; - ArcAddOns::new( - inner.with_payload_validator(payload_validator_builder), + let Self { + inner, arc_rpc, - ) + consensus_ipcpath, + } = self; + ArcAddOns { + inner: inner.with_payload_validator(payload_validator_builder), + arc_rpc, + consensus_ipcpath, + } } /// Sets rpc middleware @@ -390,16 +419,32 @@ where where T: Send, { - let Self { inner, arc_rpc } = self; - ArcAddOns::new(inner.with_rpc_middleware(rpc_middleware), arc_rpc) + let Self { + inner, + arc_rpc, + consensus_ipcpath, + } = self; + ArcAddOns { + inner: inner.with_rpc_middleware(rpc_middleware), + arc_rpc, + consensus_ipcpath, + } } /// Sets the tokio runtime for the RPC servers. /// /// Caution: This runtime must not be created from within asynchronous context. pub fn with_tokio_runtime(self, tokio_runtime: Option) -> Self { - let Self { inner, arc_rpc } = self; - Self::new(inner.with_tokio_runtime(tokio_runtime), arc_rpc) + let Self { + inner, + arc_rpc, + consensus_ipcpath, + } = self; + Self { + inner: inner.with_tokio_runtime(tokio_runtime), + arc_rpc, + consensus_ipcpath, + } } /// Replace entire ARC RPC config. @@ -407,6 +452,12 @@ where self.arc_rpc = cfg; self } + + /// IPC path for the consensus-only EthApi. Empty/`None` disables the extra server. + pub fn with_consensus_ipcpath(mut self, path: Option) -> Self { + self.consensus_ipcpath = path.filter(|p| !p.is_empty()); + self + } } impl NodeAddOns @@ -446,7 +497,21 @@ where let eth_config = EthConfigHandler::new(ctx.node.provider().clone(), ctx.node.evm_config().clone()); - self.inner + let Self { + inner, + arc_rpc, + consensus_ipcpath, + } = self; + + let consensus_eth_rpc_config = ctx + .config + .rpc + .eth_config() + .max_blocking_io_requests(CONSENSUS_IPC_MAX_BLOCKING_IO); + let engine_handle = ctx.beacon_engine_handle.clone(); + let node = ctx.node.clone(); + + let handle = inner .launch_add_ons_with(ctx, move |container| { container.modules.merge_if_module_configured( RethRpcModule::Flashbots, @@ -460,9 +525,9 @@ where // from externally reachable transports while retaining trusted local IPC access. remove_public_bundle_rpc_methods(container.modules); - if self.arc_rpc.enabled { + if arc_rpc.enabled { if let Ok(arc_module) = - crate::rpc::arc::build_arc_rpc_module(self.arc_rpc.upstream_url.clone()) + crate::rpc::arc::build_arc_rpc_module(arc_rpc.upstream_url.clone()) { container.modules.merge_configured(arc_module)?; } @@ -470,7 +535,50 @@ where Ok(()) }) - .await + .await?; + + if let Some(path) = consensus_ipcpath.filter(|p| !p.is_empty()) { + let cache = EthStateCache::spawn_with( + node.provider().clone(), + consensus_eth_rpc_config.cache.clone(), + node.task_executor().clone(), + ); + let consensus_eth = EthB::default() + .build_eth_api(EthApiCtx { + components: &node, + config: consensus_eth_rpc_config, + cache, + engine_handle, + }) + .await?; + + let mut module = consensus_eth.into_rpc(); + module + .merge(handle.rpc_registry.reth_api().into_rpc()) + .map_err(|err| { + eyre::eyre!("failed to merge reth methods onto consensus IPC: {err}") + })?; + + info!( + target: "arc::rpc", + path, + max_blocking_io = CONSENSUS_IPC_MAX_BLOCKING_IO, + "starting consensus IPC server with isolated eth_call queue" + ); + + let ipc_handle = IpcServerBuilder::default() + .build(path) + .start(module) + .await + .map_err(|err| eyre::eyre!("failed to start consensus IPC server: {err}"))?; + + node.task_executor() + .spawn_critical_task("consensus-ipc", async move { + ipc_handle.stopped().await; + }); + } + + Ok(handle) } } @@ -559,6 +667,7 @@ where fn add_ons(&self) -> Self::AddOns { ArcAddOns::default() .with_arc_rpc_config(self.rpc_cfg.clone()) + .with_consensus_ipcpath(self.consensus_ipcpath.clone()) .with_rpc_middleware(ArcRpcLayer::new( self.filter_pending_txs, self.allow_unprotected_txs, @@ -831,6 +940,7 @@ mod tests { crate::rebroadcast::DEFAULT_REBROADCAST_INTERVAL, Vec::new(), DEFAULT_TX_RELAY_TIMEOUT, + None, ); assert!(!node.rpc_cfg.enabled); @@ -863,6 +973,7 @@ mod tests { crate::rebroadcast::DEFAULT_REBROADCAST_INTERVAL, Vec::new(), DEFAULT_TX_RELAY_TIMEOUT, + None, ); assert_eq!( node.addresses_denylist_config.contract_address(), @@ -902,6 +1013,7 @@ mod tests { crate::rebroadcast::DEFAULT_REBROADCAST_INTERVAL, Vec::new(), DEFAULT_TX_RELAY_TIMEOUT, + None, ); assert!(!node.filter_pending_txs); } @@ -938,6 +1050,7 @@ mod tests { crate::rebroadcast::DEFAULT_REBROADCAST_INTERVAL, Vec::new(), DEFAULT_TX_RELAY_TIMEOUT, + None, ); assert!(!node.wait_for_payload); } @@ -974,8 +1087,10 @@ mod tests { std::time::Duration::ZERO, Vec::new(), DEFAULT_TX_RELAY_TIMEOUT, + None, ); assert!(node.rebroadcast_interval.is_zero()); + assert!(node.consensus_ipcpath.is_none()); } #[test] diff --git a/crates/node/src/main.rs b/crates/node/src/main.rs index d11cc4fe..6898c7c7 100644 --- a/crates/node/src/main.rs +++ b/crates/node/src/main.rs @@ -355,6 +355,14 @@ struct ArcExtraCli { help_heading = "Profiling" )] pprof_heap_prof: bool, + + /// Isolated IPC socket for `arc-consensus` (`eth_call` + `reth_subscribePersistedBlock`). + /// + /// Serves a second `EthApi` with its own `eth_call` semaphore so public + /// HTTP/WS traffic cannot starve consensus. Empty (default) keeps upstream + /// behaviour: consensus shares `--ipcpath`. + #[arg(long = "consensus-ipcpath", value_name = "PATH", help_heading = "IPC")] + consensus_ipcpath: Option, } /// Build [`AddressesDenylistConfig`] for the chain being run. @@ -625,6 +633,7 @@ fn main() { rebroadcast_interval, tx_relays, tx_relay_timeout, + ext.consensus_ipcpath.clone(), )) .launch_with_debug_capabilities() .await?; @@ -1101,6 +1110,21 @@ mod tests { ); } + #[test] + fn test_consensus_ipcpath_defaults_unset() { + assert!(ext_from_args([]).consensus_ipcpath.is_none()); + } + + #[test] + fn test_consensus_ipcpath_parses() { + assert_eq!( + ext_from_args(["--consensus-ipcpath", "/sockets/consensus.ipc"]) + .consensus_ipcpath + .as_deref(), + Some("/sockets/consensus.ipc") + ); + } + fn ext_from_args<'a>(args: impl IntoIterator) -> ArcExtraCli { let cli = ArcCli::try_parse_from( ["arc-node-execution", "node"] diff --git a/crates/test/integration/src/runner.rs b/crates/test/integration/src/runner.rs index d5d3d4a4..5bf69e4b 100644 --- a/crates/test/integration/src/runner.rs +++ b/crates/test/integration/src/runner.rs @@ -731,6 +731,7 @@ async fn spawn_execution_layer( std::time::Duration::from_secs(0), Vec::new(), DEFAULT_TX_RELAY_TIMEOUT, + None, ); let reth_handle = NodeBuilder::new(node_config)