net: add configurable recovered tx cache in transactions manager

Author: MatusKysel

crates/net/network/src/transactions/config.rs

@@ -24,6 +24,9 @@ pub struct TransactionsManagerConfig {
     pub transaction_fetcher_config: TransactionFetcherConfig,
     /// Max number of seen transactions to store for each peer.
     pub max_transactions_seen_by_peer_history: u32,
+    /// Max entries stored in the recovered transaction cache.
+    #[cfg_attr(feature = "serde", serde(default = "default_recovered_tx_cache_size"))]
+    pub recovered_tx_cache_size: u32,
     /// How new pending transactions are propagated.
     #[cfg_attr(feature = "serde", serde(default))]
     pub propagation_mode: TransactionPropagationMode,
@@ -37,12 +40,18 @@ impl Default for TransactionsManagerConfig {
         Self {
             transaction_fetcher_config: TransactionFetcherConfig::default(),
             max_transactions_seen_by_peer_history: DEFAULT_MAX_COUNT_TRANSACTIONS_SEEN_BY_PEER,
+            recovered_tx_cache_size: super::DEFAULT_RECOVERED_TX_CACHE_SIZE,
             propagation_mode: TransactionPropagationMode::default(),
             ingress_policy: TransactionIngressPolicy::default(),
         }
     }
 }
 
+#[cfg(feature = "serde")]
+const fn default_recovered_tx_cache_size() -> u32 {
+    super::DEFAULT_RECOVERED_TX_CACHE_SIZE
+}
+
 /// Determines how new pending transactions are propagated to other peers in full.
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
 #[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]

crates/net/network/src/transactions/mod.rs

@@ -1,6 +1,6 @@
 //! Transactions management for the p2p network.
 
-use alloy_consensus::transaction::TxHashRef;
+use alloy_consensus::transaction::{Recovered, TxHashRef};
 
 /// Aggregation on configurable parameters for [`TransactionsManager`].
 pub mod config;
@@ -15,6 +15,8 @@ pub use self::constants::{
     tx_fetcher::DEFAULT_SOFT_LIMIT_BYTE_SIZE_POOLED_TRANSACTIONS_RESP_ON_PACK_GET_POOLED_TRANSACTIONS_REQ,
     SOFT_LIMIT_BYTE_SIZE_POOLED_TRANSACTIONS_RESPONSE,
 };
