apollo_mempool: track and prune delayed declares so they can't bypass eviction

Declare transactions buffered in delayed_declares could pollute the mempool
immune to eviction (H-18):
1. add_ready_declares inserted popped declares into tx_pool without calling
   update_accounts_with_gap, so a popped declare with a nonce gap was never
   added to accounts_with_gap and thus never evictable.
2. commit_block never pruned delayed_declares, so a declare whose nonce was
   committed on-chain while waiting was later inserted stale, also immune.

Fix:
- add_ready_declares now collects the popped accounts and calls
  update_accounts_with_gap, and drops declares that are already stale (tx nonce
  below the committed nonce). Staleness uses the committed nonce only -- staged
  nonces are provisional and could otherwise discard a still-valid declare.
- commit_block prunes stale delayed declares (after state.commit, before
  update_accounts_with_gap so any masked gap is re-detected).
- AddTransactionQueue gains a size-maintaining retain for the prune.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: matanl-starkware

crates/apollo_mempool/src/fee_mempool_test.rs

@@ -1752,15 +1752,78 @@ fn stale_delayed_declare_does_not_suppress_gap_detection() {
     commit_block(&mut mempool, [(addr, 1)], []);
     assert!(mempool.accounts_with_gap().contains(&contract_address!(addr)));
 
-    // Once the delay elapses, the stale declare is moved from delayed_declares into tx_pool
-    // (add_ready_declares has no nonce guard). It sits there with stale nonce 0 < account nonce
-    // 1, so it is never queued. The next commit_block will clean it up via
-    // remove_up_to_nonce_when_committed.
+    // commit_block prunes the now-stale delayed declare (nonce 0 < committed nonce 1), so it never
+    // reaches tx_pool. Once the delay elapses there is nothing left to pop.
     fake_clock.advance(declare_delay);
     mempool.get_txs(1).unwrap();
     assert!(mempool.mempool_snapshot().unwrap().delayed_declares.is_empty());
 }
 
+#[test]
+fn future_gap_delayed_declare_is_evictable_after_pop() {
+    let declare_delay = Duration::from_secs(100);
+    let fake_clock = Arc::new(FakeClock::default());
+    let mut mempool = Mempool::new(
+        MempoolConfig {
+            static_config: MempoolStaticConfig { declare_delay, ..Default::default() },
+            ..Default::default()
+        },
+        fake_clock.clone(),
+    );
+    let addr = "0x0";
+
+    // A declare with a future nonce (6) while the account is at nonce 5: it creates a gap and is
+    // the account's only transaction.
+    let mut gapped_declare = declare_add_tx_input(declare_tx_args!(
+        tx_hash: tx_hash!(1),
+        sender_address: contract_address!(addr),
+        nonce: nonce!(6)
+    ));
+    gapped_declare.account_state.nonce = nonce!(5);
+    add_tx(&mut mempool, &gapped_declare);
+    // Still buffered (not yet in the pool), so no gap is tracked yet.
+    assert!(!mempool.accounts_with_gap().contains(&contract_address!(addr)));
+
+    // After the delay the declare is popped into the pool. Its account must be tracked in
+    // accounts_with_gap so the stuck declare is evictable (otherwise it pollutes the mempool
+    // permanently, immune to eviction).
+    fake_clock.advance(declare_delay);
+    mempool.get_txs(1).unwrap();
+    assert!(mempool.accounts_with_gap().contains(&contract_address!(addr)));
+    assert!(mempool.mempool_snapshot().unwrap().transactions.contains(&tx_hash!(1)));
+}
+
+#[test]
+fn stale_delayed_declare_pruned_on_commit() {
+    let declare_delay = Duration::from_secs(100);
+    let fake_clock = Arc::new(FakeClock::default());
+    let mut mempool = Mempool::new(
+        MempoolConfig {
+            static_config: MempoolStaticConfig { declare_delay, ..Default::default() },
+            ..Default::default()
+        },
+        fake_clock.clone(),
+    );
+    let addr = "0x0";
+
+    // Delayed declare at nonce 5 (the account's only tx).
+    let declare = declare_add_tx_input(declare_tx_args!(
+        tx_hash: tx_hash!(1),
+        sender_address: contract_address!(addr),
+        nonce: nonce!(5)
+    ));
+    add_tx(&mut mempool, &declare);
+    assert_eq!(mempool.mempool_snapshot().unwrap().delayed_declares, vec![tx_hash!(1)]);
+
+    // Another sequencer commits the account past nonce 5; the buffered declare is now stale and
+    // can never execute. It must be pruned at commit, not inserted into the pool later (where it
+    // would be immune to eviction).
+    commit_block(&mut mempool, [(addr, 6)], []);
+    let snapshot = mempool.mempool_snapshot().unwrap();
+    assert!(snapshot.delayed_declares.is_empty());
+    assert!(!snapshot.transactions.contains(&tx_hash!(1)));
+}
+
 #[rstest]
 fn returns_error_when_no_evictable_accounts() {
     let not_evictable_tx = add_tx_input!(tx_hash: 1, address: "0x0", tx_nonce: 0, account_nonce: 0);

crates/apollo_mempool/src/mempool.rs

@@ -112,6 +112,14 @@ impl MempoolState {
             .unwrap_or(incoming_account_nonce)
     }
 
+    /// Returns the committed Nonce for the address, ignoring staged (provisional, possibly
+    /// rewound) nonces. Falls back to incoming_account_nonce if no committed value is found. Used
+    /// to decide whether a transaction is stale (committed past), where staged nonces must not be
+    /// trusted since the staging block may not be committed.
+    fn committed_nonce(&self, address: ContractAddress, incoming_account_nonce: Nonce) -> Nonce {
+        self.committed.get(&address).copied().unwrap_or(incoming_account_nonce)
+    }
+
     fn contains_account(&self, address: ContractAddress) -> bool {
         self.staged.contains_key(&address) || self.committed.contains_key(&address)
     }
@@ -235,6 +243,23 @@ impl AddTransactionQueue {
         })
     }
 
