tests(mempool): add proptests

Author: 00nktk

Cargo.lock

@@ -35,3 +35,4 @@ lru = "0.12.5"
 
 [dev-dependencies]
 tracing-subscriber.workspace = true
+proptest = "1.5.0"

mempool/Cargo.toml

@@ -10,5 +10,6 @@ pub use error::MempoolError;
 pub use gas_price_calculator::GasPriceCalculatorConfig;
 pub use gas_prices::{GasPriceModel, GasPrices, GasPricesConfig, GasPricesTrait};
 pub use mempool::{Config as MempoolConfig, Mempool};
+pub use pools::EXEC_INTERVAL_MS;
 pub use pyth_price_collector::pyth_collect_symbology;
 pub use validator::{PreFlightError, PreFlightValidator};

mempool/src/lib.rs

@@ -16,7 +16,7 @@ use neon_api::NeonApi;
 
 use crate::error::MempoolError;
 use crate::pools::{ChainPool, ChainPoolConfig, NeonApiGetTxCount};
-use crate::GasPricesTrait;
+use crate::{GasPricesTrait, EXEC_INTERVAL_MS};
 
 pub type EthTxHash = B256;
 pub type GasPrice = u128;
@@ -250,6 +250,7 @@ impl<E: Execute, G: GasPricesTrait> Mempool<E, G> {
                 eviction_timeout_sec: self.config.eviction_timeout_sec,
                 tx_cache_size: self.config.tx_cache_size,
                 tx_count_cache_size: self.config.tx_count_cache_size,
+                exec_interval_ms: EXEC_INTERVAL_MS,
             };
             ChainPool::create_and_start(
                 config,

mempool/src/mempool.rs

@@ -29,7 +29,7 @@ use crate::{GasPricesTrait, MempoolError};
 
 const EXEC_RESULT_CHANNEL_SIZE: usize = 1024;
 const ONE_BLOCK_MS: u64 = 400;
-const EXEC_INTERVAL_MS: u64 = ONE_BLOCK_MS;
+pub const EXEC_INTERVAL_MS: u64 = ONE_BLOCK_MS;
 
 pub trait GetTxCountTrait: Clone + Send + Sync + 'static {
     fn get_transaction_count(
@@ -60,6 +60,7 @@ pub struct Config {
     pub eviction_timeout_sec: u64,
     pub tx_cache_size: usize,
     pub tx_count_cache_size: usize,
+    pub exec_interval_ms: u64,
 }
 
 struct HeartBeatTask {
@@ -99,6 +100,7 @@ pub struct ChainPool<E: Execute, G: GasPricesTrait, C: GetTxCountTrait> {
     eviction_timeout_sec: u64,
     tx_count_cache: Option<LruCache<Address, u64>>,
     tx_cache: Option<LruCache<EthTxHash, ()>>,
+    exec_interval_ms: u64,
 }
 
 impl<E: Execute, G: GasPricesTrait, C: GetTxCountTrait> ChainPool<E, G, C> {
@@ -128,6 +130,7 @@ impl<E: Execute, G: GasPricesTrait, C: GetTxCountTrait> ChainPool<E, G, C> {
             eviction_timeout_sec: config.eviction_timeout_sec,
             tx_count_cache: init_cache(config.tx_count_cache_size),
             tx_cache: init_cache(config.tx_cache_size),
+            exec_interval_ms: config.exec_interval_ms,
         }
     }
 
@@ -147,7 +150,12 @@ impl<E: Execute, G: GasPricesTrait, C: GetTxCountTrait> ChainPool<E, G, C> {
         let mut cmd_rx = cmd_rx;
         let (exec_result_tx, mut exec_result_rx) =
             mpsc::channel::<ExecutionResult>(EXEC_RESULT_CHANNEL_SIZE);
-        let mut exec_interval = tokio::time::interval(Duration::from_millis(EXEC_INTERVAL_MS));
+        let interval = if self.exec_interval_ms == 0 {
+            Duration::from_micros(1)
+        } else {
+            Duration::from_millis(self.exec_interval_ms)
+        };
+        let mut exec_interval = tokio::time::interval(interval);
         loop {
             tokio::select! {
                  _ = exec_interval.tick() => {
@@ -779,6 +787,7 @@ where
     Some(LruCache::new(size))
 }
 
+// TODO: move out
 #[cfg(test)]
 mod tests {
     use std::io::IsTerminal;
@@ -788,7 +797,9 @@ mod tests {
     use alloy_consensus::{SignableTransaction, TxLegacy};
     use alloy_network::TxSignerSync;
     use alloy_signer_wallet::LocalWallet;
+    use proptest::prelude::*;
     use reth_primitives::{TxKind, U256};
+    use tokio::runtime::Runtime;
     use tokio::sync::mpsc::channel;
     use tokio::time::sleep;
     use tracing_subscriber::filter::EnvFilter;
@@ -872,6 +883,7 @@ mod tests {
             eviction_timeout_sec: 60,
             tx_cache_size: 0,
             tx_count_cache_size: 0,
+            exec_interval_ms: 0,
         };
         let executor = MockExecutor::default();
         ChainPool::new(
@@ -1034,35 +1046,56 @@ mod tests {
         assert!(matches!(result, Err(MempoolError::NonceTooHigh)));
     }
 
-    #[tokio::test]
-    async fn test_random_sequence_from_basic() {
+    async fn test_sequence(nonces: Vec<u64>) {
         let sender = LocalWallet::random();
         let mut chain_pool = create_chain_pool();
-        let chain = [
-            22, 2, 5, 24, 19, 10, 6, 17, 4, 16, 11, 14, 12, 15, 0, 18, 1, 21, 8, 23, 20, 9, 3, 13,
-            7,
-        ];
-        let chain_len = chain.len();
-        for nonce in chain {
+        let len = nonces.len();
+        for nonce in nonces {
             let tx = create_req_with_addr(&sender, nonce);
             chain_pool.add_tx(tx.try_into().unwrap()).await.unwrap();
         }
 
         let txs = chain_pool.executor.txs.clone();
         let (_tx, rx) = channel(1);
         tokio::spawn(chain_pool.start(rx));
-        sleep(Duration::from_millis(
-            EXEC_INTERVAL_MS * (chain_len as u64 + 5),
-        ))
-        .await;
+        sleep(Duration::from_millis(len as u64 * 2)).await;
 
         let done = txs
             .lock()
             .unwrap()
             .iter()
             .map(|tx| tx.tx().nonce().unwrap())
             .collect::<Vec<_>>();
-        let expected = (0..25).collect::<Vec<_>>();
+        let expected = (0..len as u64).collect::<Vec<_>>();
         assert_eq!(done, expected);
     }
+
+    #[tokio::test]
+    async fn test_random_sequence_from_basic() {
+        let chain = [
+            22, 2, 5, 24, 19, 10, 6, 17, 4, 16, 11, 14, 12, 15, 0, 18, 1, 21, 8, 23, 20, 9, 3, 13,
+            7,
+        ];
+        test_sequence(chain.into()).await
+    }
+
+    #[tokio::test]
+    async fn test_simple_sequence() {
+        let chain = [0, 1];
+        test_sequence(chain.into()).await
+    }
+
+    proptest! {
+        /// Tests that all transactions from an arbitrary shuffled continuous valid sequence
+        /// get eventually queued.
+        #[test]
+        #[ignore] // Won't work with timers in chain pool
+        fn test_continuous_sequence(
+            nonces in (1..300u64)                            // Sequence of length from 1 to 300
+                .prop_map(|len| (0..len).collect::<Vec<_>>()) // of all numbers from 0 to len
+                .prop_shuffle()                               // arbitrarily shuffled
+        ) {
+            Runtime::new().unwrap().block_on(test_sequence(nonces));
+        }
+    }
 }

mempool/src/pools/chain_pool.rs

@@ -6,7 +6,7 @@ use reth_primitives::{Address, ChainId};
 
 use crate::mempool::{EthTxHash, GasPrice};
 
-pub use chain_pool::{ChainPool, Config as ChainPoolConfig, NeonApiGetTxCount};
+pub use chain_pool::{ChainPool, Config as ChainPoolConfig, NeonApiGetTxCount, EXEC_INTERVAL_MS};
 pub use sender_pool::{SenderPool, SenderPoolState};
 
 #[derive(Debug, Clone, Eq, PartialEq)]

mempool/src/pools/mod.rs

@@ -316,6 +316,10 @@ struct QueueMetrics {
 #[cfg(test)]
 #[allow(unused)] // TODO
 mod tests {
+    use std::collections::VecDeque;
+
+    use proptest::prelude::*;
+
     use super::*;
     use crate::mempool::EthTxHash;
 
@@ -733,4 +737,28 @@ mod tests {
             DoublePriorityQueue::<QueueRecord, TxNonce>::from(vec![(record3, 3)])
         );
     }
+
+    proptest! {
+        /// Tests that all transactions from an arbitrary shuffled continuous valid sequence
+        /// get eventually queued.
+        #[test]
+        fn test_continuous_sequence(
+            nonces in (1..1000u64)                            // Sequence of length from 1 to 1000
+                .prop_map(|len| (0..len).collect::<Vec<_>>()) // of all numbers from 0 to len
+                .prop_shuffle()                               // arbitrarily shuffled
+        ) {
+            let mut pool = create_sender_pool();
+            for nonce in &nonces {
+                let rec = create_record(*nonce);
+                pool.add(rec);
+            }
+            for nonce in 0..nonces.len() {
+                let rec = pool.get_for_queueing().unwrap();
+                assert_eq!(rec.nonce, nonce as u64);
+                pool.set_processing(rec.nonce);
+                pool.set_idle();
+            }
+            assert!(pool.is_empty())
+        }
+    }
 }