feat(l1): periodic mempool sweep for stale and dormant transactions

cmd/ethrex/cli.rs

@@ -11,6 +11,7 @@ use std::{
 use clap::{ArgAction, Parser as ClapParser, Subcommand as ClapSubcommand};
 use ethrex_blockchain::{
     BlockchainOptions, BlockchainType, L2Config,
+    constants::{DEFAULT_DORMANCY, DEFAULT_MAX_NONCE_GAP, DEFAULT_MEMPOOL_LIFETIME},
     error::{ChainError, InvalidBlockError},
 };
 use ethrex_common::types::{Block, DEFAULT_BUILDER_GAS_CEIL, Genesis, validate_block_body};
@@ -65,6 +66,33 @@ pub struct CLI {
     pub command: Option<Subcommand>,
 }
 
+/// Format a [`Duration`] in human-readable form (e.g. `3h`, `30m`, `45s`)
+/// matching the syntax accepted by the parser, so the value rendered in
+/// `--help` is operator-friendly instead of raw seconds.
+fn duration_default_str(d: Duration) -> clap::builder::OsStr {
+    let secs = d.as_secs();
+    let formatted = if secs % 3600 == 0 && secs > 0 {
+        format!("{}h", secs / 3600)
+    } else if secs % 60 == 0 && secs > 0 {
+        format!("{}m", secs / 60)
+    } else {
+        format!("{secs}s")
+    };
+    formatted.into()
+}
+
+/// Format the default mempool lifetime as a clap-compatible string,
+/// reusing the [`DEFAULT_MEMPOOL_LIFETIME`] constant so it stays in sync.
+fn default_mempool_lifetime_str() -> clap::builder::OsStr {
+    duration_default_str(DEFAULT_MEMPOOL_LIFETIME)
+}
+
+/// Format the default mempool dormancy as a clap-compatible string,
+/// reusing the [`DEFAULT_DORMANCY`] constant so it stays in sync.
+fn default_mempool_dormancy_str() -> clap::builder::OsStr {
+    duration_default_str(DEFAULT_DORMANCY)
+}
+
 #[derive(ClapParser, Debug, Clone)]
 pub struct Options {
     #[arg(
@@ -183,6 +211,35 @@ pub struct Options {
         env = "ETHREX_MEMPOOL_MAX_SIZE"
     )]
     pub mempool_max_size: usize,
+    #[arg(
+        help = "Maximum age of a mempool transaction before it is evicted by the periodic sweep. Accepts values like 3h, 30m, 45s.",
+        long = "mempool.lifetime",
+        default_value = default_mempool_lifetime_str(),
+        value_parser = utils::parse_duration,
+        value_name = "DURATION",
+        help_heading = "Node options",
+        env = "ETHREX_MEMPOOL_LIFETIME"
+    )]
+    pub mempool_lifetime: Duration,
+    #[arg(
+        help = "Maximum allowed gap between a sender's highest pending nonce and their on-chain nonce before the dormancy sweep is eligible to evict their pool entries.",
+        long = "mempool.max-nonce-gap",
+        default_value_t = DEFAULT_MAX_NONCE_GAP,
+        value_name = "GAP",
+        help_heading = "Node options",
+        env = "ETHREX_MEMPOOL_MAX_NONCE_GAP"
+    )]
+    pub mempool_max_nonce_gap: u64,
+    #[arg(
+        help = "Dormancy window for the nonce-gap mempool sweep. A sender is only evicted when all their pool entries are older than this and the nonce gap exceeds --mempool.max-nonce-gap. Accepts values like 3h, 30m, 45s.",
+        long = "mempool.dormancy",
+        default_value = default_mempool_dormancy_str(),
+        value_parser = utils::parse_duration,
+        value_name = "DURATION",
+        help_heading = "Node options",
+        env = "ETHREX_MEMPOOL_DORMANCY"
+    )]
+    pub mempool_dormancy: Duration,
     #[arg(
         long = "http.addr",
         default_value = "0.0.0.0",
