feat(rpc): integrate eth cache and task executor in pipeline

bin/reth/src/args/rpc_server_args.rs

@@ -11,6 +11,7 @@ use reth_rpc_builder::{
     RpcServerHandle, ServerBuilder, TransportRpcModuleConfig,
 };
 use reth_rpc_engine_api::EngineApiHandle;
+use reth_tasks::TaskSpawner;
 use reth_transaction_pool::TransactionPool;
 use std::{
     net::{IpAddr, Ipv4Addr, SocketAddr},
@@ -103,38 +104,43 @@ impl RpcServerArgs {
     }
 
     /// Convenience function for starting a rpc server with configs which extracted from cli args.
-    pub(crate) async fn start_rpc_server<Client, Pool, Network>(
+    pub(crate) async fn start_rpc_server<Client, Pool, Network, Tasks>(
         &self,
         client: Client,
         pool: Pool,
         network: Network,
+        executor: Tasks,
     ) -> Result<RpcServerHandle, RpcError>
     where
         Client: BlockProvider
             + HeaderProvider
             + StateProviderFactory
             + EvmEnvProvider
             + Clone
+            + Unpin
             + 'static,
         Pool: TransactionPool + Clone + 'static,
         Network: NetworkInfo + Peers + Clone + 'static,
+        Tasks: TaskSpawner + Clone + 'static,
     {
         reth_rpc_builder::launch(
             client,
             pool,
             network,
             self.transport_rpc_module_config(),
             self.rpc_server_config(),
+            executor,
         )
         .await
     }
 
     /// Create Engine API server.
-    pub(crate) async fn start_auth_server<Client, Pool, Network>(
+    pub(crate) async fn start_auth_server<Client, Pool, Network, Tasks>(
         &self,
         client: Client,
         pool: Pool,
         network: Network,
+        executor: Tasks,
         handle: EngineApiHandle,
     ) -> Result<ServerHandle, RpcError>
     where
@@ -143,16 +149,27 @@ impl RpcServerArgs {
             + StateProviderFactory
             + EvmEnvProvider
             + Clone
+            + Unpin
             + 'static,
         Pool: TransactionPool + Clone + 'static,
         Network: NetworkInfo + Peers + Clone + 'static,
+        Tasks: TaskSpawner + Clone + 'static,
     {
         let socket_address = SocketAddr::new(
             self.auth_addr.unwrap_or(IpAddr::V4(Ipv4Addr::UNSPECIFIED)),
             self.auth_port.unwrap_or(constants::DEFAULT_AUTH_PORT),
         );
         let secret = self.jwt_secret().map_err(|err| RpcError::Custom(err.to_string()))?;
-        reth_rpc_builder::auth::launch(client, pool, network, handle, socket_address, secret).await
+        reth_rpc_builder::auth::launch(
+            client,
+            pool,
+            network,
+            executor,
+            handle,
+            socket_address,
+            secret,
+        )
+        .await
     }
 
     /// Creates the [TransportRpcModuleConfig] from cli args.

bin/reth/src/node/mod.rs

@@ -160,7 +160,12 @@ impl Command {
 
         let _rpc_server = self
             .rpc
-            .start_rpc_server(shareable_db.clone(), test_transaction_pool.clone(), network.clone())
+            .start_rpc_server(
+                shareable_db.clone(),
+                test_transaction_pool.clone(),
+                network.clone(),
+                ctx.task_executor.clone(),
+            )
             .await?;
         info!(target: "reth::cli", "Started RPC server");
 
@@ -174,6 +179,7 @@ impl Command {
                 shareable_db,
                 test_transaction_pool,
                 network.clone(),
+                ctx.task_executor.clone(),
                 engine_api_handle,
             )
             .await?;

crates/rpc/rpc-builder/Cargo.toml

@@ -15,6 +15,7 @@ reth-rpc = { path = "../rpc" }
 reth-rpc-api = { path = "../rpc-api" }
 reth-rpc-engine-api = { path = "../rpc-engine-api" }
 reth-rpc-types = { path = "../rpc-types" }
+reth-tasks = { path = "../../tasks" }
 reth-transaction-pool = { path = "../../transaction-pool" }
 
 jsonrpsee = { version = "0.16", features = ["server"] }

crates/rpc/rpc-builder/src/auth.rs

@@ -14,11 +14,12 @@ pub use reth_ipc::server::{Builder as IpcServerBuilder, Endpoint};
 use reth_network_api::{NetworkInfo, Peers};
 use reth_provider::{BlockProvider, EvmEnvProvider, HeaderProvider, StateProviderFactory};
 use reth_rpc::{
-    AdminApi, AuthLayer, DebugApi, EngineApi, EthApi, JwtAuthValidator, JwtSecret, NetApi,
-    TraceApi, Web3Api,
+    eth::cache::EthStateCache, AdminApi, AuthLayer, DebugApi, EngineApi, EthApi, JwtAuthValidator,
+    JwtSecret, NetApi, TraceApi, Web3Api,
 };
 use reth_rpc_api::servers::*;
 use reth_rpc_engine_api::EngineApiHandle;
+use reth_tasks::TaskSpawner;
 use reth_transaction_pool::TransactionPool;
 use serde::{Deserialize, Serialize, Serializer};
 use std::{
@@ -31,22 +32,32 @@ use strum::{AsRefStr, EnumString, EnumVariantNames, ParseError, VariantNames};
 use tower::layer::util::{Identity, Stack};
 use tower_http::cors::{AllowOrigin, Any, CorsLayer};
 
-/// Configure and launch an auth server with `engine` and `eth` namespaces.
-pub async fn launch<Client, Pool, Network>(
+/// Configure and launch an auth server with `engine` and a _new_ `eth` namespace.
+pub async fn launch<Client, Pool, Network, Tasks>(
     client: Client,
     pool: Pool,
     network: Network,
+    executor: Tasks,
     handle: EngineApiHandle,
     socket_addr: SocketAddr,
     secret: JwtSecret,
 ) -> Result<ServerHandle, RpcError>
 where
-    Client:
-        BlockProvider + HeaderProvider + StateProviderFactory + EvmEnvProvider + Clone + 'static,
+    Client: BlockProvider
+        + HeaderProvider
+        + StateProviderFactory
+        + EvmEnvProvider
+        + Clone
+        + Unpin
+        + 'static,
     Pool: TransactionPool + Clone + 'static,
     Network: NetworkInfo + Peers + Clone + 'static,
+    Tasks: TaskSpawner + Clone + 'static,
 {
-    launch_with_eth_api(EthApi::new(client, pool, network), handle, socket_addr, secret).await
+    // spawn a new cache task
+    let eth_cache = EthStateCache::spawn_with(client.clone(), Default::default(), executor);
+    launch_with_eth_api(EthApi::new(client, pool, network, eth_cache), handle, socket_addr, secret)
+        .await
 }
 
 /// Configure and launch an auth server with existing EthApi implementation.
@@ -57,8 +68,13 @@ pub async fn launch_with_eth_api<Client, Pool, Network>(
     secret: JwtSecret,
 ) -> Result<ServerHandle, RpcError>
 where
-    Client:
-        BlockProvider + HeaderProvider + StateProviderFactory + EvmEnvProvider + Clone + 'static,
+    Client: BlockProvider
+        + HeaderProvider
+        + StateProviderFactory
+        + EvmEnvProvider
+        + Clone
+        + Unpin
+        + 'static,
     Pool: TransactionPool + Clone + 'static,
     Network: NetworkInfo + Peers + Clone + 'static,
 {
@@ -71,7 +87,7 @@ where
     let middleware =
         tower::ServiceBuilder::new().layer(AuthLayer::new(JwtAuthValidator::new(secret)));
 
-    // By default both http and ws are enabled.
+    // By default, both http and ws are enabled.
     let server = ServerBuilder::new().set_middleware(middleware).build(socket_addr).await?;
 
     server.start(module)