apollo_mempool: add nonce-gap metrics (accounts_with_gap, stuck_txs) (#13918)

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: matanl-starkware

crates/apollo_mempool/src/fee_mempool_test.rs

@@ -158,6 +158,7 @@ impl MempoolTestContentBuilder {
                 self.gas_price_threshold,
             )),
             accounts_with_gap: AccountsWithGap::new(),
+            n_stuck_txs: 0,
             state: MempoolState::new(
                 self.config.static_config.committed_nonce_retention_block_count,
             ),
@@ -1904,3 +1905,188 @@ fn rejects_or_accepts_tx_based_on_freed_space(
     // We do not revert the eviction attempt even if adding large_tx ultimately fails.
     assert!(!mempool.tx_pool.contains_account(contract_address!("0x1")));
 }
+
+// Stuck-tx counter tests.
+
+fn assert_gap_metrics(mempool: &Mempool, expected_accounts: usize, expected_stuck_txs: usize) {
+    assert_eq!(mempool.accounts_with_gap().len(), expected_accounts, "accounts_with_gap mismatch");
+    assert_eq!(mempool.n_stuck_txs(), expected_stuck_txs, "n_stuck_txs mismatch");
+}
+
+#[rstest]
+fn stuck_txs_counter_gap_created_and_closed_by_add_tx(mut mempool: Mempool) {
+    assert_gap_metrics(&mempool, 0, 0);
+
+    // Adding nonce 1 with account nonce 0 creates a gap; the single pool tx is stuck.
+    let tx_nonce_1 = add_tx_input!(tx_hash: 1, address: "0x0", tx_nonce: 1, account_nonce: 0);
+    add_tx(&mut mempool, &tx_nonce_1);
+    assert_gap_metrics(&mempool, 1, 1);
+
+    // Adding nonce 2 to the already-gapped account: both txs are stuck.
+    let tx_nonce_2 = add_tx_input!(tx_hash: 2, address: "0x0", tx_nonce: 2, account_nonce: 0);
+    add_tx(&mut mempool, &tx_nonce_2);
+    assert_gap_metrics(&mempool, 1, 2);
+
+    // Filling the gap (nonce 0) resolves it; no more stuck txs.
+    let tx_nonce_0 = add_tx_input!(tx_hash: 3, address: "0x0", tx_nonce: 0, account_nonce: 0);
+    add_tx(&mut mempool, &tx_nonce_0);
+    assert_gap_metrics(&mempool, 0, 0);
+}
+
+#[rstest]
+fn stuck_txs_counter_tx_inserted_between_existing_stuck_txs(mut mempool: Mempool) {
+    // Nonces 1 and 3 in pool; gap at 0. Both are stuck.
+    let tx_nonce_1 = add_tx_input!(tx_hash: 1, address: "0x0", tx_nonce: 1, account_nonce: 0);
+    let tx_nonce_3 = add_tx_input!(tx_hash: 2, address: "0x0", tx_nonce: 3, account_nonce: 0);
+    for tx in [&tx_nonce_1, &tx_nonce_3] {
+        add_tx(&mut mempool, tx);
+    }
+    assert_gap_metrics(&mempool, 1, 2);
+
+    // Inserting nonce 2 into the gap account (gap at 0 persists): all three txs are stuck.
+    let tx_nonce_2 = add_tx_input!(tx_hash: 3, address: "0x0", tx_nonce: 2, account_nonce: 0);
+    add_tx(&mut mempool, &tx_nonce_2);
+    assert_gap_metrics(&mempool, 1, 3);
+
+    // Filling the primary gap (nonce 0) resolves everything.
+    let tx_nonce_0 = add_tx_input!(tx_hash: 4, address: "0x0", tx_nonce: 0, account_nonce: 0);
+    add_tx(&mut mempool, &tx_nonce_0);
+    assert_gap_metrics(&mempool, 0, 0);
+}
+
+#[rstest]
+fn stuck_txs_counter_account_becomes_unstuck_once_leading_gap_is_filled(mut mempool: Mempool) {
+    // Nonces 1 and 3 in pool; gap at 0. Account is fully blocked: lowest pool nonce (1) >
+    // account nonce (0), so neither tx can be batched yet.
+    let tx_nonce_1 = add_tx_input!(tx_hash: 1, address: "0x0", tx_nonce: 1, account_nonce: 0);
+    let tx_nonce_3 = add_tx_input!(tx_hash: 2, address: "0x0", tx_nonce: 3, account_nonce: 0);
+    for tx in [&tx_nonce_1, &tx_nonce_3] {
+        add_tx(&mut mempool, tx);
+    }
+    assert_gap_metrics(&mempool, 1, 2);
+
+    // Adding nonce 0 aligns lowest_pool_nonce with account_nonce: the account can now make
+    // progress (nonces 0 and 1 will be batched). It exits accounts_with_gap and the metric
+    // drops to 0. Nonce 3 will be re-detected as stuck once nonces 0 and 1 are committed
+    // and account_nonce catches up to 2 (which is still missing).
+    let tx_nonce_0 = add_tx_input!(tx_hash: 3, address: "0x0", tx_nonce: 0, account_nonce: 0);
+    add_tx(&mut mempool, &tx_nonce_0);
+    assert_gap_metrics(&mempool, 0, 0);
+
+    // Committing nonces 0 and 1 (next_nonce=2) removes them from the pool and advances
+    // account_nonce to 2. Pool is now [3]; lowest_pool_nonce (3) > account_nonce (2), so the
+    // gap is re-detected and nonce 3 is the sole stuck tx.
+    commit_block(&mut mempool, [("0x0", 2)], []);
+    assert_gap_metrics(&mempool, 1, 1);
+}
+
+#[rstest]
+fn stuck_txs_counter_resolved_by_commit_block(mut mempool: Mempool) {
+    let tx_nonce_1 = add_tx_input!(tx_hash: 1, address: "0x0", tx_nonce: 1, account_nonce: 0);
+    add_tx(&mut mempool, &tx_nonce_1);
+    assert_gap_metrics(&mempool, 1, 1);
+
+    // Committing nonce 1 removes the pool tx and advances account nonce; gap is resolved.
+    commit_block(&mut mempool, [("0x0", 2)], []);
+    assert_gap_metrics(&mempool, 0, 0);
+}
+
+#[rstest]
+fn stuck_txs_counter_commit_block_advances_past_some_stuck_txs(mut mempool: Mempool) {
+    // Gap account: nonces 3 and 5 in pool (gap at 0, 1, 2).
+    let tx_nonce_3 = add_tx_input!(tx_hash: 1, address: "0x0", tx_nonce: 3, account_nonce: 0);
+    let tx_nonce_5 = add_tx_input!(tx_hash: 2, address: "0x0", tx_nonce: 5, account_nonce: 0);
+    for tx in [&tx_nonce_3, &tx_nonce_5] {
+        add_tx(&mut mempool, tx);
+    }
+    assert_gap_metrics(&mempool, 1, 2);
+
+    // Commit next_nonce=4: removes nonce 3 (< 4) from pool; nonce 5 stays.
+    // account_nonce=4, lowest_pool_nonce=5 → gap persists with 1 stuck tx.
+    commit_block(&mut mempool, [("0x0", 4)], []);
+    assert_gap_metrics(&mempool, 1, 1);
+}
+
+#[test]
+fn stuck_txs_counter_expiry_decrements_count() {
+    let fake_clock = Arc::new(FakeClock::default());
+    let mut mempool = Mempool::new(
+        MempoolConfig {
+            dynamic_config: MempoolDynamicConfig { transaction_ttl: Duration::from_secs(60) },
+            ..Default::default()
+        },
+        fake_clock.clone(),
+    );
+
+    // Gap account: nonces 2 and 3 (gap at 0 and 1). Both are stuck.
+    let tx_nonce_2 = add_tx_input!(tx_hash: 1, address: "0x0", tx_nonce: 2, account_nonce: 0);
+    add_tx(&mut mempool, &tx_nonce_2);
+    fake_clock.advance(Duration::from_secs(30));
+    let tx_nonce_3 = add_tx_input!(tx_hash: 2, address: "0x0", tx_nonce: 3, account_nonce: 0);
+    add_tx(&mut mempool, &tx_nonce_3);
+    assert_gap_metrics(&mempool, 1, 2);
+
+    // Expire nonce 2 (older than TTL) while nonce 3 stays alive.
+    fake_clock.advance(Duration::from_secs(35));
+
+    // Trigger expiry cleanup via a new tx for a different address.
+    let trigger = add_tx_input!(tx_hash: 3, address: "0x1", tx_nonce: 0, account_nonce: 0);
+    add_tx(&mut mempool, &trigger);
+
+    // Account remains in gap (nonce 3 still present); stuck count decreased by 1.
+    assert_gap_metrics(&mempool, 1, 1);
+}
+
+#[test]
+fn stuck_txs_counter_eviction_decrements_count() {
+    let tx_gap = add_tx_input!(tx_hash: 1, address: "0x0", tx_nonce: 2, account_nonce: 0);
+    let capacity = tx_gap.tx.total_bytes();
+    let mut mempool = Mempool::new(
+        MempoolConfig {
+            static_config: MempoolStaticConfig {
+                capacity_in_bytes: capacity,
+                ..Default::default()
+            },
+            ..Default::default()
+        },
+        Arc::new(FakeClock::default()),
+    );
+
+    add_tx(&mut mempool, &tx_gap);
+    assert_gap_metrics(&mempool, 1, 1);
+
+    // Adding a tx for a different address triggers eviction of the gap account's tx.
+    let new_tx = add_tx_input!(tx_hash: 2, address: "0x1", tx_nonce: 0, account_nonce: 0);
+    add_tx(&mut mempool, &new_tx);
+
+    // Gap account was fully evicted; both counters are zero.
+    assert_gap_metrics(&mempool, 0, 0);
+}
+
+#[rstest]
+fn stuck_txs_counter_multiple_gap_accounts(mut mempool: Mempool) {
+    // Two independent gap accounts.
+    // "0x0": nonce 2 in pool, account_nonce=0. One stuck tx.
+    // "0x1": nonces 1 and 3 in pool, account_nonce=0. Two stuck txs.
+    let tx_0x0_nonce2 = add_tx_input!(tx_hash: 1, address: "0x0", tx_nonce: 2, account_nonce: 0);
+    let tx_0x1_nonce1 = add_tx_input!(tx_hash: 2, address: "0x1", tx_nonce: 1, account_nonce: 0);
+    let tx_0x1_nonce3 = add_tx_input!(tx_hash: 3, address: "0x1", tx_nonce: 3, account_nonce: 0);
+    for tx in [&tx_0x0_nonce2, &tx_0x1_nonce1, &tx_0x1_nonce3] {
+        add_tx(&mut mempool, tx);
+    }
+    assert_gap_metrics(&mempool, 2, 3);
+
+    // Resolving "0x0" via commit (next_nonce=3 removes nonce 2, pool empty → no gap).
+    commit_block(&mut mempool, [("0x0", 3)], []);
+    assert_gap_metrics(&mempool, 1, 2);
+
+    // Partial commit for "0x1" (next_nonce=2 removes nonce 1; nonce 3 stays).
+    // account_nonce=2, lowest_pool_nonce=3 → gap persists with one stuck tx.
+    commit_block(&mut mempool, [("0x1", 2)], []);
+    assert_gap_metrics(&mempool, 1, 1);
+
+    // Filling the gap (nonce 2) resolves "0x1" entirely.
+    let tx_0x1_nonce2 = add_tx_input!(tx_hash: 4, address: "0x1", tx_nonce: 2, account_nonce: 2);
+    add_tx(&mut mempool, &tx_0x1_nonce2);
+    assert_gap_metrics(&mempool, 0, 0);
+}

