feat(pool): add a configurable maximum time allowed in pool

bin/rundler/src/cli/pool.rs

@@ -179,6 +179,13 @@ pub struct PoolArgs {
         default_value = "0.0"
     )]
     pub gas_limit_efficiency_reject_threshold: f32,
+
+    #[arg(
+        long = "pool.max_time_in_pool_secs",
+        name = "pool.max_time_in_pool_secs",
+        env = "POOL_MAX_TIME_IN_POOL_SECS"
+    )]
+    pub max_time_in_pool_secs: Option<u64>,
 }
 
 impl PoolArgs {
@@ -235,6 +242,7 @@ impl PoolArgs {
             drop_min_num_blocks: self.drop_min_num_blocks,
             da_gas_tracking_enabled,
             gas_limit_efficiency_reject_threshold: self.gas_limit_efficiency_reject_threshold,
+            max_time_in_pool: self.max_time_in_pool_secs.map(Duration::from_secs),
         };
 
         let mut pool_configs = vec![];

crates/pool/src/mempool/mod.rs

@@ -26,6 +26,7 @@ mod uo_pool;
 use std::{
     collections::{HashMap, HashSet},
     sync::Arc,
+    time::Duration,
 };
 
 use alloy_primitives::{Address, B256};
@@ -171,6 +172,8 @@ pub struct PoolConfig {
     /// Gas limit efficiency is defined as the ratio of the gas limit to the gas used.
     /// This applies to all the verification, call, and paymaster gas limits.
     pub gas_limit_efficiency_reject_threshold: f32,
+    /// Maximum time a UO is allowed in the pool before being dropped
+    pub max_time_in_pool: Option<Duration>,
 }
 
 /// Origin of an operation.

crates/pool/src/mempool/pool.rs

@@ -15,7 +15,7 @@ use std::{
     cmp::{self, Ordering},
     collections::{hash_map::Entry, BTreeSet, HashMap, HashSet},
     sync::Arc,
-    time::{Duration, SystemTime, UNIX_EPOCH},
+    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
 };
 
 use alloy_primitives::{Address, B256};
@@ -46,6 +46,7 @@ pub(crate) struct PoolInnerConfig {
     throttled_entity_mempool_count: u64,
     throttled_entity_live_blocks: u64,
     da_gas_tracking_enabled: bool,
+    max_time_in_pool: Option<Duration>,
 }
 
 impl From<PoolConfig> for PoolInnerConfig {
@@ -58,6 +59,7 @@ impl From<PoolConfig> for PoolInnerConfig {
             throttled_entity_mempool_count: config.throttled_entity_mempool_count,
             throttled_entity_live_blocks: config.throttled_entity_live_blocks,
             da_gas_tracking_enabled: config.da_gas_tracking_enabled,
+            max_time_in_pool: config.max_time_in_pool,
         }
     }
 }
@@ -257,6 +259,18 @@ where
                 });
                 expired.push(*hash);
                 continue;
+            } else if self
+                .config
+                .max_time_in_pool
+                .is_some_and(|m| op.elapsed_time_in_pool() > m)
+            {
+                events.push(PoolEvent::RemovedOp {
+                    op_hash: *hash,
+                    reason: OpRemovalReason::Expired {
+                        valid_until: sys_block_time.into(),
+                    },
+                });
+                expired.push(*hash);
             }
 
             if self.da_gas_oracle.is_some() && block_da_data.is_some() {
@@ -667,6 +681,7 @@ struct OrderedPoolOperation {
     po: Arc<PoolOperation>,
     submission_id: u64,
     eligible: RwLock<bool>,
+    insertion_time: Instant,
 }
 
 impl OrderedPoolOperation {
@@ -675,6 +690,7 @@ impl OrderedPoolOperation {
             po,
             submission_id,
             eligible: RwLock::new(eligible),
+            insertion_time: Instant::now(),
         }
     }
 
@@ -697,6 +713,10 @@ impl OrderedPoolOperation {
     fn set_ineligible(&self) {
         *self.eligible.write() = false;
     }
+
+    fn elapsed_time_in_pool(&self) -> Duration {
+        self.insertion_time.elapsed()
+    }
 }
 
 impl Eq for OrderedPoolOperation {}
@@ -1389,6 +1409,35 @@ mod tests {
         assert_eq!(pool.best_operations().collect::<Vec<_>>().len(), 0);
     }
 
+    #[tokio::test]
+    async fn test_max_time_in_pool() {
+        let mut conf = conf();
+        conf.max_time_in_pool = Some(Duration::from_secs(1));
+        let po = create_op(Address::random(), 0, 10);
+        let mut pool = pool_with_conf(conf.clone());
+
+        let hash = pool.add_operation(po.clone(), 0).unwrap();
+        assert!(pool.get_operation_by_hash(hash).is_some());
+        pool.do_maintenance(
+            0,
+            0.into(),
+            Some(&DAGasBlockData::default()),
+            GasFees::default(),
+            0,
+        );
+        assert!(pool.get_operation_by_hash(hash).is_some());
+
+        tokio::time::sleep(Duration::from_secs(2)).await;
+        pool.do_maintenance(
+            0,
+            0.into(),
+            Some(&DAGasBlockData::default()),
+            GasFees::default(),
+            0,
+        );
+        assert!(pool.get_operation_by_hash(hash).is_none());
+    }
+
     fn conf() -> PoolInnerConfig {
         PoolInnerConfig {
             chain_spec: ChainSpec::default(),
@@ -1398,6 +1447,7 @@ mod tests {
             throttled_entity_mempool_count: 4,
             throttled_entity_live_blocks: 10,
             da_gas_tracking_enabled: false,
+            max_time_in_pool: None,
         }
     }
 

crates/pool/src/mempool/uo_pool.rs

@@ -1841,6 +1841,7 @@ mod tests {
             reputation_tracking_enabled: true,
             drop_min_num_blocks: 10,
             gas_limit_efficiency_reject_threshold: 0.0,
+            max_time_in_pool: None,
         }
     }
 

crates/types/src/timestamp.rs

@@ -15,8 +15,7 @@
 
 use std::{
     error::Error,
-    fmt,
-    fmt::{Debug, Display, Formatter},
+    fmt::{self, Debug, Display, Formatter},
     ops::{Add, AddAssign, Sub, SubAssign},
     time::{Duration, SystemTime, UNIX_EPOCH},
 };
@@ -70,6 +69,12 @@ impl From<u64> for Timestamp {
     }
 }
 
+impl From<Duration> for Timestamp {
+    fn from(duration: Duration) -> Self {
+        Self(duration.as_secs())
+    }
+}
+
 impl Add<Duration> for Timestamp {
     type Output = Self;
 

docs/cli.md

@@ -165,6 +165,8 @@ List of command line options for configuring the Pool.
   - env: *POOL_DROP_MIN_NUM_BLOCKS*
 - `--pool.gas_limit_efficiency_reject_threshold`: The ratio of gas used to gas limit under which to reject UOs upon entry to the mempool (default: `0.0` disabled)
   - env: *POOL_GAS_LIMIT_EFFICIENCY_REJECT_THRESHOLD*
+- `--pool.max_time_in_pool_secs`: The maximum amount of time a UO is allowed to be in the mempool, in seconds. (default: `None`)
+  - env: *POOL_MAX_TIME_IN_POOL_SECS*
 
 ## Builder Options