From b7b3e59483d9216c08b83a74f6fa43a6799c8a4e Mon Sep 17 00:00:00 2001 From: Gustavo Figueiredo Date: Mon, 20 Jul 2026 15:53:43 +0200 Subject: [PATCH] test(net/tx): add transaction fetcher benchmark harness --- Cargo.lock | 1 + crates/net/network/Cargo.toml | 6 + .../network/benches/transaction_fetcher.rs | 97 +++++++++ .../network/src/test_utils/transactions.rs | 10 +- crates/net/network/src/transactions/mod.rs | 3 + .../network/src/transactions/test_harness.rs | 199 ++++++++++++++++++ 6 files changed, 314 insertions(+), 2 deletions(-) create mode 100644 crates/net/network/benches/transaction_fetcher.rs create mode 100644 crates/net/network/src/transactions/test_harness.rs diff --git a/Cargo.lock b/Cargo.lock index d5c880ba9bd..df9195cba29 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9051,6 +9051,7 @@ dependencies = [ "alloy-rlp", "aquamarine", "auto_impl", + "codspeed-criterion-compat", "derive_more", "discv5", "enr", diff --git a/crates/net/network/Cargo.toml b/crates/net/network/Cargo.toml index 298d93abced..3ec589d72c9 100644 --- a/crates/net/network/Cargo.toml +++ b/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 diff --git a/crates/net/network/benches/transaction_fetcher.rs b/crates/net/network/benches/transaction_fetcher.rs new file mode 100644 index 00000000000..4a3e359f0ba --- /dev/null +++ b/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::>(); + 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::>(); + ( + *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); diff --git a/crates/net/network/src/test_utils/transactions.rs b/crates/net/network/src/test_utils/transactions.rs index bb51cd3c56a..8c9a9acc8f6 100644 --- a/crates/net/network/src/test_utils/transactions.rs +++ b/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, NetworkManager) { + 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, NetworkManager) { 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() diff --git a/crates/net/network/src/transactions/mod.rs b/crates/net/network/src/transactions/mod.rs index be28944ccaf..18437aa1320 100644 --- a/crates/net/network/src/transactions/mod.rs +++ b/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, diff --git a/crates/net/network/src/transactions/test_harness.rs b/crates/net/network/src/transactions/test_harness.rs new file mode 100644 index 00000000000..597517c5324 --- /dev/null +++ b/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, + pool: TestPool, + _network: NetworkManager, + requests: HashMap>, +} + +impl TransactionFetchHarness { + /// Creates a harness with one ETH/68 mock session for every peer. + pub async fn new(peers: impl IntoIterator) -> 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 { + 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::>(); + let expected_hashes = hashes.iter().copied().collect::(); + 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::(), 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::(), expected_hashes); + + response.send(Ok(PooledTransactions(transactions))).unwrap(); + harness.poll_once().await; + + assert_eq!(harness.pool().get_all(hashes).len(), expected_hashes.len()); + } +}