crates/apollo_mempool/src/mempool.rs

@@ -37,10 +37,12 @@ use crate::metrics::{
     metric_count_rejected_txs,
     metric_set_get_txs_size,
     LABEL_NAME_TX_TYPE,
+    MEMPOOL_ACCOUNTS_WITH_GAP,
     MEMPOOL_DELAYED_DECLARES_SIZE,
     MEMPOOL_PENDING_QUEUE_SIZE,
     MEMPOOL_POOL_SIZE,
     MEMPOOL_PRIORITY_QUEUE_SIZE,
+    MEMPOOL_STUCK_TXS,
     MEMPOOL_TOTAL_SIZE_BYTES,
     MEMPOOL_TRANSACTIONS_RECEIVED,
 };
@@ -254,6 +256,9 @@ pub struct Mempool {
     // Accounts whose lowest transaction nonce is greater than the account nonce, which are
     // therefore candidates for eviction.
     accounts_with_gap: AccountsWithGap,
+    // Total number of transactions in the pool that belong to accounts in `accounts_with_gap`.
+    // Maintained incrementally to allow O(1) metric reporting.
+    n_stuck_txs: usize,
     state: MempoolState,
     clock: Arc<dyn Clock>,
 }
@@ -273,6 +278,7 @@ impl Mempool {
             tx_pool: TransactionPool::new(clock.clone()),
             tx_queue,
             accounts_with_gap: AccountsWithGap::new(),
+            n_stuck_txs: 0,
             state: MempoolState::new(config.static_config.committed_nonce_retention_block_count),
             clock,
         }
