Updated spammer #127
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -20,6 +20,9 @@ | |
| use crate::make_signers; | ||
| use crate::tx::{make_signed_contract_call_tx, make_signed_eip1559_tx, make_signed_eip4844_tx}; | ||
|
|
||
| /// Target pool size to maintain (in number of transactions). | ||
| const TARGET_POOL_SIZE: u64 = 30_000; | ||
|
|
||
| struct ContractPayload { | ||
| /// Contract address for contract call spamming. | ||
| address: Address, | ||
|
|
@@ -44,6 +47,8 @@ | |
| max_time: u64, | ||
| /// Maximum number of transactions to send per second. | ||
| max_rate: u64, | ||
| /// Number of ms between sending batches of txs (default: 200). | ||
| batch_interval: u64, | ||
| /// Whether to send EIP-4844 blob transactions. | ||
| blobs: bool, | ||
| /// Chain ID for the transactions. | ||
|
|
@@ -59,6 +64,7 @@ | |
| max_num_txs: u64, | ||
| max_time: u64, | ||
| max_rate: u64, | ||
| batch_interval: u64, | ||
| blobs: bool, | ||
| chain_id: u64, | ||
| ) -> Result<Self> { | ||
|
|
@@ -70,6 +76,7 @@ | |
| max_num_txs, | ||
| max_time, | ||
| max_rate, | ||
| batch_interval, | ||
| blobs, | ||
| chain_id, | ||
| contract_payload: None, | ||
|
|
@@ -83,6 +90,7 @@ | |
| max_num_txs: u64, | ||
| max_time: u64, | ||
| max_rate: u64, | ||
| batch_interval: u64, | ||
| contract: &Address, | ||
| function: &str, | ||
| args: &[String], | ||
|
|
@@ -101,6 +109,7 @@ | |
| max_num_txs, | ||
| max_time, | ||
| max_rate, | ||
| batch_interval, | ||
| blobs: false, // Contract calls don't use blobs | ||
| contract_payload: Some(contract_payload), | ||
| chain_id, | ||
|
|
@@ -155,6 +164,17 @@ | |
| Ok(u64::from_str_radix(hex_str, 16)?) | ||
| } | ||
|
|
||
| // Get current txpool status. | ||
| async fn get_txpool_status(&self) -> Result<TxpoolStatus> { | ||
| self.client.rpc_request("txpool_status", vec![]).await | ||
| } | ||
|
|
||
| // Get current number of pending and queued transactions in the pool. | ||
| async fn get_mempool_count(&self) -> Result<u64> { | ||
| let status = self.get_txpool_status().await?; | ||
| Ok(status.pending + status.queued) | ||
| } | ||
|
|
||
| /// Generate and send transactions to the Ethereum node at a controlled rate. | ||
| async fn spammer( | ||
| &self, | ||
|
|
@@ -162,94 +182,122 @@ | |
| report_sender: Sender<Instant>, | ||
| finish_sender: Sender<()>, | ||
| ) -> Result<()> { | ||
| // Fetch latest nonce for the sender address. | ||
| let address = self.signer.address(); | ||
| let latest_nonce = self.get_latest_nonce(address).await?; | ||
| let txs_per_batch = self.max_rate | ||
| .saturating_mul(self.batch_interval) | ||
| .checked_div(1000) | ||
| .unwrap_or(0); | ||
| debug!( | ||
| "Spamming {address} starting from nonce={latest_nonce} at rate {}, sending {txs_per_batch} txs every {}ms", | ||
| self.max_rate, | ||
| self.batch_interval, | ||
| ); | ||
|
|
||
| // Initialize nonce and counters. | ||
| let mut nonce = latest_nonce; | ||
| let start_time = Instant::now(); | ||
| let mut txs_sent_total = 0u64; | ||
| let mut interval = time::interval(Duration::from_millis(self.batch_interval)); | ||
|
|
||
| loop { | ||
| // Wait for next one-second tick. | ||
| let _ = interval.tick().await; | ||
| let interval_start = Instant::now(); | ||
|
|
||
| // Verify the nonce for gaps | ||
| // TODO: probably this should run as a separate task | ||
| let on_chain_nonce = self.get_latest_nonce(address).await?; | ||
| // If the span between the on-chain nonce and the one we are about to send | ||
| // is too big, then probably there is a gap that doesn't allow the | ||
| // on-chain nonce too advance. | ||
| let nonce_span = nonce.saturating_sub(on_chain_nonce); | ||
| if nonce_span > self.max_rate { | ||
| debug!("Current nonce={nonce}, on-chain nonce={on_chain_nonce}. Sending 10 txs"); | ||
| let batch_entries = self.build_batch_entries(10, on_chain_nonce).await?; | ||
| if let Some(results) = self.send_raw_batch(&batch_entries).await? { | ||
| if results.len() != batch_entries.len() { | ||
| return Err(eyre::eyre!( | ||
| "Batch response count {} does not match request count {}", | ||
| results.len(), | ||
| batch_entries.len() | ||
| )); | ||
| } | ||
|
|
||
| // Report individual results. | ||
| for ((_, tx_bytes_len), result) in batch_entries.into_iter().zip(results) { | ||
| let mapped_result = result.map(|_| tx_bytes_len); | ||
|
Contributor
There was a problem hiding this comment. Why are we discarding the Ok value of results?
Contributor
There was a problem hiding this comment. I think it's because it contains the tx hash, which isn't used later anywhere, and we don't care about single tx success because it will slow down the spammer. We should modify this if needed. cc @mpoke |
||
| result_sender.send(mapped_result).await?; | ||
| } | ||
| } else { | ||
| debug!("Batch eth_sendRawTransaction timed out; skipping this tick"); | ||
| let _ = report_sender.send(interval_start).await; | ||
|
ljoss17 marked this conversation as resolved.
Outdated
|
||
| continue; | ||
| } | ||
| } | ||
|
|
||
| // Get current pool size and calculate dynamic send rate | ||
| let current_pool_size = self.get_mempool_count().await.unwrap_or(0); | ||
| let space_available = TARGET_POOL_SIZE.saturating_sub(current_pool_size); | ||
| let txs_to_send = if space_available < txs_per_batch { | ||
| space_available | ||
| } else { | ||
| txs_per_batch | ||
| }; | ||
|
|
||
| // Continue if there is no space available | ||
| if txs_to_send == 0 { | ||
| debug!("Mempool already full. Do not send more transactions."); | ||
| let _ = report_sender.send(interval_start).await; | ||
| continue; | ||
| } | ||
|
|
||
| // Limit the max number of transactions | ||
| let tx_count = if self.max_num_txs > 0 { | ||
| txs_to_send.min(self.max_num_txs.saturating_sub(txs_sent_total)) | ||
| } else { | ||
| txs_to_send | ||
| }; | ||
|
|
||
| // Prepare batch of transactions for this interval. | ||
| let batch_entries = self.build_batch_entries(tx_count, nonce).await?; | ||
| let batch_size = batch_entries.len() as u64; | ||
|
|
||
| debug!( | ||
| "Pool: {current_pool_size}/{TARGET_POOL_SIZE}, sending {batch_size} txs from nonce {nonce} (rate: {})", | ||
| self.max_rate | ||
| ); | ||
|
|
||
| // Send all transactions in a single batch RPC call. | ||
| if !batch_entries.is_empty() { | ||
| if let Some(results) = self.send_raw_batch(&batch_entries).await? { | ||
| if results.len() != batch_entries.len() { | ||
| return Err(eyre::eyre!( | ||
| "Batch response count {} does not match request count {}", | ||
| results.len(), | ||
| batch_entries.len() | ||
| )); | ||
| } | ||
|
|
||
| // Report individual results. | ||
| for ((_, tx_bytes_len), result) in batch_entries.into_iter().zip(results) { | ||
| let mapped_result = result.map(|_| tx_bytes_len); | ||
|
|
||
| result_sender.send(mapped_result).await?; | ||
| } | ||
|
|
||
| txs_sent_total += batch_size; | ||
| nonce += batch_size; | ||
| } else { | ||
| debug!("Batch eth_sendRawTransaction timed out; skipping this tick"); | ||
| } | ||
| } | ||
|
|
||
| // Give time to the in-flight results to be received. | ||
| sleep(Duration::from_millis(20)).await; | ||
|
|
||
| // Signal tracker to report stats after this batch. | ||
| let _ = report_sender.send(interval_start).await; | ||
|
|
||
| // Check exit conditions after each tick. | ||
| if (self.max_num_txs > 0 && txs_sent_total >= self.max_num_txs) | ||
|
|
@@ -262,6 +310,68 @@ | |
| Ok(()) | ||
| } | ||
|
|
||
| async fn build_batch_entries( | ||
| &self, | ||
| tx_count: u64, | ||
| nonce: u64, | ||
| ) -> Result<Vec<(Vec<serde_json::Value>, u64)>> { | ||
| let mut batch_entries = Vec::with_capacity(tx_count as usize); | ||
| let mut next_nonce = nonce; | ||
|
|
||
| for _ in 0..tx_count { | ||
| let signed_tx = if let Some(ref payload) = self.contract_payload { | ||
| make_signed_contract_call_tx( | ||
| &self.signer, | ||
| next_nonce, | ||
| payload.address, | ||
| &payload.function_sig, | ||
| payload.args.as_slice(), | ||
| self.chain_id, | ||
| ) | ||
| .await? | ||
| } else if self.blobs { | ||
| make_signed_eip4844_tx(&self.signer, next_nonce, self.chain_id).await? | ||
| } else { | ||
| make_signed_eip1559_tx(&self.signer, next_nonce, self.chain_id).await? | ||
| }; | ||
|
|
||
| let tx_bytes = signed_tx.encoded_2718(); | ||
| let tx_bytes_len = tx_bytes.len() as u64; | ||
| let payload = hex::encode(tx_bytes); | ||
| batch_entries.push((vec![json!(payload)], tx_bytes_len)); | ||
| next_nonce += 1; | ||
| } | ||
|
|
||
| Ok(batch_entries) | ||
| } | ||
|
|
||
| async fn send_raw_batch( | ||
| &self, | ||
| batch_entries: &[(Vec<serde_json::Value>, u64)], | ||
| ) -> Result<Option<Vec<Result<String>>>> { | ||
| let params: Vec<_> = batch_entries | ||
| .iter() | ||
| .map(|(params, _)| params.clone()) | ||
| .collect(); | ||
|
|
||
| match self | ||
| .client | ||
| .rpc_batch_request("eth_sendRawTransaction", params) | ||
| .await | ||
| { | ||
| Ok(responses) => Ok(Some(responses)), | ||
| Err(err) => { | ||
| if let Some(jsonrpsee_core::client::Error::RequestTimeout) = | ||
| err.downcast_ref::<jsonrpsee_core::client::Error>() | ||
| { | ||
| Ok(None) | ||
| } else { | ||
| Err(err) | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // Track and report statistics on sent transactions. | ||
| async fn tracker( | ||
| &self, | ||
|
|
@@ -290,7 +400,7 @@ | |
| sleep(Duration::from_secs(1) - elapsed).await; | ||
| } | ||
|
|
||
| let pool_status = self.get_txpool_status().await?; | ||
| debug!("{stats_last_second}; {pool_status:?}"); | ||
|
|
||
| // Update total, then reset last second stats | ||
|
|
@@ -378,14 +488,15 @@ | |
| } | ||
| } | ||
|
|
||
| #[derive(Clone)] | ||
| struct RpcClient { | ||
| client: HttpClient, | ||
| } | ||
|
|
||
| impl RpcClient { | ||
| pub fn new(url: Url) -> Result<Self> { | ||
| let client = HttpClientBuilder::default() | ||
| .request_timeout(Duration::from_secs(1)) | ||
| .build(url)?; | ||
| Ok(Self { client }) | ||
| } | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why 10? should it not be
none_span?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Just a small number of tx to clear the nonce gap in case there is one. Note that it might be that there is no nonce gap and the blocks are committed slower than the injection rate. So we shouldn't inject a lot of transactions here as we might very well send a lot of duplicates.