fix: add test for truncating parked with large transactions (#6366)

Cargo.lock

@@ -6973,6 +6973,7 @@ dependencies = [
  "reth-provider",
  "reth-revm",
  "reth-tasks",
+ "reth-tracing",
  "revm",
  "schnellru",
  "serde",

crates/primitives/src/transaction/mod.rs

@@ -500,7 +500,7 @@ impl Transaction {
 
     /// Calculates a heuristic for the in-memory size of the [Transaction].
     #[inline]
-    fn size(&self) -> usize {
+    pub fn size(&self) -> usize {
         match self {
             Transaction::Legacy(tx) => tx.size(),
             Transaction::Eip2930(tx) => tx.size(),

crates/transaction-pool/Cargo.toml

@@ -57,6 +57,7 @@ proptest = { workspace = true, optional = true }
 [dev-dependencies]
 reth-primitives = { workspace = true, features = ["arbitrary"] }
 reth-provider = { workspace = true, features = ["test-utils"] }
+reth-tracing.workspace = true
 paste = "1.0"
 rand = "0.8"
 proptest.workspace = true

crates/transaction-pool/src/pool/parked.rs

@@ -149,39 +149,28 @@ impl<T: ParkedOrd> ParkedPool<T> {
         &mut self,
         limit: SubPoolLimit,
     ) -> Vec<Arc<ValidPoolTransaction<T::Transaction>>> {
-        if self.len() <= limit.max_txs {
+        if !limit.is_exceeded(self.len(), self.size()) {
             // if we are below the limits, we don't need to drop anything
             return Vec::new()
         }
 
         let mut removed = Vec::new();
         let mut sender_ids = self.get_senders_by_submission_id();
-        let queued = self.len();
-        let mut drop = queued - limit.max_txs;
 
-        while drop > 0 && !sender_ids.is_empty() {
+        while limit.is_exceeded(self.len(), self.size()) && !sender_ids.is_empty() {
             // SAFETY: This will not panic due to `!addresses.is_empty()`
             let sender_id = sender_ids.pop().unwrap().sender_id;
-            let mut list = self.get_txs_by_sender(sender_id);
+            let list = self.get_txs_by_sender(sender_id);
 
-            // Drop all transactions if they are less than the overflow
-            if list.len() <= drop {
-                for txid in &list {
-                    if let Some(tx) = self.remove_transaction(txid) {
-                        removed.push(tx);
-                    }
-                }
-                drop -= list.len();
-                continue
-            }
-
-            // Otherwise drop only last few transactions
-            // SAFETY: This will not panic because `list.len() > drop`
-            for txid in list.split_off(drop) {
+            // Drop transactions from this sender until the pool is under limits
+            for txid in list.into_iter().rev() {
                 if let Some(tx) = self.remove_transaction(&txid) {
                     removed.push(tx);
                 }
-                drop -= 1;
+
+                if !limit.is_exceeded(self.len(), self.size()) {
+                    break
+                }
             }
         }
 
@@ -603,6 +592,35 @@ mod tests {
         assert_eq!(parked, expected_parked);
     }
 
+    #[test]
+    fn test_truncate_parked_with_large_tx() {
+        let mut f = MockTransactionFactory::default();
+        let mut pool = ParkedPool::<BasefeeOrd<_>>::default();
+        let default_limits = SubPoolLimit::default();
+
+        // create a chain of transactions by sender A
+        // make sure they are all one over half the limit
+        let a_sender = address!("000000000000000000000000000000000000000a");
+
+        // 2 txs, that should put the pool over the size limit but not max txs
+        let a_txs = MockTransactionSet::dependent(a_sender, 0, 2, TxType::EIP1559)
+            .into_iter()
+            .map(|mut tx| {
+                tx.set_size(default_limits.max_size / 2 + 1);
+                tx
+            })
+            .collect::<Vec<_>>();
+
+        // add all the transactions to the pool
+        for tx in a_txs {
+            pool.add_transaction(f.validated_arc(tx));
+        }
+
+        // truncate the pool, it should remove at least one transaction
+        let removed = pool.truncate_pool(default_limits);
+        assert_eq!(removed.len(), 1);
+    }
+
     #[test]
     fn test_senders_by_submission_id() {
         // this test ensures that we evict from the pending pool by sender

crates/transaction-pool/src/pool/txpool.rs

@@ -32,6 +32,7 @@ use std::{
     ops::Bound::{Excluded, Unbounded},
     sync::Arc,
 };
+use tracing::trace;
 
 #[cfg_attr(doc, aquamarine::aquamarine)]
 /// A pool that manages transactions.
@@ -785,9 +786,26 @@ impl<T: TransactionOrdering> TxPool<T> {
                         .$limit
                         .is_exceeded($this.$pool.len(), $this.$pool.size())
                     {
+                        trace!(
+                            "discarding transactions from {}, limit: {:?}, curr size: {}, curr len: {}",
+                            stringify!($pool),
+                            $this.config.$limit,
+                            $this.$pool.size(),
+                            $this.$pool.len(),
+                        );
+
                         // 1. first remove the worst transaction from the subpool
                         let removed_from_subpool = $this.$pool.truncate_pool($this.config.$limit.clone());
 
+                        trace!(
+                            "removed {} transactions from {}, limit: {:?}, curr size: {}, curr len: {}",
+                            removed_from_subpool.len(),
+                            stringify!($pool),
+                            $this.config.$limit,
+                            $this.$pool.size(),
+                            $this.$pool.len()
+                        );
+
                         // 2. remove all transactions from the total set
                         for tx in removed_from_subpool {
                             $this.all_transactions.remove_transaction(tx.id());
@@ -1821,9 +1839,11 @@ impl SenderInfo {
 
 #[cfg(test)]
 mod tests {
+    use reth_primitives::{address, TxType};
+
     use super::*;
     use crate::{
-        test_utils::{MockOrdering, MockTransaction, MockTransactionFactory},
+        test_utils::{MockOrdering, MockTransaction, MockTransactionFactory, MockTransactionSet},
         traits::TransactionOrigin,
         SubPoolLimit,
     };
@@ -2679,6 +2699,86 @@ mod tests {
         assert_eq!(pool.pending_pool.len(), 1);
     }
 
+    #[test]
+    fn discard_with_large_blob_txs() {
+        // init tracing
+        reth_tracing::init_test_tracing();
+
+        // this test adds large txs to the parked pool, then attempting to discard worst
+        let mut f = MockTransactionFactory::default();
+        let mut pool = TxPool::new(MockOrdering::default(), Default::default());
+        let default_limits = pool.config.blob_limit;
+
+        // create a chain of transactions by sender A
+        // make sure they are all one over half the limit
+        let a_sender = address!("000000000000000000000000000000000000000a");
+
+        // set the base fee of the pool
+        let mut block_info = pool.block_info();
+        block_info.pending_blob_fee = Some(100);
+        block_info.pending_basefee = 100;
+
+        // update
+        pool.set_block_info(block_info);
+
+        // 2 txs, that should put the pool over the size limit but not max txs
+        let a_txs = MockTransactionSet::dependent(a_sender, 0, 2, TxType::EIP4844)
+            .into_iter()
+            .map(|mut tx| {
+                tx.set_size(default_limits.max_size / 2 + 1);
+                tx.set_max_fee((block_info.pending_basefee - 1).into());
+                tx
+            })
+            .collect::<Vec<_>>();
+
+        // add all the transactions to the parked pool
+        for tx in a_txs {
+            pool.add_transaction(f.validated(tx), U256::from(1_000), 0).unwrap();
+        }
+
+        // truncate the pool, it should remove at least one transaction
+        let removed = pool.discard_worst();
+        assert_eq!(removed.len(), 1);
+    }
+
+    #[test]
+    fn discard_with_parked_large_txs() {
+        // init tracing
+        reth_tracing::init_test_tracing();
+
+        // this test adds large txs to the parked pool, then attempting to discard worst
+        let mut f = MockTransactionFactory::default();
+        let mut pool = TxPool::new(MockOrdering::default(), Default::default());
+        let default_limits = pool.config.queued_limit;
+
+        // create a chain of transactions by sender A
+        // make sure they are all one over half the limit
+        let a_sender = address!("000000000000000000000000000000000000000a");
+
+        // set the base fee of the pool
+        let pool_base_fee = 100;
+        pool.update_basefee(pool_base_fee);
+
+        // 2 txs, that should put the pool over the size limit but not max txs
+        let a_txs = MockTransactionSet::dependent(a_sender, 0, 2, TxType::EIP1559)
+            .into_iter()
+            .map(|mut tx| {
+                tx.set_size(default_limits.max_size / 2 + 1);
+                tx.set_max_fee((pool_base_fee - 1).into());
+                tx
+            })
+            .collect::<Vec<_>>();
+
+        // add all the transactions to the parked pool
+        for tx in a_txs {
+            pool.add_transaction(f.validated(tx), U256::from(1_000), 0).unwrap();
+        }
+
+        // truncate the pool, it should remove at least one transaction
+        let removed = pool.discard_worst();
+        assert_eq!(removed.len(), 1);
+    }
+
     #[test]
     fn discard_at_capacity() {
         let mut f = MockTransactionFactory::default();

crates/transaction-pool/src/test_utils/mock.rs

@@ -21,7 +21,7 @@ use reth_primitives::{
     TxEip1559, TxEip2930, TxEip4844, TxHash, TxLegacy, TxType, B256, EIP1559_TX_TYPE_ID,
     EIP2930_TX_TYPE_ID, EIP4844_TX_TYPE_ID, LEGACY_TX_TYPE_ID, U256,
 };
-use std::{ops::Range, sync::Arc, time::Instant};
+use std::{ops::Range, sync::Arc, time::Instant, vec::IntoIter};
 
 /// A transaction pool implementation using [MockOrdering] for transaction ordering.
 ///
@@ -167,6 +167,8 @@ pub enum MockTransaction {
         value: U256,
         /// The transaction input data.
         input: Bytes,
+        /// The size of the transaction, returned in the implementation of [PoolTransaction].
+        size: usize,
     },
     /// EIP-1559 transaction type.
     Eip1559 {
@@ -190,6 +192,8 @@ pub enum MockTransaction {
         accesslist: AccessList,
         /// The transaction input data.
         input: Bytes,
+        /// The size of the transaction, returned in the implementation of [PoolTransaction].
+        size: usize,
     },
     /// EIP-4844 transaction type.
     Eip4844 {
@@ -217,6 +221,8 @@ pub enum MockTransaction {
         input: Bytes,
         /// The sidecar information for the transaction.
         sidecar: BlobTransactionSidecar,
+        /// The size of the transaction, returned in the implementation of [PoolTransaction].
+        size: usize,
     },
     /// EIP-2930 transaction type.
     Eip2930 {
@@ -238,6 +244,8 @@ pub enum MockTransaction {
         gas_price: u128,
         /// The access list associated with the transaction.
         accesslist: AccessList,
+        /// The size of the transaction, returned in the implementation of [PoolTransaction].
+        size: usize,
     },
     #[cfg(feature = "optimism")]
     /// Deposit transaction type (Optimism feature).
@@ -253,7 +261,8 @@ impl MockTransaction {
         sender => Address;
         gas_limit => u64;
         value => U256;
-        input => Bytes
+        input => Bytes;
+        size => usize
     }
 
     /// Returns a new legacy transaction with random address and hash and empty values
@@ -267,6 +276,7 @@ impl MockTransaction {
             to: TransactionKind::Call(Address::random()),
             value: Default::default(),
             input: Default::default(),
+            size: Default::default(),
         }
     }
 
@@ -283,6 +293,7 @@ impl MockTransaction {
             value: Default::default(),
             input: Bytes::new(),
             accesslist: Default::default(),
+            size: Default::default(),
         }
     }
 
@@ -301,6 +312,7 @@ impl MockTransaction {
             input: Bytes::new(),
             accesslist: Default::default(),
             sidecar: Default::default(),
+            size: Default::default(),
         }
     }
 
@@ -316,6 +328,7 @@ impl MockTransaction {
             value: Default::default(),
             gas_price: 0,
             accesslist: Default::default(),
+            size: Default::default(),
         }
     }
 
@@ -773,7 +786,14 @@ impl PoolTransaction for MockTransaction {
 
     /// Returns the size of the transaction.
     fn size(&self) -> usize {
-        0
+        match self {
+            MockTransaction::Legacy { size, .. } |
+            MockTransaction::Eip1559 { size, .. } |
+            MockTransaction::Eip4844 { size, .. } |
+            MockTransaction::Eip2930 { size, .. } => *size,
+            #[cfg(feature = "optimism")]
+            MockTransaction::Deposit(_) => 0,
+        }
     }
 
     /// Returns the transaction type as a byte identifier.
@@ -810,6 +830,7 @@ impl FromRecoveredTransaction for MockTransaction {
         let sender = tx.signer();
         let transaction = tx.into_signed();
         let hash = transaction.hash();
+        let size = transaction.size();
         match transaction.transaction {
             Transaction::Legacy(TxLegacy {
                 chain_id: _,
@@ -828,6 +849,7 @@ impl FromRecoveredTransaction for MockTransaction {
                 to,
                 value: value.into(),
                 input,
+                size,
             },
             Transaction::Eip1559(TxEip1559 {
                 chain_id: _,
@@ -850,6 +872,7 @@ impl FromRecoveredTransaction for MockTransaction {
                 value: value.into(),
                 input,
                 accesslist: access_list,
+                size,
             },
             Transaction::Eip4844(TxEip4844 {
                 chain_id: _,
@@ -876,6 +899,7 @@ impl FromRecoveredTransaction for MockTransaction {
                 input,
                 accesslist: access_list,
                 sidecar: BlobTransactionSidecar::default(),
+                size,
             },
             Transaction::Eip2930(TxEip2930 {
                 chain_id: _,
@@ -896,6 +920,7 @@ impl FromRecoveredTransaction for MockTransaction {
                 value: value.into(),
                 input,
                 accesslist: access_list,
+                size,
             },
             #[cfg(feature = "optimism")]
             Transaction::Deposit(TxDeposit {
@@ -953,6 +978,7 @@ impl From<MockTransaction> for Transaction {
                 to,
                 value,
                 input,
+                size: _,
             } => Self::Legacy(TxLegacy {
                 chain_id: Some(1),
                 nonce,
@@ -973,6 +999,7 @@ impl From<MockTransaction> for Transaction {
                 value,
                 accesslist,
                 input,
+                size: _,
             } => Self::Eip1559(TxEip1559 {
                 chain_id: 1,
                 nonce,
@@ -997,6 +1024,7 @@ impl From<MockTransaction> for Transaction {
                 accesslist,
                 input,
                 sidecar: _,
+                size: _,
             } => Self::Eip4844(TxEip4844 {
                 chain_id: 1,
                 nonce,
@@ -1020,6 +1048,7 @@ impl From<MockTransaction> for Transaction {
                 value,
                 gas_price,
                 accesslist,
+                size: _,
             } => Self::Eip2930(TxEip2930 {
                 chain_id: 1,
                 nonce,
@@ -1070,6 +1099,7 @@ impl proptest::arbitrary::Arbitrary for MockTransaction {
                     to: *to,
                     value: (*value).into(),
                     input: (*input).clone(),
+                    size: tx.size(),
                 },
                 Transaction::Eip1559(TxEip1559 {
                     nonce,
@@ -1092,6 +1122,7 @@ impl proptest::arbitrary::Arbitrary for MockTransaction {
                     value: (*value).into(),
                     input: (*input).clone(),
                     accesslist: (*access_list).clone(),
+                    size: tx.size(),
                 },
                 Transaction::Eip4844(TxEip4844 {
                     nonce,
@@ -1119,6 +1150,7 @@ impl proptest::arbitrary::Arbitrary for MockTransaction {
                     // only generate a sidecar if it is a 4844 tx - also for the sake of
                     // performance just use a default sidecar
                     sidecar: BlobTransactionSidecar::default(),
+                    size: tx.size(),
                 },
                 #[allow(unreachable_patterns)]
                 _ => unimplemented!(),
@@ -1338,11 +1370,11 @@ impl MockTransactionSet {
     /// The number of transactions created is determined by `tx_count`.
     pub fn dependent(sender: Address, from_nonce: u64, tx_count: usize, tx_type: TxType) -> Self {
         let mut txs = Vec::with_capacity(tx_count);
-        let mut curr_tx = MockTransaction::new_from_type(tx_type).with_nonce(from_nonce);
-        for i in 0..tx_count {
-            let _nonce = from_nonce + i as u64;
-            curr_tx = curr_tx.next().with_sender(sender);
+        let mut curr_tx =
+            MockTransaction::new_from_type(tx_type).with_nonce(from_nonce).with_sender(sender);
+        for _ in 0..tx_count {
             txs.push(curr_tx.clone());
+            curr_tx = curr_tx.next();
         }
 
         Self::new(txs)
@@ -1369,6 +1401,25 @@ impl MockTransactionSet {
     pub fn into_vec(self) -> Vec<MockTransaction> {
         self.transactions
     }
+
+    /// Returns an iterator over the contained transactions in the set
+    pub fn iter(&self) -> impl Iterator<Item = &MockTransaction> {
+        self.transactions.iter()
+    }
+
+    /// Returns a mutable iterator over the contained transactions in the set.
+    pub fn iter_mut(&mut self) -> impl Iterator<Item = &mut MockTransaction> {
+        self.transactions.iter_mut()
+    }
+}
+
+impl IntoIterator for MockTransactionSet {
+    type Item = MockTransaction;
+    type IntoIter = IntoIter<MockTransaction>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        self.transactions.into_iter()
+    }
 }
 
 #[test]