feat: NodePrimitivesProvider (#12855)

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

@@ -1,5 +1,6 @@
 use alloy_consensus::Header;
 use reth_evm::ConfigureEvm;
+use reth_primitives::EthPrimitives;
 use reth_provider::{BlockReader, CanonStateSubscriptions, EvmEnvProvider, StateProviderFactory};
 use reth_rpc::{EthFilter, EthPubSub};
 use reth_rpc_eth_api::EthApiTypes;
@@ -35,7 +36,7 @@ where
         + 'static,
     Pool: Send + Sync + Clone + 'static,
     Network: Clone + 'static,
-    Events: CanonStateSubscriptions + Clone + 'static,
+    Events: CanonStateSubscriptions<Primitives = EthPrimitives> + Clone + 'static,
     EthApi: EthApiTypes + 'static,
 {
     /// Returns a new instance with handlers for `eth` namespace.

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

@@ -42,7 +42,8 @@
 //!         + ChangeSetReader,
 //!     Pool: TransactionPool + Unpin + 'static,
 //!     Network: NetworkInfo + Peers + Clone + 'static,
-//!     Events: CanonStateSubscriptions + Clone + 'static,
+//!     Events:
+//!         CanonStateSubscriptions<Primitives = reth_primitives::EthPrimitives> + Clone + 'static,
 //!     EvmConfig: ConfigureEvm<Header = Header>,
 //!     BlockExecutor: BlockExecutorProvider,
 //!     Consensus: reth_consensus::Consensus + Clone + 'static,
@@ -118,7 +119,8 @@
 //!         + ChangeSetReader,
 //!     Pool: TransactionPool + Unpin + 'static,
 //!     Network: NetworkInfo + Peers + Clone + 'static,
-//!     Events: CanonStateSubscriptions + Clone + 'static,
+//!     Events:
+//!         CanonStateSubscriptions<Primitives = reth_primitives::EthPrimitives> + Clone + 'static,
 //!     EngineApi: EngineApiServer<EngineT>,
 //!     EngineT: EngineTypes,
 //!     EvmConfig: ConfigureEvm<Header = Header>,
@@ -190,6 +192,7 @@ use reth_consensus::Consensus;
 use reth_engine_primitives::EngineTypes;
 use reth_evm::{execute::BlockExecutorProvider, ConfigureEvm};
 use reth_network_api::{noop::NoopNetwork, NetworkInfo, Peers};
+use reth_primitives::EthPrimitives;
 use reth_provider::{
     AccountReader, BlockReader, CanonStateSubscriptions, ChainSpecProvider, ChangeSetReader,
     EvmEnvProvider, FullRpcProvider, StateProviderFactory,
@@ -264,7 +267,7 @@ where
     Pool: TransactionPool + 'static,
     Network: NetworkInfo + Peers + Clone + 'static,
     Tasks: TaskSpawner + Clone + 'static,
-    Events: CanonStateSubscriptions + Clone + 'static,
+    Events: CanonStateSubscriptions<Primitives = EthPrimitives> + Clone + 'static,
     EvmConfig: ConfigureEvm<Header = alloy_consensus::Header>,
     EthApi: FullEthApiServer,
     BlockExecutor: BlockExecutorProvider,
@@ -617,7 +620,7 @@ where
     Pool: TransactionPool + 'static,
     Network: NetworkInfo + Peers + Clone + 'static,
     Tasks: TaskSpawner + Clone + 'static,
-    Events: CanonStateSubscriptions + Clone + 'static,
+    Events: CanonStateSubscriptions<Primitives = EthPrimitives> + Clone + 'static,
     EvmConfig: ConfigureEvm<Header = Header>,
     BlockExecutor: BlockExecutorProvider,
     Consensus: reth_consensus::Consensus + Clone + 'static,
@@ -920,7 +923,7 @@ where
         + 'static,
     Pool: Send + Sync + Clone + 'static,
     Network: Clone + 'static,
-    Events: CanonStateSubscriptions + Clone + 'static,
+    Events: CanonStateSubscriptions<Primitives = EthPrimitives> + Clone + 'static,
     Tasks: TaskSpawner + Clone + 'static,
     EthApi: EthApiTypes + 'static,
     BlockExecutor: BlockExecutorProvider,
@@ -1282,7 +1285,7 @@ where
     Pool: TransactionPool + 'static,
     Network: NetworkInfo + Peers + Clone + 'static,
     Tasks: TaskSpawner + Clone + 'static,
-    Events: CanonStateSubscriptions + Clone + 'static,
+    Events: CanonStateSubscriptions<Primitives = EthPrimitives> + Clone + 'static,
     EthApi: FullEthApiServer,
     BlockExecutor: BlockExecutorProvider,
     Consensus: reth_consensus::Consensus + Clone + 'static,

crates/rpc/rpc-eth-types/src/builder/ctx.rs

@@ -2,6 +2,7 @@
 
 use reth_chain_state::CanonStateSubscriptions;
 use reth_chainspec::ChainSpecProvider;
+use reth_primitives::NodePrimitives;
 use reth_storage_api::BlockReaderIdExt;
 use reth_tasks::TaskSpawner;
 
@@ -41,7 +42,8 @@ where
     where
         Provider: ChainSpecProvider + 'static,
         Tasks: TaskSpawner,
-        Events: CanonStateSubscriptions,
+        Events:
+            CanonStateSubscriptions<Primitives: NodePrimitives<Receipt = reth_primitives::Receipt>>,
     {
         let fee_history_cache =
             FeeHistoryCache::new(self.cache.clone(), self.config.fee_history_cache);

crates/rpc/rpc-eth-types/src/fee_history.rs

@@ -16,7 +16,7 @@ use futures::{
 use metrics::atomics::AtomicU64;
 use reth_chain_state::CanonStateNotification;
 use reth_chainspec::{ChainSpecProvider, EthChainSpec};
-use reth_primitives::{Receipt, SealedBlock, TransactionSigned};
+use reth_primitives::{NodePrimitives, Receipt, SealedBlock, TransactionSigned};
 use reth_storage_api::BlockReaderIdExt;
 use revm_primitives::{calc_blob_gasprice, calc_excess_blob_gas};
 use serde::{Deserialize, Serialize};
@@ -205,13 +205,14 @@ struct FeeHistoryCacheInner {
 
 /// Awaits for new chain events and directly inserts them into the cache so they're available
 /// immediately before they need to be fetched from disk.
-pub async fn fee_history_cache_new_blocks_task<St, Provider>(
+pub async fn fee_history_cache_new_blocks_task<St, Provider, N>(
     fee_history_cache: FeeHistoryCache,
     mut events: St,
     provider: Provider,
 ) where
-    St: Stream<Item = CanonStateNotification> + Unpin + 'static,
+    St: Stream<Item = CanonStateNotification<N>> + Unpin + 'static,
     Provider: BlockReaderIdExt + ChainSpecProvider + 'static,
+    N: NodePrimitives<Receipt = reth_primitives::Receipt>,
 {
     // We're listening for new blocks emitted when the node is in live sync.
     // If the node transitions to stage sync, we need to fetch the missing blocks

crates/rpc/rpc/src/eth/core.rs

@@ -7,6 +7,7 @@ use alloy_eips::BlockNumberOrTag;
 use alloy_network::Ethereum;
 use alloy_primitives::U256;
 use derive_more::Deref;
+use reth_primitives::NodePrimitives;
 use reth_provider::{BlockReaderIdExt, CanonStateSubscriptions, ChainSpecProvider};
 use reth_rpc_eth_api::{
     helpers::{EthSigner, SpawnBlocking},
@@ -102,7 +103,8 @@ where
     ) -> Self
     where
         Tasks: TaskSpawner + Clone + 'static,
-        Events: CanonStateSubscriptions,
+        Events:
+            CanonStateSubscriptions<Primitives: NodePrimitives<Receipt = reth_primitives::Receipt>>,
     {
         let blocking_task_pool =
             BlockingTaskPool::build().expect("failed to build blocking task pool");

crates/rpc/rpc/src/eth/pubsub.rs

@@ -15,6 +15,7 @@ use jsonrpsee::{
     server::SubscriptionMessage, types::ErrorObject, PendingSubscriptionSink, SubscriptionSink,
 };
 use reth_network_api::NetworkInfo;
+use reth_primitives::NodePrimitives;
 use reth_provider::{BlockReader, CanonStateSubscriptions, EvmEnvProvider};
 use reth_rpc_eth_api::{pubsub::EthPubSubApiServer, TransactionCompat};
 use reth_rpc_eth_types::logs_utils;
@@ -84,7 +85,9 @@ impl<Provider, Pool, Events, Network, Eth> EthPubSubApiServer<Eth::Transaction>
 where
     Provider: BlockReader + EvmEnvProvider + Clone + 'static,
     Pool: TransactionPool + 'static,
-    Events: CanonStateSubscriptions + Clone + 'static,
+    Events: CanonStateSubscriptions<Primitives: NodePrimitives<Receipt = reth_primitives::Receipt>>
+        + Clone
+        + 'static,
     Network: NetworkInfo + Clone + 'static,
     Eth: TransactionCompat + 'static,
 {
@@ -117,7 +120,9 @@ async fn handle_accepted<Provider, Pool, Events, Network, Eth>(
 where
     Provider: BlockReader + EvmEnvProvider + Clone + 'static,
     Pool: TransactionPool + 'static,
-    Events: CanonStateSubscriptions + Clone + 'static,
+    Events: CanonStateSubscriptions<Primitives: NodePrimitives<Receipt = reth_primitives::Receipt>>
+        + Clone
+        + 'static,
     Network: NetworkInfo + Clone + 'static,
     Eth: TransactionCompat,
 {
@@ -333,7 +338,8 @@ where
 impl<Provider, Pool, Events, Network> EthPubSubInner<Provider, Pool, Events, Network>
 where
     Provider: BlockReader + EvmEnvProvider + 'static,
-    Events: CanonStateSubscriptions + 'static,
+    Events: CanonStateSubscriptions<Primitives: NodePrimitives<Receipt = reth_primitives::Receipt>>
+        + 'static,
     Network: NetworkInfo + 'static,
     Pool: 'static,
 {