+    /// Removes every queued transaction for which `keep` returns `false`, keeping `size_in_bytes`
+    /// consistent.
+    fn retain(&mut self, mut keep: impl FnMut(&AddTransactionArgs) -> bool) {
+        let mut removed_bytes = 0;
+        self.elements.retain(|(_, args)| {
+            let should_keep = keep(args);
+            if !should_keep {
+                removed_bytes += args.tx.total_bytes();
+            }
+            should_keep
+        });
+        self.size_in_bytes = self
+            .size_in_bytes
+            .checked_sub(removed_bytes)
+            .expect("Underflow when pruning AddTransactionQueue.");
+    }
+
     fn len(&self) -> usize {
         self.elements.len()
     }
@@ -599,14 +624,34 @@ impl Mempool {
 
     fn add_ready_declares(&mut self) {
         let now = self.clock.now();
+        // Collect the accounts of the popped declares so their nonce-gap status is (re)evaluated
+        // for eviction tracking. Without this, a popped declare with a nonce gap would never be
+        // added to `accounts_with_gap` and thus would be immune to eviction.
+        let mut account_nonce_updates = AddressToNonce::new();
         while let Some((submission_time, _args)) = self.delayed_declares.front() {
             if now - self.config.static_config.declare_delay < *submission_time {
                 break;
             }
             let (_submission_time, args) =
                 self.delayed_declares.pop_front().expect("Delay declare should exist.");
+            let address = args.account_state.address;
+            // Staleness is judged against the committed nonce only; staged nonces are provisional
+            // (the staging block may be rewound), so dropping against them could discard a valid
+            // declare.
+            let committed_nonce = self.state.committed_nonce(address, args.account_state.nonce);
+            if args.tx.nonce() < committed_nonce {
+                debug!(
+                    "Dropping stale delayed declare for account {address}: tx nonce {} < \
+                     committed nonce {committed_nonce}.",
+                    args.tx.nonce(),
+                );
+                continue;
+            }
+            account_nonce_updates
+                .insert(address, self.state.resolve_nonce(address, args.account_state.nonce));
             self.add_tx_inner(args);
         }
+        self.update_accounts_with_gap(account_nonce_updates);
         self.update_state_metrics();
     }
 
@@ -671,6 +716,19 @@ impl Mempool {
         // Committed nonces should overwrite rejected transactions.
         account_nonce_updates.extend(committed_nonce_updates);
 
+        // Prune delayed declares that became stale while waiting: their nonce was committed
+        // on-chain (e.g. by another sequencer) so they can never execute. Done after `state.commit`
+        // (which updated `committed` and cleared `staged`), so `resolve_nonce` reflects the
+        // committed nonce here. Pruning before `update_accounts_with_gap` lets it correctly
+        // re-detect any gap that the stale declare was masking.
+        {
+            let Mempool { state, delayed_declares, .. } = self;
+            delayed_declares.retain(|args| {
+                args.tx.nonce()
+                    >= state.resolve_nonce(args.account_state.address, args.account_state.nonce)
+            });
+        }
+
         self.update_accounts_with_gap(account_nonce_updates);
         self.update_state_metrics();
     }