feat(anvil-polkadot): add transaction pool RPCs 1/2

crates/anvil-polkadot/Cargo.toml

@@ -126,6 +126,7 @@ itertools.workspace = true
 rand_08.workspace = true
 eyre.workspace = true
 lru = "0.16.0"
+indexmap = "2.0"
 
 # cli
 clap = { version = "4", features = [

crates/anvil-polkadot/src/api_server/server.rs

@@ -1,12 +1,12 @@
 use super::revive_conversions::{ReviveBytes, ReviveFilter};
 use crate::{
     api_server::{
-        ApiRequest,
         error::{Error, Result, ToRpcResponseResult},
         revive_conversions::{
-            AlloyU256, ReviveAddress, ReviveBlockId, ReviveBlockNumberOrTag, SubstrateU256,
-            convert_to_generic_transaction,
+            convert_to_generic_transaction, AlloyU256, ReviveAddress, ReviveBlockId,
+            ReviveBlockNumberOrTag, SubstrateU256,
         },
+        ApiRequest,
     },
     logging::LoggingManager,
     macros::node_info,
@@ -15,56 +15,58 @@ use crate::{
         in_mem_rpc::InMemoryRpcClient,
         mining_engine::MiningEngine,
         service::{
-            BackendError, BackendWithOverlay, Client, Service,
             storage::{
                 AccountType, ByteCodeType, CodeInfo, ContractInfo, ReviveAccountInfo,
                 SystemAccountInfo,
             },
+            BackendError, BackendWithOverlay, Client, Service, TransactionPoolHandle,
         },
         snapshot::{RevertInfo, SnapshotManager},
     },
 };
 use alloy_eips::{BlockId, BlockNumberOrTag};
-use alloy_primitives::{Address, B256, U64, U256};
-use alloy_rpc_types::{Filter, TransactionRequest, anvil::MineOptions};
+use alloy_primitives::{Address, B256, U256, U64};
+use alloy_rpc_types::{anvil::MineOptions, txpool::TxpoolStatus, Filter, TransactionRequest};
 use alloy_serde::WithOtherFields;
 use anvil_core::eth::{EthRequest, Params as MineParams};
 use anvil_rpc::response::ResponseResult;
-use codec::{Decode, Encode};
-use futures::{StreamExt, channel::mpsc};
+use codec::{Decode, DecodeLimit, Encode};
+use futures::{channel::mpsc, StreamExt};
+use indexmap::IndexMap;
 use pallet_revive_eth_rpc::{
-    BlockInfoProvider, EthRpcError, ReceiptExtractor, ReceiptProvider, SubxtBlockInfoProvider,
     client::{Client as EthRpcClient, ClientError, SubscriptionType},
     subxt_client::{self, SrcChainConfig},
+    BlockInfoProvider, EthRpcError, ReceiptExtractor, ReceiptProvider, SubxtBlockInfoProvider,
 };
 use polkadot_sdk::{
     pallet_revive::{
-        ReviveApi,
         evm::{
             Account, Block, Bytes, FeeHistoryResult, FilterResults, ReceiptInfo, TransactionInfo,
             TransactionSigned,
         },
+        ReviveApi,
     },
     parachains_common::{AccountId, Hash, Nonce},
     polkadot_sdk_frame::runtime::types_common::OpaqueBlock,
     sc_client_api::HeaderBackend,
-    sc_service::SpawnTaskHandle,
+    sc_service::{InPoolTransaction, SpawnTaskHandle, TransactionPool},
     sp_api::{Metadata, ProvideRuntimeApi},
     sp_arithmetic::Permill,
     sp_blockchain::Info,
-    sp_core::{self, Hasher, keccak_256},
+    sp_core::{self, keccak_256, Hasher},
     sp_runtime::traits::BlakeTwo256,
 };
 use sqlx::sqlite::SqlitePoolOptions;
 use std::{collections::HashSet, sync::Arc, time::Duration};
-use substrate_runtime::Balance;
+use substrate_runtime::{Balance, RuntimeCall, UncheckedExtrinsic};
 use subxt::{
-    Metadata as SubxtMetadata, OnlineClient, backend::rpc::RpcClient,
-    client::RuntimeVersion as SubxtRuntimeVersion, config::substrate::H256,
-    ext::subxt_rpcs::LegacyRpcMethods, utils::H160,
+    backend::rpc::RpcClient, client::RuntimeVersion as SubxtRuntimeVersion,
+    config::substrate::H256, ext::subxt_rpcs::LegacyRpcMethods, utils::H160,
+    Metadata as SubxtMetadata, OnlineClient,
 };
 
 pub const CLIENT_VERSION: &str = concat!("anvil-polkadot/v", env!("CARGO_PKG_VERSION"));
+const MAX_EXTRINSIC_DEPTH: u32 = 256;
 
 pub struct Wallet {
     accounts: Vec<Account>,
@@ -81,6 +83,7 @@ pub struct ApiServer {
     wallet: Wallet,
     snapshot_manager: SnapshotManager,
     impersonation_manager: ImpersonationManager,
+    tx_pool: Arc<TransactionPoolHandle>,
 }
 
 impl ApiServer {
@@ -117,6 +120,7 @@ impl ApiServer {
             eth_rpc_client,
             snapshot_manager,
             impersonation_manager,
+            tx_pool: substrate_service.tx_pool.clone(),
             wallet: Wallet {
                 accounts: vec![
                     Account::from(subxt_signer::eth::dev::baltathar()),
@@ -283,6 +287,19 @@ impl ApiServer {
                 node_info!("eth_getLogs");
                 self.get_logs(filter).await.to_rpc_result()
             }
+            //------- Transaction Pool ---------
+            EthRequest::TxPoolStatus(_) => {
+                node_info!("txpool_status");
+                self.txpool_status().await.to_rpc_result()
+            }
+            EthRequest::DropAllTransactions() => {
+                node_info!("anvil_dropAllTransactions");
+                self.anvil_drop_all_transactions().await.to_rpc_result()
+            }
+            EthRequest::DropTransaction(eth_hash) => {
+                node_info!("anvil_dropTransaction");
+                self.anvil_drop_transaction(eth_hash).await.to_rpc_result()
+            }
             _ => Err::<(), _>(Error::RpcUnimplemented).to_rpc_result(),
         };
 
@@ -1031,6 +1048,83 @@ impl ApiServer {
 
         Ok(())
     }
+
+    /// Returns transaction pool status
+    async fn txpool_status(&self) -> Result<TxpoolStatus> {
+        let pool_status = self.tx_pool.status();
+        Ok(TxpoolStatus { pending: pool_status.ready as u64, queued: pool_status.future as u64 })
+    }
+
+    /// Drop all transactions from pool
+    async fn anvil_drop_all_transactions(&self) -> Result<()> {
+        let ready_txs = self.tx_pool.ready();
+        let future_txs = self.tx_pool.futures();
+
+        let mut invalid_txs = IndexMap::new();
+
+        for tx in ready_txs {
+            invalid_txs.insert(*tx.hash(), None);
+        }
+
+        for tx in future_txs {
+            invalid_txs.insert(*tx.hash(), None);
+        }
+
+        self.tx_pool.report_invalid(None, invalid_txs).await;
+
+        Ok(())
+    }
+
+    /// Drop a specific transaction from the pool by its ETH hash
+    async fn anvil_drop_transaction(&self, eth_hash: B256) -> Result<()> {
+        // Search in ready transactions
+        for tx in self.tx_pool.ready() {
+            if transaction_matches_eth_hash(tx.data(), eth_hash) {
+                let mut invalid_txs = IndexMap::new();
+                invalid_txs.insert(*tx.hash(), None);
+                self.tx_pool.report_invalid(None, invalid_txs).await;
+                return Ok(());
+            }
+        }
+
+        // Search in future transactions
+        for tx in self.tx_pool.futures() {
+            if transaction_matches_eth_hash(tx.data(), eth_hash) {
+                let mut invalid_txs = IndexMap::new();
+                invalid_txs.insert(*tx.hash(), None);
+                self.tx_pool.report_invalid(None, invalid_txs).await;
+                return Ok(());
+            }
+        }
+
+        // Transaction not found
+        Err(Error::InvalidParams(format!(
+            "Transaction with ETH hash {:?} not found in pool",
+            eth_hash
+        )))
+    }
+}
+
+/// Helper function to check if transaction matches ETH hash
+fn transaction_matches_eth_hash(
+    tx_data: &Arc<polkadot_sdk::sp_runtime::OpaqueExtrinsic>,
+    target_eth_hash: B256,
+) -> bool {
+    let encoded = tx_data.encode();
+    if let Ok(ext) =
+        UncheckedExtrinsic::decode_all_with_depth_limit(MAX_EXTRINSIC_DEPTH, &mut &encoded[..])
+    {
+        if let polkadot_sdk::sp_runtime::generic::UncheckedExtrinsic {
+            function:
+                RuntimeCall::Revive(polkadot_sdk::pallet_revive::Call::eth_transact { payload }),
+            ..
+        } = ext.0
+        {
+            let tx_eth_hash = keccak_256(&payload);
+            return B256::from_slice(&tx_eth_hash) == target_eth_hash;
+        }
+    }
+    false
 }
 
 fn new_contract_info(address: &Address, code_hash: H256, nonce: Nonce) -> ContractInfo {

crates/anvil-polkadot/tests/it/main.rs

@@ -6,4 +6,5 @@ mod snapshot;
 mod standard_rpc;
 mod state_injector;
 mod time_machine;
+mod txpool;
 mod utils;

crates/anvil-polkadot/tests/it/txpool.rs

@@ -0,0 +1,140 @@
+use crate::utils::{unwrap_response, TestNode};
+use alloy_primitives::{Address, B256, U256};
+use alloy_rpc_types::{txpool::TxpoolStatus, TransactionRequest};
+use anvil_core::eth::EthRequest;
+use anvil_polkadot::{
+    api_server::revive_conversions::ReviveAddress,
+    config::{AnvilNodeConfig, SubstrateNodeConfig},
+};
+use polkadot_sdk::pallet_revive::evm::Account;
+
+#[tokio::test(flavor = "multi_thread")]
+async fn test_txpool_status() {
+    let anvil_node_config = AnvilNodeConfig::test_config();
+    let substrate_node_config = SubstrateNodeConfig::new(&anvil_node_config);
+    let mut node = TestNode::new(anvil_node_config, substrate_node_config).await.unwrap();
+
+    unwrap_response::<()>(node.eth_rpc(EthRequest::SetAutomine(true)).await.unwrap()).unwrap();
+
+    let alith = Account::from(subxt_signer::eth::dev::alith());
+    let alith_addr = Address::from(ReviveAddress::new(alith.address()));
+    let recipient_addr = Address::repeat_byte(0x42);
+
+    let balance = node.get_balance(alith.address(), None).await;
+    assert!(balance > U256::ZERO, "Alith should have balance");
+
+    // Check initial pool status
+    let status: TxpoolStatus =
+        unwrap_response(node.eth_rpc(EthRequest::TxPoolStatus(())).await.unwrap()).unwrap();
+    assert_eq!(status.pending, 0, "Pool should start empty");
+    assert_eq!(status.queued, 0, "Pool should start empty");
+
+    // Disable automine so transactions stay in pool
+    unwrap_response::<()>(node.eth_rpc(EthRequest::SetAutomine(false)).await.unwrap()).unwrap();
+
+    // Send 3 transactions
+    for i in 0..3 {
+        let tx = TransactionRequest::default()
+            .from(alith_addr)
+            .to(recipient_addr)
+            .value(U256::from(1000 * (i + 1)))
+            .nonce(i);
+        node.send_transaction(tx, None).await.unwrap();
+    }
+
+    // Verify pool has 3 pending transactions
+    let status: TxpoolStatus =
+        unwrap_response(node.eth_rpc(EthRequest::TxPoolStatus(())).await.unwrap()).unwrap();
+    assert_eq!(status.pending, 3, "Pool should have 3 pending transactions");
+    assert_eq!(status.queued, 0, "Pool should have 0 queued transactions");
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn test_drop_transaction() {
+    let anvil_node_config = AnvilNodeConfig::test_config();
+    let substrate_node_config = SubstrateNodeConfig::new(&anvil_node_config);
+    let mut node = TestNode::new(anvil_node_config, substrate_node_config).await.unwrap();
+
+    unwrap_response::<()>(node.eth_rpc(EthRequest::SetAutomine(true)).await.unwrap()).unwrap();
+
+    let alith = Account::from(subxt_signer::eth::dev::alith());
+    let alith_addr = Address::from(ReviveAddress::new(alith.address()));
+    let recipient_addr = Address::repeat_byte(0x42);
+
+    let balance = node.get_balance(alith.address(), None).await;
+    assert!(balance > U256::ZERO, "Alith should have balance");
+
+    unwrap_response::<()>(node.eth_rpc(EthRequest::SetAutomine(false)).await.unwrap()).unwrap();
+
+    // Send 2 transactions with nonce 0 and 1
+    let tx1 = TransactionRequest::default()
+        .from(alith_addr)
+        .to(recipient_addr)
+        .value(U256::from(1000))
+        .nonce(0);
+    node.send_transaction(tx1, None).await.unwrap();
+
+    let tx2 = TransactionRequest::default()
+        .from(alith_addr)
+        .to(recipient_addr)
+        .value(U256::from(2000))
+        .nonce(1);
+    let tx2_hash = node.send_transaction(tx2, None).await.unwrap();
+
+    // Verify pool has 2 pending transactions
+    let status: TxpoolStatus =
+        unwrap_response(node.eth_rpc(EthRequest::TxPoolStatus(())).await.unwrap()).unwrap();
+    assert_eq!(status.pending, 2, "Pool should have 2 pending transactions");
+
+    // Drop transaction with nonce 1
+    let tx2_hash_b256 = B256::from_slice(tx2_hash.0.as_ref());
+    unwrap_response::<()>(node.eth_rpc(EthRequest::DropTransaction(tx2_hash_b256)).await.unwrap())
+        .unwrap();
+
+    // Verify only transaction with nonce 0 remains
+    let status: TxpoolStatus =
+        unwrap_response(node.eth_rpc(EthRequest::TxPoolStatus(())).await.unwrap()).unwrap();
+    assert_eq!(status.pending, 1, "Pool should have 1 pending transaction after drop");
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn test_drop_all_transactions() {
+    let anvil_node_config = AnvilNodeConfig::test_config();
+    let substrate_node_config = SubstrateNodeConfig::new(&anvil_node_config);
+    let mut node = TestNode::new(anvil_node_config, substrate_node_config).await.unwrap();
+
+    unwrap_response::<()>(node.eth_rpc(EthRequest::SetAutomine(true)).await.unwrap()).unwrap();
+
+    let alith = Account::from(subxt_signer::eth::dev::alith());
+    let alith_addr = Address::from(ReviveAddress::new(alith.address()));
+    let recipient_addr = Address::repeat_byte(0x42);
+
+    let balance = node.get_balance(alith.address(), None).await;
+    assert!(balance > U256::ZERO, "Alith should have balance");
+
+    unwrap_response::<()>(node.eth_rpc(EthRequest::SetAutomine(false)).await.unwrap()).unwrap();
+
+    // Send 3 transactions
+    for i in 0..3 {
+        let tx = TransactionRequest::default()
+            .from(alith_addr)
+            .to(recipient_addr)
+            .value(U256::from(1000 * (i + 1)))
+            .nonce(i);
+        node.send_transaction(tx, None).await.unwrap();
+    }
+
+    // Verify pool has 3 pending transactions
+    let status: TxpoolStatus =
+        unwrap_response(node.eth_rpc(EthRequest::TxPoolStatus(())).await.unwrap()).unwrap();
+    assert_eq!(status.pending, 3, "Pool should have 3 pending transactions");
+
+    // Drop all transactions
+    unwrap_response::<()>(node.eth_rpc(EthRequest::DropAllTransactions()).await.unwrap()).unwrap();
+
+    // Verify pool is empty
+    let status: TxpoolStatus =
+        unwrap_response(node.eth_rpc(EthRequest::TxPoolStatus(())).await.unwrap()).unwrap();
+    assert_eq!(status.pending, 0, "Pool should be empty after dropping all");
+    assert_eq!(status.queued, 0, "Pool should be empty after dropping all");
+}