From 9a695d1fae20b3f0e6031067f1695c35ae1906b8 Mon Sep 17 00:00:00 2001 From: Brian Picciano <933154+mediocregopher@users.noreply.github.com> Date: Mon, 7 Sep 2026 12:42:03 +0000 Subject: [PATCH 1/3] fix(bench): dispatch transactions round-robin across RPC endpoints Co-authored-by: Derek Cofausper <256792747+decofe@users.noreply.github.com> --- README.md | 2 ++ crates/bench-cli/src/main.rs | 2 +- crates/bench-core/src/sender.rs | 52 +++++++++++++++++++++++++++++---- 3 files changed, 49 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index 4468b69..7af7ea1 100644 --- a/README.md +++ b/README.md @@ -251,6 +251,8 @@ txgen-ethereum extract-big-blocks \ Send pre-generated transactions from NDJSON file or stdin. +When multiple `--rpc-url` endpoints are supplied, transactions are dispatched to them in round-robin order. + After sending completes, queries the node for per-block statistics (transaction count, gas used) and includes them in the report. ```bash diff --git a/crates/bench-cli/src/main.rs b/crates/bench-cli/src/main.rs index ee94560..b491d53 100644 --- a/crates/bench-cli/src/main.rs +++ b/crates/bench-cli/src/main.rs @@ -28,7 +28,7 @@ pub struct SendArgs { #[arg(short, long)] pub input: Option, - /// RPC endpoint URLs (comma-separated or repeated) + /// RPC endpoint URLs, used round-robin (comma-separated or repeated) #[arg(long = "rpc-url", value_delimiter = ',', default_values_t = vec!["http://localhost:8545".to_string()])] pub rpc_urls: Vec, diff --git a/crates/bench-core/src/sender.rs b/crates/bench-core/src/sender.rs index 1c5efc1..9d998bd 100644 --- a/crates/bench-core/src/sender.rs +++ b/crates/bench-core/src/sender.rs @@ -14,7 +14,6 @@ use alloy_primitives::{keccak256, Address, Bytes, TxHash, U256}; use alloy_provider::{DynProvider, Provider}; use alloy_transport::RpcError; use eyre::{Context, Result}; -use rand::seq::IndexedRandom; use reqwest::header::HeaderMap; use std::{ collections::{HashSet, VecDeque}, @@ -689,6 +688,8 @@ impl PendingTx { /// Transaction sender. pub struct Sender { endpoints: Vec, + /// Next RPC endpoint in round-robin dispatch order. + next_endpoint: usize, request_auth: Option>, metrics: Arc, semaphore: Arc, @@ -751,6 +752,7 @@ impl Sender { Self { endpoints, + next_endpoint: 0, request_auth, metrics, semaphore, @@ -935,11 +937,7 @@ impl Sender { // transactions observe a freshly reloaded sender map. Do this while // the transaction is still queued so any error is propagated and no // HTTP request is made. - let endpoint = self - .endpoints - .choose(&mut rand::rng()) - .expect("sender has at least one endpoint") - .clone(); + let endpoint = self.endpoints[self.next_endpoint].clone(); let pending = self.pending.get(index).expect("pending index exists"); let id = pending.id.as_deref().unwrap_or("").to_string(); let submission_headers = match self @@ -957,6 +955,7 @@ impl Sender { let pending = self.pending.remove(index).expect("pending index exists"); self.activate_keys(&pending); + self.next_endpoint = (self.next_endpoint + 1) % self.endpoints.len(); self.dispatch(pending, endpoint, submission_headers, permit); } @@ -1345,6 +1344,47 @@ mod tests { ProviderBuilder::new_with_network::().connect_mocked_client(asserter).erased() } + #[tokio::test] + async fn sender_round_robins_across_all_endpoints() { + for endpoint_count in [1, 10] { + let raw = Bytes::from_static(&[0x02, 0xf8, 0x70]); + let tx_hash = keccak256(&raw); + let endpoints = (0..endpoint_count) + .map(|index| { + let asserter = Asserter::new(); + for _ in 0..25 { + asserter.push_success(&tx_hash); + } + RpcEndpoint::new(format!("rpc-{index}"), mocked_provider(asserter)) + }) + .collect(); + let auth = Arc::new(RecordingAuth::default()); + let mut sender = Sender::new_with_request_auth( + endpoints, + SenderConfig { rate_limit: 0, max_concurrent: 3 }, + MetricsCollector::new(RunClock::new()), + Some(auth.clone()), + ); + for _ in 0..25 { + sender + .send(GeneratedTx { + phase: TxPhase::Workload, + id: None, + sender: Some(Address::repeat_byte(0x11)), + raw: raw.clone(), + submission_keys: Vec::new(), + inclusion_keys: Vec::new(), + }) + .await + .unwrap(); + } + sender.flush().await.unwrap(); + let expected = + (0..25).map(|index| format!("rpc-{}", index % endpoint_count)).collect::>(); + assert_eq!(*auth.endpoints.lock().unwrap(), expected); + } + } + fn receipt_json( transaction_hash: TxHash, effective_gas_price: Option<&str>, From f3972a8e4ee4e8b626469ab915c3d9b8a64d2e74 Mon Sep 17 00:00:00 2001 From: Brian Picciano <933154+mediocregopher@users.noreply.github.com> Date: Mon, 7 Sep 2026 12:42:47 +0000 Subject: [PATCH 2/3] test(bench): use scheduling keys in round-robin coverage Co-authored-by: Derek Cofausper <256792747+decofe@users.noreply.github.com> --- crates/bench-core/src/sender.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/bench-core/src/sender.rs b/crates/bench-core/src/sender.rs index 9d998bd..d9e9dff 100644 --- a/crates/bench-core/src/sender.rs +++ b/crates/bench-core/src/sender.rs @@ -1365,14 +1365,14 @@ mod tests { MetricsCollector::new(RunClock::new()), Some(auth.clone()), ); - for _ in 0..25 { + for index in 0..25 { sender .send(GeneratedTx { phase: TxPhase::Workload, id: None, sender: Some(Address::repeat_byte(0x11)), raw: raw.clone(), - submission_keys: Vec::new(), + submission_keys: vec![SchedulingKey::from([index; 20])], inclusion_keys: Vec::new(), }) .await From 75d41d66a8fe7527c7e52c4b6ec141510ea71d0c Mon Sep 17 00:00:00 2001 From: Brian Picciano <933154+mediocregopher@users.noreply.github.com> Date: Mon, 7 Sep 2026 13:24:26 +0000 Subject: [PATCH 3/3] fix(bench): limit block report concurrency over RPC tunnels Co-authored-by: Derek Cofausper <256792747+decofe@users.noreply.github.com> --- crates/bench-core/src/metrics.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/crates/bench-core/src/metrics.rs b/crates/bench-core/src/metrics.rs index fc70224..93224e2 100644 --- a/crates/bench-core/src/metrics.rs +++ b/crates/bench-core/src/metrics.rs @@ -27,7 +27,10 @@ use tokio::{ task::JoinHandle, }; -const BLOCK_STATS_FETCH_CONCURRENCY: usize = 32; +// Reports may fetch large blocks through a single bandwidth-limited RPC tunnel. +// Keep this separate from transaction submission concurrency: these reads happen +// after the workload window and should not saturate the tunnel and time out. +const BLOCK_STATS_FETCH_CONCURRENCY: usize = 4; /// Metrics collected during a benchmark run. #[derive(Debug, Clone, Serialize, Deserialize)]