test(net/tx): add transaction fetcher benchmark harness

crates/net/network/Cargo.toml

@@ -95,6 +95,7 @@ reth-transaction-pool = { workspace = true, features = ["test-utils"] }
 alloy-genesis.workspace = true
 
 # misc
+criterion.workspace = true
 url.workspace = true
 secp256k1 = { workspace = true, features = ["rand"] }
 
@@ -139,3 +140,8 @@ test-utils = [
     "reth-evm-ethereum?/test-utils",
     "reth-tasks/test-utils",
 ]
+
+[[bench]]
+name = "transaction_fetcher"
+required-features = ["test-utils"]
+harness = false

crates/net/network/benches/transaction_fetcher.rs

@@ -0,0 +1,97 @@
+#![allow(missing_docs)]
+
+use alloy_primitives::B256;
+use criterion::{criterion_group, criterion_main, BatchSize, Criterion, Throughput};
+use reth_eth_wire::{NewPooledTransactionHashes, NewPooledTransactionHashes68};
+use reth_network::transactions::test_harness::{
+    TransactionFetchHarness, TransactionFetchRequestStats,
+};
+use reth_network_peers::PeerId;
+use tokio::runtime::Runtime;
+
+// The disjoint workload intentionally matches the manager's default 4,096 pending-import
+// capacity, so both implementations complete the same work in one poll.
+const PEER_COUNT: usize = 64;
+const HASHES_PER_PEER: usize = 64;
+const ANNOUNCEMENT_ENTRY_COUNT: usize = PEER_COUNT * HASHES_PER_PEER;
+// Keep request packing controlled by hash count rather than the 2 MiB response-size soft limit.
+const TRANSACTION_ENCODED_SIZE: usize = 512;
+const EIP1559_TRANSACTION_TYPE: u8 = 2;
+
+struct BenchCase {
+    harness: TransactionFetchHarness,
+    announcements: Vec<(PeerId, NewPooledTransactionHashes)>,
+}
+
+impl BenchCase {
+    async fn new(overlapping: bool) -> Self {
+        let peers = (0..PEER_COUNT).map(peer_id).collect::<Vec<_>>();
+        let announcements = peers
+            .iter()
+            .enumerate()
+            .map(|(peer_index, peer_id)| {
+                let start = if overlapping { 0 } else { peer_index * HASHES_PER_PEER };
+                let hashes = (start..start + HASHES_PER_PEER).map(tx_hash).collect::<Vec<_>>();
+                (
+                    *peer_id,
+                    NewPooledTransactionHashes::Eth68(NewPooledTransactionHashes68 {
+                        types: vec![EIP1559_TRANSACTION_TYPE; hashes.len()],
+                        sizes: vec![TRANSACTION_ENCODED_SIZE; hashes.len()],
+                        hashes,
+                    }),
+                )
+            })
+            .collect();
+        let harness = TransactionFetchHarness::new(peers).await;
+        Self { harness, announcements }
+    }
+
+    async fn run(&mut self) {
+        for (peer_id, announcement) in self.announcements.drain(..) {
+            self.harness.announce(peer_id, announcement);
+        }
+        self.harness.poll_once().await;
+    }
+}
+
+const fn peer_id(index: usize) -> PeerId {
+    PeerId::new([(index + 1) as u8; 64])
+}
+
+fn tx_hash(index: usize) -> B256 {
+    let mut bytes = [0u8; 32];
+    bytes[24..].copy_from_slice(&(index as u64 + 1).to_be_bytes());
+    B256::from(bytes)
+}
+
+fn bench_transaction_fetcher(c: &mut Criterion) {
+    let runtime = Runtime::new().unwrap();
+    let mut group = c.benchmark_group("transaction_fetcher");
+    group.throughput(Throughput::Elements(ANNOUNCEMENT_ENTRY_COUNT as u64));
+
+    for (name, overlapping) in [("disjoint_announcements", false), ("overlapping_gossip", true)] {
+        let unique_hash_count =
+            if overlapping { HASHES_PER_PEER } else { ANNOUNCEMENT_ENTRY_COUNT };
+        let expected = TransactionFetchRequestStats {
+            request_count: if overlapping { 1 } else { PEER_COUNT },
+            requested_hash_count: unique_hash_count,
+            unique_requested_hash_count: unique_hash_count,
+        };
+        let mut validation_case = runtime.block_on(BenchCase::new(overlapping));
+        runtime.block_on(validation_case.run());
+        assert_eq!(validation_case.harness.drain_request_stats(), expected);
+
+        group.bench_function(name, |b| {
+            b.iter_batched_ref(
+                || runtime.block_on(BenchCase::new(overlapping)),
+                |case| runtime.block_on(case.run()),
+                BatchSize::LargeInput,
+            )
+        });
+    }
+
+    group.finish();
+}
+
+criterion_group!(benches, bench_transaction_fetcher);
+criterion_main!(benches);

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