@@ -392,6 +449,9 @@ impl Options {
             discv4_enabled: true,
             discv5_enabled: true,
             mempool_max_size: 10_000,
+            mempool_lifetime: DEFAULT_MEMPOOL_LIFETIME,
+            mempool_max_nonce_gap: DEFAULT_MAX_NONCE_GAP,
+            mempool_dormancy: DEFAULT_DORMANCY,
             ..Default::default()
         }
     }
@@ -414,6 +474,9 @@ impl Options {
             discv4_enabled: true,
             discv5_enabled: true,
             mempool_max_size: 10_000,
+            mempool_lifetime: DEFAULT_MEMPOOL_LIFETIME,
+            mempool_max_nonce_gap: DEFAULT_MAX_NONCE_GAP,
+            mempool_dormancy: DEFAULT_DORMANCY,
             ..Default::default()
         }
     }
@@ -450,6 +513,9 @@ impl Default for Options {
             dev: Default::default(),
             force: false,
             mempool_max_size: Default::default(),
+            mempool_lifetime: DEFAULT_MEMPOOL_LIFETIME,
+            mempool_max_nonce_gap: DEFAULT_MAX_NONCE_GAP,
+            mempool_dormancy: DEFAULT_DORMANCY,
             tx_broadcasting_time_interval: Default::default(),
             target_peers: Default::default(),
             lookup_interval: Default::default(),
@@ -626,6 +692,9 @@ impl Subcommand {
                     BlockchainOptions {
                         max_mempool_size: opts.mempool_max_size,
                         r#type: blockchain_type,
+                        mempool_lifetime: opts.mempool_lifetime,
+                        max_nonce_gap: opts.mempool_max_nonce_gap,
+                        dormancy: opts.mempool_dormancy,
                         ..Default::default()
                     },
                 )

cmd/ethrex/initializers.rs

@@ -529,9 +529,14 @@ pub async fn init_l1(
             max_blobs_per_block: opts.max_blobs_per_block,
             precompute_witnesses: opts.precompute_witnesses,
             precompile_cache_enabled: !opts.no_precompile_cache,
+            mempool_lifetime: opts.mempool_lifetime,
+            max_nonce_gap: opts.mempool_max_nonce_gap,
+            dormancy: opts.mempool_dormancy,
         },
     );
 
+    blockchain.spawn_mempool_sweep();
+
     regenerate_head_state(&store, &blockchain).await?;
 
     let signer = get_signer(&datadir);

cmd/ethrex/l2/initializers.rs

@@ -224,10 +224,15 @@ pub async fn init_l2(
         max_blobs_per_block: None, // L2 doesn't support blob transactions
         precompute_witnesses: opts.node_opts.precompute_witnesses,
         precompile_cache_enabled: true,
+        mempool_lifetime: opts.node_opts.mempool_lifetime,
+        max_nonce_gap: opts.node_opts.mempool_max_nonce_gap,
+        dormancy: opts.node_opts.mempool_dormancy,
     };
 
     let blockchain = init_blockchain(store.clone(), blockchain_opts.clone());
 
+    blockchain.spawn_mempool_sweep();
+
     regenerate_state(&store, &rollup_store, &blockchain, None).await?;
 
     let signer = get_signer(&datadir);

cmd/ethrex/utils.rs

@@ -70,6 +70,16 @@ pub fn read_chain_file(chain_rlp_path: &str) -> Vec<Block> {
     decode::chain_file(chain_file).expect("Failed to decode chain rlp file")
 }
 
