jleony/nanoreth
1
0
Fork 0
mirror of https://github.com/hl-archive-node/nanoreth.git synced 2025-12-06 10:59:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(rpc): EthApi builder (#9041)

Browse Source 
Co-authored-by: Matthias Seitz <matthias.seitz@outlook.de>
This commit is contained in:
Emilia Hane
2024-07-03 17:30:29 +02:00
committed by GitHub
parent 84c5c3376e
commit 335b93425e
29 changed files with 736 additions and 557 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
Cargo.lock generated
View File

@ -8151,6 +8151,7 @@ dependencies = [
"alloy-rlp",
"assert_matches",
"async-trait",
"derive_more",
"futures",
"http 1.1.0",
"http-body",
@ -8252,6 +8253,7 @@ dependencies = [
"reth-rpc",
"reth-rpc-api",
"reth-rpc-engine-api",
"reth-rpc-eth-api",
"reth-rpc-eth-types",
"reth-rpc-layer",
"reth-rpc-server-types",

2
crates/ethereum/node/tests/e2e/dev.rs
View File

@ -19,7 +19,7 @@ async fn can_run_dev_node() -> eyre::Result<()> {
Ok(())
}
async fn assert_chain_advances(mut node: EthNode) {
async fn assert_chain_advances(node: EthNode) {
let mut notifications = node.inner.provider.canonical_state_stream();
// submit tx through rpc

2
crates/node/builder/src/launch/mod.rs
View File

@ -308,7 +308,7 @@ where
let jwt_secret = ctx.auth_jwt_secret()?;
// Start RPC servers
let (rpc_server_handles, mut rpc_registry) = crate::rpc::launch_rpc_servers(
let (rpc_server_handles, rpc_registry) = crate::rpc::launch_rpc_servers(
ctx.node_adapter().clone(),
engine_api,
ctx.node_config(),

29
crates/node/builder/src/rpc.rs
View File

@ -1,22 +1,24 @@
//! Builder support for rpc components.
use std::{
fmt,
ops::{Deref, DerefMut},
};
use futures::TryFutureExt;
use reth_network::NetworkHandle;
use reth_node_api::FullNodeComponents;
use reth_node_core::{node_config::NodeConfig, rpc::api::EngineApiServer};
use reth_payload_builder::PayloadBuilderHandle;
use reth_rpc::eth::EthApi;
use reth_rpc_builder::{
auth::{AuthRpcModule, AuthServerHandle},
config::RethRpcServerConfig,
RethModuleRegistry, RpcModuleBuilder, RpcServerHandle, TransportRpcModules,
EthApiBuild, RpcModuleBuilder, RpcRegistryInner, RpcServerHandle, TransportRpcModules,
};
use reth_rpc_layer::JwtSecret;
use reth_tasks::TaskExecutor;
use reth_tracing::tracing::{debug, info};
use std::{
fmt,
ops::{Deref, DerefMut},
};
/// Contains the handles to the spawned RPC servers.
///
@ -145,27 +147,28 @@ impl<Node: FullNodeComponents> ExtendRpcModules<Node> for () {
}
}
/// Helper wrapper type to encapsulate the [`RethModuleRegistry`] over components trait.
/// Helper wrapper type to encapsulate the [`RpcRegistryInner`] over components trait.
#[derive(Debug)]
#[allow(clippy::type_complexity)]
pub struct RpcRegistry<Node: FullNodeComponents> {
pub(crate) registry: RethModuleRegistry<
pub(crate) registry: RpcRegistryInner<
Node::Provider,
Node::Pool,
NetworkHandle,
TaskExecutor,
Node::Provider,
Node::Evm,
EthApi<Node::Provider, Node::Pool, NetworkHandle, Node::Evm>,
>,
}
impl<Node: FullNodeComponents> Deref for RpcRegistry<Node> {
type Target = RethModuleRegistry<
type Target = RpcRegistryInner<
Node::Provider,
Node::Pool,
NetworkHandle,
TaskExecutor,
Node::Provider,
Node::Evm,
EthApi<Node::Provider, Node::Pool, NetworkHandle, Node::Evm>,
>;
fn deref(&self) -> &Self::Target {
@ -185,7 +188,7 @@ impl<Node: FullNodeComponents> Clone for RpcRegistry<Node> {
}
}
/// Helper container to encapsulate [`RethModuleRegistry`], [`TransportRpcModules`] and
/// Helper container to encapsulate [`RpcRegistryInner`], [`TransportRpcModules`] and
/// [`AuthRpcModule`].
///
/// This can be used to access installed modules, or create commonly used handlers like
@ -202,7 +205,7 @@ pub struct RpcContext<'a, Node: FullNodeComponents> {
/// A Helper type the holds instances of the configured modules.
///
/// This provides easy access to rpc handlers, such as [`RethModuleRegistry::eth_api`].
/// This provides easy access to rpc handlers, such as [`RpcRegistryInner::eth_api`].
pub registry: &'a mut RpcRegistry<Node>,
/// Holds installed modules per transport type.
///
@ -272,7 +275,7 @@ where
.with_events(node.provider().clone())
.with_executor(node.task_executor().clone())
.with_evm_config(node.evm_config().clone())
.build_with_auth_server(module_config, engine_api);
.build_with_auth_server(module_config, engine_api, EthApiBuild::build);
let mut registry = RpcRegistry { registry };
let ctx = RpcContext {

1
crates/rpc/rpc-builder/Cargo.toml
View File

@ -19,6 +19,7 @@ reth-node-core.workspace = true
reth-provider.workspace = true
reth-rpc.workspace = true
reth-rpc-api.workspace = true
reth-rpc-eth-api.workspace = true
reth-rpc-layer.workspace = true
reth-rpc-eth-types.workspace = true
reth-rpc-server-types.workspace = true

5
crates/rpc/rpc-builder/src/auth.rs
View File

@ -1,5 +1,3 @@
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use crate::error::{RpcError, ServerKind};
use http::header::AUTHORIZATION;
use jsonrpsee::{
@ -9,13 +7,14 @@ use jsonrpsee::{
Methods,
};
use reth_engine_primitives::EngineTypes;
use reth_rpc_api::*;
use reth_rpc_api::servers::*;
use reth_rpc_eth_types::EthSubscriptionIdProvider;
use reth_rpc_layer::{
secret_to_bearer_header, AuthClientLayer, AuthClientService, AuthLayer, JwtAuthValidator,
JwtSecret,
};
use reth_rpc_server_types::constants;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use tower::layer::util::Identity;
pub use jsonrpsee::server::ServerBuilder;

8
crates/rpc/rpc-builder/src/config.rs
View File

@ -214,13 +214,11 @@ impl RethRpcServerConfig for RpcServerArgs {
#[cfg(test)]
mod tests {
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use clap::{Args, Parser};
use reth_node_core::args::RpcServerArgs;
use reth_rpc_server_types::{
constants, constants::gas_oracle::RPC_DEFAULT_GAS_CAP, RethRpcModule, RpcModuleSelection,
};
use reth_rpc_eth_types::RPC_DEFAULT_GAS_CAP;
use reth_rpc_server_types::{constants, RethRpcModule, RpcModuleSelection};
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use crate::config::RethRpcServerConfig;

437
crates/rpc/rpc-builder/src/eth.rs
View File

@ -1,232 +1,136 @@
use std::sync::Arc;
use std::{fmt::Debug, time::Duration};
use reth_evm::ConfigureEvm;
use reth_network_api::{NetworkInfo, Peers};
use reth_network_api::NetworkInfo;
use reth_provider::{
AccountReader, BlockReaderIdExt, CanonStateSubscriptions, ChainSpecProvider, ChangeSetReader,
EvmEnvProvider, StateProviderFactory,
BlockReader, BlockReaderIdExt, CanonStateSubscriptions, ChainSpecProvider, EvmEnvProvider,
FullRpcProvider, StateProviderFactory,
};
use reth_rpc::eth::{EthApi, EthFilter, EthFilterConfig, EthPubSub, RawTransactionForwarder};
use reth_rpc::{eth::EthFilterConfig, EthApi, EthFilter, EthPubSub};
use reth_rpc_eth_types::{
cache::cache_new_blocks_task, fee_history::fee_history_cache_new_blocks_task, EthStateCache,
EthStateCacheConfig, FeeHistoryCache, FeeHistoryCacheConfig, GasPriceOracle,
GasPriceOracleConfig,
GasPriceOracleConfig, RPC_DEFAULT_GAS_CAP,
};
use reth_rpc_server_types::constants::{
default_max_tracing_requests, gas_oracle::RPC_DEFAULT_GAS_CAP, DEFAULT_MAX_BLOCKS_PER_FILTER,
DEFAULT_MAX_LOGS_PER_RESPONSE,
default_max_tracing_requests, DEFAULT_MAX_BLOCKS_PER_FILTER, DEFAULT_MAX_LOGS_PER_RESPONSE,
};
use reth_tasks::{pool::BlockingTaskPool, TaskSpawner};
use reth_transaction_pool::TransactionPool;
use serde::{Deserialize, Serialize};
use crate::RpcModuleConfig;
/// Default value for stale filter ttl
const DEFAULT_STALE_FILTER_TTL: Duration = Duration::from_secs(5 * 60);
/// All handlers for the `eth` namespace
/// Alias for function that builds the core `eth` namespace API.
pub type EthApiBuilder<Provider, Pool, EvmConfig, Network, Tasks, Events, EthApi> =
Box<dyn FnOnce(&EthApiBuilderCtx<Provider, Pool, EvmConfig, Network, Tasks, Events>) -> EthApi>;
/// Handlers for core, filter and pubsub `eth` namespace APIs.
#[derive(Debug, Clone)]
pub struct EthHandlers<Provider, Pool, Network, Events, EvmConfig> {
pub struct EthHandlers<Provider, Pool, Network, Events, EthApi> {
/// Main `eth_` request handler
pub api: EthApi<Provider, Pool, Network, EvmConfig>,
pub api: EthApi,
/// The async caching layer used by the eth handlers
pub cache: EthStateCache,
/// Polling based filter handler available on all transports
pub filter: EthFilter<Provider, Pool>,
/// Handler for subscriptions only available for transports that support it (ws, ipc)
pub pubsub: EthPubSub<Provider, Pool, Events, Network>,
/// The configured tracing call pool
pub blocking_task_pool: BlockingTaskPool,
}
/// Configuration for `EthHandlersBuilder`
#[derive(Clone, Debug)]
pub(crate) struct EthHandlersConfig<Provider, Pool, Network, Tasks, Events, EvmConfig> {
/// The provider for blockchain data, responsible for reading blocks, accounts, state, etc.
pub(crate) provider: Provider,
/// The transaction pool for managing pending transactions.
pub(crate) pool: Pool,
/// The network information, handling peer connections and network state.
pub(crate) network: Network,
/// The task executor for spawning asynchronous tasks.
pub(crate) executor: Tasks,
/// The event subscriptions for canonical state changes.
pub(crate) events: Events,
/// The EVM configuration for Ethereum Virtual Machine settings.
pub(crate) evm_config: EvmConfig,
/// An optional forwarder for raw transactions.
pub(crate) eth_raw_transaction_forwarder: Option<Arc<dyn RawTransactionForwarder>>,
impl<Provider, Pool, Network, Events, EthApi> EthHandlers<Provider, Pool, Network, Events, EthApi> {
/// Returns a new [`EthHandlers`] builder.
#[allow(clippy::too_many_arguments)]
pub fn builder<EvmConfig, Tasks, EthApiB>(
provider: Provider,
pool: Pool,
network: Network,
evm_config: EvmConfig,
config: EthConfig,
executor: Tasks,
events: Events,
eth_api_builder: EthApiB,
) -> EthHandlersBuilder<Provider, Pool, Network, Tasks, Events, EvmConfig, EthApi>
where
EthApiB: FnOnce(&EthApiBuilderCtx<Provider, Pool, EvmConfig, Network, Tasks, Events>) -> EthApi
+ 'static,
{
EthHandlersBuilder {
provider,
pool,
network,
evm_config,
config,
executor,
events,
eth_api_builder: Box::new(eth_api_builder),
}
}
}
/// Represents the builder for the `EthHandlers` struct, used to configure and create instances of
/// `EthHandlers`.
#[derive(Debug, Clone)]
pub(crate) struct EthHandlersBuilder<Provider, Pool, Network, Tasks, Events, EvmConfig> {
eth_handlers_config: EthHandlersConfig<Provider, Pool, Network, Tasks, Events, EvmConfig>,
/// Configuration for the RPC module
rpc_config: RpcModuleConfig,
/// Builds [`EthHandlers`] for core, filter, and pubsub `eth_` apis.
#[allow(missing_debug_implementations)]
pub struct EthHandlersBuilder<Provider, Pool, Network, Tasks, Events, EvmConfig, EthApi> {
provider: Provider,
pool: Pool,
network: Network,
evm_config: EvmConfig,
config: EthConfig,
executor: Tasks,
events: Events,
eth_api_builder: EthApiBuilder<Provider, Pool, EvmConfig, Network, Tasks, Events, EthApi>,
}
impl<Provider, Pool, Network, Tasks, Events, EvmConfig>
EthHandlersBuilder<Provider, Pool, Network, Tasks, Events, EvmConfig>
impl<Provider, Pool, Network, Tasks, Events, EvmConfig, EthApi>
EthHandlersBuilder<Provider, Pool, Network, Tasks, Events, EvmConfig, EthApi>
where
Provider: BlockReaderIdExt
+ AccountReader
+ StateProviderFactory
+ EvmEnvProvider
+ ChainSpecProvider
+ ChangeSetReader
+ Clone
+ Unpin
+ 'static,
Pool: TransactionPool + Clone + 'static,
Network: NetworkInfo + Peers + Clone + 'static,
Provider: StateProviderFactory + BlockReader + EvmEnvProvider + Clone + Unpin + 'static,
Pool: Send + Sync + Clone + 'static,
EvmConfig: ConfigureEvm,
Network: Clone,
Tasks: TaskSpawner + Clone + 'static,
Events: CanonStateSubscriptions + Clone + 'static,
EvmConfig: ConfigureEvm + 'static,
Events: CanonStateSubscriptions + Clone,
{
/// Creates a new `EthHandlersBuilder` with the provided components.
pub(crate) const fn new(
eth_handlers_config: EthHandlersConfig<Provider, Pool, Network, Tasks, Events, EvmConfig>,
rpc_config: RpcModuleConfig,
) -> Self {
Self { eth_handlers_config, rpc_config }
}
/// Returns a new instance with handlers for `eth` namespace.
pub fn build(self) -> EthHandlers<Provider, Pool, Network, Events, EthApi> {
let Self { provider, pool, network, evm_config, config, executor, events, eth_api_builder } =
self;
/// Builds and returns an `EthHandlers` instance.
pub(crate) fn build(self) -> EthHandlers<Provider, Pool, Network, Events, EvmConfig> {
// Initialize the cache
let cache = self.init_cache();
let cache = EthStateCache::spawn_with(
provider.clone(),
config.cache,
executor.clone(),
evm_config.clone(),
);
// Initialize the fee history cache
let fee_history_cache = self.init_fee_history_cache(&cache);
// Spawn background tasks for cache
self.spawn_cache_tasks(&cache, &fee_history_cache);
// Initialize the gas oracle
let gas_oracle = self.init_gas_oracle(&cache);
// Initialize the blocking task pool
let blocking_task_pool = self.init_blocking_task_pool();
// Initialize the Eth API
let api = self.init_api(&cache, gas_oracle, &fee_history_cache, &blocking_task_pool);
// Initialize the filter
let filter = self.init_filter(&cache);
// Initialize the pubsub
let pubsub = self.init_pubsub();
EthHandlers { api, cache, filter, pubsub, blocking_task_pool }
}
/// Initializes the `EthStateCache`.
fn init_cache(&self) -> EthStateCache {
EthStateCache::spawn_with(
self.eth_handlers_config.provider.clone(),
self.rpc_config.eth.cache.clone(),
self.eth_handlers_config.executor.clone(),
self.eth_handlers_config.evm_config.clone(),
)
}
/// Initializes the `FeeHistoryCache`.
fn init_fee_history_cache(&self, cache: &EthStateCache) -> FeeHistoryCache {
FeeHistoryCache::new(cache.clone(), self.rpc_config.eth.fee_history_cache.clone())
}
/// Spawns background tasks for updating caches.
fn spawn_cache_tasks(&self, cache: &EthStateCache, fee_history_cache: &FeeHistoryCache) {
// Get the stream of new canonical blocks
let new_canonical_blocks = self.eth_handlers_config.events.canonical_state_stream();
// Clone the cache for the task
let cache_clone = cache.clone();
// Spawn a critical task to update the cache with new blocks
self.eth_handlers_config.executor.spawn_critical(
let new_canonical_blocks = events.canonical_state_stream();
let c = cache.clone();
executor.spawn_critical(
"cache canonical blocks task",
Box::pin(async move {
cache_new_blocks_task(cache_clone, new_canonical_blocks).await;
cache_new_blocks_task(c, new_canonical_blocks).await;
}),
);
// Get another stream of new canonical blocks
let new_canonical_blocks = self.eth_handlers_config.events.canonical_state_stream();
let ctx = EthApiBuilderCtx {
provider,
pool,
network,
evm_config,
config,
executor,
events,
cache,
};
// Clone the fee history cache for the task
let fhc_clone = fee_history_cache.clone();
let api = eth_api_builder(&ctx);
// Clone the provider for the task
let provider_clone = self.eth_handlers_config.provider.clone();
let filter = EthFilterApiBuilder::build(&ctx);
// Spawn a critical task to update the fee history cache with new blocks
self.eth_handlers_config.executor.spawn_critical(
"cache canonical blocks for fee history task",
Box::pin(async move {
fee_history_cache_new_blocks_task(fhc_clone, new_canonical_blocks, provider_clone)
.await;
}),
);
}
let pubsub = EthPubSubApiBuilder::build(&ctx);
/// Initializes the `GasPriceOracle`.
fn init_gas_oracle(&self, cache: &EthStateCache) -> GasPriceOracle<Provider> {
GasPriceOracle::new(
self.eth_handlers_config.provider.clone(),
self.rpc_config.eth.gas_oracle.clone(),
cache.clone(),
)
}
/// Initializes the `BlockingTaskPool`.
fn init_blocking_task_pool(&self) -> BlockingTaskPool {
BlockingTaskPool::build().expect("failed to build tracing pool")
}
/// Initializes the `EthApi`.
fn init_api(
&self,
cache: &EthStateCache,
gas_oracle: GasPriceOracle<Provider>,
fee_history_cache: &FeeHistoryCache,
blocking_task_pool: &BlockingTaskPool,
) -> EthApi<Provider, Pool, Network, EvmConfig> {
EthApi::with_spawner(
self.eth_handlers_config.provider.clone(),
self.eth_handlers_config.pool.clone(),
self.eth_handlers_config.network.clone(),
cache.clone(),
gas_oracle,
self.rpc_config.eth.rpc_gas_cap,
Box::new(self.eth_handlers_config.executor.clone()),
blocking_task_pool.clone(),
fee_history_cache.clone(),
self.eth_handlers_config.evm_config.clone(),
self.eth_handlers_config.eth_raw_transaction_forwarder.clone(),
)
}
/// Initializes the `EthFilter`.
fn init_filter(&self, cache: &EthStateCache) -> EthFilter<Provider, Pool> {
EthFilter::new(
self.eth_handlers_config.provider.clone(),
self.eth_handlers_config.pool.clone(),
cache.clone(),
self.rpc_config.eth.filter_config(),
Box::new(self.eth_handlers_config.executor.clone()),
)
}
/// Initializes the `EthPubSub`.
fn init_pubsub(&self) -> EthPubSub<Provider, Pool, Events, Network> {
EthPubSub::with_spawner(
self.eth_handlers_config.provider.clone(),
self.eth_handlers_config.pool.clone(),
self.eth_handlers_config.events.clone(),
self.eth_handlers_config.network.clone(),
Box::new(self.eth_handlers_config.executor.clone()),
)
EthHandlers { api, cache: ctx.cache, filter, pubsub }
}
}
@ -249,7 +153,7 @@ pub struct EthConfig {
pub rpc_gas_cap: u64,
///
/// Sets TTL for stale filters
pub stale_filter_ttl: std::time::Duration,
pub stale_filter_ttl: Duration,
/// Settings for the fee history cache
pub fee_history_cache: FeeHistoryCacheConfig,
}
@ -264,9 +168,6 @@ impl EthConfig {
}
}
/// Default value for stale filter ttl
const DEFAULT_STALE_FILTER_TTL: std::time::Duration = std::time::Duration::from_secs(5 * 60);
impl Default for EthConfig {
fn default() -> Self {
Self {
@ -275,7 +176,7 @@ impl Default for EthConfig {
max_tracing_requests: default_max_tracing_requests(),
max_blocks_per_filter: DEFAULT_MAX_BLOCKS_PER_FILTER,
max_logs_per_response: DEFAULT_MAX_LOGS_PER_RESPONSE,
rpc_gas_cap: RPC_DEFAULT_GAS_CAP,
rpc_gas_cap: RPC_DEFAULT_GAS_CAP.into(),
stale_filter_ttl: DEFAULT_STALE_FILTER_TTL,
fee_history_cache: FeeHistoryCacheConfig::default(),
}
@ -319,3 +220,157 @@ impl EthConfig {
self
}
}
/// Context for building the `eth` namespace API.
#[derive(Debug, Clone)]
pub struct EthApiBuilderCtx<Provider, Pool, EvmConfig, Network, Tasks, Events> {
/// Database handle.
pub provider: Provider,
/// Mempool handle.
pub pool: Pool,
/// Network handle.
pub network: Network,
/// EVM configuration.
pub evm_config: EvmConfig,
/// RPC config for `eth` namespace.
pub config: EthConfig,
/// Runtime handle.
pub executor: Tasks,
/// Events handle.
pub events: Events,
/// RPC cache handle.
pub cache: EthStateCache,
}
/// Ethereum layer one `eth` RPC server builder.
#[derive(Default, Debug, Clone, Copy)]
pub struct EthApiBuild;
impl EthApiBuild {
/// Builds the [`EthApiServer`](reth_rpc_eth_api::EthApiServer), for given context.
pub fn build<Provider, Pool, EvmConfig, Network, Tasks, Events>(
ctx: &EthApiBuilderCtx<Provider, Pool, EvmConfig, Network, Tasks, Events>,
) -> EthApi<Provider, Pool, Network, EvmConfig>
where
Provider: FullRpcProvider,
Pool: TransactionPool,
Network: NetworkInfo + Clone,
Tasks: TaskSpawner + Clone + 'static,
Events: CanonStateSubscriptions,
EvmConfig: ConfigureEvm,
{
let gas_oracle = GasPriceOracleBuilder::build(ctx);
let fee_history_cache = FeeHistoryCacheBuilder::build(ctx);
EthApi::with_spawner(
ctx.provider.clone(),
ctx.pool.clone(),
ctx.network.clone(),
ctx.cache.clone(),
gas_oracle,
ctx.config.rpc_gas_cap,
Box::new(ctx.executor.clone()),
BlockingTaskPool::build().expect("failed to build blocking task pool"),
fee_history_cache,
ctx.evm_config.clone(),
None,
)
}
}
/// Builds the `eth_` namespace API [`EthFilterApiServer`](reth_rpc_eth_api::EthFilterApiServer).
#[derive(Debug)]
pub struct EthFilterApiBuilder;
impl EthFilterApiBuilder {
/// Builds the [`EthFilterApiServer`](reth_rpc_eth_api::EthFilterApiServer), for given context.
pub fn build<Provider, Pool, EvmConfig, Network, Tasks, Events>(
ctx: &EthApiBuilderCtx<Provider, Pool, EvmConfig, Network, Tasks, Events>,
) -> EthFilter<Provider, Pool>
where
Provider: Send + Sync + Clone + 'static,
Pool: Send + Sync + Clone + 'static,
Tasks: TaskSpawner + Clone + 'static,
{
EthFilter::new(
ctx.provider.clone(),
ctx.pool.clone(),
ctx.cache.clone(),
ctx.config.filter_config(),
Box::new(ctx.executor.clone()),
)
}
}
/// Builds the `eth_` namespace API [`EthPubSubApiServer`](reth_rpc_eth_api::EthFilterApiServer).
#[derive(Debug)]
pub struct EthPubSubApiBuilder;
impl EthPubSubApiBuilder {
/// Builds the [`EthPubSubApiServer`](reth_rpc_eth_api::EthPubSubApiServer), for given context.
pub fn build<Provider, Pool, EvmConfig, Network, Tasks, Events>(
ctx: &EthApiBuilderCtx<Provider, Pool, EvmConfig, Network, Tasks, Events>,
) -> EthPubSub<Provider, Pool, Events, Network>
where
Provider: Clone,
Pool: Clone,
Events: Clone,
Network: Clone,
Tasks: TaskSpawner + Clone + 'static,
{
EthPubSub::with_spawner(
ctx.provider.clone(),
ctx.pool.clone(),
ctx.events.clone(),
ctx.network.clone(),
Box::new(ctx.executor.clone()),
)
}
}
/// Builds `eth_` core api component [`GasPriceOracle`], for given context.
#[derive(Debug)]
pub struct GasPriceOracleBuilder;
impl GasPriceOracleBuilder {
/// Builds a [`GasPriceOracle`], for given context.
pub fn build<Provider, Pool, EvmConfig, Network, Tasks, Events>(
ctx: &EthApiBuilderCtx<Provider, Pool, EvmConfig, Network, Tasks, Events>,
) -> GasPriceOracle<Provider>
where
Provider: BlockReaderIdExt + Clone,
{
GasPriceOracle::new(ctx.provider.clone(), ctx.config.gas_oracle, ctx.cache.clone())
}
}
/// Builds `eth_` core api component [`FeeHistoryCache`], for given context.
#[derive(Debug)]
pub struct FeeHistoryCacheBuilder;
impl FeeHistoryCacheBuilder {
/// Builds a [`FeeHistoryCache`], for given context.
pub fn build<Provider, Pool, EvmConfig, Network, Tasks, Events>(
ctx: &EthApiBuilderCtx<Provider, Pool, EvmConfig, Network, Tasks, Events>,
) -> FeeHistoryCache
where
Provider: ChainSpecProvider + BlockReaderIdExt + Clone + 'static,
Tasks: TaskSpawner,
Events: CanonStateSubscriptions,
{
let fee_history_cache =
FeeHistoryCache::new(ctx.cache.clone(), ctx.config.fee_history_cache);
let new_canonical_blocks = ctx.events.canonical_state_stream();
let fhc = fee_history_cache.clone();
let provider = ctx.provider.clone();
ctx.executor.spawn_critical(
"cache canonical blocks for fee history task",
Box::pin(async move {
fee_history_cache_new_blocks_task(fhc, new_canonical_blocks, provider).await;
}),
);
fee_history_cache
}
}

592
crates/rpc/rpc-builder/src/lib.rs
View File

@ -19,13 +19,12 @@
//! ```
//! use reth_evm::ConfigureEvm;
//! use reth_network_api::{NetworkInfo, Peers};
//! use reth_provider::{
//! AccountReader, BlockReaderIdExt, CanonStateSubscriptions, ChainSpecProvider,
//! ChangeSetReader, EvmEnvProvider, StateProviderFactory,
//! };
//! use reth_provider::{AccountReader, CanonStateSubscriptions, ChangeSetReader, FullRpcProvider};
//! use reth_rpc_builder::{
//! RethRpcModule, RpcModuleBuilder, RpcServerConfig, ServerBuilder, TransportRpcModuleConfig,
//! EthApiBuild, RethRpcModule, RpcModuleBuilder, RpcServerConfig, ServerBuilder,
//! TransportRpcModuleConfig,
//! };
//!
//! use reth_tasks::TokioTaskExecutor;
//! use reth_transaction_pool::TransactionPool;
//! pub async fn launch<Provider, Pool, Network, Events, EvmConfig>(
@ -35,16 +34,8 @@
//! events: Events,
//! evm_config: EvmConfig,
//! ) where
//! Provider: AccountReader
//! + BlockReaderIdExt
//! + ChainSpecProvider
//! + ChangeSetReader
//! + StateProviderFactory
//! + EvmEnvProvider
//! + Clone
//! + Unpin
//! + 'static,
//! Pool: TransactionPool + Clone + 'static,
//! Provider: FullRpcProvider + AccountReader + ChangeSetReader,
//! Pool: TransactionPool + 'static,
//! Network: NetworkInfo + Peers + Clone + 'static,
//! Events: CanonStateSubscriptions + Clone + 'static,
//! EvmConfig: ConfigureEvm,
@ -64,7 +55,7 @@
//! events,
//! evm_config,
//! )
//! .build(transports);
//! .build(transports, EthApiBuild::build);
//! let handle = RpcServerConfig::default()
//! .with_http(ServerBuilder::default())
//! .start(transport_modules)
@ -80,13 +71,10 @@
//! use reth_engine_primitives::EngineTypes;
//! use reth_evm::ConfigureEvm;
//! use reth_network_api::{NetworkInfo, Peers};
//! use reth_provider::{
//! AccountReader, BlockReaderIdExt, CanonStateSubscriptions, ChainSpecProvider,
//! ChangeSetReader, EvmEnvProvider, StateProviderFactory,
//! };
//! use reth_provider::{AccountReader, CanonStateSubscriptions, ChangeSetReader, FullRpcProvider};
//! use reth_rpc_api::EngineApiServer;
//! use reth_rpc_builder::{
//! auth::AuthServerConfig, RethRpcModule, RpcModuleBuilder, RpcServerConfig,
//! auth::AuthServerConfig, EthApiBuild, RethRpcModule, RpcModuleBuilder, RpcServerConfig,
//! TransportRpcModuleConfig,
//! };
//! use reth_rpc_layer::JwtSecret;
@ -101,16 +89,8 @@
//! engine_api: EngineApi,
//! evm_config: EvmConfig,
//! ) where
//! Provider: AccountReader
//! + BlockReaderIdExt
//! + ChainSpecProvider
//! + ChangeSetReader
//! + StateProviderFactory
//! + EvmEnvProvider
//! + Clone
//! + Unpin
//! + 'static,
//! Pool: TransactionPool + Clone + 'static,
//! Provider: FullRpcProvider + AccountReader + ChangeSetReader,
//! Pool: TransactionPool + 'static,
//! Network: NetworkInfo + Peers + Clone + 'static,
//! Events: CanonStateSubscriptions + Clone + 'static,
//! EngineApi: EngineApiServer<EngineT>,
@ -135,7 +115,7 @@
//!
//! // configure the server modules
//! let (modules, auth_module, _registry) =
//! builder.build_with_auth_server(transports, engine_api);
//! builder.build_with_auth_server(transports, engine_api, EthApiBuild::build);
//!
//! // start the servers
//! let auth_config = AuthServerConfig::builder(JwtSecret::random()).build();
@ -155,13 +135,14 @@
#![cfg_attr(not(test), warn(unused_crate_dependencies))]
#![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))]
use crate::{
auth::AuthRpcModule,
cors::CorsDomainError,
error::WsHttpSamePortError,
eth::{EthHandlersBuilder, EthHandlersConfig},
metrics::RpcRequestMetrics,
use std::{
collections::HashMap,
fmt,
net::{Ipv4Addr, SocketAddr, SocketAddrV4},
sync::Arc,
time::{Duration, SystemTime, UNIX_EPOCH},
};
use error::{ConflictingModules, RpcError, ServerKind};
use http::{header::AUTHORIZATION, HeaderMap};
use jsonrpsee::{
@ -174,30 +155,33 @@ use reth_evm::ConfigureEvm;
use reth_ipc::server::IpcServer;
use reth_network_api::{noop::NoopNetwork, NetworkInfo, Peers};
use reth_provider::{
AccountReader, BlockReader, BlockReaderIdExt, CanonStateSubscriptions, ChainSpecProvider,
ChangeSetReader, EvmEnvProvider, StateProviderFactory,
AccountReader, BlockReader, CanonStateSubscriptions, ChainSpecProvider, ChangeSetReader,
EvmEnvProvider, FullRpcProvider, StateProviderFactory,
};
use reth_rpc::{
eth::{EthApi, EthBundle, RawTransactionForwarder},
AdminApi, DebugApi, EngineEthApi, NetApi, OtterscanApi, RPCApi, RethApi, TraceApi, TxPoolApi,
Web3Api,
AdminApi, DebugApi, EngineEthApi, EthBundle, NetApi, OtterscanApi, RPCApi, RethApi, TraceApi,
TxPoolApi, Web3Api,
};
use reth_rpc_api::servers::*;
use reth_rpc_eth_api::{
helpers::{
Call, EthApiSpec, EthTransactions, LoadPendingBlock, TraceExt, UpdateRawTxForwarder,
},
EthApiServer, FullEthApiServer, RawTransactionForwarder,
};
use reth_rpc_api::*;
use reth_rpc_eth_types::{EthStateCache, EthSubscriptionIdProvider};
use reth_rpc_layer::{AuthLayer, Claims, JwtAuthValidator, JwtSecret};
use reth_tasks::{pool::BlockingTaskGuard, TaskSpawner, TokioTaskExecutor};
use reth_transaction_pool::{noop::NoopTransactionPool, TransactionPool};
use serde::{Deserialize, Serialize};
use std::{
collections::HashMap,
fmt,
net::{Ipv4Addr, SocketAddr, SocketAddrV4},
sync::Arc,
time::{Duration, SystemTime, UNIX_EPOCH},
};
use tower_http::cors::CorsLayer;
use tracing::{instrument, trace};
use crate::{
auth::AuthRpcModule, cors::CorsDomainError, error::WsHttpSamePortError,
metrics::RpcRequestMetrics,
};
// re-export for convenience
pub use jsonrpsee::server::ServerBuilder;
pub use reth_ipc::server::{
@ -219,15 +203,18 @@ mod cors;
pub mod error;
/// Eth utils
mod eth;
pub use eth::{EthConfig, EthHandlers};
pub mod eth;
pub use eth::{
EthApiBuild, EthApiBuilderCtx, EthConfig, EthHandlers, FeeHistoryCacheBuilder,
GasPriceOracleBuilder,
};
// Rpc server metrics
mod metrics;
/// Convenience function for starting a server in one step.
#[allow(clippy::too_many_arguments)]
pub async fn launch<Provider, Pool, Network, Tasks, Events, EvmConfig>(
pub async fn launch<Provider, Pool, Network, Tasks, Events, EvmConfig, EthApi, EthApiB>(
provider: Provider,
pool: Pool,
network: Network,
@ -236,27 +223,23 @@ pub async fn launch<Provider, Pool, Network, Tasks, Events, EvmConfig>(
executor: Tasks,
events: Events,
evm_config: EvmConfig,
eth: EthApiB,
) -> Result<RpcServerHandle, RpcError>
where
Provider: BlockReaderIdExt
+ AccountReader
+ StateProviderFactory
+ EvmEnvProvider
+ ChainSpecProvider
+ ChangeSetReader
+ Clone
+ Unpin
+ 'static,
Pool: TransactionPool + Clone + 'static,
Provider: FullRpcProvider + AccountReader + ChangeSetReader,
Pool: TransactionPool + 'static,
Network: NetworkInfo + Peers + Clone + 'static,
Tasks: TaskSpawner + Clone + 'static,
Events: CanonStateSubscriptions + Clone + 'static,
EvmConfig: ConfigureEvm,
EthApiB: FnOnce(&EthApiBuilderCtx<Provider, Pool, EvmConfig, Network, Tasks, Events>) -> EthApi
+ 'static,
EthApi: FullEthApiServer,
{
let module_config = module_config.into();
let server_config = server_config.into();
RpcModuleBuilder::new(provider, pool, network, executor, events, evm_config)
.build(module_config)
.build(module_config, eth)
.start_server(server_config)
.await
}
@ -324,8 +307,8 @@ impl<Provider, Pool, Network, Tasks, Events, EvmConfig>
/// Configure a [`NoopTransactionPool`] instance.
///
/// Caution: This will configure a pool API that does absolutely nothing.
/// This is only intended for allow easier setup of namespaces that depend on the [`EthApi`]
/// which requires a [`TransactionPool`] implementation.
/// This is only intended for allow easier setup of namespaces that depend on the
/// [`EthApi`](reth_rpc::eth::EthApi) which requires a [`TransactionPool`] implementation.
pub fn with_noop_pool(
self,
) -> RpcModuleBuilder<Provider, NoopTransactionPool, Network, Tasks, Events, EvmConfig> {
@ -355,8 +338,8 @@ impl<Provider, Pool, Network, Tasks, Events, EvmConfig>
/// Configure a [`NoopNetwork`] instance.
///
/// Caution: This will configure a network API that does absolutely nothing.
/// This is only intended for allow easier setup of namespaces that depend on the [`EthApi`]
/// which requires a [`NetworkInfo`] implementation.
/// This is only intended for allow easier setup of namespaces that depend on the
/// [`EthApi`](reth_rpc::eth::EthApi) which requires a [`NetworkInfo`] implementation.
pub fn with_noop_network(
self,
) -> RpcModuleBuilder<Provider, Pool, NoopNetwork, Tasks, Events, EvmConfig> {
@ -429,16 +412,8 @@ impl<Provider, Pool, Network, Tasks, Events, EvmConfig>
impl<Provider, Pool, Network, Tasks, Events, EvmConfig>
RpcModuleBuilder<Provider, Pool, Network, Tasks, Events, EvmConfig>
where
Provider: BlockReaderIdExt
+ AccountReader
+ StateProviderFactory
+ EvmEnvProvider
+ ChainSpecProvider
+ ChangeSetReader
+ Clone
+ Unpin
+ 'static,
Pool: TransactionPool + Clone + 'static,
Provider: FullRpcProvider + AccountReader + ChangeSetReader,
Pool: TransactionPool + 'static,
Network: NetworkInfo + Peers + Clone + 'static,
Tasks: TaskSpawner + Clone + 'static,
Events: CanonStateSubscriptions + Clone + 'static,
@ -450,25 +425,31 @@ where
/// This behaves exactly as [`RpcModuleBuilder::build`] for the [`TransportRpcModules`], but
/// also configures the auth (engine api) server, which exposes a subset of the `eth_`
/// namespace.
pub fn build_with_auth_server<EngineApi, EngineT>(
#[allow(clippy::type_complexity)]
pub fn build_with_auth_server<EngineApi, EngineT, EthApi, EthApiB>(
self,
module_config: TransportRpcModuleConfig,
engine: EngineApi,
eth: EthApiB,
) -> (
TransportRpcModules,
AuthRpcModule,
RethModuleRegistry<Provider, Pool, Network, Tasks, Events, EvmConfig>,
RpcRegistryInner<Provider, Pool, Network, Tasks, Events, EthApi>,
)
where
EngineT: EngineTypes + 'static,
EngineApi: EngineApiServer<EngineT>,
EthApiB: FnOnce(&EthApiBuilderCtx<Provider, Pool, EvmConfig, Network, Tasks, Events>) -> EthApi
+ 'static,
EthApi: FullEthApiServer,
{
let Self { provider, pool, network, executor, events, evm_config } = self;
let config = module_config.config.clone().unwrap_or_default();
let mut registry =
RethModuleRegistry::new(provider, pool, network, executor, events, config, evm_config);
let mut registry = RpcRegistryInner::new(
provider, pool, network, executor, events, config, evm_config, eth,
);
let modules = registry.create_transport_rpc_modules(module_config);
@ -477,7 +458,7 @@ where
(modules, auth_module, registry)
}
/// Converts the builder into a [`RethModuleRegistry`] which can be used to create all
/// Converts the builder into a [`RpcRegistryInner`] which can be used to create all
/// components.
///
/// This is useful for getting access to API handlers directly:
@ -488,7 +469,7 @@ where
/// use reth_evm::ConfigureEvm;
/// use reth_network_api::noop::NoopNetwork;
/// use reth_provider::test_utils::{NoopProvider, TestCanonStateSubscriptions};
/// use reth_rpc_builder::RpcModuleBuilder;
/// use reth_rpc_builder::{EthApiBuild, RpcModuleBuilder};
/// use reth_tasks::TokioTaskExecutor;
/// use reth_transaction_pool::noop::NoopTransactionPool;
///
@ -500,24 +481,38 @@ where
/// .with_executor(TokioTaskExecutor::default())
/// .with_events(TestCanonStateSubscriptions::default())
/// .with_evm_config(evm)
/// .into_registry(Default::default());
/// .into_registry(Default::default(), EthApiBuild::build);
///
/// let eth_api = registry.eth_api();
/// }
/// ```
pub fn into_registry(
pub fn into_registry<EthApi, EthApiB>(
self,
config: RpcModuleConfig,
) -> RethModuleRegistry<Provider, Pool, Network, Tasks, Events, EvmConfig> {
eth: EthApiB,
) -> RpcRegistryInner<Provider, Pool, Network, Tasks, Events, EthApi>
where
EthApiB: FnOnce(&EthApiBuilderCtx<Provider, Pool, EvmConfig, Network, Tasks, Events>) -> EthApi
+ 'static,
{
let Self { provider, pool, network, executor, events, evm_config } = self;
RethModuleRegistry::new(provider, pool, network, executor, events, config, evm_config)
RpcRegistryInner::new(provider, pool, network, executor, events, config, evm_config, eth)
}
/// Configures all [`RpcModule`]s specific to the given [`TransportRpcModuleConfig`] which can
/// be used to start the transport server(s).
///
/// See also [`RpcServer::start`]
pub fn build(self, module_config: TransportRpcModuleConfig) -> TransportRpcModules<()> {
pub fn build<EthApi, EthApiB>(
self,
module_config: TransportRpcModuleConfig,
eth: EthApiB,
) -> TransportRpcModules<()>
where
EthApiB: FnOnce(&EthApiBuilderCtx<Provider, Pool, EvmConfig, Network, Tasks, Events>) -> EthApi
+ 'static,
EthApi: FullEthApiServer,
{
let mut modules = TransportRpcModules::default();
let Self { provider, pool, network, executor, events, evm_config } = self;
@ -525,7 +520,7 @@ where
if !module_config.is_empty() {
let TransportRpcModuleConfig { http, ws, ipc, config } = module_config.clone();
let mut registry = RethModuleRegistry::new(
let mut registry = RpcRegistryInner::new(
provider,
pool,
network,
@ -533,6 +528,7 @@ where
events,
config.unwrap_or_default(),
evm_config,
eth,
);
modules.config = module_config;
@ -621,34 +617,34 @@ impl RpcModuleConfigBuilder {
/// A Helper type the holds instances of the configured modules.
#[derive(Debug, Clone)]
pub struct RethModuleRegistry<Provider, Pool, Network, Tasks, Events, EvmConfig> {
pub struct RpcRegistryInner<Provider, Pool, Network, Tasks, Events, EthApi> {
provider: Provider,
pool: Pool,
network: Network,
executor: Tasks,
events: Events,
/// Defines how to configure the EVM before execution.
evm_config: EvmConfig,
/// Additional settings for handlers.
config: RpcModuleConfig,
/// Holds a clone of all the eth namespace handlers
eth: Option<EthHandlers<Provider, Pool, Network, Events, EvmConfig>>,
/// Holds a all `eth_` namespace handlers
eth: EthHandlers<Provider, Pool, Network, Events, EthApi>,
/// to put trace calls behind semaphore
blocking_pool_guard: BlockingTaskGuard,
/// Contains the [Methods] of a module
modules: HashMap<RethRpcModule, Methods>,
/// Optional forwarder for `eth_sendRawTransaction`
// TODO(mattsse): find a more ergonomic way to configure eth/rpc customizations
eth_raw_transaction_forwarder: Option<Arc<dyn RawTransactionForwarder>>,
}
// === impl RethModuleRegistry ===
// === impl RpcRegistryInner ===
impl<Provider, Pool, Network, Tasks, Events, EvmConfig>
RethModuleRegistry<Provider, Pool, Network, Tasks, Events, EvmConfig>
impl<Provider, Pool, Network, Tasks, Events, EthApi>
RpcRegistryInner<Provider, Pool, Network, Tasks, Events, EthApi>
where
Provider: StateProviderFactory + BlockReader + EvmEnvProvider + Clone + Unpin + 'static,
Pool: Send + Sync + Clone + 'static,
Network: Clone,
Events: CanonStateSubscriptions + Clone,
Tasks: TaskSpawner + Clone + 'static,
{
/// Creates a new, empty instance.
pub fn new(
#[allow(clippy::too_many_arguments)]
pub fn new<EvmConfig, EthApiB>(
provider: Provider,
pool: Pool,
network: Network,
@ -656,34 +652,59 @@ impl<Provider, Pool, Network, Tasks, Events, EvmConfig>
events: Events,
config: RpcModuleConfig,
evm_config: EvmConfig,
) -> Self {
eth_api_builder: EthApiB,
) -> Self
where
EvmConfig: ConfigureEvm,
EthApiB: FnOnce(&EthApiBuilderCtx<Provider, Pool, EvmConfig, Network, Tasks, Events>) -> EthApi
+ 'static,
{
let blocking_pool_guard = BlockingTaskGuard::new(config.eth.max_tracing_requests);
let eth = EthHandlers::builder(
provider.clone(),
pool.clone(),
network.clone(),
evm_config,
config.eth,
executor.clone(),
events.clone(),
eth_api_builder,
)
.build();
Self {
provider,
pool,
network,
evm_config,
eth: None,
eth,
executor,
modules: Default::default(),
blocking_pool_guard: BlockingTaskGuard::new(config.eth.max_tracing_requests),
config,
blocking_pool_guard,
events,
eth_raw_transaction_forwarder: None,
}
}
}
/// Sets a forwarder for `eth_sendRawTransaction`
impl<Provider, Pool, Network, Tasks, Events, EthApi>
RpcRegistryInner<Provider, Pool, Network, Tasks, Events, EthApi>
{
/// Returns a reference to the installed [`EthApi`](reth_rpc::eth::EthApi).
pub const fn eth_api(&self) -> &EthApi {
&self.eth.api
}
/// Returns a reference to the installed [`EthHandlers`].
pub const fn eth_handlers(&self) -> &EthHandlers<Provider, Pool, Network, Events, EthApi> {
&self.eth
}
/// Returns the [`EthStateCache`] frontend
///
/// Note: this might be removed in the future in favor of a more generic approach.
pub fn set_eth_raw_transaction_forwarder(
&mut self,
forwarder: Arc<dyn RawTransactionForwarder>,
) {
if let Some(eth) = self.eth.as_ref() {
// in case the eth api has been created before the forwarder was set: <https://github.com/paradigmxyz/reth/issues/8661>
eth.api.set_eth_raw_transaction_forwarder(forwarder.clone());
}
self.eth_raw_transaction_forwarder = Some(forwarder);
/// This will spawn exactly one [`EthStateCache`] service if this is the first time the cache is
/// requested.
pub const fn eth_cache(&self) -> &EthStateCache {
&self.eth.cache
}
/// Returns a reference to the pool
@ -721,13 +742,30 @@ impl<Provider, Pool, Network, Tasks, Events, EvmConfig>
}
}
impl<Provider: ChainSpecProvider, Pool, Network, Tasks, Events, EvmConfig>
RethModuleRegistry<Provider, Pool, Network, Tasks, Events, EvmConfig>
impl<Provider, Pool, Network, Tasks, Events, EthApi>
RpcRegistryInner<Provider, Pool, Network, Tasks, Events, EthApi>
where
Network: NetworkInfo + Peers + Clone + 'static,
EthApi: UpdateRawTxForwarder,
{
/// Sets a forwarder for `eth_sendRawTransaction`
///
/// Note: this might be removed in the future in favor of a more generic approach.
pub fn set_eth_raw_transaction_forwarder(&self, forwarder: Arc<dyn RawTransactionForwarder>) {
// in case the eth api has been created before the forwarder was set: <https://github.com/paradigmxyz/reth/issues/8661>
self.eth.api.set_eth_raw_transaction_forwarder(forwarder.clone());
}
}
impl<Provider: ChainSpecProvider, Pool, Network, Tasks, Events, EthApi>
RpcRegistryInner<Provider, Pool, Network, Tasks, Events, EthApi>
where
Network: NetworkInfo + Clone + 'static,
{
/// Instantiates `AdminApi`
pub fn admin_api(&self) -> AdminApi<Network> {
pub fn admin_api(&self) -> AdminApi<Network>
where
Network: Peers,
{
AdminApi::new(self.network.clone(), self.provider.chain_spec())
}
@ -737,7 +775,10 @@ where
}
/// Register Admin Namespace
pub fn register_admin(&mut self) -> &mut Self {
pub fn register_admin(&mut self) -> &mut Self
where
Network: Peers,
{
let adminapi = self.admin_api();
self.modules.insert(RethRpcModule::Admin, adminapi.into_rpc().into());
self
@ -751,31 +792,24 @@ where
}
}
impl<Provider, Pool, Network, Tasks, Events, EvmConfig>
RethModuleRegistry<Provider, Pool, Network, Tasks, Events, EvmConfig>
impl<Provider, Pool, Network, Tasks, Events, EthApi>
RpcRegistryInner<Provider, Pool, Network, Tasks, Events, EthApi>
where
Provider: BlockReaderIdExt
+ AccountReader
+ StateProviderFactory
+ EvmEnvProvider
+ ChainSpecProvider
+ ChangeSetReader
+ Clone
+ Unpin
+ 'static,
Pool: TransactionPool + Clone + 'static,
Provider: FullRpcProvider + AccountReader + ChangeSetReader,
Network: NetworkInfo + Peers + Clone + 'static,
Tasks: TaskSpawner + Clone + 'static,
Events: CanonStateSubscriptions + Clone + 'static,
EvmConfig: ConfigureEvm,
EthApi: Clone,
{
/// Register Eth Namespace
///
/// # Panics
///
/// If called outside of the tokio runtime. See also [`Self::eth_api`]
pub fn register_eth(&mut self) -> &mut Self {
let eth_api = self.eth_api();
pub fn register_eth(&mut self) -> &mut Self
where
EthApi: EthApiServer,
{
let eth_api = self.eth_api().clone();
self.modules.insert(RethRpcModule::Eth, eth_api.into_rpc().into());
self
}
@ -785,7 +819,10 @@ where
/// # Panics
///
/// If called outside of the tokio runtime. See also [`Self::eth_api`]
pub fn register_ots(&mut self) -> &mut Self {
pub fn register_ots(&mut self) -> &mut Self
where
EthApi: EthApiServer + TraceExt,
{
let otterscan_api = self.otterscan_api();
self.modules.insert(RethRpcModule::Ots, otterscan_api.into_rpc().into());
self
@ -796,7 +833,10 @@ where
/// # Panics
///
/// If called outside of the tokio runtime. See also [`Self::eth_api`]
pub fn register_debug(&mut self) -> &mut Self {
pub fn register_debug(&mut self) -> &mut Self
where
EthApi: EthApiSpec + EthTransactions + TraceExt,
{
let debug_api = self.debug_api();
self.modules.insert(RethRpcModule::Debug, debug_api.into_rpc().into());
self
@ -807,34 +847,15 @@ where
/// # Panics
///
/// If called outside of the tokio runtime. See also [`Self::eth_api`]
pub fn register_trace(&mut self) -> &mut Self {
pub fn register_trace(&mut self) -> &mut Self
where
EthApi: TraceExt,
{
let trace_api = self.trace_api();
self.modules.insert(RethRpcModule::Trace, trace_api.into_rpc().into());
self
}
/// Configures the auth module that includes the
/// * `engine_` namespace
/// * `api_` namespace
///
/// Note: This does _not_ register the `engine_` in this registry.
pub fn create_auth_module<EngineApi, EngineT>(&mut self, engine_api: EngineApi) -> AuthRpcModule
where
EngineT: EngineTypes + 'static,
EngineApi: EngineApiServer<EngineT>,
{
let eth_handlers = self.eth_handlers();
let mut module = RpcModule::new(());
module.merge(engine_api.into_rpc()).expect("No conflicting methods");
// also merge a subset of `eth_` handlers
let engine_eth = EngineEthApi::new(eth_handlers.api.clone(), eth_handlers.filter);
module.merge(engine_eth.into_rpc()).expect("No conflicting methods");
AuthRpcModule { inner: module }
}
/// Register Net Namespace
///
/// See also [`Self::eth_api`]
@ -842,7 +863,10 @@ where
/// # Panics
///
/// If called outside of the tokio runtime.
pub fn register_net(&mut self) -> &mut Self {
pub fn register_net(&mut self) -> &mut Self
where
EthApi: EthApiSpec + 'static,
{
let netapi = self.net_api();
self.modules.insert(RethRpcModule::Net, netapi.into_rpc().into());
self
@ -861,6 +885,113 @@ where
self
}
/// Instantiates `TraceApi`
///
/// # Panics
///
/// If called outside of the tokio runtime. See also [`Self::eth_api`]
pub fn trace_api(&self) -> TraceApi<Provider, EthApi>
where
EthApi: TraceExt,
{
TraceApi::new(
self.provider.clone(),
self.eth_api().clone(),
self.blocking_pool_guard.clone(),
)
}
/// Instantiates [`EthBundle`] Api
///
/// # Panics
///
/// If called outside of the tokio runtime. See also [`Self::eth_api`]
pub fn bundle_api(&self) -> EthBundle<EthApi>
where
EthApi: EthTransactions + LoadPendingBlock + Call,
{
let eth_api = self.eth_api().clone();
EthBundle::new(eth_api, self.blocking_pool_guard.clone())
}
/// Instantiates `OtterscanApi`
///
/// # Panics
///
/// If called outside of the tokio runtime. See also [`Self::eth_api`]
pub fn otterscan_api(&self) -> OtterscanApi<EthApi>
where
EthApi: EthApiServer,
{
let eth_api = self.eth_api().clone();
OtterscanApi::new(eth_api)
}
/// Instantiates `DebugApi`
///
/// # Panics
///
/// If called outside of the tokio runtime. See also [`Self::eth_api`]
pub fn debug_api(&self) -> DebugApi<Provider, EthApi>
where
EthApi: EthApiSpec + EthTransactions + TraceExt,
{
let eth_api = self.eth_api().clone();
DebugApi::new(self.provider.clone(), eth_api, self.blocking_pool_guard.clone())
}
/// Instantiates `NetApi`
///
/// # Panics
///
/// If called outside of the tokio runtime. See also [`Self::eth_api`]
pub fn net_api(&self) -> NetApi<Network, EthApi>
where
EthApi: EthApiSpec + 'static,
{
let eth_api = self.eth_api().clone();
NetApi::new(self.network.clone(), eth_api)
}
/// Instantiates `RethApi`
pub fn reth_api(&self) -> RethApi<Provider> {
RethApi::new(self.provider.clone(), Box::new(self.executor.clone()))
}
}
impl<Provider, Pool, Network, Tasks, Events, EthApi>
RpcRegistryInner<Provider, Pool, Network, Tasks, Events, EthApi>
where
Provider: FullRpcProvider + AccountReader + ChangeSetReader,
Pool: TransactionPool + 'static,
Network: NetworkInfo + Peers + Clone + 'static,
Tasks: TaskSpawner + Clone + 'static,
Events: CanonStateSubscriptions + Clone + 'static,
EthApi: FullEthApiServer,
{
/// Configures the auth module that includes the
/// * `engine_` namespace
/// * `api_` namespace
///
/// Note: This does _not_ register the `engine_` in this registry.
pub fn create_auth_module<EngineApi, EngineT>(&self, engine_api: EngineApi) -> AuthRpcModule
where
EngineT: EngineTypes + 'static,
EngineApi: EngineApiServer<EngineT>,
{
let mut module = RpcModule::new(());
module.merge(engine_api.into_rpc()).expect("No conflicting methods");
// also merge a subset of `eth_` handlers
let eth_handlers = self.eth_handlers();
let engine_eth = EngineEthApi::new(eth_handlers.api.clone(), eth_handlers.filter.clone());
module.merge(engine_eth.into_rpc()).expect("No conflicting methods");
AuthRpcModule { inner: module }
}
/// Helper function to create a [`RpcModule`] if it's not `None`
fn maybe_module(&mut self, config: Option<&RpcModuleSelection>) -> Option<RpcModule<()>> {
config.map(|config| self.module_for(config))
@ -908,13 +1039,8 @@ where
&mut self,
namespaces: impl Iterator<Item = RethRpcModule>,
) -> Vec<Methods> {
let EthHandlers {
api: eth_api,
filter: eth_filter,
pubsub: eth_pubsub,
cache: _,
blocking_task_pool: _,
} = self.with_eth(|eth| eth.clone());
let EthHandlers { api: eth_api, filter: eth_filter, pubsub: eth_pubsub, .. } =
self.eth_handlers().clone();
// Create a copy, so we can list out all the methods for rpc_ api
let namespaces: Vec<_> = namespaces.collect();
@ -983,120 +1109,6 @@ where
})
.collect::<Vec<_>>()
}
/// Returns the [`EthStateCache`] frontend
///
/// This will spawn exactly one [`EthStateCache`] service if this is the first time the cache is
/// requested.
pub fn eth_cache(&mut self) -> EthStateCache {
self.with_eth(|handlers| handlers.cache.clone())
}
/// Creates the [`EthHandlers`] type the first time this is called.
///
/// This will spawn the required service tasks for [`EthApi`] for:
/// - [`EthStateCache`]
/// - [`FeeHistoryCache`](reth_rpc_eth_types::FeeHistoryCache)
fn with_eth<F, R>(&mut self, f: F) -> R
where
F: FnOnce(&EthHandlers<Provider, Pool, Network, Events, EvmConfig>) -> R,
{
f(match &self.eth {
Some(eth) => eth,
None => self.eth.insert(self.init_eth()),
})
}
fn init_eth(&self) -> EthHandlers<Provider, Pool, Network, Events, EvmConfig> {
EthHandlersBuilder::new(
EthHandlersConfig {
provider: self.provider.clone(),
pool: self.pool.clone(),
network: self.network.clone(),
executor: self.executor.clone(),
events: self.events.clone(),
evm_config: self.evm_config.clone(),
eth_raw_transaction_forwarder: self.eth_raw_transaction_forwarder.clone(),
},
self.config.clone(),
)
.build()
}
/// Returns the configured [`EthHandlers`] or creates it if it does not exist yet
///
/// # Panics
///
/// If called outside of the tokio runtime. See also [`Self::eth_api`]
pub fn eth_handlers(&mut self) -> EthHandlers<Provider, Pool, Network, Events, EvmConfig> {
self.with_eth(|handlers| handlers.clone())
}
/// Returns the configured [`EthApi`] or creates it if it does not exist yet
///
/// Caution: This will spawn the necessary tasks required by the [`EthApi`]: [`EthStateCache`].
///
/// # Panics
///
/// If called outside of the tokio runtime.
pub fn eth_api(&mut self) -> EthApi<Provider, Pool, Network, EvmConfig> {
self.with_eth(|handlers| handlers.api.clone())
}
/// Instantiates `TraceApi`
///
/// # Panics
///
/// If called outside of the tokio runtime. See also [`Self::eth_api`]
pub fn trace_api(&mut self) -> TraceApi<Provider, EthApi<Provider, Pool, Network, EvmConfig>> {
let eth = self.eth_handlers();
TraceApi::new(self.provider.clone(), eth.api, self.blocking_pool_guard.clone())
}
/// Instantiates [`EthBundle`] Api
///
/// # Panics
///
/// If called outside of the tokio runtime. See also [`Self::eth_api`]
pub fn bundle_api(&mut self) -> EthBundle<EthApi<Provider, Pool, Network, EvmConfig>> {
let eth_api = self.eth_api();
EthBundle::new(eth_api, self.blocking_pool_guard.clone())
}
/// Instantiates `OtterscanApi`
///
/// # Panics
///
/// If called outside of the tokio runtime. See also [`Self::eth_api`]
pub fn otterscan_api(&mut self) -> OtterscanApi<EthApi<Provider, Pool, Network, EvmConfig>> {
let eth_api = self.eth_api();
OtterscanApi::new(eth_api)
}
/// Instantiates `DebugApi`
///
/// # Panics
///
/// If called outside of the tokio runtime. See also [`Self::eth_api`]
pub fn debug_api(&mut self) -> DebugApi<Provider, EthApi<Provider, Pool, Network, EvmConfig>> {
let eth_api = self.eth_api();
DebugApi::new(self.provider.clone(), eth_api, self.blocking_pool_guard.clone())
}
/// Instantiates `NetApi`
///
/// # Panics
///
/// If called outside of the tokio runtime. See also [`Self::eth_api`]
pub fn net_api(&mut self) -> NetApi<Network, EthApi<Provider, Pool, Network, EvmConfig>> {
let eth_api = self.eth_api();
NetApi::new(self.network.clone(), eth_api)
}
/// Instantiates `RethApi`
pub fn reth_api(&self) -> RethApi<Provider> {
RethApi::new(self.provider.clone(), Box::new(self.executor.clone()))
}
}
/// A builder type for configuring and launching the servers that will handle RPC requests.

23
crates/rpc/rpc-builder/tests/it/startup.rs
View File

@ -1,14 +1,16 @@
//! Startup tests
use std::io;
use reth_rpc_builder::{
error::{RpcError, ServerKind, WsHttpSamePortError},
EthApiBuild, RpcServerConfig, TransportRpcModuleConfig,
};
use reth_rpc_server_types::RethRpcModule;
use crate::utils::{
launch_http, launch_http_ws_same_port, launch_ws, test_address, test_rpc_builder,
};
use reth_rpc_builder::{
error::{RpcError, ServerKind, WsHttpSamePortError},
RpcServerConfig, TransportRpcModuleConfig,
};
use reth_rpc_server_types::RethRpcModule;
use std::io;
fn is_addr_in_use_kind(err: &RpcError, kind: ServerKind) -> bool {
match err {
@ -24,7 +26,8 @@ async fn test_http_addr_in_use() {
let handle = launch_http(vec![RethRpcModule::Admin]).await;
let addr = handle.http_local_addr().unwrap();
let builder = test_rpc_builder();
let server = builder.build(TransportRpcModuleConfig::set_http(vec![RethRpcModule::Admin]));
let server = builder
.build(TransportRpcModuleConfig::set_http(vec![RethRpcModule::Admin]), EthApiBuild::build);
let result = server
.start_server(RpcServerConfig::http(Default::default()).with_http_address(addr))
.await;
@ -37,7 +40,8 @@ async fn test_ws_addr_in_use() {
let handle = launch_ws(vec![RethRpcModule::Admin]).await;
let addr = handle.ws_local_addr().unwrap();
let builder = test_rpc_builder();
let server = builder.build(TransportRpcModuleConfig::set_ws(vec![RethRpcModule::Admin]));
let server = builder
.build(TransportRpcModuleConfig::set_ws(vec![RethRpcModule::Admin]), EthApiBuild::build);
let result =
server.start_server(RpcServerConfig::ws(Default::default()).with_ws_address(addr)).await;
let err = result.unwrap_err();
@ -58,6 +62,7 @@ async fn test_launch_same_port_different_modules() {
let server = builder.build(
TransportRpcModuleConfig::set_ws(vec![RethRpcModule::Admin])
.with_http(vec![RethRpcModule::Eth]),
EthApiBuild::build,
);
let addr = test_address();
let res = server
@ -81,6 +86,7 @@ async fn test_launch_same_port_same_cors() {
let server = builder.build(
TransportRpcModuleConfig::set_ws(vec![RethRpcModule::Eth])
.with_http(vec![RethRpcModule::Eth]),
EthApiBuild::build,
);
let addr = test_address();
let res = server
@ -102,6 +108,7 @@ async fn test_launch_same_port_different_cors() {
let server = builder.build(
TransportRpcModuleConfig::set_ws(vec![RethRpcModule::Eth])
.with_http(vec![RethRpcModule::Eth]),
EthApiBuild::build,
);
let addr = test_address();
let res = server

21
crates/rpc/rpc-builder/tests/it/utils.rs
View File

@ -1,3 +1,5 @@
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use reth_beacon_consensus::BeaconConsensusEngineHandle;
use reth_chainspec::MAINNET;
use reth_ethereum_engine_primitives::EthEngineTypes;
@ -7,7 +9,7 @@ use reth_payload_builder::test_utils::spawn_test_payload_service;
use reth_provider::test_utils::{NoopProvider, TestCanonStateSubscriptions};
use reth_rpc_builder::{
auth::{AuthRpcModule, AuthServerConfig, AuthServerHandle},
RpcModuleBuilder, RpcServerConfig, RpcServerHandle, TransportRpcModuleConfig,
EthApiBuild, RpcModuleBuilder, RpcServerConfig, RpcServerHandle, TransportRpcModuleConfig,
};
use reth_rpc_engine_api::EngineApi;
use reth_rpc_layer::JwtSecret;
@ -15,7 +17,6 @@ use reth_rpc_server_types::RpcModuleSelection;
use reth_rpc_types::engine::{ClientCode, ClientVersionV1};
use reth_tasks::TokioTaskExecutor;
use reth_transaction_pool::test_utils::{TestPool, TestPoolBuilder};
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use tokio::sync::mpsc::unbounded_channel;
/// Localhost with port 0 so a free port is used.
@ -51,7 +52,7 @@ pub async fn launch_auth(secret: JwtSecret) -> AuthServerHandle {
/// Launches a new server with http only with the given modules
pub async fn launch_http(modules: impl Into<RpcModuleSelection>) -> RpcServerHandle {
let builder = test_rpc_builder();
let server = builder.build(TransportRpcModuleConfig::set_http(modules));
let server = builder.build(TransportRpcModuleConfig::set_http(modules), EthApiBuild::build);
server
.start_server(RpcServerConfig::http(Default::default()).with_http_address(test_address()))
.await
@ -61,7 +62,7 @@ pub async fn launch_http(modules: impl Into<RpcModuleSelection>) -> RpcServerHan
/// Launches a new server with ws only with the given modules
pub async fn launch_ws(modules: impl Into<RpcModuleSelection>) -> RpcServerHandle {
let builder = test_rpc_builder();
let server = builder.build(TransportRpcModuleConfig::set_ws(modules));
let server = builder.build(TransportRpcModuleConfig::set_ws(modules), EthApiBuild::build);
server
.start_server(RpcServerConfig::ws(Default::default()).with_ws_address(test_address()))
.await
@ -72,8 +73,10 @@ pub async fn launch_ws(modules: impl Into<RpcModuleSelection>) -> RpcServerHandl
pub async fn launch_http_ws(modules: impl Into<RpcModuleSelection>) -> RpcServerHandle {
let builder = test_rpc_builder();
let modules = modules.into();
let server =
builder.build(TransportRpcModuleConfig::set_ws(modules.clone()).with_http(modules));
let server = builder.build(
TransportRpcModuleConfig::set_ws(modules.clone()).with_http(modules),
EthApiBuild::build,
);
server
.start_server(
RpcServerConfig::ws(Default::default())
@ -89,8 +92,10 @@ pub async fn launch_http_ws(modules: impl Into<RpcModuleSelection>) -> RpcServer
pub async fn launch_http_ws_same_port(modules: impl Into<RpcModuleSelection>) -> RpcServerHandle {
let builder = test_rpc_builder();
let modules = modules.into();
let server =
builder.build(TransportRpcModuleConfig::set_ws(modules.clone()).with_http(modules));
let server = builder.build(
TransportRpcModuleConfig::set_ws(modules.clone()).with_http(modules),
EthApiBuild::build,
);
let addr = test_address();
server
.start_server(

18
crates/rpc/rpc-eth-api/src/core.rs
View File

@ -16,9 +16,16 @@ use reth_rpc_types::{
use tracing::trace;
use crate::helpers::{
EthApiSpec, EthBlocks, EthCall, EthFees, EthState, EthTransactions, LoadReceipt, Trace,
transaction::UpdateRawTxForwarder, EthApiSpec, EthBlocks, EthCall, EthFees, EthState,
EthTransactions, FullEthApi,
};
/// Helper trait, unifies functionality that must be supported to implement all RPC methods for
/// server.
pub trait FullEthApiServer: EthApiServer + FullEthApi + UpdateRawTxForwarder + Clone {}
impl<T> FullEthApiServer for T where T: EthApiServer + FullEthApi + UpdateRawTxForwarder + Clone {}
/// Eth rpc interface: <https://ethereum.github.io/execution-apis/api-documentation/>
#[cfg_attr(not(feature = "client"), rpc(server, namespace = "eth"))]
#[cfg_attr(feature = "client", rpc(server, client, namespace = "eth"))]
@ -324,14 +331,7 @@ pub trait EthApi {
#[async_trait::async_trait]
impl<T> EthApiServer for T
where
Self: EthApiSpec
+ EthTransactions
+ EthBlocks
+ EthState
+ EthCall
+ EthFees
+ Trace
+ LoadReceipt,
Self: FullEthApi,
{
/// Handler for: `eth_protocolVersion`
async fn protocol_version(&self) -> RpcResult<U64> {

20
crates/rpc/rpc-eth-api/src/helpers/mod.rs
View File

@ -36,7 +36,7 @@ pub use signer::EthSigner;
pub use spec::EthApiSpec;
pub use state::{EthState, LoadState};
pub use trace::Trace;
pub use transaction::{EthTransactions, LoadTransaction};
pub use transaction::{EthTransactions, LoadTransaction, UpdateRawTxForwarder};
/// Extension trait that bundles traits needed for tracing transactions.
pub trait TraceExt:
@ -45,3 +45,21 @@ pub trait TraceExt:
}
impl<T> TraceExt for T where T: LoadTransaction + LoadBlock + LoadPendingBlock + Trace + Call {}
/// Helper trait to unify all `eth` rpc server building block traits, for simplicity.
pub trait FullEthApi:
EthApiSpec + EthTransactions + EthBlocks + EthState + EthCall + EthFees + Trace + LoadReceipt
{
}
impl<T> FullEthApi for T where
T: EthApiSpec
+ EthTransactions
+ EthBlocks
+ EthState
+ EthCall
+ EthFees
+ Trace
+ LoadReceipt
{
}

20
crates/rpc/rpc-eth-api/src/helpers/transaction.rs
View File

@ -1,7 +1,7 @@
//! Database access for `eth_` transaction RPC methods. Loads transaction and receipt data w.r.t.
//! network.
use std::{fmt, sync::Arc};
use std::{fmt, ops::Deref, sync::Arc};
use alloy_dyn_abi::TypedData;
use futures::Future;
@ -648,3 +648,21 @@ pub trait RawTransactionForwarder: fmt::Debug + Send + Sync + 'static {
/// Forwards raw transaction bytes for `eth_sendRawTransaction`
async fn forward_raw_transaction(&self, raw: &[u8]) -> EthResult<()>;
}
/// Configure server's forwarder for `eth_sendRawTransaction`, at runtime.
pub trait UpdateRawTxForwarder {
/// Sets a forwarder for `eth_sendRawTransaction`
///
/// Note: this might be removed in the future in favor of a more generic approach.
fn set_eth_raw_transaction_forwarder(&self, forwarder: Arc<dyn RawTransactionForwarder>);
}
impl<T, K> UpdateRawTxForwarder for T
where
T: Deref<Target = Arc<K>>,
K: UpdateRawTxForwarder,
{
fn set_eth_raw_transaction_forwarder(&self, forwarder: Arc<dyn RawTransactionForwarder>) {
self.deref().deref().set_eth_raw_transaction_forwarder(forwarder);
}
}

2
crates/rpc/rpc-eth-api/src/lib.rs
View File

@ -19,7 +19,7 @@ pub mod helpers;
pub mod pubsub;
pub use bundle::{EthBundleApiServer, EthCallBundleApiServer};
pub use core::EthApiServer;
pub use core::{EthApiServer, FullEthApiServer};
pub use filter::EthFilterApiServer;
pub use pubsub::EthPubSubApiServer;

2
crates/rpc/rpc-eth-types/src/cache/config.rs vendored
View File

@ -8,7 +8,7 @@ use reth_rpc_server_types::constants::cache::{
};
/// Settings for the [`EthStateCache`](super::EthStateCache).
#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct EthStateCacheConfig {
/// Max number of blocks in cache.

2
crates/rpc/rpc-eth-types/src/fee_history.rs
View File

@ -168,7 +168,7 @@ impl FeeHistoryCache {
}
/// Settings for the [`FeeHistoryCache`].
#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct FeeHistoryCacheConfig {
/// Max number of blocks in cache.

2
crates/rpc/rpc-eth-types/src/gas_oracle.rs
View File

@ -24,7 +24,7 @@ use super::{EthApiError, EthResult, EthStateCache, RpcInvalidTransactionError};
pub const RPC_DEFAULT_GAS_CAP: GasCap = GasCap(constants::gas_oracle::RPC_DEFAULT_GAS_CAP);
/// Settings for the [`GasPriceOracle`]
#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct GasPriceOracleConfig {
/// The number of populated blocks to produce the gas price estimate

4
crates/rpc/rpc-eth-types/src/lib.rs
View File

@ -26,7 +26,9 @@ pub use cache::{
};
pub use error::{EthApiError, EthResult, RevertError, RpcInvalidTransactionError, SignError};
pub use fee_history::{FeeHistoryCache, FeeHistoryCacheConfig, FeeHistoryEntry};
pub use gas_oracle::{GasCap, GasPriceOracle, GasPriceOracleConfig, GasPriceOracleResult};
pub use gas_oracle::{
GasCap, GasPriceOracle, GasPriceOracleConfig, GasPriceOracleResult, RPC_DEFAULT_GAS_CAP,
};
pub use id_provider::EthSubscriptionIdProvider;
pub use logs_utils::EthFilterError;
pub use pending_block::{PendingBlock, PendingBlockEnv, PendingBlockEnvOrigin};

1
crates/rpc/rpc/Cargo.toml
View File

@ -71,6 +71,7 @@ futures.workspace = true
rand.workspace = true
serde.workspace = true
thiserror.workspace = true
derive_more.workspace = true
[dev-dependencies]
reth-evm-ethereum.workspace = true

12
crates/rpc/rpc/src/eth/core.rs
View File

@ -3,10 +3,11 @@
use std::sync::Arc;
use derive_more::Deref;
use reth_primitives::{BlockNumberOrTag, U256};
use reth_provider::{BlockReaderIdExt, ChainSpecProvider};
use reth_rpc_eth_api::{
helpers::{EthSigner, SpawnBlocking},
helpers::{transaction::UpdateRawTxForwarder, EthSigner, SpawnBlocking},
RawTransactionForwarder,
};
use reth_rpc_eth_types::{EthStateCache, FeeHistoryCache, GasCap, GasPriceOracle, PendingBlock};
@ -24,6 +25,7 @@ use crate::eth::DevSigner;
/// separately in submodules. The rpc handler implementation can then delegate to the main impls.
/// This way [`EthApi`] is not limited to [`jsonrpsee`] and can be used standalone or in other
/// network handlers (for example ipc).
#[derive(Deref)]
pub struct EthApi<Provider, Pool, Network, EvmConfig> {
/// All nested fields bundled together.
pub(super) inner: Arc<EthApiInner<Provider, Pool, Network, EvmConfig>>,
@ -302,6 +304,14 @@ impl<Provider, Pool, Network, EvmConfig> EthApiInner<Provider, Pool, Network, Ev
}
}
impl<Provider, Pool, Network, EvmConfig> UpdateRawTxForwarder
for EthApiInner<Provider, Pool, Network, EvmConfig>
{
fn set_eth_raw_transaction_forwarder(&self, forwarder: Arc<dyn RawTransactionForwarder>) {
self.raw_transaction_forwarder.write().replace(forwarder);
}
}
#[cfg(test)]
mod tests {
use jsonrpsee_types::error::INVALID_PARAMS_CODE;

2
crates/storage/db-api/src/lib.rs
View File

@ -79,3 +79,5 @@ pub mod models;
mod scale;
mod utils;
pub use database::Database;

37
crates/storage/provider/src/test_utils/noop.rs
View File

@ -1,11 +1,8 @@
use crate::{
traits::{BlockSource, ReceiptProvider},
AccountReader, BlockHashReader, BlockIdReader, BlockNumReader, BlockReader, BlockReaderIdExt,
ChainSpecProvider, ChangeSetReader, EvmEnvProvider, HeaderProvider, PruneCheckpointReader,
ReceiptProviderIdExt, RequestsProvider, StageCheckpointReader, StateProvider, StateProviderBox,
StateProviderFactory, StateRootProvider, TransactionVariant, TransactionsProvider,
WithdrawalsProvider,
use std::{
ops::{RangeBounds, RangeInclusive},
sync::Arc,
};
use reth_chainspec::{ChainInfo, ChainSpec, MAINNET};
use reth_db_api::models::{AccountBeforeTx, StoredBlockBodyIndices};
use reth_evm::ConfigureEvmEnv;
@ -23,9 +20,17 @@ use revm::{
db::BundleState,
primitives::{BlockEnv, CfgEnvWithHandlerCfg},
};
use std::{
ops::{RangeBounds, RangeInclusive},
sync::Arc,
use tokio::sync::broadcast;
use crate::{
providers::StaticFileProvider,
traits::{BlockSource, ReceiptProvider},
AccountReader, BlockHashReader, BlockIdReader, BlockNumReader, BlockReader, BlockReaderIdExt,
CanonStateNotifications, CanonStateSubscriptions, ChainSpecProvider, ChangeSetReader,
EvmEnvProvider, HeaderProvider, PruneCheckpointReader, ReceiptProviderIdExt, RequestsProvider,
StageCheckpointReader, StateProvider, StateProviderBox, StateProviderFactory,
StateRootProvider, StaticFileProviderFactory, TransactionVariant, TransactionsProvider,
WithdrawalsProvider,
};
/// Supports various api interfaces for testing purposes.
@ -465,3 +470,15 @@ impl PruneCheckpointReader for NoopProvider {
Ok(None)
}
}
impl StaticFileProviderFactory for NoopProvider {
fn static_file_provider(&self) -> StaticFileProvider {
StaticFileProvider::default()
}
}
impl CanonStateSubscriptions for NoopProvider {
fn subscribe_to_canonical_state(&self) -> CanonStateNotifications {
broadcast::channel(1).1
}
}

32
crates/storage/provider/src/traits/full.rs
View File

@ -2,8 +2,8 @@
use crate::{
AccountReader, BlockReaderIdExt, CanonStateSubscriptions, ChainSpecProvider, ChangeSetReader,
DatabaseProviderFactory, EvmEnvProvider, StageCheckpointReader, StateProviderFactory,
StaticFileProviderFactory,
DatabaseProviderFactory, EvmEnvProvider, HeaderProvider, StageCheckpointReader,
StateProviderFactory, StaticFileProviderFactory, TransactionsProvider,
};
use reth_db_api::database::Database;
@ -41,3 +41,31 @@ impl<T, DB: Database> FullProvider<DB> for T where
+ 'static
{
}
/// Helper trait to unify all provider traits required to support `eth` RPC server behaviour, for
/// simplicity.
pub trait FullRpcProvider:
StateProviderFactory
+ EvmEnvProvider
+ ChainSpecProvider
+ BlockReaderIdExt
+ HeaderProvider
+ TransactionsProvider
+ Clone
+ Unpin
+ 'static
{
}
impl<T> FullRpcProvider for T where
T: StateProviderFactory
+ EvmEnvProvider
+ ChainSpecProvider
+ BlockReaderIdExt
+ HeaderProvider
+ TransactionsProvider
+ Clone
+ Unpin
+ 'static
{
}

2
crates/storage/provider/src/traits/mod.rs
View File

@ -43,7 +43,7 @@ mod stats;
pub use stats::StatsReader;
mod full;
pub use full::FullProvider;
pub use full::{FullProvider, FullRpcProvider};
mod tree_viewer;
pub use tree_viewer::TreeViewer;

2
examples/custom-dev-node/src/main.rs
View File

@ -27,7 +27,7 @@ async fn main() -> eyre::Result<()> {
.with_rpc(RpcServerArgs::default().with_http())
.with_chain(custom_chain());
let NodeHandle { mut node, node_exit_future: _ } = NodeBuilder::new(node_config)
let NodeHandle { node, node_exit_future: _ } = NodeBuilder::new(node_config)
.testing_node(tasks.executor())
.node(EthereumNode::default())
.launch()

4
examples/custom-inspector/src/main.rs
View File

@ -31,14 +31,14 @@ fn main() {
Cli::<RethCliTxpoolExt>::parse()
.run(|builder, args| async move {
// launch the node
let NodeHandle { mut node, node_exit_future } =
let NodeHandle { node, node_exit_future } =
builder.node(EthereumNode::default()).launch().await?;
// create a new subscription to pending transactions
let mut pending_transactions = node.pool.new_pending_pool_transactions_listener();
// get an instance of the `trace_` API handler
let eth_api = node.rpc_registry.eth_api();
let eth_api = node.rpc_registry.eth_api().clone();
println!("Spawning trace task!");

7
examples/rpc-db/src/main.rs
View File

@ -12,6 +12,8 @@
//! cast rpc myrpcExt_customMethod
//! ```
use std::{path::Path, sync::Arc};
use reth::{
providers::{
providers::{BlockchainProvider, StaticFileProvider},
@ -25,7 +27,7 @@ use reth_db_api::models::ClientVersion;
// Bringing up the RPC
use reth::rpc::builder::{
RethRpcModule, RpcModuleBuilder, RpcServerConfig, TransportRpcModuleConfig,
EthApiBuild, RethRpcModule, RpcModuleBuilder, RpcServerConfig, TransportRpcModuleConfig,
};
// Configuring the network parts, ideally also wouldn't need to think about this.
use myrpc_ext::{MyRpcExt, MyRpcExtApiServer};
@ -34,7 +36,6 @@ use reth::{
tasks::TokioTaskExecutor,
};
use reth_node_ethereum::EthEvmConfig;
use std::{path::Path, sync::Arc};
// Custom rpc extension
pub mod myrpc_ext;
@ -71,7 +72,7 @@ async fn main() -> eyre::Result<()> {
// Pick which namespaces to expose.
let config = TransportRpcModuleConfig::default().with_http([RethRpcModule::Eth]);
let mut server = rpc_builder.build(config);
let mut server = rpc_builder.build(config, EthApiBuild::build);
// Add a custom rpc namespace
let custom_rpc = MyRpcExt { provider };

2
examples/txpool-tracing/src/main.rs
View File

@ -28,7 +28,7 @@ fn main() {
Cli::<RethCliTxpoolExt>::parse()
.run(|builder, args| async move {
// launch the node
let NodeHandle { mut node, node_exit_future } =
let NodeHandle { node, node_exit_future } =
builder.node(EthereumNode::default()).launch().await?;
// create a new subscription to pending transactions