reth_rpc/
txpool.rs

1use core::fmt;
2use std::collections::BTreeMap;
3
4use alloy_consensus::Transaction;
5use alloy_primitives::Address;
6use alloy_rpc_types_txpool::{
7    TxpoolContent, TxpoolContentFrom, TxpoolInspect, TxpoolInspectSummary, TxpoolStatus,
8};
9use jsonrpsee::core::RpcResult;
10use reth_primitives_traits::NodePrimitives;
11use reth_rpc_api::TxPoolApiServer;
12use reth_rpc_convert::RpcConvert;
13use reth_rpc_eth_api::RpcTransaction;
14use reth_transaction_pool::{
15    AllPoolTransactions, PoolConsensusTx, PoolTransaction, TransactionPool,
16};
17use tracing::trace;
18
19/// `txpool` API implementation.
20///
21/// This type provides the functionality for handling `txpool` related requests.
22#[derive(Clone)]
23pub struct TxPoolApi<Pool, Eth> {
24    /// An interface to interact with the pool
25    pool: Pool,
26    converter: Eth,
27}
28
29impl<Pool, Eth> TxPoolApi<Pool, Eth> {
30    /// Creates a new instance of `TxpoolApi`.
31    pub const fn new(pool: Pool, converter: Eth) -> Self {
32        Self { pool, converter }
33    }
34}
35
36impl<Pool, Eth> TxPoolApi<Pool, Eth>
37where
38    Pool: TransactionPool<Transaction: PoolTransaction<Consensus: Transaction>> + 'static,
39    Eth: RpcConvert<Primitives: NodePrimitives<SignedTx = PoolConsensusTx<Pool>>>,
40{
41    fn content(&self) -> Result<TxpoolContent<RpcTransaction<Eth::Network>>, Eth::Error> {
42        let AllPoolTransactions { pending, queued } = self.pool.all_transactions();
43
44        let mut content = TxpoolContent::default();
45        for tx in pending {
46            let sender = tx.transaction.sender();
47            self.insert_by_nonce(&tx.transaction, content.pending.entry(sender).or_default())?;
48        }
49        for tx in queued {
50            let sender = tx.transaction.sender();
51            self.insert_by_nonce(&tx.transaction, content.queued.entry(sender).or_default())?;
52        }
53
54        Ok(content)
55    }
56
57    fn content_from(
58        &self,
59        from: Address,
60    ) -> Result<TxpoolContentFrom<RpcTransaction<Eth::Network>>, Eth::Error> {
61        let mut content = TxpoolContentFrom::default();
62        // one snapshot for both sides, so a transaction moving between sub-pools meanwhile is
63        // reported on exactly one of them
64        let AllPoolTransactions { pending, queued } = self.pool.all_transactions_by_sender(from);
65        for tx in pending {
66            self.insert_by_nonce(&tx.transaction, &mut content.pending)?;
67        }
68        for tx in queued {
69            self.insert_by_nonce(&tx.transaction, &mut content.queued)?;
70        }
71
72        Ok(content)
73    }
74
75    /// Converts the pool transaction and inserts it into the given map, keyed by its nonce.
76    #[inline]
77    fn insert_by_nonce(
78        &self,
79        tx: &Pool::Transaction,
80        txs: &mut BTreeMap<String, RpcTransaction<Eth::Network>>,
81    ) -> Result<(), Eth::Error> {
82        txs.insert(tx.nonce().to_string(), self.converter.fill_pending(tx.clone_into_consensus())?);
83
84        Ok(())
85    }
86}
87
88impl<Pool, Eth> TxPoolApiServer<RpcTransaction<Eth::Network>> for TxPoolApi<Pool, Eth>
89where
90    Pool: TransactionPool<Transaction: PoolTransaction<Consensus: Transaction>> + 'static,
91    Eth: RpcConvert<Primitives: NodePrimitives<SignedTx = PoolConsensusTx<Pool>>>,
92{
93    /// Returns the number of transactions currently pending for inclusion in the next block(s), as
94    /// well as the ones that are being scheduled for future execution only.
95    /// Ref: [Here](https://geth.ethereum.org/docs/rpc/ns-txpool#txpool_status)
96    ///
97    /// Handler for `txpool_status`
98    async fn txpool_status(&self) -> RpcResult<TxpoolStatus> {
99        trace!(target: "rpc::eth", "Serving txpool_status");
100        let (pending, queued) = self.pool.pending_and_queued_txn_count();
101        Ok(TxpoolStatus { pending: pending as u64, queued: queued as u64 })
102    }
103
104    /// Returns a summary of all the transactions currently pending for inclusion in the next
105    /// block(s), as well as the ones that are being scheduled for future execution only.
106    ///
107    /// See [here](https://geth.ethereum.org/docs/rpc/ns-txpool#txpool_inspect) for more details
108    ///
109    /// Handler for `txpool_inspect`
110    async fn txpool_inspect(&self) -> RpcResult<TxpoolInspect> {
111        trace!(target: "rpc::eth", "Serving txpool_inspect");
112
113        #[inline]
114        fn insert<T: PoolTransaction<Consensus: Transaction>>(
115            tx: &T,
116            inspect: &mut BTreeMap<Address, BTreeMap<String, TxpoolInspectSummary>>,
117        ) {
118            let entry = inspect.entry(tx.sender()).or_default();
119            let tx = tx.clone_into_consensus();
120            entry.insert(tx.nonce().to_string(), tx.into_inner().into());
121        }
122
123        let AllPoolTransactions { pending, queued } = self.pool.all_transactions();
124
125        Ok(TxpoolInspect {
126            pending: pending.iter().fold(Default::default(), |mut acc, tx| {
127                insert(&tx.transaction, &mut acc);
128                acc
129            }),
130            queued: queued.iter().fold(Default::default(), |mut acc, tx| {
131                insert(&tx.transaction, &mut acc);
132                acc
133            }),
134        })
135    }
136
137    /// Retrieves the transactions contained within the txpool, returning pending as well as queued
138    /// transactions of this address, grouped by nonce.
139    ///
140    /// See [here](https://geth.ethereum.org/docs/rpc/ns-txpool#txpool_contentFrom) for more details
141    /// Handler for `txpool_contentFrom`
142    async fn txpool_content_from(
143        &self,
144        from: Address,
145    ) -> RpcResult<TxpoolContentFrom<RpcTransaction<Eth::Network>>> {
146        trace!(target: "rpc::eth", ?from, "Serving txpool_contentFrom");
147        self.content_from(from).map_err(Into::into)
148    }
149
150    /// Returns the details of all transactions currently pending for inclusion in the next
151    /// block(s), as well as the ones that are being scheduled for future execution only.
152    ///
153    /// See [here](https://geth.ethereum.org/docs/rpc/ns-txpool#txpool_content) for more details
154    /// Handler for `txpool_content`
155    async fn txpool_content(&self) -> RpcResult<TxpoolContent<RpcTransaction<Eth::Network>>> {
156        trace!(target: "rpc::eth", "Serving txpool_content");
157        self.content().map_err(Into::into)
158    }
159}
160
161impl<Pool, Eth> fmt::Debug for TxPoolApi<Pool, Eth> {
162    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
163        f.debug_struct("TxpoolApi").finish_non_exhaustive()
164    }
165}
166
167#[cfg(test)]
168mod tests {
169    use super::*;
170    use crate::eth::helpers::types::EthRpcConverter;
171    use reth_chainspec::MAINNET;
172    use reth_rpc_eth_types::receipt::EthReceiptConverter;
173    use reth_transaction_pool::{
174        test_utils::{testing_pool, MockTransaction},
175        TransactionOrigin,
176    };
177
178    #[tokio::test]
179    async fn content_from_matches_content() {
180        let senders = [Address::with_last_byte(1), Address::with_last_byte(2)];
181
182        let pool = testing_pool();
183        for sender in senders {
184            // nonces 0 and 1 end up in the pending sub-pool, the gapped nonce 9 in a parked one
185            for nonce in [0, 1, 9] {
186                let tx = MockTransaction::legacy()
187                    .with_sender(sender)
188                    .with_nonce(nonce)
189                    .with_gas_price(100);
190                pool.add_transaction(TransactionOrigin::External, tx).await.unwrap();
191            }
192        }
193
194        let api =
195            TxPoolApi::new(pool, EthRpcConverter::new(EthReceiptConverter::new(MAINNET.clone())));
196        let mut content = api.content().unwrap();
197        // guards the fixture: both sub-pools must be populated for the comparison to mean anything
198        assert_eq!((content.pending.len(), content.queued.len()), (senders.len(), senders.len()));
199
200        // the unknown sender must yield the same empty result as the whole-pool path
201        for sender in senders.into_iter().chain([Address::with_last_byte(3)]) {
202            assert_eq!(api.content_from(sender).unwrap(), content.remove_from(&sender));
203        }
204    }
205}
reth_rpc/txpool.rs

reth_rpc/
txpool.rs
