feat(net): share sender recovery cache across transaction paths

Author: shekhirin

Cargo.lock

@@ -532,7 +532,7 @@ tracing-appender = "0.2.5"
 url = { version = "2.3", default-features = false }
 zstd = "0.13"
 byteorder = "1"
-fixed-cache = { version = "0.1.10", features = ["stats"] }
+fixed-cache = { version = "0.1.10", default-features = false, features = ["stats"] }
 moka = "0.12"
 tar-no-std = { version = "0.4.2", default-features = false }
 miniz_oxide = { version = "0.9.0", default-features = false }

Cargo.toml

@@ -30,7 +30,7 @@ use reth_chainspec::{ChainSpec, EthChainSpec, MAINNET};
 use reth_ethereum_primitives::{Block, EthPrimitives, TransactionSigned};
 use reth_evm::{
     eth::NextEvmEnvAttributes, precompiles::PrecompilesMap, ConfigureEvm, EvmEnv, EvmFactory,
-    JitBackend, NextBlockEnvAttributes, TransactionEnvMut,
+    JitBackend, NextBlockEnvAttributes, SenderRecoveryCache, TransactionEnvMut,
 };
 use reth_primitives_traits::{SealedBlock, SealedHeader};
 use revm::{context::BlockEnv, primitives::hardfork::SpecId};