+/// Default maximum recovered transaction cache size.
+pub const DEFAULT_RECOVERED_TX_CACHE_SIZE: u32 = 32_768;
 use config::AnnouncementAcceptance;
 pub use config::{
     AnnouncementFilteringPolicy, TransactionFetcherConfig, TransactionIngressPolicy,
@@ -30,7 +32,7 @@ use crate::{
         DEFAULT_BUDGET_TRY_DRAIN_NETWORK_TRANSACTION_EVENTS,
         DEFAULT_BUDGET_TRY_DRAIN_PENDING_POOL_IMPORTS, DEFAULT_BUDGET_TRY_DRAIN_STREAM,
     },
-    cache::LruCache,
+    cache::{LruCache, LruMap},
     duration_metered_exec, metered_poll_nested_stream_with_budget,
     metrics::{
         AnnouncedTxTypesMetrics, TransactionsManagerMetrics, NETWORK_POOL_TRANSACTIONS_SCOPE,
@@ -311,6 +313,8 @@ pub struct TransactionsManager<Pool, N: NetworkPrimitives = EthNetworkPrimitives
     pending_pool_imports_info: PendingPoolImportsInfo,
     /// Bad imports.
     bad_imports: LruCache<TxHash>,
+    /// Cache recovered transactions to avoid repeated signature recovery for duplicates.
+    recovered_txs: LruMap<TxHash, Recovered<N::PooledTransaction>>,
     /// All the connected peers.
     peers: HashMap<PeerId, PeerMetadata<N>>,
     /// Send half for the command channel.
@@ -405,6 +409,7 @@ impl<Pool: TransactionPool, N: NetworkPrimitives> TransactionsManager<Pool, N> {
                 DEFAULT_MAX_COUNT_PENDING_POOL_IMPORTS,
             ),
             bad_imports: LruCache::new(DEFAULT_MAX_COUNT_BAD_IMPORTS),
+            recovered_txs: LruMap::new(transactions_manager_config.recovered_tx_cache_size.max(1)),
             peers: Default::default(),
             command_tx,
             command_rx: UnboundedReceiverStream::new(command_rx),
@@ -1372,18 +1377,41 @@ where
         //    reallocations
         let mut new_txs = Vec::with_capacity(transactions.len());
         for tx in transactions {
-            // recover transaction
-            let tx = match tx.try_into_recovered() {
-                Ok(tx) => tx,
-                Err(badtx) => {
-                    trace!(target: "net::tx",
-                        peer_id=format!("{peer_id:#}"),
-                        hash=%badtx.tx_hash(),
-                        client_version=%peer.client_version,
-                        "failed ecrecovery for transaction"
-                    );
-                    has_bad_transactions = true;
-                    continue
+            // recover transaction, prefer cached value if present when enabled
+            let tx_hash = *tx.tx_hash();
+            let tx = if self.config.recovered_tx_cache_size == 0 {
+                match tx.try_into_recovered() {
+                    Ok(tx) => tx,
+                    Err(badtx) => {
+                        trace!(target: "net::tx",
+                            peer_id=format!("{peer_id:#}"),
+                            hash=%badtx.tx_hash(),
+                            client_version=%peer.client_version,
+                            "failed ecrecovery for transaction"
+                        );
+                        has_bad_transactions = true;
+                        continue
+                    }
+                }
+            } else {
+                match self.recovered_txs.get(&tx_hash) {
+                    Some(cached) => cached.clone(),
+                    None => match tx.try_into_recovered() {
+                        Ok(tx) => {
+                            _ = self.recovered_txs.insert(tx_hash, tx.clone());
+                            tx
+                        }
+                        Err(badtx) => {
+                            trace!(target: "net::tx",
+                                peer_id=format!("{peer_id:#}"),
+                                hash=%badtx.tx_hash(),
+                                client_version=%peer.client_version,
+                                "failed ecrecovery for transaction"
+                            );
+                            has_bad_transactions = true;
+                            continue
+                        }
+                    },
                 }
             };
 
@@ -2128,6 +2156,7 @@ mod tests {
     use alloy_rlp::Decodable;
     use futures::FutureExt;
     use reth_chainspec::MIN_TRANSACTION_GAS;
+    use reth_eth_wire::{EthVersion, PooledTransactions};
     use reth_ethereum_primitives::{PooledTransactionVariant, Transaction, TransactionSigned};
     use reth_network_api::{NetworkInfo, PeerKind};
     use reth_network_p2p::{
@@ -2889,6 +2918,37 @@ mod tests {
         assert!(propagated.0.is_empty());
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_recovered_tx_cache_stores_recovered_once() {
+        reth_tracing::init_test_tracing();
+
+        // Build a tx manager and shrink the recovered cache to a tiny size.
+        let (mut tx_manager, _network) = new_tx_manager().await;
+        tx_manager.config.recovered_tx_cache_size = 1;
+        tx_manager.recovered_txs = LruMap::new(1);
+
+        // Register a peer so the import path can proceed.
+        let peer_id = PeerId::random();
+        let (peer_meta, _rx) = new_mock_session(peer_id, EthVersion::Eth68);
+        tx_manager.peers.insert(peer_id, peer_meta);
+
+        // A simple valid transaction (same as other tests).
+        let input = hex!(
+            "02f871018302a90f808504890aef60826b6c94ddf4c5025d1a5742cf12f74eec246d4432c295e487e09c3bbcc12b2b80c080a0f21a4eacd0bf8fea9c5105c543be5a1d8c796516875710fafafdf16d16d8ee23a001280915021bb446d1973501a67f93d2b38894a514b976e7b46dc2fe54598d76"
+        );
+        let signed_tx = TransactionSigned::decode(&mut &input[..]).unwrap();
+        let pooled: PooledTransactionVariant = signed_tx.try_into().unwrap();
+
+        // Feed the same transaction twice; the cache should only keep one entry.
+        tx_manager.import_transactions(
+            peer_id,
+            PooledTransactions(vec![pooled.clone(), pooled]),
+            TransactionSource::Broadcast,
+        );
+
+        assert_eq!(tx_manager.recovered_txs.len(), 1);
+    }
+
     #[tokio::test]
     async fn test_relaxed_filter_ignores_unknown_tx_types() {
         reth_tracing::init_test_tracing();

crates/node/core/src/args/network.rs

@@ -33,6 +33,7 @@ use reth_network::{
             },
         },
         TransactionFetcherConfig, TransactionPropagationMode, TransactionsManagerConfig,
+        DEFAULT_RECOVERED_TX_CACHE_SIZE,
         DEFAULT_SOFT_LIMIT_BYTE_SIZE_POOLED_TRANSACTIONS_RESP_ON_PACK_GET_POOLED_TRANSACTIONS_REQ,
         SOFT_LIMIT_BYTE_SIZE_POOLED_TRANSACTIONS_RESPONSE,
     },
@@ -256,6 +257,7 @@ impl NetworkArgs {
                 self.max_capacity_cache_txns_pending_fetch,
             ),
             max_transactions_seen_by_peer_history: self.max_seen_tx_history,
+            recovered_tx_cache_size: DEFAULT_RECOVERED_TX_CACHE_SIZE,
             propagation_mode: self.propagation_mode,
             ingress_policy: self.tx_ingress_policy,
         }