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

reth_rpc/
txpool.rs
