feat: added TransactionStream to subscribe to pending transactions (#10482)

* added new method to subscribe to pending tx

* wip

* clippy -.-

* clippy -.-

* unbounded channel

* wip

* wip test

* touchup

* check

---------

Co-authored-by: Matthias Seitz <[email protected]>

Author: Soubhik-10

crates/anvil/src/eth/api.rs

@@ -89,10 +89,14 @@ use foundry_evm::{
         primitives::BlockEnv,
     },
 };
-use futures::channel::{mpsc::Receiver, oneshot};
+use futures::{
+    channel::{mpsc::Receiver, oneshot},
+    StreamExt,
+};
 use parking_lot::RwLock;
 use revm::primitives::Bytecode;
 use std::{future::Future, sync::Arc, time::Duration};
+use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver};
 
 /// The client version: `anvil/v{major}.{minor}.{patch}`
 pub const CLIENT_VERSION: &str = concat!("anvil/v", env!("CARGO_PKG_VERSION"));
@@ -2863,6 +2867,26 @@ impl EthApi {
         self.pool.add_ready_listener()
     }
 
+    /// Returns a listener for pending transactions, yielding full transactions
+    pub fn full_pending_transactions(&self) -> UnboundedReceiver<AnyRpcTransaction> {
+        let (tx, rx) = unbounded_channel();
+        let mut hashes = self.new_ready_transactions();
+
+        let this = self.clone();
+
+        tokio::spawn(async move {
+            while let Some(hash) = hashes.next().await {
+                if let Ok(Some(txn)) = this.transaction_by_hash(hash).await {
+                    if tx.send(txn).is_err() {
+                        break;
+                    }
+                }
+            }
+        });
+
+        rx
+    }
+
     /// Returns a new accessor for certain storage elements
     pub fn storage_info(&self) -> StorageInfo {
         StorageInfo::new(Arc::clone(&self.backend))

crates/anvil/src/pubsub.rs

@@ -2,6 +2,7 @@ use crate::{
     eth::{backend::notifications::NewBlockNotifications, error::to_rpc_result},
     StorageInfo,
 };
+use alloy_network::AnyRpcTransaction;
 use alloy_primitives::{TxHash, B256};
 use alloy_rpc_types::{pubsub::SubscriptionResult, FilteredParams, Log, Transaction};
 use anvil_core::eth::{block::Block, subscription::SubscriptionId, transaction::TypedReceipt};
@@ -13,6 +14,7 @@ use std::{
     pin::Pin,
     task::{Context, Poll},
 };
+use tokio::sync::mpsc::UnboundedReceiver;
 
 /// Listens for new blocks and matching logs emitted in that block
 #[derive(Debug)]
@@ -86,6 +88,7 @@ pub enum EthSubscription {
     Logs(Box<LogsSubscription>),
     Header(NewBlockNotifications, StorageInfo, SubscriptionId),
     PendingTransactions(Receiver<TxHash>, SubscriptionId),
+    FullPendingTransactions(UnboundedReceiver<AnyRpcTransaction>, SubscriptionId),
 }
 
 impl EthSubscription {
@@ -120,6 +123,13 @@ impl EthSubscription {
                     });
                 Poll::Ready(res)
             }
+            Self::FullPendingTransactions(tx, id) => {
+                let res = ready!(tx.poll_recv(cx)).map(to_rpc_result).map(|result| {
+                    let params = EthSubscriptionParams { subscription: id.clone(), result };
+                    EthSubscriptionResponse::new(params)
+                });
+                Poll::Ready(res)
+            }
         }
     }
 }

crates/anvil/src/server/handler.rs

@@ -58,20 +58,22 @@ impl PubSubEthRpcHandler {
                 let canceled = cx.remove_subscription(&id).is_some();
                 ResponseResult::Success(canceled.into())
             }