@@ -366,8 +372,8 @@ impl Mempool {
         }
 
         metric_set_get_txs_size(n_returned_txs);
-        self.update_state_metrics();
         self.update_accounts_with_gap(account_nonce_updates);
+        self.update_state_metrics();
 
         Ok(eligible_tx_references
             .iter()
@@ -483,8 +489,8 @@ impl Mempool {
             self.add_tx_inner(args);
         }
 
-        self.update_state_metrics();
         self.update_accounts_with_gap(account_nonce_updates);
+        self.update_state_metrics();
         Ok(())
     }
 
@@ -541,6 +547,7 @@ impl Mempool {
             if let Ok(tx) = self.tx_pool.remove(tx_hash) {
                 self.tx_queue.remove_by_address(tx.contract_address());
                 rejected_txs_counter += 1;
+                self.decrement_stuck_txs_if_gap_account(tx.contract_address(), 1);
                 account_nonce_updates
                     .entry(tx.contract_address())
                     .and_modify(|nonce| *nonce = (*nonce).min(tx.nonce()))
@@ -563,6 +570,12 @@ impl Mempool {
 
         let tx_reference = TransactionReference::new(&tx);
 
+        // Pre-count this tx as stuck; update_accounts_with_gap will correct the count if this tx
+        // resolves the gap.
+        if self.accounts_with_gap.contains(&account_state.address) {
+            self.n_stuck_txs += 1;
+        }
+
         self.tx_pool
             .insert(tx)
             .expect("Duplicate transactions should cause an error during the validation stage.");
@@ -631,6 +644,7 @@ impl Mempool {
             // Remove from pool.
             let n_removed_txs = self.tx_pool.remove_up_to_nonce_when_committed(address, next_nonce);
             metric_count_committed_txs(n_removed_txs);
+            self.decrement_stuck_txs_if_gap_account(address, n_removed_txs);
 
             // Close nonce gap, if exists.
             // In FIFO mode, we handle gap filling when rewinding transactions.
@@ -657,8 +671,8 @@ impl Mempool {
         // Committed nonces should overwrite rejected transactions.
         account_nonce_updates.extend(committed_nonce_updates);
 
-        self.update_state_metrics();
         self.update_accounts_with_gap(account_nonce_updates);
+        self.update_state_metrics();
     }
 
     pub fn account_tx_in_pool_or_recent_block(&self, account_address: ContractAddress) -> bool {
@@ -769,6 +783,7 @@ impl Mempool {
         self.tx_pool
             .remove(existing_tx_reference.tx_hash)
             .expect("Transaction hash from pool must exist.");
+        self.decrement_stuck_txs_if_gap_account(address, 1);
 
         Ok(())
     }
@@ -820,6 +835,11 @@ impl Mempool {
         let removed_txs = self
             .tx_pool
             .remove_txs_older_than(self.config.dynamic_config.transaction_ttl, &self.state.staged);
+
+        for tx_ref in &removed_txs {
+            self.decrement_stuck_txs_if_gap_account(tx_ref.address, 1);
+        }
+
         let queued_txs = self.tx_queue.remove_txs(&removed_txs);
 
         self.log_and_count_expired_txs(&removed_txs);
@@ -858,6 +878,7 @@ impl Mempool {
                 self.tx_pool
                     .remove(tx.tx_hash)
                     .expect("Transaction hash from queue must appear in pool.");
+                self.decrement_stuck_txs_if_gap_account(tx.address, 1);
                 (tx.address, self.state.resolve_nonce(tx.address, tx.nonce))
             })
             .collect();
@@ -883,6 +904,20 @@ impl Mempool {
         self.tx_pool.size_in_bytes() + self.delayed_declares.size_in_bytes()
     }
 
+    fn decrement_stuck_txs_if_gap_account(&mut self, address: ContractAddress, n: usize) {
+        if self.accounts_with_gap.contains(&address) {
+            self.n_stuck_txs -= n;
+        }
+    }
+
+    // Removes address from the gap-account set and deducts its remaining pool txs from
+    // n_stuck_txs. No-op if the address was not tracked.
+    fn remove_from_accounts_with_gap(&mut self, address: ContractAddress) {
+        if self.accounts_with_gap.swap_remove(&address) {
+            self.n_stuck_txs -= self.tx_pool.n_txs_for_address(address);
+        }
+    }
+
     // Returns true if the mempool will exceeds its capacity by adding the given transaction.
     fn exceeds_capacity(&self, tx: &InternalRpcTransaction) -> bool {
         self.size_in_bytes() + tx.total_bytes() > self.config.static_config.capacity_in_bytes
@@ -893,7 +928,7 @@ impl Mempool {
             // If a delayed declare transaction exists at the account nonce, it is next to execute,
             // so no gap exists.
             if self.delayed_declares.contains(address, account_nonce) {
-                self.accounts_with_gap.swap_remove(&address);
+                self.remove_from_accounts_with_gap(address);
                 continue;
             }
 
@@ -905,9 +940,14 @@ impl Mempool {
 
             // Update the eviction tracking set accordingly.
             if gap_exists {
-                self.accounts_with_gap.insert(address);
+                if self.accounts_with_gap.insert(address) {
+                    // Newly entered gap: all current pool txs for this account are now stuck.
+                    self.n_stuck_txs += self.tx_pool.n_txs_for_address(address);
+                }
+                // Stayed in gap: per-tx deltas were already applied at add/remove sites.
             } else {
-                self.accounts_with_gap.swap_remove(&address);
+                // Left gap: remaining pool txs for this account are no longer stuck.
+                self.remove_from_accounts_with_gap(address);
             }
         }
     }
@@ -939,6 +979,7 @@ impl Mempool {
                     .expect("Transaction must exist in the pool.");
                 total_space_freed += tx.total_bytes();
                 metric_count_evicted_txs(1);
+                self.decrement_stuck_txs_if_gap_account(address, 1);
                 if total_space_freed >= required_space {
                     break;
                 }
@@ -986,12 +1027,19 @@ impl Mempool {
         &self.accounts_with_gap
     }
 
+    #[cfg(test)]
+    pub(crate) fn n_stuck_txs(&self) -> usize {
+        self.n_stuck_txs
+    }
+
     fn update_state_metrics(&self) {
         MEMPOOL_POOL_SIZE.set_lossy(self.tx_pool.len());
         MEMPOOL_PRIORITY_QUEUE_SIZE.set_lossy(self.tx_queue.priority_queue_len());
         MEMPOOL_PENDING_QUEUE_SIZE.set_lossy(self.tx_queue.pending_queue_len());
         MEMPOOL_DELAYED_DECLARES_SIZE.set_lossy(self.delayed_declares.len());
         MEMPOOL_TOTAL_SIZE_BYTES.set_lossy(self.size_in_bytes());
+        MEMPOOL_ACCOUNTS_WITH_GAP.set_lossy(self.accounts_with_gap.len());
+        MEMPOOL_STUCK_TXS.set_lossy(self.n_stuck_txs);
     }
 }