From 1a67c5a992230936c4badd626c7b36128661f38d Mon Sep 17 00:00:00 2001 From: keanji-x Date: Mon, 27 Apr 2026 22:32:19 +0800 Subject: [PATCH] chore(fmt): apply rustfmt + strip trailing whitespace across the tree `cargo fmt --all -- --check` was failing on main because: * `src/actors/monitor/monitor_actor.rs` and `src/actors/monitor/txn_tracker.rs` had trailing whitespace (rustfmt declines to format files containing trailing whitespace, surfacing it as `error[internal]: left behind trailing whitespace`). * Several files had drifted away from rustfmt's current output. This commit: 1. Strips trailing whitespace from every `.rs` under `src/` (`find src -name '*.rs' -exec sed -i 's/[[:space:]]*\$//' {} +`). 2. Runs `cargo fmt --all` so the tree matches rustfmt.toml. No logic changes. After this commit `cargo fmt --all -- --check` exits 0 and the Rust CI / Check job passes. This unblocks PRs that otherwise inherit the pre-existing CI failure on main. --- src/actors/consumer/actor.rs | 59 +++----- src/actors/consumer/dispatcher.rs | 6 +- src/actors/monitor/mempool_tracker.rs | 18 +-- src/actors/monitor/mod.rs | 3 - src/actors/monitor/monitor_actor.rs | 70 ++++----- src/actors/monitor/txn_tracker.rs | 119 ++++++--------- src/actors/producer/messages.rs | 13 +- src/actors/producer/producer_actor.rs | 68 +++------ src/config/bench_config.rs | 1 - src/config/contract_config.rs | 12 +- src/eth/eth_cli.rs | 74 +++------ src/eth/txn_builder.rs | 5 +- src/main.rs | 140 ++++++------------ .../addr_pool/managed_address_pool.rs | 13 +- .../addr_pool/weighted_address_pool.rs | 5 +- src/txn_plan/constructor/approve.rs | 11 +- src/txn_plan/constructor/erc20_transfer.rs | 12 +- src/txn_plan/constructor/faucet.rs | 4 +- .../constructor/swap_token_2_token.rs | 8 +- src/txn_plan/faucet_plan.rs | 120 +++++++-------- src/txn_plan/plan.rs | 17 +-- src/util/gen_account.rs | 56 ++----- 22 files changed, 281 insertions(+), 553 deletions(-) diff --git a/src/actors/consumer/actor.rs b/src/actors/consumer/actor.rs index 6fb8ac7..30d9bce 100644 --- a/src/actors/consumer/actor.rs +++ b/src/actors/consumer/actor.rs @@ -124,10 +124,7 @@ impl RateLimiter { /// Get current status pub fn get_status(&self) -> (u32, u32) { - ( - self.current_tokens.load(Ordering::Relaxed) as u32, - self.bucket_capacity, - ) + (self.current_tokens.load(Ordering::Relaxed) as u32, self.bucket_capacity) } } @@ -199,10 +196,7 @@ impl Consumer { transactions_sending: Arc, ) { let metadata = signed_txn.metadata; - debug!( - "Acquired permit, processing transaction: {:?}", - metadata.txn_id - ); + debug!("Acquired permit, processing transaction: {:?}", metadata.txn_id); transactions_sending.fetch_add(1, Ordering::Relaxed); let mut last_error: Option = None; @@ -215,10 +209,7 @@ impl Consumer { MAX_RETRIES, metadata.txn_id ); - match dispatcher - .send_tx(signed_txn.bytes.clone(), metadata.txn_id) - .await - { + match dispatcher.send_tx(signed_txn.bytes.clone(), metadata.txn_id).await { // Transaction sent successfully Ok((tx_hash, rpc_url)) => { tracing::debug!( @@ -320,22 +311,22 @@ impl Consumer { { // The RPC already told us "nonce too low" - trust this directly. // Try to get current nonce for better logging, but don't fail if we can't. - let (expect_nonce, actual_nonce, from_account) = - if let Ok(next_nonce) = dispatcher + let (expect_nonce, actual_nonce, from_account) = if let Ok(next_nonce) = + dispatcher .provider(&url) .await .unwrap() .get_pending_txn_count(metadata.from_account.as_ref().clone()) .await - { - (next_nonce, metadata.nonce, metadata.from_account.clone()) - } else { - // Failed to get nonce, but RPC already said "nonce too low" - // Use 0 as placeholder - the important thing is NOT to retry - warn!("Nonce too low but failed to get current nonce for {:?}, treating as resolved", metadata.txn_id); - (0, metadata.nonce, metadata.from_account.clone()) - }; - + { + (next_nonce, metadata.nonce, metadata.from_account.clone()) + } else { + // Failed to get nonce, but RPC already said "nonce too low" + // Use 0 as placeholder - the important thing is NOT to retry + warn!("Nonce too low but failed to get current nonce for {:?}, treating as resolved", metadata.txn_id); + (0, metadata.nonce, metadata.from_account.clone()) + }; + monitor_addr.do_send(UpdateSubmissionResult { metadata, result: Arc::new(SubmissionResult::NonceTooLow { @@ -348,7 +339,7 @@ impl Consumer { send_time: Instant::now(), signed_bytes: Arc::new(signed_txn.bytes.clone()), }); - + // After encountering Nonce error, should stop retrying and return regardless transactions_sending.fetch_sub(1, Ordering::Relaxed); return; @@ -394,13 +385,7 @@ impl Consumer { max_tps: Option, ) -> Consumer { let dispatcher = Arc::new(SimpleDispatcher::new(providers)); - Consumer::new( - dispatcher, - max_concurrent_senders, - monitor_addr, - max_pool_size, - max_tps, - ) + Consumer::new(dispatcher, max_concurrent_senders, monitor_addr, max_pool_size, max_tps) } /// Start transaction pool consumer @@ -420,7 +405,7 @@ impl Consumer { std::thread::spawn(move || { let rt = tokio::runtime::Runtime::new().unwrap(); rt.block_on(async move { - info!("Transaction pool consumer started with JoinSet and rate limiting (max_tps: {}).", + info!("Transaction pool consumer started with JoinSet and rate limiting (max_tps: {}).", if rate_limiter.max_tps == 0 { "unlimited".to_string() } else { rate_limiter.max_tps.to_string() }); let mut in_flight_tasks = tokio::task::JoinSet::new(); @@ -502,9 +487,7 @@ impl Actor for Consumer { fn started(&mut self, ctx: &mut Self::Context) { // Register self with Monitor - self.monitor_addr.do_send(RegisterConsumer { - addr: ctx.address(), - }); + self.monitor_addr.do_send(RegisterConsumer { addr: ctx.address() }); let rate_limiter = self.rate_limiter.clone(); let dispatcher = self.dispatcher.clone(); @@ -588,10 +571,8 @@ impl Handler for Consumer { debug!("Retrying transaction: {:?}", msg.metadata.txn_id); // Convert to SignedTxnWithMetadata and send through normal channel - let signed_txn = SignedTxnWithMetadata { - bytes: (*msg.signed_bytes).clone(), - metadata: msg.metadata, - }; + let signed_txn = + SignedTxnWithMetadata { bytes: (*msg.signed_bytes).clone(), metadata: msg.metadata }; let sender = self.pool_sender.clone(); let pool_size = self.stats.pool_size.clone(); diff --git a/src/actors/consumer/dispatcher.rs b/src/actors/consumer/dispatcher.rs index 8b78684..d7d8834 100644 --- a/src/actors/consumer/dispatcher.rs +++ b/src/actors/consumer/dispatcher.rs @@ -51,10 +51,8 @@ impl Dispatcher for SimpleDispatcher { ) -> std::result::Result<(TxHash, String), (anyhow::Error, String)> { let provider = self.select_provider(txn_id); let rpc_url = provider.rpc().as_ref().clone(); - let tx_hash = provider - .send_raw_tx(bytes) - .await - .map_err(|e| (e, provider.rpc().as_ref().clone()))?; + let tx_hash = + provider.send_raw_tx(bytes).await.map_err(|e| (e, provider.rpc().as_ref().clone()))?; Ok((tx_hash, rpc_url)) } diff --git a/src/actors/monitor/mempool_tracker.rs b/src/actors/monitor/mempool_tracker.rs index 13414f5..7ed9d94 100644 --- a/src/actors/monitor/mempool_tracker.rs +++ b/src/actors/monitor/mempool_tracker.rs @@ -7,8 +7,6 @@ use crate::{ eth::{EthHttpCli, MempoolStatus, TxPoolContent}, }; - - /// Action to take after analyzing mempool status #[derive(Debug)] pub enum MempoolAction { @@ -105,7 +103,9 @@ impl MempoolTracker { /// Identify accounts with nonce gaps from txpool_content /// Returns list of addresses that need correction - pub fn identify_problematic_accounts(content: &TxPoolContent) -> Vec { + pub fn identify_problematic_accounts( + content: &TxPoolContent, + ) -> Vec { let mut problematic_accounts = Vec::new(); for (address, nonces) in &content.queued { @@ -114,21 +114,17 @@ impl MempoolTracker { // We check if they also have pending transactions. // If they have NO pending transactions but HAVE queued, strictly implies a gap. let has_pending = content.pending.contains_key(address); - + // Check if there are any valid nonces in queued - if nonces.keys().any(|s| s.parse::().is_ok()) { + if nonces.keys().any(|s| s.parse::().is_ok()) { if !has_pending { - problematic_accounts.push(*address); + problematic_accounts.push(*address); } } } - tracing::info!( - "Identified {} accounts with likely nonce gaps", - problematic_accounts.len() - ); + tracing::info!("Identified {} accounts with likely nonce gaps", problematic_accounts.len()); problematic_accounts } } - diff --git a/src/actors/monitor/mod.rs b/src/actors/monitor/mod.rs index 05d8e37..62bcbbe 100644 --- a/src/actors/monitor/mod.rs +++ b/src/actors/monitor/mod.rs @@ -71,7 +71,6 @@ pub struct PlanCompleted { pub plan_id: PlanId, } - #[derive(Message)] #[rtype(result = "()")] pub struct PlanFailed { @@ -79,8 +78,6 @@ pub struct PlanFailed { pub reason: String, } - - /// Message to retry a timed-out transaction #[derive(Message, Clone)] #[rtype(result = "()")] diff --git a/src/actors/monitor/monitor_actor.rs b/src/actors/monitor/monitor_actor.rs index 42b41d8..b6e0090 100644 --- a/src/actors/monitor/monitor_actor.rs +++ b/src/actors/monitor/monitor_actor.rs @@ -12,11 +12,11 @@ use crate::actors::producer::Producer; use crate::eth::EthHttpCli; use crate::txn_plan::PlanId; -use super::txn_tracker::{BackpressureAction, PlanStatus, TxnTracker}; use super::mempool_tracker::MempoolAction; +use super::txn_tracker::{BackpressureAction, PlanStatus, TxnTracker}; use super::{ - PlanCompleted, PlanFailed, RegisterConsumer, RegisterPlan, RegisterProducer, - ReportProducerStats, RetryTxn, Tick, UpdateSubmissionResult, CorrectNonces, + CorrectNonces, PlanCompleted, PlanFailed, RegisterConsumer, RegisterPlan, RegisterProducer, + ReportProducerStats, RetryTxn, Tick, UpdateSubmissionResult, }; use crate::actors::{PauseProducer, ResumeProducer}; @@ -30,7 +30,7 @@ struct LogStats; pub struct Monitor { /// Registered Producer address producer_addr: Option>, - /// Registered Consumer address + /// Registered Consumer address consumer_addr: Option>, /// Transaction and plan tracker txn_tracker: TxnTracker, @@ -39,7 +39,6 @@ pub struct Monitor { } impl Monitor { - pub fn new_with_clients( clients: Vec>, max_pool_size: usize, @@ -50,12 +49,7 @@ impl Monitor { consumer_addr: None, txn_tracker: TxnTracker::new(clients.clone(), sampling_policy), mempool_tracker: MempoolTracker::new(max_pool_size), - clients: Arc::new( - clients - .into_iter() - .map(|client| (client.rpc(), client)) - .collect(), - ), + clients: Arc::new(clients.into_iter().map(|client| (client.rpc(), client)).collect()), } } @@ -102,7 +96,7 @@ impl Actor for Monitor { match act.mempool_tracker.process_pool_status(res, producer_addr) { Ok((pending, queued, action)) => { act.txn_tracker.update_mempool_stats(pending, queued); - + // Handle nonce correction if needed if matches!(action, MempoolAction::NeedsNonceCorrection) { let clients = act.clients.clone(); @@ -200,24 +194,27 @@ impl Handler for Monitor { // If the transaction failed submission, retry it endlessly to prevent nonce gaps // and premature plan completion. Do NOT tell TxnTracker about the failure yet. tracing::warn!( - "Transaction failed submission (ErrorWithRetry). Retrying via Consumer. plan_id={}, tx_hash={:?}", + "Transaction failed submission (ErrorWithRetry). Retrying via Consumer. plan_id={}, tx_hash={:?}", msg.metadata.plan_id, msg.metadata.txn_id ); - + if let Some(consumer) = &self.consumer_addr { consumer.do_send(RetryTxn { signed_bytes: msg.signed_bytes.clone(), metadata: msg.metadata.clone(), }); } else { - tracing::error!("Cannot retry transaction, no consumer address: {:?}", msg.metadata.txn_id); + tracing::error!( + "Cannot retry transaction, no consumer address: {:?}", + msg.metadata.txn_id + ); // Fallback to tracker if no consumer (will mark as failed) self.txn_tracker.handle_submission_result(&msg); } } _ => { - self.txn_tracker.handle_submission_result(&msg); + self.txn_tracker.handle_submission_result(&msg); } } } @@ -246,24 +243,20 @@ impl Handler for Monitor { let tasks = self.txn_tracker.perform_sampling_check(); let consumer_addr = self.consumer_addr.clone(); if !tasks.is_empty() { - ctx.spawn( - future::join_all(tasks) - .into_actor(self) - .map(move |results, act, _ctx| { - // Process results and get retry queue - let retry_queue = act.txn_tracker.handle_receipt_result(results); - - // 3. Send retries to consumer - if let Some(consumer) = &consumer_addr { - for retry_txn in retry_queue { - consumer.do_send(RetryTxn { - signed_bytes: retry_txn.signed_bytes, - metadata: retry_txn.metadata, - }); - } - } - }), - ); + ctx.spawn(future::join_all(tasks).into_actor(self).map(move |results, act, _ctx| { + // Process results and get retry queue + let retry_queue = act.txn_tracker.handle_receipt_result(results); + + // 3. Send retries to consumer + if let Some(consumer) = &consumer_addr { + for retry_txn in retry_queue { + consumer.do_send(RetryTxn { + signed_bytes: retry_txn.signed_bytes, + metadata: retry_txn.metadata, + }); + } + } + })); } // Check completion status of all plans @@ -301,8 +294,7 @@ impl Handler for Monitor { type Result = (); fn handle(&mut self, msg: ProduceTxns, _ctx: &mut Self::Context) { - self.txn_tracker - .handler_produce_txns(msg.plan_id, msg.count); + self.txn_tracker.handler_produce_txns(msg.plan_id, msg.count); } } @@ -310,8 +302,7 @@ impl Handler for Monitor { type Result = (); fn handle(&mut self, msg: PlanProduced, _ctx: &mut Self::Context) { - self.txn_tracker - .handle_plan_produced(msg.plan_id, msg.count); + self.txn_tracker.handle_plan_produced(msg.plan_id, msg.count); } } @@ -319,8 +310,7 @@ impl Handler for Monitor { type Result = (); fn handle(&mut self, msg: ReportProducerStats, _ctx: &mut Self::Context) { - self.txn_tracker - .update_producer_stats(msg.ready_accounts, msg.sending_txns); + self.txn_tracker.update_producer_stats(msg.ready_accounts, msg.sending_txns); } } diff --git a/src/actors/monitor/txn_tracker.rs b/src/actors/monitor/txn_tracker.rs index 7619fd7..9fc69dc 100644 --- a/src/actors/monitor/txn_tracker.rs +++ b/src/actors/monitor/txn_tracker.rs @@ -3,7 +3,6 @@ use std::collections::{BTreeSet, HashMap, HashSet, VecDeque}; use std::sync::Arc; use std::time::{Duration, Instant}; - use alloy::primitives::TxHash; use comfy_table::{presets::UTF8_FULL, Cell, Table}; use tracing::{debug, error, warn}; @@ -16,7 +15,6 @@ use crate::config::SamplingPolicy; use super::UpdateSubmissionResult; - const TXN_TIMEOUT: Duration = Duration::from_secs(600); // 10 minutes timeout const TPS_WINDOW: Duration = Duration::from_secs(17); @@ -91,7 +89,7 @@ struct PlanTracker { plan_failed_production: bool, plan_name: String, - + /// Set of transaction hashes that have been resolved to avoid double counting resolved_hashes: HashSet, } @@ -128,9 +126,7 @@ impl Ord for PendingTxInfo { /// 1. Primarily sorted by submission time (`submit_time`) in ascending order /// 2. If submission times are the same, sort by transaction hash (`tx_hash`) to ensure uniqueness fn cmp(&self, other: &Self) -> Ordering { - self.submit_time - .cmp(&other.submit_time) - .then_with(|| self.tx_hash.cmp(&other.tx_hash)) + self.submit_time.cmp(&other.submit_time).then_with(|| self.tx_hash.cmp(&other.tx_hash)) } } @@ -204,10 +200,7 @@ impl TxnTracker { let current = self.pending_txns.len(); if current >= MAX_PENDING_TXNS && !self.producer_paused_by_pending { self.producer_paused_by_pending = true; - warn!( - "Pending txns {} >= {}, pausing producer", - current, MAX_PENDING_TXNS - ); + warn!("Pending txns {} >= {}, pausing producer", current, MAX_PENDING_TXNS); BackpressureAction::Pause } else if current < BACKPRESSURE_RESUME_THRESHOLD && self.producer_paused_by_pending { self.producer_paused_by_pending = false; @@ -234,8 +227,11 @@ impl TxnTracker { tracker.plan_produced = true; // Force sync produced transactions count to catch up any lagging messages if tracker.produce_transactions != count { - warn!("PlanProduced sync: plan {} count adjusted from {} to source-of-truth {}", plan_id, tracker.produce_transactions, count); - tracker.produce_transactions = count; + warn!( + "PlanProduced sync: plan {} count adjusted from {} to source-of-truth {}", + plan_id, tracker.produce_transactions, count + ); + tracker.produce_transactions = count; } } } @@ -243,10 +239,7 @@ impl TxnTracker { /// Register new plan (or update existing one if retried) pub fn register_plan(&mut self, plan_id: PlanId, plan_name: String) { if let Some(tracker) = self.plan_trackers.get_mut(&plan_id) { - debug!( - "Plan already registered (likely retry): plan_id={}. Resetting flags.", - plan_id - ); + debug!("Plan already registered (likely retry): plan_id={}. Resetting flags.", plan_id); // Result of retry logic: we are producing more transactions for this plan. // Reset flags to keep plan open until the new attempt finishes. tracker.plan_produced = false; @@ -282,7 +275,12 @@ impl TxnTracker { pub fn handle_submission_result(&mut self, msg: &UpdateSubmissionResult) { let plan_id = &msg.metadata.plan_id; if !self.plan_trackers.contains_key(plan_id) { - warn!("Plan not found: plan_id={}, tx_hash={:?}, result={:?}", plan_id, msg.result, msg.result.as_ref()); + warn!( + "Plan not found: plan_id={}, tx_hash={:?}, result={:?}", + plan_id, + msg.result, + msg.result.as_ref() + ); return; } let plan_tracker = self.plan_trackers.get_mut(plan_id).unwrap(); @@ -306,12 +304,7 @@ impl TxnTracker { // Insert transaction into the global, time-sorted BTreeSet self.pending_txns.insert(pending_info); } - SubmissionResult::NonceTooLow { - tx_hash, - expect_nonce, - actual_nonce, - from_account, - } => { + SubmissionResult::NonceTooLow { tx_hash, expect_nonce, actual_nonce, from_account } => { let pending_info = PendingTxInfo { tx_hash: *tx_hash, metadata: msg.metadata.clone(), @@ -326,15 +319,15 @@ impl TxnTracker { ); } e => { - warn!( - "Transaction submission failed: plan_id={}, error={:?}", - plan_id, e - ); + warn!("Transaction submission failed: plan_id={}, error={:?}", plan_id, e); if let Some(tracker) = self.plan_trackers.get_mut(plan_id) { tracker.resolved_transactions += 1; tracker.failed_submissions += 1; self.total_failed_submissions += 1; - warn!("Incrementing failed_submissions for plan {}: resolved={}, failed={}", plan_id, tracker.resolved_transactions, tracker.failed_submissions); + warn!( + "Incrementing failed_submissions for plan {}: resolved={}, failed={}", + plan_id, tracker.resolved_transactions, tracker.failed_submissions + ); self.resolved_txn_timestamps.push_back(Instant::now()); self.total_resolved_transactions += 1; } @@ -366,10 +359,10 @@ impl TxnTracker { } if let PlanStatus::Completed = status { if let Some(completed_tracker) = self.plan_trackers.remove(plan_id) { - warn!("Removing completed plan {}: produced={}, resolved={}, consumed={}, failed_sub={}, failed_exec={}", - plan_id, - completed_tracker.produce_transactions, - completed_tracker.resolved_transactions, + warn!("Removing completed plan {}: produced={}, resolved={}, consumed={}, failed_sub={}, failed_exec={}", + plan_id, + completed_tracker.produce_transactions, + completed_tracker.resolved_transactions, completed_tracker.consumed_transactions, completed_tracker.failed_submissions, completed_tracker.failed_executions @@ -406,7 +399,8 @@ impl TxnTracker { // --- Core sampling logic --- // Filter out transactions that are already being checked - let candidates: Vec<_> = self.pending_txns + let candidates: Vec<_> = self + .pending_txns .iter() .filter(|info| !self.inflight_checks.contains(&info.tx_hash)) .cloned() @@ -446,14 +440,9 @@ impl TxnTracker { let task = async move { let result = client.get_transaction_receipt(task_info.tx_hash).await; - let account = client - .get_latest_txn_count(task_info.metadata.from_account.as_ref()) - .await; - tracing::debug!( - "checked tx_hash={:?} result={:?}", - task_info.tx_hash, - result - ); + let account = + client.get_latest_txn_count(task_info.metadata.from_account.as_ref()).await; + tracing::debug!("checked tx_hash={:?} result={:?}", task_info.tx_hash, result); (task_info, account, result) }; tasks.push(task); @@ -500,26 +489,18 @@ impl TxnTracker { } Err(e) => { // RPC query failed - warn!( - "Failed to get receipt for tx_hash={:?}: {}", - info.tx_hash, e - ); + warn!("Failed to get receipt for tx_hash={:?}: {}", info.tx_hash, e); failed_txns.push(info); } } } if !failed_txns.is_empty() { - debug!( - "Failed to get receipt for {} transactions", - failed_txns.len() - ); + debug!("Failed to get receipt for {} transactions", failed_txns.len()); } - let successful_txns_hash = successful_txns - .iter() - .map(|(info, _)| info.tx_hash) - .collect::>(); + let successful_txns_hash = + successful_txns.iter().map(|(info, _)| info.tx_hash).collect::>(); // 2. If there are successful transactions, calculate median time and clean up // Only use heuristic batch cleaning in Partial mode. In Full mode, we track every txn explicitly. @@ -581,7 +562,10 @@ impl TxnTracker { ); } } else { - debug!("Duplicate resolution skipped: plan_id={}, tx_hash={:?}", info.metadata.plan_id, info.tx_hash); + debug!( + "Duplicate resolution skipped: plan_id={}, tx_hash={:?}", + info.metadata.plan_id, info.tx_hash + ); } } } @@ -694,11 +678,7 @@ impl TxnTracker { let max = *self.latencies.iter().max().unwrap(); (avg, min, max) } else { - ( - Duration::from_secs(0), - Duration::from_secs(0), - Duration::from_secs(0), - ) + (Duration::from_secs(0), Duration::from_secs(0), Duration::from_secs(0)) }; // Calculate success rate @@ -718,12 +698,12 @@ impl TxnTracker { for (_plan_id, tracker) in &self.plan_trackers { if tracker.plan_failed_production { - // Counted separately or as completed failed? + // Counted separately or as completed failed? // Let's count it as completed (failed) for the "Completed Plans" metric if we consider it "Done" - // But for clarity, let's keep it separate or part of "Not Produced" logic? - // The requirement is to explain "Not Produced". + // But for clarity, let's keep it separate or part of "Not Produced" logic? + // The requirement is to explain "Not Produced". // If we set plan_produced=true in mark_plan_failed, it lands here. - + // If failed production, it contributes to "Prod Failures" but we should decide if it's "Completed". // Since it will never produce more txns, it is effectively completed (failed). completed_plans += 1; @@ -752,12 +732,7 @@ impl TxnTracker { table.load_preset(UTF8_FULL); // Set table header - summary statistics - table.set_header(vec![ - "Metric", - "Value", - "Metric", - "Value", - ]); + table.set_header(vec!["Metric", "Value", "Metric", "Value"]); // Row 1: Txn progress and TPS table.add_row(vec![ @@ -792,7 +767,11 @@ impl TxnTracker { Cell::new("Produced Plans"), Cell::new(&format_large_number(produced_plans)), Cell::new("Not Produced"), - Cell::new(&format!("{}/{}F", format_large_number(not_produced_plans), format_large_number(self.total_failed_production_plans))), + Cell::new(&format!( + "{}/{}F", + format_large_number(not_produced_plans), + format_large_number(self.total_failed_production_plans) + )), ]); // Row 5: Completed plans and in progress plans diff --git a/src/actors/producer/messages.rs b/src/actors/producer/messages.rs index b7b21c4..4f3712a 100644 --- a/src/actors/producer/messages.rs +++ b/src/actors/producer/messages.rs @@ -12,18 +12,9 @@ pub struct RegisterTxnPlan { impl RegisterTxnPlan { pub fn new( plan: Box, - ) -> ( - Self, - tokio::sync::oneshot::Receiver>, - ) { + ) -> (Self, tokio::sync::oneshot::Receiver>) { let (tx, rx) = tokio::sync::oneshot::channel(); - ( - Self { - plan, - responder: tx, - }, - rx, - ) + (Self { plan, responder: tx }, rx) } } diff --git a/src/actors/producer/producer_actor.rs b/src/actors/producer/producer_actor.rs index d5b0639..9e8b703 100644 --- a/src/actors/producer/producer_actor.rs +++ b/src/actors/producer/producer_actor.rs @@ -8,8 +8,8 @@ use std::time::Duration; use crate::actors::consumer::Consumer; use crate::actors::monitor::monitor_actor::{PlanProduced, ProduceTxns}; use crate::actors::monitor::{ - Monitor, PlanCompleted, PlanFailed, RegisterPlan, RegisterProducer, ReportProducerStats, - SubmissionResult, UpdateSubmissionResult, CorrectNonces, + CorrectNonces, Monitor, PlanCompleted, PlanFailed, RegisterPlan, RegisterProducer, + ReportProducerStats, SubmissionResult, UpdateSubmissionResult, }; use crate::actors::{ExeFrontPlan, PauseProducer, ResumeProducer}; use crate::txn_plan::{addr_pool::AddressPool, PlanExecutionMode, PlanId, TxnPlan}; @@ -34,9 +34,7 @@ pub struct ProducerState { impl ProducerState { pub fn running() -> Self { - Self { - state: Arc::new(AtomicU32::new(1)), - } + Self { state: Arc::new(AtomicU32::new(1)) } } pub fn set_running(&self) { @@ -167,9 +165,7 @@ impl Producer { // Fetch accounts and build transactions let ready_accounts = address_pool.fetch_senders(plan.size().unwrap_or_else(|| address_pool.len())); - let iterator = plan - .as_mut() - .build_txns(ready_accounts, account_generator.clone())?; + let iterator = plan.as_mut().build_txns(ready_accounts, account_generator.clone())?; // If the plan doesn't consume nonces, accounts can be used by other processes immediately. if !iterator.consume_nonce { @@ -177,10 +173,7 @@ impl Producer { } // must send to monitor before sending to consumer monitor_addr - .send(RegisterPlan { - plan_id: plan_id.clone(), - plan_name: plan.name().to_string(), - }) + .send(RegisterPlan { plan_id: plan_id.clone(), plan_name: plan.name().to_string() }) .await .unwrap(); let mut count = 0; @@ -207,22 +200,13 @@ impl Producer { plan_id: plan_id.clone(), reason: format!("Consumer send error: {}", e), }); - return Err(anyhow::anyhow!( - "Failed to send transaction to Consumer: {}", - e - )); + return Err(anyhow::anyhow!("Failed to send transaction to Consumer: {}", e)); } - monitor_addr.do_send(ProduceTxns { - plan_id: plan_id.clone(), - count: 1, - }); + monitor_addr.do_send(ProduceTxns { plan_id: plan_id.clone(), count: 1 }); count += 1; sending_txns.fetch_add(1, Ordering::Relaxed); } - monitor_addr.do_send(PlanProduced { - plan_id: plan_id.clone(), - count, - }); + monitor_addr.do_send(PlanProduced { plan_id: plan_id.clone(), count }); tracing::debug!( "All transactions for plan '{}' (id={}) ({} txns) have been sent to the consumer.", @@ -239,9 +223,7 @@ impl Actor for Producer { type Context = Context; fn started(&mut self, ctx: &mut Self::Context) { - self.monitor_addr.do_send(RegisterProducer { - addr: ctx.address(), - }); + self.monitor_addr.do_send(RegisterProducer { addr: ctx.address() }); let address_pool = self.address_pool.clone(); async move { let count = address_pool.len(); @@ -311,10 +293,7 @@ impl Handler for Producer { { tracing::error!("Execution of plan '{}' failed: {}", plan_id, e); // Notify self of failure to handle cleanup and trigger the next plan. - self_addr.do_send(PlanFailed { - plan_id, - reason: e.to_string(), - }); + self_addr.do_send(PlanFailed { plan_id, reason: e.to_string() }); return Ok(None); } @@ -360,11 +339,7 @@ impl Handler for Producer { self.stats.remain_plans_num += 1; let plan_id = msg.plan.id().clone(); - tracing::debug!( - "Registering new plan '{}' (id={}).", - msg.plan.name(), - plan_id - ); + tracing::debug!("Registering new plan '{}' (id={}).", msg.plan.name(), plan_id); // Add the plan to the back of the queue. self.plan_queue.push_back(msg.plan); @@ -432,9 +407,7 @@ impl Handler for Producer { let address_pool = self.address_pool.clone(); self.stats .sending_txns - .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |val| { - Some(val.saturating_sub(1)) - }) + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |val| Some(val.saturating_sub(1))) .ok(); match msg.result.as_ref() { SubmissionResult::Success(_) => { @@ -523,7 +496,9 @@ impl Handler for Producer { for correction in msg.corrections { // Find the account_id from the address - if let Some(account_id) = self.account_generator.find_account_id_by_address(&correction.account) { + if let Some(account_id) = + self.account_generator.find_account_id_by_address(&correction.account) + { tracing::debug!( "Correcting nonce for account {:?} to {}", correction.account, @@ -532,19 +507,14 @@ impl Handler for Producer { // Update nonce cache self.nonce_cache.insert(account_id, correction.expected_nonce as u32); // Unlock the account with the correct nonce - self.address_pool.unlock_correct_nonce(account_id, correction.expected_nonce as u32); + self.address_pool + .unlock_correct_nonce(account_id, correction.expected_nonce as u32); } else { - tracing::warn!( - "Could not find account_id for address {:?}", - correction.account - ); + tracing::warn!("Could not find account_id for address {:?}", correction.account); } } // Update ready accounts count - self.stats.ready_accounts.store( - self.address_pool.ready_len() as u64, - Ordering::Relaxed, - ); + self.stats.ready_accounts.store(self.address_pool.ready_len() as u64, Ordering::Relaxed); } } diff --git a/src/config/bench_config.rs b/src/config/bench_config.rs index b08c5b9..35bfdb1 100644 --- a/src/config/bench_config.rs +++ b/src/config/bench_config.rs @@ -107,7 +107,6 @@ impl SamplingPolicy { } } - impl BenchConfig { /// Load configuration from TOML file pub fn load>(path: P) -> Result { diff --git a/src/config/contract_config.rs b/src/config/contract_config.rs index 7b2e6dc..73df408 100644 --- a/src/config/contract_config.rs +++ b/src/config/contract_config.rs @@ -107,22 +107,14 @@ impl ContractConfig { #[allow(unused)] pub fn get_weth_address(&self) -> anyhow::Result
{ Address::from_str( - &self - .addresses - .weth9 - .as_ref() - .unwrap_or_else(|| panic!("WETH address not found")), + &self.addresses.weth9.as_ref().unwrap_or_else(|| panic!("WETH address not found")), ) .map_err(|e| anyhow::anyhow!("Invalid WETH address: {}", e)) } /// Get all token addresses pub fn get_all_token(&self) -> Vec { - self.addresses - .tokens - .iter() - .map(|token| token.clone()) - .collect() + self.addresses.tokens.iter().map(|token| token.clone()).collect() } pub fn get_all_token_addresses(&self) -> Vec
{ diff --git a/src/eth/eth_cli.rs b/src/eth/eth_cli.rs index 50fa25a..5b969dc 100644 --- a/src/eth/eth_cli.rs +++ b/src/eth/eth_cli.rs @@ -111,10 +111,7 @@ impl EthHttpCli { /// Create new TxnSender instance pub fn new(rpc_url: &str, chain_id: u64) -> Result { - debug!( - "Creating TxnSender for URL: {}, Chain ID: {}", - rpc_url, chain_id - ); + debug!("Creating TxnSender for URL: {}, Chain ID: {}", rpc_url, chain_id); // Parse URL let url = @@ -131,7 +128,8 @@ impl EthHttpCli { let http = Http::with_client(client, url.clone()); let rpc_client = RpcClient::new(http, true); - let provider: RootProvider = ProviderBuilder::default().connect_client(rpc_client); + let provider: RootProvider = + ProviderBuilder::default().connect_client(rpc_client); inner.push(Arc::new(provider)); } @@ -175,8 +173,7 @@ impl EthHttpCli { }) .await; - self.update_metrics("eth_blockNumber", result.is_ok(), start.elapsed()) - .await; + self.update_metrics("eth_blockNumber", result.is_ok(), start.elapsed()).await; result.with_context(|| "Failed to verify connection to Ethereum node") } @@ -186,12 +183,10 @@ impl EthHttpCli { pub async fn get_balance(&self, address: &Address) -> Result { let start = Instant::now(); - let result = self - .retry_with_backoff(|| async { self.inner[0].get_balance(*address).await }) - .await; + let result = + self.retry_with_backoff(|| async { self.inner[0].get_balance(*address).await }).await; - self.update_metrics("eth_getBalance", result.is_ok(), start.elapsed()) - .await; + self.update_metrics("eth_getBalance", result.is_ok(), start.elapsed()).await; result.with_context(|| format!("Failed to get balance for address: {:?}", address)) } @@ -201,12 +196,10 @@ impl EthHttpCli { pub async fn get_gas_price(&self) -> Result { let start = Instant::now(); - let result = self - .retry_with_backoff(|| async { self.inner[0].get_gas_price().await }) - .await; + let result = + self.retry_with_backoff(|| async { self.inner[0].get_gas_price().await }).await; - self.update_metrics("eth_gasPrice", result.is_ok(), start.elapsed()) - .await; + self.update_metrics("eth_gasPrice", result.is_ok(), start.elapsed()).await; result .map_err(|e| anyhow::anyhow!("Failed to get gas price: {:?}", e)) @@ -226,8 +219,7 @@ impl EthHttpCli { }) .await; - self.update_metrics("txpool_status", result.is_ok(), start.elapsed()) - .await; + self.update_metrics("txpool_status", result.is_ok(), start.elapsed()).await; result.with_context(|| "Failed to get mempool status") } @@ -237,12 +229,10 @@ impl EthHttpCli { pub async fn get_block_number(&self) -> Result { let start = Instant::now(); - let result = self - .retry_with_backoff(|| async { self.inner[0].get_block_number().await }) - .await; + let result = + self.retry_with_backoff(|| async { self.inner[0].get_block_number().await }).await; - self.update_metrics("eth_blockNumber", result.is_ok(), start.elapsed()) - .await; + self.update_metrics("eth_blockNumber", result.is_ok(), start.elapsed()).await; result.with_context(|| "Failed to get block number") } @@ -364,23 +354,12 @@ impl EthHttpCli { // Add summary row for RPC metrics let total_sent: u64 = metrics.per_method.values().map(|m| m.requests_sent).sum(); - let total_succeeded: u64 = metrics - .per_method - .values() - .map(|m| m.requests_succeeded) - .sum(); + let total_succeeded: u64 = metrics.per_method.values().map(|m| m.requests_succeeded).sum(); let total_failed: u64 = metrics.per_method.values().map(|m| m.requests_failed).sum(); - let overall_success_rate = if total_sent > 0 { - total_succeeded as f64 / total_sent as f64 * 100.0 - } else { - 0.0 - }; + let overall_success_rate = + if total_sent > 0 { total_succeeded as f64 / total_sent as f64 * 100.0 } else { 0.0 }; let overall_avg_latency = if total_sent > 0 { - let total_latency: u64 = metrics - .per_method - .values() - .map(|m| m.total_latency_ms) - .sum(); + let total_latency: u64 = metrics.per_method.values().map(|m| m.total_latency_ms).sum(); total_latency as f64 / total_sent as f64 } else { 0.0 @@ -422,12 +401,7 @@ impl EthHttpCli { Err(e) => Err(anyhow::Error::from(e)), }; - self.update_metrics( - "eth_sendRawTransaction", - final_result.is_ok(), - start.elapsed(), - ) - .await; + self.update_metrics("eth_sendRawTransaction", final_result.is_ok(), start.elapsed()).await; final_result } @@ -450,8 +424,7 @@ impl EthHttpCli { }) .await; - self.update_metrics("eth_sendRawTransaction", result.is_ok(), start.elapsed()) - .await; + self.update_metrics("eth_sendRawTransaction", result.is_ok(), start.elapsed()).await; result.with_context(|| "Failed to send transaction envelope") } @@ -467,8 +440,7 @@ impl EthHttpCli { .retry_with_backoff(|| async { self.inner[idx].get_transaction_receipt(tx_hash).await }) .await; - self.update_metrics("eth_getTransactionReceipt", result.is_ok(), start.elapsed()) - .await; + self.update_metrics("eth_getTransactionReceipt", result.is_ok(), start.elapsed()).await; result.with_context(|| format!("Failed to get transaction receipt for hash: {:?}", tx_hash)) } @@ -481,7 +453,6 @@ impl EthHttpCli { .await? } - // pub async fn get_account(&self, address: Address) -> Result { // self.retry_with_backoff(|| async { self.inner[0].get_account(address).await }) // .await @@ -501,8 +472,7 @@ impl EthHttpCli { }) .await; - self.update_metrics("txpool_content", result.is_ok(), start.elapsed()) - .await; + self.update_metrics("txpool_content", result.is_ok(), start.elapsed()).await; result.with_context(|| "Failed to get txpool content") } diff --git a/src/eth/txn_builder.rs b/src/eth/txn_builder.rs index 69719a5..1faf33b 100644 --- a/src/eth/txn_builder.rs +++ b/src/eth/txn_builder.rs @@ -26,10 +26,7 @@ impl TxnBuilder { tx_request: TransactionRequest, signer: &PrivateKeySigner, ) -> Result { - debug!( - "Building and signing transaction with request: {:?}", - tx_request - ); + debug!("Building and signing transaction with request: {:?}", tx_request); debug!("Signer address: {:?}", signer.address()); let mut unsigned_tx = tx_request.build_unsigned().unwrap(); let sig = signer.sign_transaction_sync(&mut unsigned_tx)?; diff --git a/src/main.rs b/src/main.rs index 2d155e0..6a2b2e9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -4,10 +4,10 @@ use alloy::{ signers::local::PrivateKeySigner, }; use anyhow::Result; -use serde::{Deserialize, Serialize}; use clap::Parser; use futures::stream::{self, StreamExt}; use indicatif::{ProgressBar, ProgressStyle}; +use serde::{Deserialize, Serialize}; use std::{ collections::HashMap, process::{Command, Output}, @@ -110,16 +110,10 @@ async fn execute_faucet_distribution( init_nonce_map: Arc>, ) -> Result<()> { let total_faucet_levels = faucet_builder.total_levels(); - info!( - "{} faucet distribution will proceed in {} levels.", - faucet_name, total_faucet_levels - ); + info!("{} faucet distribution will proceed in {} levels.", faucet_name, total_faucet_levels); for level in 0..total_faucet_levels { - info!( - "Starting {} faucet distribution for LEVEL {}...", - faucet_name, level - ); + info!("Starting {} faucet distribution for LEVEL {}...", faucet_name, level); let faucet_level_plan = faucet_builder.create_plan_for_level(level, init_nonce_map.clone(), chain_id); @@ -132,15 +126,9 @@ async fn execute_faucet_distribution( if wait_duration_secs > 0 { tokio::time::sleep(std::time::Duration::from_secs(wait_duration_secs)).await; } - info!( - "{} faucet distribution for LEVEL {} completed successfully.", - faucet_name, level - ); + info!("{} faucet distribution for LEVEL {} completed successfully.", faucet_name, level); } - info!( - "All {} faucet distribution levels are complete.", - faucet_name - ); + info!("All {} faucet distribution levels are complete.", faucet_name); Ok(()) } @@ -170,10 +158,7 @@ async fn test_uniswap( let start_time = Instant::now(); loop { if duration_secs > 0 && start_time.elapsed() >= Duration::from_secs(duration_secs) { - info!( - "Benchmark duration of {} seconds reached. Stopping.", - duration_secs - ); + info!("Benchmark duration of {} seconds reached. Stopping.", duration_secs); break; } let plan = PlanBuilder::swap_token_to_token( @@ -214,10 +199,7 @@ async fn test_erc20_transfer( let start_time = Instant::now(); loop { if duration_secs > 0 && start_time.elapsed() >= Duration::from_secs(duration_secs) { - info!( - "Benchmark duration of {} seconds reached. Stopping.", - duration_secs - ); + info!("Benchmark duration of {} seconds reached. Stopping.", duration_secs); break; } // bench erc20 transfer @@ -265,10 +247,8 @@ async fn get_init_nonce_map( let mut init_nonce_map = accout_generator.init_nonce_map(); let faucet_signer = PrivateKeySigner::from_str(faucet_private_key).unwrap(); let faucet_address = faucet_signer.address(); - init_nonce_map.insert( - faucet_address, - eth_client.get_pending_txn_count(faucet_address).await.unwrap(), - ); + init_nonce_map + .insert(faucet_address, eth_client.get_pending_txn_count(faucet_address).await.unwrap()); Arc::new(init_nonce_map) } @@ -279,8 +259,7 @@ async fn start_bench() -> Result<()> { // Initialize tracing let log_path = benchmark_config.log_path.trim(); - let env_filter = EnvFilter::try_from_default_env() - .unwrap_or_else(|_| EnvFilter::new("info")); + let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")); let _guard = if log_path.is_empty() || log_path.eq_ignore_ascii_case("console") { // Console only @@ -297,12 +276,17 @@ async fn start_bench() -> Result<()> { let directory = path.parent().unwrap_or_else(|| std::path::Path::new(".")); let file_stem = path.file_stem().unwrap_or_else(|| std::ffi::OsStr::new("gravity_bench")); let extension = path.extension().unwrap_or_else(|| std::ffi::OsStr::new("log")); - + // Ensure directory exists std::fs::create_dir_all(directory).unwrap(); let timestamp = chrono::Local::now().format("%Y-%m-%d-%H-%M-%S").to_string(); - let new_filename = format!("{}.{}.{}", file_stem.to_string_lossy(), timestamp, extension.to_string_lossy()); + let new_filename = format!( + "{}.{}.{}", + file_stem.to_string_lossy(), + timestamp, + extension.to_string_lossy() + ); let full_path = directory.join(new_filename); let file = std::fs::File::create(&full_path).unwrap(); @@ -310,13 +294,9 @@ async fn start_bench() -> Result<()> { tracing_subscriber::registry() .with(env_filter) - .with( - tracing_subscriber::fmt::layer() - .with_writer(non_blocking) - .with_ansi(false) - ) + .with(tracing_subscriber::fmt::layer().with_writer(non_blocking).with_ansi(false)) .init(); - + println!("Logging to file: {:?}", full_path); Some(guard) }; @@ -331,7 +311,8 @@ async fn start_bench() -> Result<()> { let snapshot: Snapshot = serde_json::from_str(&snapshot_json).unwrap_or_else(|e| { panic!("Failed to parse snapshot.json: {}", e); }); - let seed_bytes = hex::decode(snapshot.seed.trim()).expect("Invalid hex seed in snapshot.json"); + let seed_bytes = + hex::decode(snapshot.seed.trim()).expect("Invalid hex seed in snapshot.json"); let mut seed = [0u8; 32]; seed.copy_from_slice(&seed_bytes); info!("Recovered faucet_start_nonce: {}", snapshot.faucet_start_nonce); @@ -356,7 +337,7 @@ async fn start_bench() -> Result<()> { .unwrap_or_else(|e| { panic!("Contract config file not found {}", e); }); - + let seed: [u8; 32] = rand::random(); (contract_config, seed, None) }; @@ -366,9 +347,8 @@ async fn start_bench() -> Result<()> { PrivateKeySigner::from_str(&benchmark_config.faucet.private_key).unwrap(), seed, ); - let account_ids = accout_generator - .gen_account(0, benchmark_config.accounts.num_accounts as u64) - .unwrap(); + let account_ids = + accout_generator.gen_account(0, benchmark_config.accounts.num_accounts as u64).unwrap(); let account_addresses = Arc::new({ account_ids .iter() @@ -388,7 +368,8 @@ async fn start_bench() -> Result<()> { let chain_id = benchmark_config.nodes[0].chain_id; info!("Initializing Faucet constructor..."); - let faucet_address = PrivateKeySigner::from_str(&benchmark_config.faucet.private_key).unwrap().address(); + let faucet_address = + PrivateKeySigner::from_str(&benchmark_config.faucet.private_key).unwrap().address(); let on_chain_nonce = eth_clients[0].get_pending_txn_count(faucet_address).await.unwrap(); // In recover mode, use the saved start_nonce so that the skip logic // correctly identifies already-completed Level 0 transactions. @@ -398,10 +379,7 @@ async fn start_bench() -> Result<()> { // Save snapshot in normal mode (after we know the start_nonce) if !args.recover { - let snapshot = Snapshot { - seed: hex::encode(seed), - faucet_start_nonce: start_nonce, - }; + let snapshot = Snapshot { seed: hex::encode(seed), faucet_start_nonce: start_nonce }; let snapshot_json = serde_json::to_string_pretty(&snapshot).unwrap(); std::fs::write("snapshot.json", &snapshot_json).unwrap_or_else(|e| { panic!("Failed to write snapshot.json: {}", e); @@ -464,21 +442,17 @@ async fn start_bench() -> Result<()> { let address_pool: Arc = match benchmark_config.address_pool_type { config::AddressPoolType::Random => { info!("Using RandomAddressPool"); - Arc::new( - txn_plan::addr_pool::managed_address_pool::RandomAddressPool::new( - account_ids.clone(), - account_manager.clone(), - ), - ) + Arc::new(txn_plan::addr_pool::managed_address_pool::RandomAddressPool::new( + account_ids.clone(), + account_manager.clone(), + )) } config::AddressPoolType::Weighted => { info!("Using WeightedAddressPool"); - Arc::new( - txn_plan::addr_pool::weighted_address_pool::WeightedAddressPool::new( - account_ids.clone(), - account_manager.clone(), - ), - ) + Arc::new(txn_plan::addr_pool::weighted_address_pool::WeightedAddressPool::new( + account_ids.clone(), + account_manager.clone(), + )) } }; @@ -503,15 +477,10 @@ async fn start_bench() -> Result<()> { ) .await; - let producer = Producer::new( - address_pool.clone(), - consumer, - monitor, - account_manager.clone(), - ) - .await - .unwrap() - .start(); + let producer = Producer::new(address_pool.clone(), consumer, monitor, account_manager.clone()) + .await + .unwrap() + .start(); execute_faucet_distribution( eth_faucet_builder, chain_id, @@ -543,26 +512,12 @@ async fn start_bench() -> Result<()> { let duration_secs = benchmark_config.performance.duration_secs; if benchmark_config.enable_swap_token { info!("bench uniswap"); - test_uniswap( - address_pool, - chain_id, - contract_config, - &producer, - tps, - duration_secs, - ) - .await?; + test_uniswap(address_pool, chain_id, contract_config, &producer, tps, duration_secs) + .await?; } else { info!("bench erc20 transfer"); - test_erc20_transfer( - address_pool, - chain_id, - contract_config, - &producer, - tps, - duration_secs, - ) - .await?; + test_erc20_transfer(address_pool, chain_id, contract_config, &producer, tps, duration_secs) + .await?; } Ok(()) } @@ -618,10 +573,7 @@ async fn init_nonce(accout_generator: &mut AccountGenerator, eth_client: Arc>() - .await; + stream::iter(tasks).buffer_unordered(1024).collect::>().await; pb.finish_with_message("Done"); let elapsed = start_time.elapsed(); @@ -669,9 +621,7 @@ async fn main() -> Result<()> { }; let res = async { start_bench().await }; let ctrl_c = async { - tokio::signal::ctrl_c() - .await - .expect("Failed to install CTRL+C signal handler"); + tokio::signal::ctrl_c().await.expect("Failed to install CTRL+C signal handler"); println!("Received Ctrl+C, saving heap profile..."); }; tokio::select! { diff --git a/src/txn_plan/addr_pool/managed_address_pool.rs b/src/txn_plan/addr_pool/managed_address_pool.rs index ce71530..2f846be 100644 --- a/src/txn_plan/addr_pool/managed_address_pool.rs +++ b/src/txn_plan/addr_pool/managed_address_pool.rs @@ -30,16 +30,9 @@ impl RandomAddressPool { ready_accounts.push((account_id, nonce)); } - let inner = Inner { - account_status, - ready_accounts, - all_account_ids: account_ids, - }; - - Self { - inner: Mutex::new(inner), - account_generator, - } + let inner = Inner { account_status, ready_accounts, all_account_ids: account_ids }; + + Self { inner: Mutex::new(inner), account_generator } } } diff --git a/src/txn_plan/addr_pool/weighted_address_pool.rs b/src/txn_plan/addr_pool/weighted_address_pool.rs index c737883..108ea7a 100644 --- a/src/txn_plan/addr_pool/weighted_address_pool.rs +++ b/src/txn_plan/addr_pool/weighted_address_pool.rs @@ -84,10 +84,7 @@ impl WeightedAddressPool { long_tail_ready_accounts, }; - Self { - inner: Mutex::new(inner), - account_generator, - } + Self { inner: Mutex::new(inner), account_generator } } fn unlock_account(&self, account: AccountId, nonce: Option) { diff --git a/src/txn_plan/constructor/approve.rs b/src/txn_plan/constructor/approve.rs index deaa43f..a12e964 100644 --- a/src/txn_plan/constructor/approve.rs +++ b/src/txn_plan/constructor/approve.rs @@ -23,11 +23,7 @@ pub struct ApproveTokenConstructor { impl ApproveTokenConstructor { pub fn new(chain_id: u64, token_address: Address, spender_address: Address) -> Self { - Self { - chain_id, - token_address, - spender_address, - } + Self { chain_id, token_address, spender_address } } } @@ -38,10 +34,7 @@ impl FromTxnConstructor for ApproveTokenConstructor { account_generator: AccountManager, nonce: u64, ) -> Result { - let approve_call = IERC20::approveCall { - spender: self.spender_address, - amount: U256::MAX, - }; + let approve_call = IERC20::approveCall { spender: self.spender_address, amount: U256::MAX }; let call_data = approve_call.abi_encode(); let call_data = Bytes::from(call_data); diff --git a/src/txn_plan/constructor/erc20_transfer.rs b/src/txn_plan/constructor/erc20_transfer.rs index 5e950a6..a780a27 100644 --- a/src/txn_plan/constructor/erc20_transfer.rs +++ b/src/txn_plan/constructor/erc20_transfer.rs @@ -28,12 +28,7 @@ impl Erc20TransferConstructor { chain_id: u64, address_pool: Arc, ) -> Self { - Self { - token_list, - transfer_amount, - chain_id, - address_pool, - } + Self { token_list, transfer_amount, chain_id, address_pool } } } @@ -49,10 +44,7 @@ impl FromTxnConstructor for Erc20TransferConstructor { let to_address = account_generator.get_address_by_id(to_address); let from_address = account_generator.get_address_by_id(from_account_id); // build ERC20 transfer call - let transfer_call = IERC20::transferCall { - to: to_address, - amount: self.transfer_amount, - }; + let transfer_call = IERC20::transferCall { to: to_address, amount: self.transfer_amount }; let call_data = transfer_call.abi_encode(); let call_data = Bytes::from(call_data); diff --git a/src/txn_plan/constructor/faucet.rs b/src/txn_plan/constructor/faucet.rs index 822af85..99c751c 100644 --- a/src/txn_plan/constructor/faucet.rs +++ b/src/txn_plan/constructor/faucet.rs @@ -139,9 +139,7 @@ impl FaucetTreePlanBuilder { for level in &account_levels { for acc in level { let address = account_generator.get_address_by_id(*acc); - nonce_map - .entry(address) - .or_insert_with(|| Arc::new(AtomicU64::new(0))); + nonce_map.entry(address).or_insert_with(|| Arc::new(AtomicU64::new(0))); } } } diff --git a/src/txn_plan/constructor/swap_token_2_token.rs b/src/txn_plan/constructor/swap_token_2_token.rs index 7127bdc..091fc52 100644 --- a/src/txn_plan/constructor/swap_token_2_token.rs +++ b/src/txn_plan/constructor/swap_token_2_token.rs @@ -28,13 +28,7 @@ impl SwapTokenToTokenConstructor { transter_amount: U256, router_address: Address, ) -> Self { - Self { - token_list, - chain_id, - address_pool, - transter_amount, - router_address, - } + Self { token_list, chain_id, address_pool, transter_amount, router_address } } } diff --git a/src/txn_plan/faucet_plan.rs b/src/txn_plan/faucet_plan.rs index 3ee2ee7..1fcb36d 100644 --- a/src/txn_plan/faucet_plan.rs +++ b/src/txn_plan/faucet_plan.rs @@ -127,81 +127,65 @@ impl TxnPlan for LevelFaucetPlan { let txn_builder = self.txn_builder.clone(); let account_init_nonce = self.account_init_nonce.clone(); let handle = tokio::task::spawn_blocking(move || { - senders - .chunks(1024) - .enumerate() - .for_each(|(chunk_index, chunk)| { - chunk.into_par_iter().enumerate().for_each( - |(sender_index, sender_signer_id)| { - let sender_signer = - account_generator.get_signer_by_id(*sender_signer_id); - let start_index = (chunk_index * 1024 + sender_index) * degree; - let end_index = (start_index + degree).min(final_recipients.len()); - if end_index < start_index { - return; - } - for i in start_index..end_index { - let (to_address, value) = if is_final_level { - let to = final_recipients[i].clone(); - let val = amount_per_recipient; - (to, val) - } else { - let to_id = account_levels[level][i]; - let to = account_generator.get_address_by_id(to_id); - let val = intermediate_funding_amounts[level]; - (Arc::new(to), val) - }; - let nonce_map_guard = nonce_map.lock().unwrap(); - let nonce = nonce_map_guard - .get(&sender_signer.address()) - .unwrap() - .fetch_add(1, Ordering::Relaxed); - let init_nonce = account_init_nonce - .get(&sender_signer.address()) - .unwrap_or(&0); - // Skip transaction if it was already executed (recovery mode). - // In normal mode, init_nonce is 0 for all accounts, so nothing is skipped. - // In recovery mode, init_nonce is the on-chain nonce, so we skip if init_nonce > nonce. - if *init_nonce > nonce && init_nonce != &0 { - continue; - } - let tx_request = txn_builder.build_faucet_txn( - *to_address, - value, - nonce, - chain_id, - ); - let tx_envelope = TxnBuilder::build_and_sign_transaction( - tx_request, - &sender_signer, - ) + senders.chunks(1024).enumerate().for_each(|(chunk_index, chunk)| { + chunk.into_par_iter().enumerate().for_each(|(sender_index, sender_signer_id)| { + let sender_signer = account_generator.get_signer_by_id(*sender_signer_id); + let start_index = (chunk_index * 1024 + sender_index) * degree; + let end_index = (start_index + degree).min(final_recipients.len()); + if end_index < start_index { + return; + } + for i in start_index..end_index { + let (to_address, value) = if is_final_level { + let to = final_recipients[i].clone(); + let val = amount_per_recipient; + (to, val) + } else { + let to_id = account_levels[level][i]; + let to = account_generator.get_address_by_id(to_id); + let val = intermediate_funding_amounts[level]; + (Arc::new(to), val) + }; + let nonce_map_guard = nonce_map.lock().unwrap(); + let nonce = nonce_map_guard + .get(&sender_signer.address()) + .unwrap() + .fetch_add(1, Ordering::Relaxed); + let init_nonce = + account_init_nonce.get(&sender_signer.address()).unwrap_or(&0); + // Skip transaction if it was already executed (recovery mode). + // In normal mode, init_nonce is 0 for all accounts, so nothing is skipped. + // In recovery mode, init_nonce is the on-chain nonce, so we skip if init_nonce > nonce. + if *init_nonce > nonce && init_nonce != &0 { + continue; + } + let tx_request = + txn_builder.build_faucet_txn(*to_address, value, nonce, chain_id); + let tx_envelope = + TxnBuilder::build_and_sign_transaction(tx_request, &sender_signer) .unwrap(); - let metadata = Arc::new(TxnMetadata { - from_account: Arc::new(sender_signer.address()), - nonce, - from_account_id: *sender_signer_id, - txn_id: Uuid::new_v4(), - plan_id: plan_id.clone(), - }); + let metadata = Arc::new(TxnMetadata { + from_account: Arc::new(sender_signer.address()), + nonce, + from_account_id: *sender_signer_id, + txn_id: Uuid::new_v4(), + plan_id: plan_id.clone(), + }); - tx.send(SignedTxnWithMetadata { - bytes: tx_envelope.encoded_2718(), - metadata, - }) - .unwrap(); - } - }, - ) - }); + tx.send(SignedTxnWithMetadata { + bytes: tx_envelope.encoded_2718(), + metadata, + }) + .unwrap(); + } + }) + }); drop(tx); }); tokio::spawn(async move { handle.await.unwrap(); }); - Ok(TxnIter { - iterator: rx, - consume_nonce: false, - }) + Ok(TxnIter { iterator: rx, consume_nonce: false }) } } diff --git a/src/txn_plan/plan.rs b/src/txn_plan/plan.rs index 868b3c6..99a7926 100644 --- a/src/txn_plan/plan.rs +++ b/src/txn_plan/plan.rs @@ -107,10 +107,7 @@ impl TxnPlan for ManyToOnePlan { let tx_envelope = TxnBuilder::build_and_sign_transaction(tx_request, &signer) .unwrap(); - SignedTxnWithMetadata { - bytes: tx_envelope.encoded_2718(), - metadata, - } + SignedTxnWithMetadata { bytes: tx_envelope.encoded_2718(), metadata } }) .collect::>() }) @@ -124,10 +121,7 @@ impl TxnPlan for ManyToOnePlan { handle.await.unwrap(); }); - Ok(TxnIter { - iterator: rx, - consume_nonce: true, - }) + Ok(TxnIter { iterator: rx, consume_nonce: true }) } } @@ -214,7 +208,7 @@ impl TxnPlan for OneToManyPlan { from_account: Arc::new( account_generator.get_address_by_id(from_account_id), ), - from_account_id: from_account_id, + from_account_id, nonce: 0, txn_id: Uuid::new_v4(), plan_id: plan_id.clone(), @@ -239,9 +233,6 @@ impl TxnPlan for OneToManyPlan { handle.await.unwrap(); }); - Ok(TxnIter { - iterator: rx, - consume_nonce: false, - }) + Ok(TxnIter { iterator: rx, consume_nonce: false }) } } diff --git a/src/util/gen_account.rs b/src/util/gen_account.rs index 7b64551..eb92c1c 100644 --- a/src/util/gen_account.rs +++ b/src/util/gen_account.rs @@ -27,11 +27,7 @@ pub struct AccountSignerCache { impl AccountSignerCache { pub(crate) fn new(size: usize, seed: [u8; 32]) -> Self { - Self { - signers: Vec::with_capacity(size), - size, - seed, - } + Self { signers: Vec::with_capacity(size), size, seed } } pub(crate) fn save_signer(&mut self, signer: PrivateKeySigner, account_id: AccountId) { @@ -124,9 +120,7 @@ impl AccountGenerator { } pub fn accouts_nonce_iter(&self) -> impl Iterator)> { - self.accout_addresses - .iter() - .zip(self.init_nonces.iter().cloned()) + self.accout_addresses.iter().zip(self.init_nonces.iter().cloned()) } pub fn account_ids_with_nonce(&self) -> impl Iterator)> + '_ { @@ -148,8 +142,7 @@ impl AccountGenerator { self.address_to_id.insert(addr, account_id); self.accout_signers.save_signer(signer.clone(), account_id); } - self.init_nonces - .extend((0..size).map(|_| Arc::new(AtomicU64::new(0)))); + self.init_nonces.extend((0..size).map(|_| Arc::new(AtomicU64::new(0)))); } let mut res = Vec::with_capacity(size as usize); for i in 0..size { @@ -326,13 +319,12 @@ mod tests { // Generate accounts let faucet_pk = ""; let faucet_bytes = hex::decode(faucet_pk).expect("invalid faucet private key hex"); - let faucet_signer = PrivateKeySigner::from_slice(&faucet_bytes) - .expect("failed to create faucet signer"); + let faucet_signer = + PrivateKeySigner::from_slice(&faucet_bytes).expect("failed to create faucet signer"); let mut generator = AccountGenerator::with_capacity(faucet_signer.clone(), [0u8; 32]); - let account_ids = generator - .gen_account(0, num_accounts) - .expect("failed to generate accounts"); + let account_ids = + generator.gen_account(0, num_accounts).expect("failed to generate accounts"); // Connect to RPC let eth_client = EthHttpCli::new(rpc_url, chain_id) @@ -344,10 +336,8 @@ mod tests { // Check faucet balance let faucet_address = faucet_signer.address(); - let faucet_balance = eth_client - .get_balance(&faucet_address) - .await - .expect("failed to get faucet balance"); + let faucet_balance = + eth_client.get_balance(&faucet_address).await.expect("failed to get faucet balance"); println!( "Faucet : {:?} balance = {} wei ({} ETH)", faucet_address, @@ -396,9 +386,8 @@ mod tests { // Generate accounts let mut generator = AccountGenerator::with_capacity(faucet_signer.clone(), [0u8; 32]); - let account_ids = generator - .gen_account(0, num_accounts) - .expect("failed to generate accounts"); + let account_ids = + generator.gen_account(0, num_accounts).expect("failed to generate accounts"); // Connect to RPC let eth_client = @@ -424,14 +413,9 @@ mod tests { .expect("failed to build tx request"); let tx_envelope = TxnBuilder::build_and_sign_transaction(tx_request, &faucet_signer) .expect("failed to sign tx"); - let tx_hash = eth_client - .send_tx_envelope(tx_envelope) - .await - .expect("failed to send tx"); - println!( - "ID {:>6}: {:?} tx = {:?}", - id.0, to, tx_hash, - ); + let tx_hash = + eth_client.send_tx_envelope(tx_envelope).await.expect("failed to send tx"); + println!("ID {:>6}: {:?} tx = {:?}", id.0, to, tx_hash,); nonce += 1; } @@ -443,16 +427,8 @@ mod tests { println!("\n=== Balances after faucet ==="); for &id in &account_ids { let address = generator.get_address_by_id(id); - let balance = eth_client - .get_balance(&address) - .await - .expect("failed to get balance"); - println!( - "ID {:>6}: {:?} balance = {} ETH", - id.0, - address, - format_eth(balance), - ); + let balance = eth_client.get_balance(&address).await.expect("failed to get balance"); + println!("ID {:>6}: {:?} balance = {} ETH", id.0, address, format_eth(balance),); } println!("Done."); }