-            EthPubSub::EthSubscribe(kind, params) => {
-                let filter = match *params {
+            EthPubSub::EthSubscribe(kind, raw_params) => {
+                let filter = match &*raw_params {
                     Params::None => None,
-                    Params::Logs(filter) => Some(*filter),
-                    Params::Bool(_) => {
-                        return ResponseResult::Error(RpcError::invalid_params(
-                            "Expected params for logs subscription",
-                        ))
-                    }
+                    Params::Logs(filter) => Some(filter.clone()),
+                    Params::Bool(_) => None,
                 };
-                let params = FilteredParams::new(filter);
+                let params = FilteredParams::new(filter.map(|b| *b));
 
                 let subscription = match kind {
                     SubscriptionKind::Logs => {
+                        if raw_params.is_bool() {
+                            return ResponseResult::Error(RpcError::invalid_params(
+                                "Expected params for logs subscription",
+                            ))
+                        }
+
                         trace!(target: "rpc::ws", "received logs subscription {:?}", params);
                         let blocks = self.api.new_block_notifications();
                         let storage = self.api.storage_info();
@@ -91,10 +93,23 @@ impl PubSubEthRpcHandler {
                     }
                     SubscriptionKind::NewPendingTransactions => {
                         trace!(target: "rpc::ws", "received pending transactions subscription");
-                        EthSubscription::PendingTransactions(
-                            self.api.new_ready_transactions(),
-                            id.clone(),
-                        )
+                        match *raw_params {
+                            Params::Bool(true) => EthSubscription::FullPendingTransactions(
+                                self.api.full_pending_transactions(),
+                                id.clone(),
+                            ),
+                            Params::Bool(false) | Params::None => {
+                                EthSubscription::PendingTransactions(
+                                    self.api.new_ready_transactions(),
+                                    id.clone(),
+                                )
+                            }
+                            _ => {
+                                return ResponseResult::Error(RpcError::invalid_params(
+                                    "Expected boolean parameter for newPendingTransactions",
+                                ))
+                            }
+                        }
                     }
                     SubscriptionKind::Syncing => {
                         return RpcError::internal_error_with("Not implemented").into()

crates/anvil/tests/it/transaction.rs

@@ -4,7 +4,7 @@ use crate::{
 };
 use alloy_network::{EthereumWallet, TransactionBuilder, TransactionResponse};
 use alloy_primitives::{address, hex, map::B256HashSet, Address, Bytes, FixedBytes, U256};
-use alloy_provider::Provider;
+use alloy_provider::{Provider, WsConnect};
 use alloy_rpc_types::{
     state::{AccountOverride, StateOverride},
     AccessList, AccessListItem, BlockId, BlockNumberOrTag, BlockTransactions, TransactionRequest,
@@ -715,6 +715,34 @@ async fn can_get_pending_transaction() {
     assert_eq!(mined.tx_hash(), pending.unwrap().unwrap().tx_hash());
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn can_listen_full_pending_transaction() {
+    let (api, handle) = spawn(NodeConfig::test()).await;
+
+    // Disable auto-mining so transactions remain pending
+    api.anvil_set_auto_mine(false).await.unwrap();
+
+    let provider = alloy_provider::ProviderBuilder::new()
+        .on_ws(WsConnect::new(handle.ws_endpoint()))
+        .await
+        .unwrap();
+
+    // Subscribe to full pending transactions
+    let sub = provider.subscribe_full_pending_transactions().await;
+    tokio::time::sleep(Duration::from_millis(1000)).await;
+    let mut stream = sub.expect("Failed to subscribe to pending tx").into_stream().take(5);
+
+    let from = handle.dev_wallets().next().unwrap().address();
+    let tx = TransactionRequest::default().from(from).value(U256::from(1337)).to(Address::random());
+
+    let tx = provider.send_transaction(tx).await.unwrap();
+
+    // Wait for the subscription to yield a transaction
+    let received = stream.next().await.expect("Failed to receive pending tx");
+
+    assert_eq!(received.tx_hash(), *tx.tx_hash());
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn can_get_raw_transaction() {
     let (api, handle) = spawn(NodeConfig::test()).await;