@@ -10,7 +10,7 @@ use crate::{
             tx_manager::DEFAULT_MAX_COUNT_TRANSACTIONS_SEEN_BY_PEER,
         },
         fetcher::{TransactionFetcher, TxFetchMetadata},
-        PeerMetadata, TransactionsManager,
+        PeerMetadata, TransactionsManager, TransactionsManagerConfig,
     },
     NetworkConfigBuilder, NetworkManager,
 };
@@ -29,6 +29,13 @@ use tracing::trace;
 
 /// A new tx manager for testing.
 pub async fn new_tx_manager(
+) -> (TransactionsManager<TestPool, EthNetworkPrimitives>, NetworkManager<EthNetworkPrimitives>) {
+    new_tx_manager_with_config(TransactionsManagerConfig::default()).await
+}
+
+/// A new transaction manager for testing with the provided configuration.
+pub(crate) async fn new_tx_manager_with_config(
+    transactions_manager_config: TransactionsManagerConfig,
 ) -> (TransactionsManager<TestPool, EthNetworkPrimitives>, NetworkManager<EthNetworkPrimitives>) {
     let secret_key = SecretKey::new(&mut rand_08::thread_rng());
     let client = NoopProvider::default();
@@ -41,7 +48,6 @@ pub async fn new_tx_manager(
 
     let pool = testing_pool();
 
-    let transactions_manager_config = config.transactions_manager_config.clone();
     let (_network_handle, network, transactions, _) = NetworkManager::new(config)
         .await
         .unwrap()

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

@@ -12,6 +12,9 @@ pub mod constants;
 pub mod fetcher;
 /// Defines the traits for transaction-related policies.
 pub mod policy;
+/// Shared transaction-fetching test harness.
+#[cfg(any(test, feature = "test-utils"))]
+pub mod test_harness;
 
 pub use self::constants::{
     tx_fetcher::DEFAULT_SOFT_LIMIT_BYTE_SIZE_POOLED_TRANSACTIONS_RESP_ON_PACK_GET_POOLED_TRANSACTIONS_REQ,

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

@@ -0,0 +1,199 @@
+//! Shared test harness for transaction-fetching implementations.
+
+use super::{
+    NetworkTransactionEvent, TransactionPropagationMode, TransactionsManager,
+    TransactionsManagerConfig,
+};
+use crate::{
+    test_utils::transactions::{new_mock_session, new_tx_manager_with_config},
+    NetworkManager,
+};
+use alloy_primitives::map::{B256Set, HashMap};
+use reth_eth_wire::{EthNetworkPrimitives, EthVersion, NewPooledTransactionHashes};
+use reth_network_api::PeerRequest;
+use reth_network_p2p::sync::{NetworkSyncUpdater, SyncState};
+use reth_network_peers::PeerId;
+use reth_transaction_pool::test_utils::TestPool;
+use std::{
+    future::{poll_fn, Future},
+    pin::Pin,
+    task::Poll,
+};
+use tokio::sync::mpsc;
+
+#[cfg(test)]
+const FIRST_PEER: PeerId = PeerId::new([1; 64]);
+#[cfg(test)]
+const SECOND_PEER: PeerId = PeerId::new([2; 64]);
+#[cfg(test)]
+const EIP1559_TRANSACTION_TYPE: u8 = 2;
+// Both fixed EIP-1559 fixtures below are 116 bytes.
+#[cfg(test)]
+const TEST_TRANSACTION_ENCODED_SIZE: usize = 116;
+
+/// A manager-level transaction-fetching harness backed by deterministic mock peer sessions.
+///
+/// Keeping the harness at the manager boundary lets the same workload exercise either fetcher
+/// implementation without exposing their internal state.
+#[derive(Debug)]
+pub struct TransactionFetchHarness {
+    manager: TransactionsManager<TestPool, EthNetworkPrimitives>,
+    pool: TestPool,
+    _network: NetworkManager<EthNetworkPrimitives>,
+    requests: HashMap<PeerId, mpsc::Receiver<PeerRequest>>,
+}
+
+impl TransactionFetchHarness {
+    /// Creates a harness with one ETH/68 mock session for every peer.
+    pub async fn new(peers: impl IntoIterator<Item = PeerId>) -> Self {
+        let config = TransactionsManagerConfig {
+            propagation_mode: TransactionPropagationMode::Max(0),
+            ..Default::default()
+        };
+        let (mut manager, network) = new_tx_manager_with_config(config).await;
+        manager.network.update_sync_state(SyncState::Idle);
+
+        let mut requests = HashMap::default();
+        for peer_id in peers {
+            let (peer, receiver) = new_mock_session(peer_id, EthVersion::Eth68);
+            manager.peers.insert(peer_id, peer);
+            requests.insert(peer_id, receiver);
+        }
+
+        let pool = manager.pool.clone();
+        Self { manager, pool, _network: network, requests }
+    }
+
+    /// Delivers a pooled-transaction-hash announcement to the manager.
+    pub fn announce(&mut self, peer_id: PeerId, msg: NewPooledTransactionHashes) {
+        self.manager.on_network_tx_event(
+            NetworkTransactionEvent::IncomingPooledTransactionHashes { peer_id, msg },
+        );
+    }
+
+    /// Polls the manager once, driving request scheduling and completed responses.
+    pub async fn poll_once(&mut self) {
+        poll_fn(|cx| {
+            let _ = Pin::new(&mut self.manager).poll(cx);
+            Poll::Ready(())
+        })
+        .await;
+    }
+
+    /// Takes the next request sent to `peer_id`.
+    pub fn take_request(&mut self, peer_id: PeerId) -> PeerRequest {
+        self.requests
+            .get_mut(&peer_id)
+            .expect("peer is registered")
+            .try_recv()
+            .expect("manager sent a request to the mock peer")
+    }
+
+    /// Drains all immediately available pooled-transaction requests.
+    pub fn drain_request_stats(&mut self) -> TransactionFetchRequestStats {
+        let mut request_count = 0;
+        let mut requested_hash_count = 0;
+        let mut unique_hashes = B256Set::default();
+
+        for receiver in self.requests.values_mut() {
+            while let Ok(request) = receiver.try_recv() {
+                if let PeerRequest::GetPooledTransactions { request, .. } = request {
+                    request_count += 1;
+                    requested_hash_count += request.0.len();
+                    unique_hashes.extend(request.0);
+                }
+            }
+        }
+
+        TransactionFetchRequestStats {
+            request_count,
+            requested_hash_count,
+            unique_requested_hash_count: unique_hashes.len(),
+        }
+    }
+
+    /// Returns the test transaction pool.
+    pub const fn pool(&self) -> &TestPool {
+        &self.pool
+    }
+}
+
+/// Observable request totals for a transaction-fetching workload.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct TransactionFetchRequestStats {
+    /// Number of `GetPooledTransactions` requests emitted.
+    pub request_count: usize,
+    /// Total hashes across all emitted requests.
+    pub requested_hash_count: usize,
+    /// Number of distinct hashes across all emitted requests.
+    pub unique_requested_hash_count: usize,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::transactions::PooledTransactions;
+    use alloy_primitives::{hex, TxHash};
+    use alloy_rlp::Decodable;
+    use reth_eth_wire::NewPooledTransactionHashes68;
+    use reth_ethereum_primitives::{PooledTransactionVariant, TransactionSigned};
+    use reth_network_p2p::error::RequestError;
+    use reth_transaction_pool::TransactionPool;
+
+    fn announcement(hashes: &[TxHash]) -> NewPooledTransactionHashes {
+        NewPooledTransactionHashes::Eth68(NewPooledTransactionHashes68 {
+            types: vec![EIP1559_TRANSACTION_TYPE; hashes.len()],
+            sizes: vec![TEST_TRANSACTION_ENCODED_SIZE; hashes.len()],
+            hashes: hashes.to_vec(),
+        })
+    }
+
+    fn transactions() -> Vec<PooledTransactionVariant> {
+        let first = hex!(
+            "02f871018302a90f808504890aef60826b6c94ddf4c5025d1a5742cf12f74eec246d4432c295e487e09c3bbcc12b2b80c080a0f21a4eacd0bf8fea9c5105c543be5a1d8c796516875710fafafdf16d16d8ee23a001280915021bb446d1973501a67f93d2b38894a514b976e7b46dc2fe54598daa"
+        );
+        let second = hex!(
+            "02f871018302a90f808504890aef60826b6c94ddf4c5025d1a5742cf12f74eec246d4432c295e487e09c3bbcc12b2b80c080a0f21a4eacd0bf8fea9c5105c543be5a1d8c796516875710fafafdf16d16d8ee23a001280915021bb446d1973501a67f93d2b38894a514b976e7b46dc2fe54598d76"
+        );
+        [first.as_slice(), second.as_slice()]
+            .into_iter()
+            .map(|raw| TransactionSigned::decode(&mut &raw[..]).unwrap().try_into().unwrap())
+            .collect()
+    }
+
+    #[tokio::test]
+    async fn retries_failed_request_from_alternate_peer() {
+        let transactions = transactions();
+        let hashes = transactions.iter().map(|tx| *tx.tx_hash()).collect::<Vec<_>>();
+        let expected_hashes = hashes.iter().copied().collect::<B256Set>();
+        let mut harness = TransactionFetchHarness::new([FIRST_PEER, SECOND_PEER]).await;
+
+        harness.announce(FIRST_PEER, announcement(&hashes));
+        harness.poll_once().await;
+
+        let PeerRequest::GetPooledTransactions { request, response } =
+            harness.take_request(FIRST_PEER)
+        else {
+            panic!("unexpected peer request")
+        };
+        assert_eq!(request.0.iter().copied().collect::<B256Set>(), expected_hashes);
+
+        response.send(Err(RequestError::Timeout)).unwrap();
+        harness.poll_once().await;
+
+        harness.announce(SECOND_PEER, announcement(&hashes));
+        harness.poll_once().await;
+
+        let PeerRequest::GetPooledTransactions { request, response } =
+            harness.take_request(SECOND_PEER)
+        else {
+            panic!("unexpected peer request")
+        };
+        assert_eq!(request.0.iter().copied().collect::<B256Set>(), expected_hashes);
+
+        response.send(Ok(PooledTransactions(transactions))).unwrap();
+        harness.poll_once().await;
+
+        assert_eq!(harness.pool().get_all(hashes).len(), expected_hashes.len());
+    }
+}