+/// Parse a duration string like `3h`, `30m`, `45s` into a [`Duration`].
+///
+/// Delegates to [`ethrex_common::serde_utils::parse_duration`], which accepts
+/// concatenations of unit-suffixed values (`h`, `m`, `s`, `ms`, `us`/`µs`,
+/// `ns`). For example `1h30m` parses to 1 hour 30 minutes.
+pub fn parse_duration(s: &str) -> eyre::Result<std::time::Duration> {
+    ethrex_common::serde_utils::parse_duration(s.to_string())
+        .ok_or_else(|| eyre::eyre!("invalid duration {s:?}"))
+}
+
 pub fn parse_sync_mode(s: &str) -> eyre::Result<SyncMode> {
     match s {
         "full" => Ok(SyncMode::Full),

crates/blockchain/Cargo.toml

@@ -35,6 +35,7 @@ tokio-util.workspace = true
 
 [dev-dependencies]
 tokio = { workspace = true, features = ["full"] }
+serde_json.workspace = true
 
 [lib]
 path = "./blockchain.rs"

crates/blockchain/blockchain.rs

@@ -52,8 +52,8 @@ pub mod vm;
 
 use ::tracing::{debug, error, info, instrument, warn};
 use constants::{
-    AMSTERDAM_MAX_INITCODE_SIZE, MAX_INITCODE_SIZE, MAX_TRANSACTION_DATA_SIZE,
-    POST_OSAKA_GAS_LIMIT_CAP,
+    AMSTERDAM_MAX_INITCODE_SIZE, DEFAULT_DORMANCY, DEFAULT_MAX_NONCE_GAP, DEFAULT_MEMPOOL_LIFETIME,
+    MAX_INITCODE_SIZE, MAX_TRANSACTION_DATA_SIZE, MEMPOOL_SWEEP_INTERVAL, POST_OSAKA_GAS_LIMIT_CAP,
 };
 use error::MempoolError;
 use error::{ChainError, InvalidBlockError};
@@ -100,7 +100,7 @@ use std::collections::{BTreeMap, HashMap, HashSet};
 use std::sync::LazyLock;
 use std::sync::mpsc::Sender;
 use std::sync::{
-    Arc, RwLock,
+    Arc, RwLock, Weak,
     atomic::{AtomicBool, AtomicUsize, Ordering},
     mpsc::{Receiver, channel},
 };
@@ -231,6 +231,17 @@ pub struct BlockchainOptions {
     /// warmer thread and the executor. Set to false (via `--no-precompile-cache`) to
     /// disable the cache for benchmarking purposes.
     pub precompile_cache_enabled: bool,
+    /// Maximum age of a mempool transaction before it is evicted by the
+    /// periodic TTL sweep.
+    pub mempool_lifetime: Duration,
+    /// Threshold for the dormancy sweep: senders whose top pending nonce
+    /// exceeds their on-chain nonce by more than this value are eligible
+    /// for eviction once they have been dormant for [`Self::dormancy`].
+    pub max_nonce_gap: u64,
+    /// Dormancy window used by the nonce-gap sweep. A sender is only evicted
+    /// when all their pool entries are older than this and the nonce gap is
+    /// above [`Self::max_nonce_gap`].
+    pub dormancy: Duration,
 }
 
 impl Default for BlockchainOptions {
@@ -242,6 +253,9 @@ impl Default for BlockchainOptions {
             max_blobs_per_block: None,
             precompute_witnesses: false,
             precompile_cache_enabled: true,
+            mempool_lifetime: DEFAULT_MEMPOOL_LIFETIME,
+            max_nonce_gap: DEFAULT_MAX_NONCE_GAP,
+            dormancy: DEFAULT_DORMANCY,
         }
     }
 }
@@ -355,6 +369,23 @@ impl Blockchain {
         }
     }
 
+    /// Spawn the periodic mempool sweep task.
+    ///
+    /// Every [`MEMPOOL_SWEEP_INTERVAL`] the task runs two evictions:
+    ///   1. TTL sweep: drops txs older than [`BlockchainOptions::mempool_lifetime`].
+    ///   2. Dormancy sweep: drops every entry belonging to senders whose top
+    ///      pending nonce exceeds the on-chain nonce by more than
+    ///      [`BlockchainOptions::max_nonce_gap`] and who have been dormant
+    ///      for at least [`BlockchainOptions::dormancy`].
+    ///
+    /// The task holds only a [`Weak`] to the `Blockchain`, so it exits as
+    /// soon as the last `Arc` is dropped — there is no need to thread a
+    /// cancellation token through the call sites.
+    pub fn spawn_mempool_sweep(self: &Arc<Self>) {
+        let weak = Arc::downgrade(self);
+        tokio::spawn(run_mempool_sweep(weak));
+    }
+
     /// Executes a block withing a new vm instance and state
     fn execute_block(
         &self,
@@ -3166,3 +3197,52 @@ fn collect_trie(index: u8, mut trie: Trie) -> Result<(Box<BranchNode>, Vec<TrieN
     };
     Ok((root, nodes))
 }
