feat(anvil): remove all txs from tx pool by sender origin (#7480)

* feat(anvil): remove all txs from pool by sender origin

* refactor(anvil): combine pending and ready transactions iterators in one

crates/anvil/core/src/eth/mod.rs

@@ -717,6 +717,13 @@ pub enum EthRequest {
     /// contract.
     #[cfg_attr(feature = "serde", serde(rename = "ots_getContractCreator", with = "sequence"))]
     OtsGetContractCreator(Address),
+
+    /// Removes transactions from the pool by sender origin.
+    #[cfg_attr(
+        feature = "serde",
+        serde(rename = "anvil_removePoolTransactions", with = "sequence")
+    )]
+    RemovePoolTransactions(Address),
 }
 
 /// Represents ethereum JSON-RPC API
@@ -1506,4 +1513,11 @@ true}]}"#;
         let value: serde_json::Value = serde_json::from_str(s).unwrap();
         let _req = serde_json::from_value::<EthRequest>(value).unwrap();
     }
+
+    #[test]
+    fn test_remove_pool_transactions() {
+        let s = r#"{"method": "anvil_removePoolTransactions",  "params":["0x364d6D0333432C3Ac016Ca832fb8594A8cE43Ca6"]}"#;
+        let value: serde_json::Value = serde_json::from_str(s).unwrap();
+        let _req = serde_json::from_value::<EthRequest>(value).unwrap();
+    }
 }

crates/anvil/src/eth/api.rs

@@ -413,6 +413,9 @@ impl EthApi {
             EthRequest::OtsGetContractCreator(address) => {
                 self.ots_get_contract_creator(address).await.to_rpc_result()
             }
+            EthRequest::RemovePoolTransactions(address) => {
+                self.anvil_remove_pool_transactions(address).await.to_rpc_result()
+            }
         }
     }
 
@@ -1781,6 +1784,12 @@ impl EthApi {
         })
     }
 
+    pub async fn anvil_remove_pool_transactions(&self, address: Address) -> Result<()> {
+        node_info!("anvil_removePoolTransactions");
+        self.pool.remove_transactions_by_address(address);
+        Ok(())
+    }
+
     /// Snapshot the state of the blockchain at the current block.
     ///
     /// Handler for RPC call: `evm_snapshot`

crates/anvil/src/eth/pool/mod.rs

@@ -36,7 +36,7 @@ use crate::{
     },
     mem::storage::MinedBlockOutcome,
 };
-use alloy_primitives::{TxHash, U64};
+use alloy_primitives::{Address, TxHash, U64};
 use alloy_rpc_types::txpool::TxpoolStatus;
 use anvil_core::eth::transaction::PendingTransaction;
 use futures::channel::mpsc::{channel, Receiver, Sender};
@@ -141,6 +141,11 @@ impl Pool {
         self.inner.write().remove_invalid(tx_hashes)
     }
 
+    /// Remove transactions by sender
+    pub fn remove_transactions_by_address(&self, sender: Address) -> Vec<Arc<PoolTransaction>> {
+        self.inner.write().remove_transactions_by_address(sender)
+    }
+
     /// Removes a single transaction from the pool
     ///
     /// This is similar to `[Pool::remove_invalid()]` but for a single transaction.
@@ -218,6 +223,24 @@ impl PoolInner {
         )
     }
 
+    /// Returns an iterator over all transactions in the pool filtered by the sender
+    pub fn transactions_by_sender(
+        &self,
+        sender: Address,
+    ) -> impl Iterator<Item = Arc<PoolTransaction>> + '_ {
+        let pending_txs = self
+            .pending_transactions
+            .transactions()
+            .filter(move |tx| tx.pending_transaction.sender().eq(&sender));
+
+        let ready_txs = self
+            .ready_transactions
+            .get_transactions()
+            .filter(move |tx| tx.pending_transaction.sender().eq(&sender));
+
+        pending_txs.chain(ready_txs)
+    }
+
     /// Returns true if this pool already contains the transaction
     fn contains(&self, tx_hash: &TxHash) -> bool {
         self.pending_transactions.contains(tx_hash) || self.ready_transactions.contains(tx_hash)
@@ -342,6 +365,25 @@ impl PoolInner {
 
         removed
     }
+
+    /// Remove transactions by sender address
+    pub fn remove_transactions_by_address(&mut self, sender: Address) -> Vec<Arc<PoolTransaction>> {
+        let tx_hashes =
+            self.transactions_by_sender(sender).map(move |tx| tx.hash()).collect::<Vec<TxHash>>();
+
+        if tx_hashes.is_empty() {
+            return vec![]
+        }
+
+        trace!(target: "txpool", "Removing transactions: {:?}", tx_hashes);
+
+        let mut removed = self.ready_transactions.remove_with_markers(tx_hashes.clone(), None);
+        removed.extend(self.pending_transactions.remove(tx_hashes));
+
+        trace!(target: "txpool", "Removed transactions: {:?}", removed);
+
+        removed
+    }
 }
 
 /// Represents the outcome of a prune

crates/anvil/tests/it/anvil_api.rs

@@ -9,6 +9,7 @@ use anvil_core::{
 use ethers::{
     abi::{ethereum_types::BigEndianHash, AbiDecode},
     prelude::{Middleware, SignerMiddleware},
+    signers::Signer,
     types::{
         transaction::eip2718::TypedTransaction, Address, BlockNumber, Eip1559TransactionRequest,
         TransactionRequest, H256, U256, U64,
@@ -631,3 +632,27 @@ async fn test_fork_revert_call_latest_block_timestamp() {
         latest_block.header.miner.to_ethers()
     );
 }
+
+#[tokio::test(flavor = "multi_thread")]
+async fn can_remove_pool_transactions() {
+    let (api, handle) = spawn(NodeConfig::test()).await;
+    let provider = ethers_http_provider(&handle.http_endpoint());
+    let wallet = handle.dev_wallets().next().unwrap().to_ethers();
+    let provider = Arc::new(SignerMiddleware::new(provider, wallet.clone()));
+
+    let sender = Address::random();
+    let to = Address::random();
+    let val = 1337u64;
+
+    let tx = TransactionRequest::new().from(sender).to(to).value(val);
+
+    provider.send_transaction(tx.from(wallet.address()), None).await.unwrap();
+
+    let initial_txs = provider.txpool_inspect().await.unwrap();
+    assert_eq!(initial_txs.pending.len(), 1);
+
+    api.anvil_remove_pool_transactions(wallet.address().to_alloy()).await.unwrap();
+
+    let final_txs = provider.txpool_inspect().await.unwrap();
+    assert_eq!(final_txs.pending.len(), 0);
+}