@@ -87,6 +87,8 @@ pub struct EthEvmConfig<C = ChainSpec, EvmFactory = EthEvmFactory> {
     pub executor_factory: EthBlockExecutorFactory<RethReceiptBuilder, Arc<C>, EvmFactory>,
     /// Ethereum block assembler.
     pub block_assembler: EthBlockAssembler<C>,
+    /// Cache of recovered transaction senders.
+    pub sender_recovery_cache: SenderRecoveryCache,
 }
 
 impl EthEvmConfig {
@@ -113,6 +115,7 @@ impl<ChainSpec, EvmFactory> EthEvmConfig<ChainSpec, EvmFactory> {
     pub fn new_with_evm_factory(chain_spec: Arc<ChainSpec>, evm_factory: EvmFactory) -> Self {
         Self {
             block_assembler: EthBlockAssembler::new(chain_spec.clone()),
+            sender_recovery_cache: SenderRecoveryCache::default(),
             executor_factory: EthBlockExecutorFactory::new(
                 RethReceiptBuilder::default(),
                 chain_spec,
@@ -125,6 +128,12 @@ impl<ChainSpec, EvmFactory> EthEvmConfig<ChainSpec, EvmFactory> {
     pub const fn chain_spec(&self) -> &Arc<ChainSpec> {
         self.executor_factory.spec()
     }
+
+    /// Uses the provided sender recovery cache.
+    pub fn with_sender_recovery_cache(mut self, cache: SenderRecoveryCache) -> Self {
+        self.sender_recovery_cache = cache;
+        self
+    }
 }
 
 impl<ChainSpec, EvmF> ConfigureEvm for EthEvmConfig<ChainSpec, EvmF>
@@ -345,10 +354,11 @@ where
         payload: &ExecutionData,
     ) -> Result<impl ExecutableTxIterator<Self>, Self::Error> {
         let txs = payload.payload.transactions().clone();
-        let convert = |tx: Bytes| {
+        let sender_recovery_cache = self.sender_recovery_cache.clone();
+        let convert = move |tx: Bytes| {
             let tx =
                 TxTy::<Self::Primitives>::decode_2718_exact(tx.as_ref()).map_err(AnyError::new)?;
-            let signer = tx.try_recover().map_err(AnyError::new)?;
+            let signer = sender_recovery_cache.recover(&tx).map_err(AnyError::new)?;
             Ok::<_, AnyError>(tx.with_signer(signer))
         };
 

crates/ethereum/evm/src/lib.rs

@@ -608,6 +608,7 @@ where
         let dump_dir = jit.debug.then(|| ctx.config().datadir().data_dir().join("jit"));
 
         let (evm_config, revmc_metrics) = build_evm_config(ctx.chain_spec(), jit, dump_dir)?;
+        let evm_config = evm_config.with_sender_recovery_cache(ctx.sender_recovery_cache().clone());
 
         #[cfg(not(feature = "jit"))]
         let _ = revmc_metrics;

crates/ethereum/node/src/node.rs

@@ -34,6 +34,7 @@ derive_more.workspace = true
 futures-util.workspace = true
 metrics = { workspace = true, optional = true }
 rayon = { workspace = true, optional = true }
+fixed-cache.workspace = true
 
 [dev-dependencies]
 reth-ethereum-primitives.workspace = true
@@ -57,6 +58,7 @@ std = [
     "reth-trie-common/std",
     "reth-ethereum-primitives/std",
     "alloy-eip7928/std",
+    "fixed-cache/std",
 ]
 metrics = ["std", "dep:metrics", "dep:reth-metrics"]
 test-utils = [

crates/evm/evm/Cargo.toml

@@ -44,6 +44,8 @@ pub use aliases::*;
 mod engine;
 #[cfg(feature = "std")]
 pub use engine::{ConfigureEngineEvm, ConvertTx, ExecutableTxIterator, ExecutableTxTuple};
+mod sender_recovery;
+pub use sender_recovery::SenderRecoveryCache;
 
 #[cfg(feature = "metrics")]
 pub mod metrics;

crates/evm/evm/src/lib.rs

@@ -0,0 +1,84 @@
+use alloc::sync::Arc;
+use alloy_primitives::{map::FbBuildHasher, Address, B256};
+use reth_primitives_traits::{transaction::signed::RecoveryError, SignedTransaction};
+
+/// Number of entries retained in the default sender recovery cache.
+const SENDER_RECOVERY_CACHE_CAPACITY: usize = 1 << 17;
+
+/// Shared cache of recovered transaction senders.
+///
+/// Sender recovery is performed when a transaction enters the pool and again when the same
+/// transaction is received in an execution payload. Sharing this bounded, lock-free cache lets
+/// payload prewarming reuse the result produced by transaction-pool ingress.
+#[derive(Clone, Debug)]
+pub struct SenderRecoveryCache {
+    cache: Arc<fixed_cache::Cache<B256, Address, FbBuildHasher<32>>>,
+}
+
+impl SenderRecoveryCache {
+    /// Creates a sender recovery cache with the given capacity.
+    ///
+    /// The capacity must satisfy [`fixed_cache::Cache`] capacity requirements.
+    pub fn new(capacity: usize) -> Self {
+        Self { cache: Arc::new(fixed_cache::Cache::new(capacity, FbBuildHasher::<32>::default())) }
+    }
+
+    /// Returns a cached sender for the transaction hash.
+    #[inline]
+    pub fn get(&self, tx_hash: &B256) -> Option<Address> {
+        self.cache.get(tx_hash)
+    }
+
+    /// Returns the cached sender or recovers and caches it on a miss.
+    ///
+    /// Failed recoveries are not cached.
+    #[inline]
+    pub fn recover<T: SignedTransaction>(&self, transaction: &T) -> Result<Address, RecoveryError> {
+        self.cache.get_or_try_insert_with_ref(
+            transaction.tx_hash(),
+            |_| transaction.try_recover(),
+            |hash| *hash,
+        )
+    }
+}
+
+impl Default for SenderRecoveryCache {
+    fn default() -> Self {
+        Self::new(SENDER_RECOVERY_CACHE_CAPACITY)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use alloy_consensus::TxLegacy;
+    use alloy_primitives::{Signature, U256};
+    use reth_ethereum_primitives::{Transaction, TransactionSigned};
+
+    #[test]
+    fn recover_populates_cache() {
+        let transaction = TransactionSigned::new_unhashed(
+            Transaction::Legacy(TxLegacy::default()),
+            Signature::test_signature(),
+        );
+        let cache = SenderRecoveryCache::new(4);
+        let shared_cache = cache.clone();
+
+        let sender = cache.recover(&transaction).unwrap();
+
+        assert_eq!(shared_cache.get(transaction.tx_hash()), Some(sender));
+        assert_eq!(shared_cache.recover(&transaction).unwrap(), sender);
+    }
+
+    #[test]
+    fn failed_recovery_is_not_cached() {
+        let transaction = TransactionSigned::new_unhashed(
+            Transaction::Legacy(TxLegacy::default()),
+            Signature::new(U256::ZERO, U256::ZERO, false),
+        );
+        let cache = SenderRecoveryCache::new(4);
+
+        assert!(cache.recover(&transaction).is_err());
+        assert_eq!(cache.get(transaction.tx_hash()), None);
+    }
+}

crates/evm/evm/src/sender_recovery.rs

@@ -24,6 +24,7 @@ reth-discv4.workspace = true
 reth-discv5.workspace = true
 reth-dns-discovery.workspace = true
 reth-ethereum-forks.workspace = true
+reth-evm.workspace = true
 reth-eth-wire.workspace = true
 reth-eth-wire-types.workspace = true
 reth-ecies.workspace = true
@@ -139,4 +140,5 @@ test-utils = [
     "dep:reth-evm-ethereum",
     "reth-evm-ethereum?/test-utils",
     "reth-tasks/test-utils",
+    "reth-evm/test-utils",
 ]

crates/net/network/Cargo.toml

@@ -34,6 +34,15 @@ pub struct NetworkBuilder<Tx, Eth, N: NetworkPrimitives = EthNetworkPrimitives>
 // === impl NetworkBuilder ===
 
 impl<Tx, Eth, N: NetworkPrimitives> NetworkBuilder<Tx, Eth, N> {
+    /// Maps the transactions component.
+    pub fn map_transactions<F, NewTx>(self, f: F) -> NetworkBuilder<NewTx, Eth, N>
+    where
+        F: FnOnce(Tx) -> NewTx,
+    {
+        let Self { network, transactions, request_handler } = self;
+        NetworkBuilder { network, transactions: f(transactions), request_handler }
+    }
+
     /// Consumes the type and returns all fields.
     pub fn split(self) -> (NetworkManager<N>, Tx, Eth) {
         let Self { network, transactions, request_handler } = self;

crates/net/network/src/builder.rs

@@ -55,6 +55,7 @@ use reth_eth_wire::{
     PooledTransactions, RequestTxHashes, Transactions, ValidAnnouncementData,
 };
 use reth_ethereum_primitives::TxType;
+use reth_evm::SenderRecoveryCache;
 use reth_metrics::common::mpsc::MemoryBoundedReceiver;
 use reth_network_api::{
     events::{PeerEvent, SessionInfo},
@@ -285,6 +286,8 @@ impl<N: NetworkPrimitives> TransactionsHandle<N> {
 pub struct TransactionsManager<Pool, N: NetworkPrimitives = EthNetworkPrimitives> {
     /// Access to the transaction pool.
     pool: Pool,
+    /// Cache of recovered transaction senders shared with payload prewarming.
+    sender_recovery_cache: SenderRecoveryCache,
     /// Network access.
     network: NetworkHandle<N>,
     /// Subscriptions to all network related events.
@@ -400,6 +403,7 @@ impl<Pool: TransactionPool, N: NetworkPrimitives> TransactionsManager<Pool, N> {
 
         Self {
             pool,
+            sender_recovery_cache: SenderRecoveryCache::default(),
             network,
             network_events,
             transaction_fetcher,
@@ -424,6 +428,12 @@ impl<Pool: TransactionPool, N: NetworkPrimitives> TransactionsManager<Pool, N> {
         TransactionsHandle { manager_tx: self.command_tx.clone() }
     }
 
+    /// Uses the provided sender recovery cache.
+    pub fn with_sender_recovery_cache(mut self, cache: SenderRecoveryCache) -> Self {
+        self.sender_recovery_cache = cache;
+        self
+    }
+
     /// Returns `true` if [`TransactionsManager`] has capacity to request pending hashes. Returns
     /// `false` if [`TransactionsManager`] is operating close to full capacity.
     fn has_capacity_for_fetching_pending_hashes(&self) -> bool {
@@ -1454,18 +1464,19 @@ where
 
         let txs_len = transactions.len();
 
-        let recover = |tx| match Pool::Transaction::try_recover(tx) {
-            Ok(tx) => Some(tx),
-            Err(badtx) => {
-                trace!(target: "net::tx",
-                    peer_id=format!("{peer_id:#}"),
-                    hash=%badtx.tx_hash(),
-                    client_version=%client_version,
-                    "failed ecrecovery for transaction"
-                );
-                None
-            }
-        };
+        let recover =
+            |tx| match Pool::Transaction::try_recover_with_cache(tx, &self.sender_recovery_cache) {
+                Ok(tx) => Some(tx),
+                Err(badtx) => {
+                    trace!(target: "net::tx",
+                        peer_id=format!("{peer_id:#}"),
+                        hash=%badtx.tx_hash(),
+                        client_version=%client_version,
+                        "failed ecrecovery for transaction"
+                    );
+                    None
+                }
+            };
 
         let new_txs = transactions.into_par_iter().filter_map(recover).collect::<Vec<_>>();