+
+/// Periodic mempool sweep loop: ticks every [`MEMPOOL_SWEEP_INTERVAL`] and
+/// runs the TTL + dormancy evictions on the live mempool.
+///
+/// Holds only a [`Weak`] to the [`Blockchain`] so the task tears itself down
+/// once the last strong reference is dropped (no explicit shutdown signal).
+async fn run_mempool_sweep(blockchain: Weak<Blockchain>) {
+    let mut interval = tokio::time::interval(MEMPOOL_SWEEP_INTERVAL);
+    // `Skip` so a sweep iteration that takes longer than the configured
+    // interval doesn't trigger a burst of back-to-back ticks afterward
+    // (default `Burst` behavior). Under load we'd rather skip a sweep cycle
+    // than queue them up.
+    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+    // Drop the first immediate tick — let the node finish initialising before
+    // doing any work.
+    interval.tick().await;
+    loop {
+        interval.tick().await;
+        let Some(chain) = blockchain.upgrade() else {
+            // Blockchain dropped; tear down.
+            return;
+        };
+
+        let ttl = chain.options.mempool_lifetime;
+        let max_gap = chain.options.max_nonce_gap;
+        let dormancy = chain.options.dormancy;
+
+        match chain.mempool.evict_stale(ttl) {
+            Ok(0) => {}
+            Ok(n) => info!(evicted = n, ttl_secs = ttl.as_secs(), "mempool TTL sweep"),
+            Err(e) => warn!(error = %e, "mempool TTL sweep failed"),
+        }
+
+        match chain
+            .mempool
+            .evict_dormant(&chain.storage, max_gap, dormancy)
+            .await
+        {
+            Ok(0) => {}
+            Ok(n) => info!(
+                evicted = n,
+                max_nonce_gap = max_gap,
+                dormancy_secs = dormancy.as_secs(),
+                "mempool dormancy sweep"
+            ),
+            Err(e) => warn!(error = %e, "mempool dormancy sweep failed"),
+        }
+    }
+}

crates/blockchain/constants.rs

@@ -52,3 +52,24 @@ pub const MIN_GAS_LIMIT: u64 = 5000;
 // === EIP-7825 constants ===
 // https://eips.ethereum.org/EIPS/eip-7825
 pub const POST_OSAKA_GAS_LIMIT_CAP: u64 = 16777216;
+
+// === Mempool sweep defaults ===
+
+use std::time::Duration;
+
+/// Default maximum age of a mempool transaction before the periodic sweep
+/// evicts it. Transactions older than this are dropped regardless of pool
+/// occupancy.
+pub const DEFAULT_MEMPOOL_LIFETIME: Duration = Duration::from_secs(3 * 60 * 60);
+
+/// Default maximum gap allowed between a sender's top pending nonce and
+/// their on-chain nonce before the dormancy sweep considers them stalled.
+pub const DEFAULT_MAX_NONCE_GAP: u64 = 64;
+
+/// Default dormancy window for the nonce-gap sweep. A sender must have made
+/// no on-chain progress for at least this long (i.e. all their pool entries
+/// are older than this) before they are eligible for eviction.
+pub const DEFAULT_DORMANCY: Duration = Duration::from_secs(3 * 60 * 60);
+
+/// How often the periodic mempool sweep runs.
+pub const MEMPOOL_SWEEP_INTERVAL: Duration = Duration::from_secs(60);