tempo_transaction_pool/

tt_2d_pool.rs

/// Basic 2D nonce pool for user nonces (nonce_key > 0) that are tracked on chain.
use crate::{
    metrics::AA2dPoolMetrics, ordering::TempoTipOrdering, transaction::TempoPooledTransaction,
};
use alloy_consensus::Transaction;
use alloy_primitives::{
    Address, B256, TxHash, U256,
    map::{AddressMap, B256Map, HashMap, HashSet, U256Map, hash_map},
};
use reth_primitives_traits::transaction::error::InvalidTransactionError;
use reth_tracing::tracing::trace;
use reth_transaction_pool::{
    AllPoolTransactions, BestTransactions, GetPooledTransactionLimit, PoolResult, PoolTransaction,
    PriceBumpConfig, Priority, SubPool, SubPoolLimit, TransactionOrdering, TransactionOrigin,
    ValidPoolTransaction,
    error::{InvalidPoolTransactionError, PoolError, PoolErrorKind},
    pool::{AddedPendingTransaction, AddedTransaction, QueuedReason, pending::PendingTransaction},
};
use revm::database::BundleAccount;
use std::{
    borrow::Borrow,
    collections::{
        BTreeMap, BTreeSet,
        Bound::{Excluded, Unbounded},
        btree_map::Entry,
    },
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
};
use tempo_precompiles::NONCE_PRECOMPILE_ADDRESS;
use tokio::sync::broadcast;

type TxOrdering = TempoTipOrdering<TempoPooledTransaction>;
type PoolUpdateResult = (
    Vec<Arc<ValidPoolTransaction<TempoPooledTransaction>>>,
    Vec<Arc<ValidPoolTransaction<TempoPooledTransaction>>>,
);
/// A sub-pool that keeps track of 2D nonce transactions.
///
/// It maintains both pending and queued transactions.
///
/// A 2d nonce transaction is pending if it dosn't have a nonce gap for its nonce key, and is queued if its nonce key set has nonce gaps.
///
/// This pool relies on state changes to track the nonces.
///
/// # Limitations
///
/// * We assume new AA transactions either create a new nonce key (nonce 0) or use an existing nonce key. To keep track of the known keys by accounts this pool relies on state changes to promote transactions to pending.
#[derive(Debug)]
pub struct AA2dPool {
    /// Keeps track of transactions inserted in the pool.
    ///
    /// This way we can determine when transactions were submitted to the pool.
    submission_id: u64,
    /// Regular 2D nonce independent, pending, executable transactions, one per
    /// sequence id.
    ///
    /// Expiring nonce transactions are not included here because they are keyed
    /// by expiring nonce hash, not `AASequenceId`.
    independent_transactions: HashMap<AASequenceId, AA2dStoredTransaction>,
    /// _All_ regular 2D nonce transactions that are currently inside the pool,
    /// grouped by their unique identifier.
    ///
    /// Expiring nonce transactions are not stored in `by_id`; they are tracked
    /// separately by `expiring_nonce_txs`.
    by_id: BTreeMap<AA2dTransactionId, Arc<AA2dInternalTransaction>>,
    /// _All_ transactions by hash.
    by_hash: B256Map<Arc<ValidPoolTransaction<TempoPooledTransaction>>>,
    /// Expiring nonce transactions, keyed by expiring nonce hash (always pending/independent).
    /// These use expiring nonce replay protection instead of sequential nonces.
    expiring_nonce_txs: B256Map<AA2dStoredTransaction>,
    /// Expiring nonce transactions in eviction order.
    ///
    /// Regular 2D transactions use `by_eviction_order`, which is keyed by
    /// `AA2dTransactionId`. Expiring nonce transactions are always pending but
    /// are not stored in `by_id`, so they need a separate ordered index. Each
    /// key carries the transaction and a priority snapshot. The first entry is
    /// the expiring nonce transaction that should be evicted next:
    /// lowest priority first, then newest submission first when priorities tie.
    expiring_nonce_eviction_order: BTreeSet<ExpiringNonceEvictionKey>,
    /// A mapping of `expiring_nonce_seen` slot to expiring nonce hash.
    ///
    /// Used to track inclusion of expiring nonce transactions.
    slot_to_expiring_nonce_hash: U256Map<B256>,
    /// Scratch buffer reused while processing nonce state updates.
    state_update_nonce_changes: HashMap<AASequenceId, u64>,
    /// Scratch buffer reused while processing included expiring nonce transactions.
    state_update_included_expiring_nonce_hashes: Vec<B256>,
    /// Reverse index for the storage slot of an account's nonce
    ///
    /// ```solidity
    ///  mapping(address => mapping(uint256 => uint64)) public nonces
    /// ```
    ///
    /// This identifies the account and nonce key based on the slot in the `NonceManager`.
    slot_to_seq_id: U256Map<AASequenceId>,
    /// Settings for this sub-pool.
    config: AA2dPoolConfig,
    /// Metrics for tracking pool statistics
    metrics: AA2dPoolMetrics,
    /// All transactions ordered by eviction priority (lowest priority first).
    ///
    /// Rebuilt when the pool base fee changes. At eviction time, we scan this
    /// set checking `is_pending` to find queued or pending transactions. Keys
    /// own a priority snapshot so repricing does not mutate canonical
    /// transaction storage.
    by_eviction_order: BTreeSet<EvictionKey>,
    /// Base fee used for transaction insertion and eviction-order priorities.
    base_fee: u64,
    /// Tracks the number of transactions per sender for DoS protection.
    ///
    /// Bounded by pool size (max unique senders = pending_limit + queued_limit).
    /// Entries are removed when count reaches 0 via `decrement_sender_count`.
    txs_by_sender: AddressMap<usize>,
    /// Number of pending transactions, including expiring nonce transactions.
    pending_count: usize,
    /// Number of queued regular 2D nonce transactions.
    queued_count: usize,
    /// Used to broadcast new pending transactions to active [`BestAA2dTransactions`] iterators.
    new_transaction_notifier: broadcast::Sender<AA2dStoredTransaction>,
}

impl Default for AA2dPool {
    fn default() -> Self {
        Self::new(AA2dPoolConfig::default())
    }
}

impl AA2dPool {
    /// Creates a new instance with the givenconfig and nonce keys
    pub fn new(config: AA2dPoolConfig) -> Self {
        let (new_transaction_notifier, _) = broadcast::channel(200);
        Self {
            submission_id: 0,
            independent_transactions: Default::default(),
            by_id: Default::default(),
            by_hash: Default::default(),
            expiring_nonce_txs: Default::default(),
            expiring_nonce_eviction_order: Default::default(),
            slot_to_expiring_nonce_hash: Default::default(),
            state_update_nonce_changes: Default::default(),
            state_update_included_expiring_nonce_hashes: Default::default(),
            slot_to_seq_id: Default::default(),
            config,
            metrics: AA2dPoolMetrics::default(),
            by_eviction_order: Default::default(),
            base_fee: 0,
            txs_by_sender: Default::default(),
            pending_count: 0,
            queued_count: 0,
            new_transaction_notifier,
        }
    }

    /// Broadcasts a new pending transaction to all active [`BestAA2dTransactions`] iterators.
    fn notify_new_pending(&self, tx: &AA2dStoredTransaction) {
        if self.new_transaction_notifier.receiver_count() > 0 {
            let _ = self.new_transaction_notifier.send(tx.clone());
        }
    }

    /// Updates all metrics to reflect the current state of the pool
    fn update_metrics(&self) {
        let (pending, queued) = self.pending_and_queued_txn_count();
        let total = self.by_id.len() + self.expiring_nonce_txs.len();
        self.metrics.set_transaction_counts(total, pending, queued);
    }

    pub(crate) fn set_base_fee(&mut self, base_fee: u64) {
        if self.base_fee == base_fee {
            return;
        }

        self.base_fee = base_fee;
        self.rebuild_eviction_order();
    }

    fn rebuild_eviction_order(&mut self) {
        self.by_eviction_order.clear();
        for (id, tx) in &self.by_id {
            self.by_eviction_order.insert(EvictionKey::with_base_fee(
                Arc::clone(tx),
                *id,
                self.base_fee,
            ));
        }

        self.expiring_nonce_eviction_order.clear();
        for tx in self.expiring_nonce_txs.values() {
            self.expiring_nonce_eviction_order.insert(
                ExpiringNonceEvictionKey::from_pending_with_base_fee(tx, self.base_fee),
            );
        }
    }

    /// Entrypoint for adding a 2d AA transaction.
    ///
    /// `on_chain_nonce` is expected to be the nonce of the sender at the time of validation.
    /// If transaction is using 2D nonces, this is expected to be the nonce corresponding
    /// to the transaction's nonce key.
    ///
    /// `hardfork` indicates the active Tempo hardfork. When T1 or later, expiring nonce
    /// transactions (nonce_key == U256::MAX) are handled specially. Otherwise, they are
    /// treated as regular 2D nonce transactions.
    pub(crate) fn add_transaction(
        &mut self,
        transaction: Arc<ValidPoolTransaction<TempoPooledTransaction>>,
        on_chain_nonce: u64,
        hardfork: tempo_chainspec::hardfork::TempoHardfork,
    ) -> PoolResult<AddedTransaction<TempoPooledTransaction>> {
        debug_assert!(
            transaction.transaction.is_aa(),
            "only AA transactions are supported"
        );
        // Handle expiring nonce transactions separately - they use expiring nonce hash as unique ID
        // Only treat as expiring nonce if T1 hardfork is active.
        //
        // No `by_hash` duplicate check needed here: a duplicate transaction maps to the same
        // expiring nonce hash, which `add_expiring_nonce_transaction` rejects.
        if hardfork.is_t1() && transaction.transaction.is_expiring_nonce() {
            return self.add_expiring_nonce_transaction(transaction);
        }

        if self.contains(transaction.hash()) {
            return Err(PoolError::new(
                *transaction.hash(),
                PoolErrorKind::AlreadyImported,
            ));
        }

        let tx_id = transaction
            .transaction
            .aa_transaction_id()
            .expect("Transaction added to AA2D pool must be an AA transaction");

        if transaction.nonce() < on_chain_nonce {
            // outdated transaction
            return Err(PoolError::new(
                *transaction.hash(),
                PoolErrorKind::InvalidTransaction(InvalidPoolTransactionError::Consensus(
                    InvalidTransactionError::NonceNotConsistent {
                        tx: transaction.nonce(),
                        state: on_chain_nonce,
                    },
                )),
            ));
        }

        // assume the transaction is not pending, will get updated later
        let tx = Arc::new(AA2dInternalTransaction {
            inner: AA2dStoredTransaction::new(self.next_id(), transaction.clone()),
            is_pending: AtomicBool::new(false),
        });

        // Use entry API once to both check for replacement and insert.
        // This avoids a separate contains_key lookup.
        let sender = transaction.sender();
        let replaced = match self.by_id.entry(tx_id) {
            Entry::Occupied(mut entry) => {
                // Ensure the replacement transaction is not underpriced
                if entry
                    .get()
                    .inner
                    .transaction
                    .is_underpriced(&tx.inner.transaction, &self.config.price_bump_config)
                {
                    return Err(PoolError::new(
                        *transaction.hash(),
                        PoolErrorKind::ReplacementUnderpriced,
                    ));
                }

                let replaced = entry.insert(Arc::clone(&tx));
                self.remove_from_counts(replaced.is_pending());
                self.queued_count += 1;
                Some(replaced)
            }
            Entry::Vacant(entry) => {
                // Check per-sender limit and increment the count for new (non-replacement)
                // transactions with a single map lookup
                match self.txs_by_sender.entry(sender) {
                    hash_map::Entry::Occupied(mut count) => {
                        if *count.get() >= self.config.max_txs_per_sender {
                            return Err(PoolError::new(
                                *transaction.hash(),
                                PoolErrorKind::SpammerExceededCapacity(sender),
                            ));
                        }
                        *count.get_mut() += 1;
                    }
                    hash_map::Entry::Vacant(count) => {
                        if self.config.max_txs_per_sender == 0 {
                            return Err(PoolError::new(
                                *transaction.hash(),
                                PoolErrorKind::SpammerExceededCapacity(sender),
                            ));
                        }
                        count.insert(1);
                    }
                }

                entry.insert(Arc::clone(&tx));
                self.queued_count += 1;
                None
            }
        };

        // Cache the nonce key slot for reverse lookup, if this transaction uses 2D nonce.
        // This must happen after successful by_id insertion to avoid leaking slot entries
        // when the transaction is rejected (e.g., by per-sender limit or replacement check).
        if transaction.transaction.is_aa_2d() {
            self.record_2d_slot(&transaction.transaction);
        }

        // clean up replaced
        if let Some(replaced) = &replaced {
            // we only need to remove it from the hash list, because we already replaced it in the by id set,
            // and if this is the independent transaction, it will be replaced by the new transaction below
            self.by_hash.remove(replaced.inner.transaction.hash());
            // Remove from eviction set
            self.remove_eviction_key(replaced);
        }

        // insert transaction by hash
        self.by_hash
            .insert(*tx.inner.transaction.hash(), tx.inner.transaction.clone());

        // contains transactions directly impacted by the new transaction (filled nonce gap)
        let mut promoted = Vec::new();
        // Track whether this transaction was inserted as pending
        let mut inserted_as_pending = false;
        let mut newly_pending = 0usize;
        // now we need to scan the range and mark transactions as pending, if any
        let on_chain_id = AA2dTransactionId::new(tx_id.seq_id, on_chain_nonce);
        // track the next nonce we expect if the transactions are gapless
        let mut next_nonce = on_chain_id.nonce;

        // scan all the transactions with the same nonce key starting with the on chain nonce
        // to check if our new transaction was inserted as pending and perhaps promoted more transactions
        for (existing_id, existing_tx) in self.descendant_txs(&on_chain_id) {
            if existing_id.nonce == next_nonce {
                match existing_id.nonce.cmp(&tx_id.nonce) {
                    std::cmp::Ordering::Less => {
                        // unaffected by our transaction
                    }
                    std::cmp::Ordering::Equal => {
                        if !existing_tx.set_pending(true) {
                            newly_pending += 1;
                        }
                        inserted_as_pending = true;
                    }
                    std::cmp::Ordering::Greater => {
                        // if this was previously not pending we need to promote the transaction
                        let was_pending = existing_tx.set_pending(true);
                        if !was_pending {
                            newly_pending += 1;
                            promoted.push(existing_tx.inner.clone());
                        } else {
                            // already pending, so the rest of the contiguous sequence is
                            // already pending as well
                            break;
                        }
                    }
                }
                // continue ungapped sequence
                next_nonce = existing_id.nonce.saturating_add(1);
            } else {
                // can exit early here because we hit a nonce gap
                break;
            }
        }
        self.pending_count += newly_pending;
        self.queued_count -= newly_pending;

        // Record metrics
        self.metrics.inc_inserted();

        // Create eviction key for the new transaction and add to the single eviction set
        let new_tx_eviction_key = EvictionKey::with_base_fee(Arc::clone(&tx), tx_id, self.base_fee);
        self.by_eviction_order.insert(new_tx_eviction_key);

        if inserted_as_pending {
            if !promoted.is_empty() {
                self.metrics.inc_promoted(promoted.len());
            }
            // if this is the next nonce in line we can mark it as independent
            if tx_id.nonce == on_chain_nonce {
                self.independent_transactions
                    .insert(tx_id.seq_id, tx.inner.clone());
            }

            // Notify active BestAA2dTransactions iterators about new pending transactions.
            self.notify_new_pending(&tx.inner);
            for promoted_tx in &promoted {
                self.notify_new_pending(promoted_tx);
            }

            return Ok(AddedTransaction::Pending(AddedPendingTransaction {
                transaction,
                replaced: replaced.map(|tx| tx.inner.transaction.clone()),
                promoted: promoted.into_iter().map(|tx| tx.transaction).collect(),
                discarded: self.discard(),
            }));
        }

        // Call discard for queued transactions too
        let _ = self.discard();

        Ok(AddedTransaction::Parked {
            transaction,
            replaced: replaced.map(|tx| tx.inner.transaction.clone()),
            subpool: SubPool::Queued,
            queued_reason: Some(QueuedReason::NonceGap),
        })
    }

    /// Adds an expiring nonce transaction to the pool.
    ///
    /// Expiring nonce transactions use the expiring nonce hash as their unique identifier instead
    /// of (sender, nonce_key, nonce). They are always immediately pending since they don't have
    /// sequential nonce dependencies.
    fn add_expiring_nonce_transaction(
        &mut self,
        transaction: Arc<ValidPoolTransaction<TempoPooledTransaction>>,
    ) -> PoolResult<AddedTransaction<TempoPooledTransaction>> {
        let tx_hash = *transaction.hash();
        let expiring_nonce_hash = transaction.transaction.precomputed_expiring_nonce_hash();

        let expiring_nonce_entry = match self.expiring_nonce_txs.entry(expiring_nonce_hash) {
            hash_map::Entry::Occupied(_) => {
                return Err(PoolError::new(tx_hash, PoolErrorKind::AlreadyImported));
            }
            hash_map::Entry::Vacant(entry) => entry,
        };

        // Check per-sender limit and increment the count with a single map lookup
        let sender = transaction.sender();
        match self.txs_by_sender.entry(sender) {
            hash_map::Entry::Occupied(mut count) => {
                if *count.get() >= self.config.max_txs_per_sender {
                    return Err(PoolError::new(
                        tx_hash,
                        PoolErrorKind::SpammerExceededCapacity(sender),
                    ));
                }
                *count.get_mut() += 1;
            }
            hash_map::Entry::Vacant(count) => {
                if self.config.max_txs_per_sender == 0 {
                    return Err(PoolError::new(
                        tx_hash,
                        PoolErrorKind::SpammerExceededCapacity(sender),
                    ));
                }
                count.insert(1);
            }
        }

        // Create pending transaction
        let pending_tx = AA2dStoredTransaction {
            submission_id: {
                let id = self.submission_id;
                self.submission_id = self.submission_id.wrapping_add(1);
                id
            },
            transaction: transaction.clone(),
        };
        let eviction_key =
            ExpiringNonceEvictionKey::from_pending_with_base_fee(&pending_tx, self.base_fee);
        let pending_tx_update = if self.new_transaction_notifier.receiver_count() > 0 {
            Some(pending_tx.clone())
        } else {
            None
        };

        // Insert into expiring nonce map and by_hash
        expiring_nonce_entry.insert(pending_tx);
        self.expiring_nonce_eviction_order.insert(eviction_key);
        if let Some(slot) = transaction.transaction.expiring_nonce_slot() {
            self.slot_to_expiring_nonce_hash
                .insert(slot, expiring_nonce_hash);
        }
        self.by_hash.insert(tx_hash, transaction.clone());

        self.pending_count += 1;

        trace!(target: "txpool", hash = %tx_hash, "Added expiring nonce transaction");

        self.update_metrics();

        // Notify active BestAA2dTransactions iterators about the new pending transaction
        if let Some(pending_tx) = pending_tx_update {
            let _ = self.new_transaction_notifier.send(pending_tx);
        }

        // Expiring nonce transactions are always immediately pending
        Ok(AddedTransaction::Pending(AddedPendingTransaction {
            transaction,
            replaced: None,
            promoted: vec![],
            discarded: self.discard(),
        }))
    }

    /// Returns how many pending and queued transactions are in the pool.
    pub(crate) fn pending_and_queued_txn_count(&self) -> (usize, usize) {
        (self.pending_count, self.queued_count)
    }

    /// Scans the pool to recompute counts for invariant checks.
    #[cfg(test)]
    fn scan_pending_and_queued_txn_count(&self) -> (usize, usize) {
        let (pending_2d, queued_2d) = self.by_id.values().fold((0, 0), |mut acc, tx| {
            if tx.is_pending() {
                acc.0 += 1;
            } else {
                acc.1 += 1;
            }
            acc
        });
        // Expiring nonce txs are always pending
        let expiring_pending = self.expiring_nonce_txs.len();
        (pending_2d + expiring_pending, queued_2d)
    }

    /// Returns all transactions that where submitted with the given [`TransactionOrigin`]
    pub(crate) fn get_transactions_by_origin_iter(
        &self,
        origin: TransactionOrigin,
    ) -> impl Iterator<Item = Arc<ValidPoolTransaction<TempoPooledTransaction>>> + '_ {
        let regular = self
            .by_id
            .values()
            .filter(move |tx| tx.inner.transaction.origin == origin)
            .map(|tx| tx.inner.transaction.clone());
        let expiring = self
            .expiring_nonce_txs
            .values()
            .filter(move |tx| tx.transaction.origin == origin)
            .map(|tx| tx.transaction.clone());
        regular.chain(expiring)
    }

    /// Returns all transactions that where submitted with the given [`TransactionOrigin`]
    pub(crate) fn get_pending_transactions_by_origin_iter(
        &self,
        origin: TransactionOrigin,
    ) -> impl Iterator<Item = Arc<ValidPoolTransaction<TempoPooledTransaction>>> + '_ {
        let regular = self
            .by_id
            .values()
            .filter(move |tx| tx.is_pending() && tx.inner.transaction.origin == origin)
            .map(|tx| tx.inner.transaction.clone());
        // Expiring nonce txs are always pending
        let expiring = self
            .expiring_nonce_txs
            .values()
            .filter(move |tx| tx.transaction.origin == origin)
            .map(|tx| tx.transaction.clone());
        regular.chain(expiring)
    }

    /// Returns all transactions of the address
    pub(crate) fn get_transactions_by_sender_iter(
        &self,
        sender: Address,
    ) -> impl Iterator<Item = Arc<ValidPoolTransaction<TempoPooledTransaction>>> + '_ {
        let regular = self
            .by_id
            .values()
            .filter(move |tx| tx.inner.transaction.sender() == sender)
            .map(|tx| tx.inner.transaction.clone());
        let expiring = self
            .expiring_nonce_txs
            .values()
            .filter(move |tx| tx.transaction.sender() == sender)
            .map(|tx| tx.transaction.clone());
        regular.chain(expiring)
    }

    /// Returns an iterator over all transaction hashes in this pool
    pub(crate) fn all_transaction_hashes_iter(&self) -> impl Iterator<Item = TxHash> {
        self.by_hash.keys().copied()
    }

    /// Returns all transactions from that are queued.
    pub(crate) fn queued_transactions(
        &self,
    ) -> impl Iterator<Item = Arc<ValidPoolTransaction<TempoPooledTransaction>>> {
        self.by_id
            .values()
            .filter(|tx| !tx.is_pending())
            .map(|tx| tx.inner.transaction.clone())
    }

    /// Returns all transactions that are pending.
    pub(crate) fn pending_transactions(
        &self,
    ) -> impl Iterator<Item = Arc<ValidPoolTransaction<TempoPooledTransaction>>> + '_ {
        // Include both regular pending 2D nonce txs and expiring nonce txs
        let regular_pending = self
            .by_id
            .values()
            .filter(|tx| tx.is_pending())
            .map(|tx| tx.inner.transaction.clone());
        let expiring_pending = self
            .expiring_nonce_txs
            .values()
            .map(|tx| tx.transaction.clone());
        regular_pending.chain(expiring_pending)
    }

    /// Appends all transactions to the provided collection.
    pub(crate) fn append_all_transactions(
        &self,
        transactions: &mut AllPoolTransactions<TempoPooledTransaction>,
    ) {
        transactions.pending.reserve(self.pending_count);
        transactions.queued.reserve(self.queued_count);

        for tx in self.by_id.values() {
            if tx.is_pending() {
                transactions.pending.push(tx.inner.transaction.clone());
            } else {
                transactions.queued.push(tx.inner.transaction.clone());
            }
        }

        transactions.pending.extend(
            self.expiring_nonce_txs
                .values()
                .map(|tx| tx.transaction.clone()),
        );
    }

    /// Returns the best, executable transactions for this sub-pool
    pub(crate) fn best_transactions(&self) -> BestAA2dTransactions {
        self.best_transactions_with_base_fee(self.base_fee)
    }

    /// Returns the best, executable transactions for this sub-pool at `base_fee`.
    #[expect(clippy::mutable_key_type)]
    pub(crate) fn best_transactions_with_base_fee(&self, base_fee: u64) -> BestAA2dTransactions {
        let expiring_nonce_order = if base_fee == self.base_fee {
            self.expiring_nonce_eviction_order.clone()
        } else {
            self.expiring_nonce_txs
                .values()
                .map(|tx| ExpiringNonceEvictionKey::from_pending_with_base_fee(tx, base_fee))
                .collect()
        };
        let independent = self
            .independent_transactions
            .values()
            .filter_map(|tx| {
                let id = tx
                    .transaction
                    .transaction
                    .aa_transaction_id()
                    .expect("Independent transaction must have AA transaction ID");
                let tx = self.by_id.get(&id)?;
                Some(tx.inner.clone_into_pending(base_fee))
            })
            .collect();

        BestAA2dTransactions {
            independent,
            by_id: self
                .by_id
                .iter()
                .filter(|(_, tx)| tx.is_pending())
                .map(|(id, tx)| (*id, tx.inner.clone()))
                .collect(),
            expiring_nonce_order,
            invalid: Default::default(),
            new_transaction_receiver: Some(self.new_transaction_notifier.subscribe()),
            last_priority: None,
            base_fee,
        }
    }

    /// Returns the transaction by hash.
    pub(crate) fn get(
        &self,
        tx_hash: &TxHash,
    ) -> Option<Arc<ValidPoolTransaction<TempoPooledTransaction>>> {
        self.by_hash.get(tx_hash).cloned()
    }

    /// Returns the transaction by hash.
    pub(crate) fn get_all<'a, I>(
        &self,
        tx_hashes: I,
    ) -> Vec<Arc<ValidPoolTransaction<TempoPooledTransaction>>>
    where
        I: Iterator<Item = &'a TxHash> + 'a,
    {
        let mut ret = Vec::new();
        for tx_hash in tx_hashes {
            if let Some(tx) = self.get(tx_hash) {
                ret.push(tx);
            }
        }
        ret
    }

    /// Returns pooled transaction elements for the given hashes while respecting the size limit.
    ///
    /// This method collects transactions from the pool, converts them to pooled format,
    /// and tracks the accumulated size. It stops collecting when the limit is exceeded.
    ///
    /// The `accumulated_size` is updated with the total encoded size of returned transactions.
    pub(crate) fn append_pooled_transaction_elements<'a>(
        &self,
        tx_hashes: impl IntoIterator<Item = &'a TxHash>,
        limit: GetPooledTransactionLimit,
        accumulated_size: &mut usize,
        out: &mut Vec<<TempoPooledTransaction as PoolTransaction>::Pooled>,
    ) {
        for tx_hash in tx_hashes {
            let Some(tx) = self.by_hash.get(tx_hash) else {
                continue;
            };

            let encoded_len = tx.transaction.encoded_length();
            let Some(pooled) = tx.transaction.clone_into_pooled().ok() else {
                continue;
            };

            *accumulated_size += encoded_len;
            out.push(pooled.into_inner());

            if limit.exceeds(*accumulated_size) {
                break;
            }
        }
    }

    /// Returns an iterator over all senders in this pool.
    pub(crate) fn senders_iter(&self) -> impl Iterator<Item = &Address> {
        let regular = self
            .by_id
            .values()
            .map(|tx| tx.inner.transaction.sender_ref());
        let expiring = self
            .expiring_nonce_txs
            .values()
            .map(|tx| tx.transaction.sender_ref());
        regular.chain(expiring)
    }

    /// Returns all transactions that _follow_ after the given id but have the same sender.
    ///
    /// NOTE: The range is _inclusive_: if the transaction that belongs to `id` it will be the
    /// first value.
    fn descendant_txs<'a, 'b: 'a>(
        &'a self,
        id: &'b AA2dTransactionId,
    ) -> impl Iterator<Item = (&'a AA2dTransactionId, &'a Arc<AA2dInternalTransaction>)> + 'a {
        self.by_id
            .range(id..)
            .take_while(|(other, _)| id.seq_id == other.seq_id)
    }

    /// Returns all transactions that _follow_ after the given id and have the same sender.
    ///
    /// NOTE: The range is _exclusive_
    fn descendant_txs_exclusive<'a, 'b: 'a>(
        &'a self,
        id: &'b AA2dTransactionId,
    ) -> impl Iterator<Item = (&'a AA2dTransactionId, &'a Arc<AA2dInternalTransaction>)> + 'a {
        self.by_id
            .range((Excluded(id), Unbounded))
            .take_while(|(other, _)| id.seq_id == other.seq_id)
    }

    /// Removes the transaction with the given id from all sets.
    ///
    /// This does __not__ shift the independent transaction forward or mark descendants as pending.
    fn remove_transaction_by_id(
        &mut self,
        id: &AA2dTransactionId,
    ) -> Option<Arc<ValidPoolTransaction<TempoPooledTransaction>>> {
        let tx = self.by_id.remove(id)?;

        // Remove from eviction set
        self.remove_eviction_key(&tx);

        // Clean up cached nonce key slots if this was the last transaction of the sequence
        if self.by_id.range(id.seq_id.range()).next().is_none()
            && let Some(slot) = tx.inner.transaction.transaction.nonce_key_slot()
        {
            self.slot_to_seq_id.remove(&slot);
        }

        self.remove_independent(id);
        let removed_tx = tx.inner.transaction.clone();
        self.by_hash.remove(removed_tx.hash());
        self.remove_from_counts(tx.is_pending());

        // Decrement sender count
        self.decrement_sender_count(removed_tx.sender());

        Some(removed_tx)
    }

    /// Decrements the transaction count for a sender, removing the entry if it reaches zero.
    fn decrement_sender_count(&mut self, sender: Address) {
        if let hash_map::Entry::Occupied(mut entry) = self.txs_by_sender.entry(sender) {
            let count = entry.get_mut();
            *count -= 1;
            if *count == 0 {
                entry.remove();
            }
        }
    }

    fn remove_eviction_key(&mut self, tx: &Arc<AA2dInternalTransaction>) {
        self.by_eviction_order.remove(&EvictionOrderKey::new(
            TempoTipOrdering::default().priority(&tx.inner.transaction.transaction, self.base_fee),
            tx.inner.submission_id,
        ));
    }

    /// Removes the independent transaction if it matches the given id.
    fn remove_independent(&mut self, id: &AA2dTransactionId) -> Option<AA2dStoredTransaction> {
        // Only remove from independent_transactions if this is the independent transaction
        match self.independent_transactions.entry(id.seq_id) {
            hash_map::Entry::Occupied(entry) => {
                // we know it's the independent tx if the tracked tx has the same nonce
                if entry.get().transaction.nonce() == id.nonce {
                    return Some(entry.remove());
                }
            }
            hash_map::Entry::Vacant(_) => {}
        };
        None
    }

    /// Removes the transaction by its hash from all internal sets.
    ///
    /// This batches demotion by seq_id to avoid O(N*N) complexity when removing many
    /// transactions from the same sequence.
    pub(crate) fn remove_transactions<'a, I>(
        &mut self,
        tx_hashes: I,
    ) -> Vec<Arc<ValidPoolTransaction<TempoPooledTransaction>>>
    where
        I: Iterator<Item = &'a TxHash> + 'a,
    {
        let mut txs = Vec::new();
        let mut seq_ids_to_demote: HashMap<AASequenceId, u64> = HashMap::default();

        for tx_hash in tx_hashes {
            if let Some((tx, seq_id)) = self.remove_transaction_by_hash_no_demote(tx_hash) {
                if let Some(id) = seq_id {
                    seq_ids_to_demote
                        .entry(id.seq_id)
                        .and_modify(|min_nonce| {
                            if id.nonce < *min_nonce {
                                *min_nonce = id.nonce;
                            }
                        })
                        .or_insert(id.nonce);
                }
                txs.push(tx);
            }
        }

        // Demote once per seq_id, starting from the minimum removed nonce
        for (seq_id, min_nonce) in seq_ids_to_demote {
            self.demote_from_nonce(&seq_id, min_nonce);
        }

        txs
    }

    /// Removes the transaction by its hash from all internal sets.
    ///
    /// This does __not__ shift the independent transaction forward but it does demote descendants
    /// to queued status since removing a transaction creates a nonce gap.
    fn remove_transaction_by_hash(
        &mut self,
        tx_hash: &B256,
    ) -> Option<Arc<ValidPoolTransaction<TempoPooledTransaction>>> {
        let (tx, id) = self.remove_transaction_by_hash_no_demote(tx_hash)?;

        // Demote all descendants to queued status since removing this transaction creates a gap
        if let Some(id) = id {
            self.demote_descendants(&id);
        }

        Some(tx)
    }

    /// Internal helper that removes a transaction without demoting descendants.
    ///
    /// Returns the removed transaction and its AA2dTransactionId (if it was a 2D nonce tx).
    fn remove_transaction_by_hash_no_demote(
        &mut self,
        tx_hash: &B256,
    ) -> Option<(
        Arc<ValidPoolTransaction<TempoPooledTransaction>>,
        Option<AA2dTransactionId>,
    )> {
        let tx = self.by_hash.remove(tx_hash)?;

        // Check if this is an expiring nonce transaction
        if tx.transaction.is_expiring_nonce() {
            let tx =
                self.remove_expiring_nonce_tx(&tx.transaction.precomputed_expiring_nonce_hash())?;
            return Some((tx, None));
        }

        // Regular 2D nonce transaction
        let id = tx
            .transaction
            .aa_transaction_id()
            .expect("is AA transaction");
        self.remove_transaction_by_id(&id)?;

        Some((tx, Some(id)))
    }

    /// Demotes all descendants of the given transaction to queued status (`is_pending = false`).
    ///
    /// This should be called after removing a transaction to ensure descendants don't remain
    /// marked as pending when they're no longer executable due to the nonce gap.
    fn demote_descendants(&mut self, id: &AA2dTransactionId) {
        self.demote_from_nonce(&id.seq_id, id.nonce);
    }

    /// Demotes all transactions for a seq_id with nonce > min_nonce to queued status.
    ///
    /// This is used both for single-tx removal (demote_descendants) and batch removal
    /// where we want to demote once per seq_id starting from the minimum removed nonce.
    fn demote_from_nonce(&mut self, seq_id: &AASequenceId, min_nonce: u64) {
        let start_id = AA2dTransactionId::new(*seq_id, min_nonce);
        for (_, tx) in self
            .by_id
            .range((Excluded(&start_id), Unbounded))
            .take_while(|(other, _)| *seq_id == other.seq_id)
        {
            if tx.set_pending(false) {
                self.pending_count -= 1;
                self.queued_count += 1;
            }
        }
    }

    /// Removes and returns all matching transactions and their dependent transactions from the
    /// pool.
    pub(crate) fn remove_transactions_and_descendants<'a, I>(
        &mut self,
        hashes: I,
    ) -> Vec<Arc<ValidPoolTransaction<TempoPooledTransaction>>>
    where
        I: Iterator<Item = &'a TxHash> + 'a,
    {
        let mut removed = Vec::new();
        for hash in hashes {
            if let Some(tx) = self.remove_transaction_by_hash(hash) {
                let id = tx.transaction.aa_transaction_id();
                removed.push(tx);
                if let Some(id) = id {
                    self.remove_descendants(&id, &mut removed);
                }
            }
        }
        removed
    }

    /// Removes all transactions from the given sender.
    pub(crate) fn remove_transactions_by_sender(
        &mut self,
        sender_id: Address,
    ) -> Vec<Arc<ValidPoolTransaction<TempoPooledTransaction>>> {
        let mut removed = Vec::new();
        let txs = self
            .get_transactions_by_sender_iter(sender_id)
            .collect::<Vec<_>>();
        for tx in txs {
            if tx.transaction.is_expiring_nonce() {
                if let Some(tx) =
                    self.remove_expiring_nonce_tx(&tx.transaction.precomputed_expiring_nonce_hash())
                {
                    removed.push(tx);
                }
            } else if let Some(tx) = tx
                .transaction
                .aa_transaction_id()
                .and_then(|id| self.remove_transaction_by_id(&id))
            {
                removed.push(tx);
            }
        }
        removed
    }

    /// Removes _only_ the descendants of the given transaction from this pool.
    ///
    /// All removed transactions are added to the `removed` vec.
    fn remove_descendants(
        &mut self,
        tx: &AA2dTransactionId,
        removed: &mut Vec<Arc<ValidPoolTransaction<TempoPooledTransaction>>>,
    ) {
        let mut id = *tx;

        // this will essentially pop _all_ descendant transactions one by one
        loop {
            let descendant = self.descendant_txs_exclusive(&id).map(|(id, _)| *id).next();
            if let Some(descendant) = descendant {
                if let Some(tx) = self.remove_transaction_by_id(&descendant) {
                    removed.push(tx)
                }
                id = descendant;
            } else {
                return;
            }
        }
    }

    /// Updates the internal state based on the state changes of the `NonceManager` [`NONCE_PRECOMPILE_ADDRESS`].
    ///
    /// This takes changed [`AASequenceId`]s with their current on-chain nonce.
    ///
    /// This will prune mined transactions and promote unblocked transactions if any, returns `(promoted, mined)`
    #[cfg(test)]
    pub(crate) fn on_nonce_changes(
        &mut self,
        on_chain_ids: HashMap<AASequenceId, u64>,
    ) -> PoolUpdateResult {
        trace!(target: "txpool::2d", ?on_chain_ids, "processing nonce changes");

        self.on_nonce_changes_iter(on_chain_ids)
    }

    fn on_nonce_changes_iter(
        &mut self,
        on_chain_ids: impl IntoIterator<Item = (AASequenceId, u64)>,
    ) -> PoolUpdateResult {
        let mut promoted = Vec::new();
        let mut mined_ids = Vec::new();

        // we assume the set of changed senders is smaller than the individual accounts
        'changes: for (sender_id, on_chain_nonce) in on_chain_ids {
            let mut iter = self
                .by_id
                .range_mut((sender_id.start_bound(), Unbounded))
                .take_while(move |(other, _)| sender_id == other.seq_id)
                .peekable();

            let Some(mut current) = iter.next() else {
                continue;
            };

            // track mined transactions
            'mined: loop {
                if current.0.nonce < on_chain_nonce {
                    mined_ids.push(*current.0);
                    let Some(next) = iter.next() else {
                        continue 'changes;
                    };
                    current = next;
                } else {
                    break 'mined;
                }
            }

            // Process remaining transactions starting from `current` (which is >= on_chain_nonce)
            let mut next_nonce = on_chain_nonce;
            let mut newly_pending = 0usize;
            let mut newly_queued = 0usize;
            for (existing_id, existing_tx) in std::iter::once(current).chain(iter) {
                if existing_id.nonce == next_nonce {
                    // Promote if transaction was previously queued (not pending)
                    let was_pending = existing_tx.set_pending(true);
                    if !was_pending {
                        newly_pending += 1;
                        promoted.push(existing_tx.inner.transaction.clone());
                    }

                    if existing_id.nonce == on_chain_nonce {
                        // if this is the on chain nonce we can mark it as the next independent transaction
                        self.independent_transactions
                            .insert(existing_id.seq_id, existing_tx.inner.clone());
                    }

                    next_nonce = next_nonce.saturating_add(1);
                } else {
                    // Gap detected - mark this and all remaining transactions as non-pending
                    if existing_tx.set_pending(false) {
                        newly_queued += 1;
                    }
                }
            }
            self.pending_count += newly_pending;
            self.queued_count -= newly_pending;
            self.pending_count -= newly_queued;
            self.queued_count += newly_queued;

            // If no transaction was found at the on-chain nonce (next_nonce unchanged),
            // remove any stale independent transaction entry for this seq_id.
            // This handles reorgs where the on-chain nonce decreases.
            if next_nonce == on_chain_nonce {
                self.independent_transactions.remove(&sender_id);
            }
        }

        // actually remove mined transactions
        let mut mined = Vec::with_capacity(mined_ids.len());
        for id in mined_ids {
            if let Some(removed) = self.remove_transaction_by_id(&id) {
                mined.push(removed);
            }
        }

        (promoted, mined)
    }

    /// Removes lowest-priority transactions if the pool is above capacity.
    ///
    /// This evicts transactions with the lowest priority (based on [`TempoTipOrdering`])
    /// to prevent DoS attacks where adversaries use vanity addresses with many leading zeroes
    /// to avoid eviction.
    ///
    /// Evicts queued transactions first (up to queued_limit), then pending if needed.
    /// Counts are computed lazily by scanning the eviction set. Eviction order
    /// is only rebuilt when the pool is actually over limit.
    ///
    /// Note: Only `max_txs` is enforced here; `max_size` is intentionally not checked for 2D pools
    /// since the protocol pool already enforces size-based limits as a primary defense.
    fn discard(&mut self) -> Vec<Arc<ValidPoolTransaction<TempoPooledTransaction>>> {
        let mut removed = Vec::new();

        // Compute counts lazily by scanning the pool
        let (pending_count, queued_count) = self.pending_and_queued_txn_count();

        // Evict queued transactions if over queued limit (lowest priority first)
        if queued_count > self.config.queued_limit.max_txs {
            let queued_excess = queued_count - self.config.queued_limit.max_txs;
            self.evict_lowest_priority(queued_excess, false, &mut removed);
        }

        // Evict pending transactions if over pending limit (lowest priority first)
        if pending_count > self.config.pending_limit.max_txs {
            let pending_excess = pending_count - self.config.pending_limit.max_txs;
            self.evict_lowest_priority(pending_excess, true, &mut removed);
        }

        if !removed.is_empty() {
            self.metrics.inc_removed(removed.len());
        }

        removed
    }

    /// Evicts the lowest-priority transactions from the pool.
    ///
    /// Scans the single eviction set (ordered by priority) and filters by `is_pending`
    /// to find queued or pending transactions to evict. This is a best-effort scan
    /// that checks a bool for each transaction.
    fn evict_lowest_priority(
        &mut self,
        count: usize,
        evict_pending: bool,
        removed: &mut Vec<Arc<ValidPoolTransaction<TempoPooledTransaction>>>,
    ) {
        if count == 0 {
            return;
        }

        removed.reserve(count);

        if evict_pending {
            // For pending eviction, consider both regular 2D txs and expiring nonce txs
            for _ in 0..count {
                if let Some(tx) = self.evict_one_pending() {
                    removed.push(tx);
                } else {
                    break;
                }
            }
        } else {
            // For queued, only look at by_eviction_order (expiring nonce txs are always pending)
            let to_remove: Vec<_> = self
                .by_eviction_order
                .iter()
                .filter(|key| !key.is_pending())
                .map(|key| key.tx_id)
                .take(count)
                .collect();

            for id in to_remove {
                if let Some(tx) = self.remove_transaction_by_id(&id) {
                    removed.push(tx);
                }
            }
        }
    }

    /// Evicts one pending transaction, considering both regular 2D and expiring nonce txs.
    /// Evicts the transaction with lowest priority; ties broken by submission order (newer first).
    fn evict_one_pending(&mut self) -> Option<Arc<ValidPoolTransaction<TempoPooledTransaction>>> {
        let worst_2d = self
            .by_eviction_order
            .iter()
            .find(|key| key.is_pending())
            .map(|key| (key.tx_id, key.priority().clone(), key.submission_id()));

        let worst_expiring = self
            .expiring_nonce_eviction_order
            .first()
            .map(|key| (key.priority().clone(), key.submission_id()));

        match (worst_2d, worst_expiring) {
            (Some((id, pri_2d, sid_2d)), Some((pri_exp, sid_exp))) => {
                // Same ordering as EvictionKey::Ord: lower priority first, newer first.
                let evict_expiring = pri_exp
                    .cmp(&pri_2d)
                    .then_with(|| sid_2d.cmp(&sid_exp))
                    .is_le();
                if evict_expiring {
                    self.evict_worst_expiring_nonce_tx()
                } else {
                    self.evict_2d_pending_tx(&id)
                }
            }
            (Some((id, ..)), None) => self.evict_2d_pending_tx(&id),
            (None, Some(_)) => self.evict_worst_expiring_nonce_tx(),
            (None, None) => None,
        }
    }

    /// Evicts a regular 2D pending transaction by ID.
    fn evict_2d_pending_tx(
        &mut self,
        id: &AA2dTransactionId,
    ) -> Option<Arc<ValidPoolTransaction<TempoPooledTransaction>>> {
        let tx = self.remove_transaction_by_id(id)?;
        self.demote_descendants(id);
        Some(tx)
    }

    /// Evicts the worst expiring nonce transaction.
    ///
    /// Use when eviction has selected the front of
    /// `expiring_nonce_eviction_order`; this pops it directly instead of doing a
    /// keyed removal.
    fn evict_worst_expiring_nonce_tx(
        &mut self,
    ) -> Option<Arc<ValidPoolTransaction<TempoPooledTransaction>>> {
        let eviction_key = self.expiring_nonce_eviction_order.pop_first()?;
        let pending_tx = self
            .expiring_nonce_txs
            .remove(&eviction_key.expiring_hash())?;

        Some(self.remove_expiring_nonce_pending_tx(pending_tx))
    }

    /// Removes an expiring nonce transaction by hash.
    ///
    /// Use when removal starts from a hash, such as direct removal, sender
    /// removal, or nonce-state inclusion. This path removes the matching
    /// eviction key by lookup.
    fn remove_expiring_nonce_tx(
        &mut self,
        expiring_hash: &B256,
    ) -> Option<Arc<ValidPoolTransaction<TempoPooledTransaction>>> {
        let pending_tx = self.expiring_nonce_txs.remove(expiring_hash)?;
        self.expiring_nonce_eviction_order
            .remove(&EvictionOrderKey::new(
                TempoTipOrdering::default()
                    .priority(&pending_tx.transaction.transaction, self.base_fee),
                pending_tx.submission_id,
            ));
        Some(self.remove_expiring_nonce_pending_tx(pending_tx))
    }

    /// Removes secondary state for an already-detached expiring nonce transaction.
    ///
    /// Call only after removing the transaction from `expiring_nonce_txs` and
    /// handling its eviction-order entry. Shared by the pop-first eviction path
    /// and the hash-based removal path.
    fn remove_expiring_nonce_pending_tx(
        &mut self,
        pending_tx: AA2dStoredTransaction,
    ) -> Arc<ValidPoolTransaction<TempoPooledTransaction>> {
        self.by_hash.remove(pending_tx.transaction.hash());
        if let Some(slot) = pending_tx.transaction.transaction.expiring_nonce_slot() {
            self.slot_to_expiring_nonce_hash.remove(&slot);
        }
        self.decrement_sender_count(pending_tx.transaction.sender());
        self.pending_count -= 1;
        pending_tx.transaction
    }

    fn remove_from_counts(&mut self, pending: bool) {
        if pending {
            self.pending_count -= 1;
        } else {
            self.queued_count -= 1;
        }
    }

    /// Returns a reference to the metrics for this pool
    pub fn metrics(&self) -> &AA2dPoolMetrics {
        &self.metrics
    }

    /// Returns `true` if the transaction with the given hash is already included in this pool.
    pub(crate) fn contains(&self, tx_hash: &TxHash) -> bool {
        self.by_hash.contains_key(tx_hash)
    }

    /// Returns hashes of transactions in the pool that can be propagated.
    pub(crate) fn pooled_transactions_hashes_iter(&self) -> impl Iterator<Item = TxHash> {
        self.by_hash
            .values()
            .filter(|tx| tx.propagate)
            .map(|tx| *tx.hash())
    }

    /// Returns transactions in the pool that can be propagated
    pub(crate) fn pooled_transactions_iter(
        &self,
    ) -> impl Iterator<Item = Arc<ValidPoolTransaction<TempoPooledTransaction>>> {
        self.by_hash.values().filter(|tx| tx.propagate).cloned()
    }

    const fn next_id(&mut self) -> u64 {
        let id = self.submission_id;
        self.submission_id = self.submission_id.wrapping_add(1);
        id
    }

    /// Caches the 2D nonce key slot for the given sender and nonce key.
    fn record_2d_slot(&mut self, transaction: &TempoPooledTransaction) {
        let address = transaction.sender();
        let nonce_key = transaction.nonce_key().unwrap_or_default();
        let Some(slot) = transaction.nonce_key_slot() else {
            return;
        };

        trace!(target: "txpool::2d", ?address, ?nonce_key, "recording 2d nonce slot");
        let seq_id = AASequenceId::new(address, nonce_key);

        if self.slot_to_seq_id.insert(slot, seq_id).is_none() {
            self.metrics.inc_nonce_key_count(1);
        }
    }

    /// Processes nonce-precompile storage updates and updates internal state accordingly.
    pub(crate) fn on_state_updates(
        &mut self,
        state: &AddressMap<BundleAccount>,
    ) -> PoolUpdateResult {
        self.state_update_nonce_changes.clear();
        self.state_update_included_expiring_nonce_hashes.clear();

        let Some(nonce_state) = state.get(&NONCE_PRECOMPILE_ADDRESS) else {
            return (Vec::new(), Vec::new());
        };

        let mut changes = std::mem::take(&mut self.state_update_nonce_changes);
        let mut included_expiring_nonce_hashes =
            std::mem::take(&mut self.state_update_included_expiring_nonce_hashes);

        // Process known 2D nonce slot changes.
        for (slot, value) in nonce_state.storage.iter() {
            if let Some(seq_id) = self.slot_to_seq_id.get(slot) {
                changes.insert(*seq_id, value.present_value.saturating_to());
            }
            // Detect included expiring nonce transactions via their
            // `expiring_nonce_seen` slot being set to a non-zero value.
            if !value.present_value.is_zero()
                && let Some(expiring_nonce_hash) = self.slot_to_expiring_nonce_hash.get(slot)
            {
                included_expiring_nonce_hashes.push(*expiring_nonce_hash);
            }
        }

        let (promoted, mut mined) = self.on_nonce_changes_iter(changes.drain());

        // Remove included expiring nonce transactions
        for expiring_nonce_hash in included_expiring_nonce_hashes.drain(..) {
            if let Some(tx) = self.remove_expiring_nonce_tx(&expiring_nonce_hash) {
                mined.push(tx);
            }
        }
        self.state_update_nonce_changes = changes;
        self.state_update_included_expiring_nonce_hashes = included_expiring_nonce_hashes;

        // Record metrics for all changes
        if !promoted.is_empty() {
            self.metrics.inc_promoted(promoted.len());
        }
        if !mined.is_empty() {
            self.metrics.inc_removed(mined.len());
        }
        self.update_metrics();

        (promoted, mined)
    }

    /// Asserts that all assumptions are valid.
    #[cfg(test)]
    pub(crate) fn assert_invariants(&self) {
        // Basic size constraints
        assert!(
            self.independent_transactions.len() <= self.by_id.len(),
            "independent_transactions.len() ({}) > by_id.len() ({})",
            self.independent_transactions.len(),
            self.by_id.len()
        );
        // by_hash contains both regular 2D nonce txs (in by_id) and expiring nonce txs
        assert_eq!(
            self.by_id.len() + self.expiring_nonce_txs.len(),
            self.by_hash.len(),
            "by_id.len() ({}) + expiring_nonce_txs.len() ({}) != by_hash.len() ({})",
            self.by_id.len(),
            self.expiring_nonce_txs.len(),
            self.by_hash.len()
        );
        assert_eq!(
            self.expiring_nonce_txs.len(),
            self.expiring_nonce_eviction_order.len(),
            "expiring_nonce_txs.len() ({}) != expiring_nonce_eviction_order.len() ({})",
            self.expiring_nonce_txs.len(),
            self.expiring_nonce_eviction_order.len()
        );
        assert_eq!(
            self.by_id.len(),
            self.by_eviction_order.len(),
            "by_id.len() ({}) != by_eviction_order.len() ({})",
            self.by_id.len(),
            self.by_eviction_order.len()
        );

        // All independent transactions must exist in by_id
        for (seq_id, independent_tx) in &self.independent_transactions {
            let tx_id = independent_tx
                .transaction
                .transaction
                .aa_transaction_id()
                .expect("Independent transaction must have AA transaction ID");
            assert!(
                self.by_id.contains_key(&tx_id),
                "Independent transaction {tx_id:?} not in by_id"
            );
            assert_eq!(
                seq_id, &tx_id.seq_id,
                "Independent transactions sequence ID {seq_id:?} does not match transaction sequence ID {tx_id:?}"
            );

            // Independent transactions must be pending
            let tx_in_pool = self.by_id.get(&tx_id).unwrap();
            assert!(
                tx_in_pool.is_pending(),
                "Independent transaction {tx_id:?} is not pending"
            );

            // Independent transaction should match the one in by_id
            assert_eq!(
                independent_tx.transaction.hash(),
                tx_in_pool.inner.transaction.hash(),
                "Independent transaction hash mismatch for {tx_id:?}"
            );
        }

        // Each sender should have at most one transaction in independent set
        let mut seen_senders = std::collections::HashSet::new();
        for id in self.independent_transactions.keys() {
            assert!(
                seen_senders.insert(*id),
                "Duplicate sender {id:?} in independent transactions"
            );
        }

        // Verify by_hash integrity
        for (hash, tx) in &self.by_hash {
            // Hash should match transaction hash
            assert_eq!(
                tx.hash(),
                hash,
                "Hash mismatch in by_hash: expected {:?}, got {:?}",
                hash,
                tx.hash()
            );

            // Expiring nonce txs are stored in expiring_nonce_txs, not by_id
            if tx.transaction.is_expiring_nonce() {
                assert!(
                    self.expiring_nonce_txs
                        .contains_key(&tx.transaction.precomputed_expiring_nonce_hash()),
                    "Expiring nonce transaction with hash {hash:?} in by_hash but not in expiring_nonce_txs"
                );
                continue;
            }

            // Transaction in by_hash should exist in by_id
            let id = tx
                .transaction
                .aa_transaction_id()
                .expect("Transaction in pool should be AA transaction");
            assert!(
                self.by_id.contains_key(&id),
                "Transaction with hash {hash:?} in by_hash but not in by_id"
            );

            // The transaction in by_id should have the same hash
            let tx_in_by_id = &self.by_id.get(&id).unwrap().inner.transaction;
            assert_eq!(
                tx.hash(),
                tx_in_by_id.hash(),
                "Transaction hash mismatch between by_hash and by_id for id {id:?}"
            );
        }

        // Verify by_id integrity
        for (id, tx) in &self.by_id {
            // Transaction in by_id should exist in by_hash
            let hash = tx.inner.transaction.hash();
            assert!(
                self.by_hash.contains_key(hash),
                "Transaction with id {id:?} in by_id but not in by_hash"
            );

            // The transaction should have the correct AA ID
            let tx_id = tx
                .inner
                .transaction
                .transaction
                .aa_transaction_id()
                .expect("Transaction in pool should be AA transaction");
            assert_eq!(
                &tx_id, id,
                "Transaction ID mismatch: expected {id:?}, got {tx_id:?}"
            );

            // If THIS transaction is the independent transaction for its sequence, it must be pending
            if let Some(independent_tx) = self.independent_transactions.get(&id.seq_id)
                && independent_tx.transaction.hash() == tx.inner.transaction.hash()
            {
                assert!(
                    tx.is_pending(),
                    "Transaction {id:?} is in independent set but not pending"
                );
            }

            let expected_priority = TempoTipOrdering::default()
                .priority(&tx.inner.transaction.transaction, self.base_fee);
            let expected_order =
                EvictionOrderKey::new(expected_priority.clone(), tx.inner.submission_id);
            let Some(eviction_key) = self.by_eviction_order.get(&expected_order) else {
                panic!("Transaction with id {id:?} in by_id but not in by_eviction_order");
            };
            assert_eq!(
                eviction_key.tx_id, *id,
                "Eviction key for transaction {id:?} has mismatched transaction id"
            );
            assert_eq!(
                eviction_key.priority(),
                &expected_priority,
                "Eviction key for transaction {id:?} has stale priority"
            );
            assert_eq!(
                eviction_key.tx.inner.transaction.hash(),
                tx.inner.transaction.hash(),
                "Eviction key for transaction {id:?} has mismatched transaction hash"
            );
        }

        for key in &self.by_eviction_order {
            let Some(tx) = self.by_id.get(&key.tx_id) else {
                panic!("Eviction key {:?} not in by_id", key.tx_id);
            };
            assert_eq!(
                key.submission_id(),
                tx.inner.submission_id,
                "Eviction key {:?} has mismatched submission id",
                key.tx_id
            );
            let expected_priority = TempoTipOrdering::default()
                .priority(&tx.inner.transaction.transaction, self.base_fee);
            assert_eq!(
                key.priority(),
                &expected_priority,
                "Eviction key {:?} has stale priority",
                key.tx_id
            );
            assert_eq!(
                key.tx.inner.transaction.hash(),
                tx.inner.transaction.hash(),
                "Eviction key {:?} has mismatched transaction hash",
                key.tx_id
            );
        }

        // Verify pending/queued consistency
        // pending_and_queued_txn_count includes expiring nonce txs in pending count
        let (pending_count, queued_count) = self.pending_and_queued_txn_count();
        let (scanned_pending, scanned_queued) = self.scan_pending_and_queued_txn_count();
        assert_eq!(
            (pending_count, queued_count),
            (scanned_pending, scanned_queued),
            "cached counts ({pending_count}, {queued_count}) != scanned counts ({scanned_pending}, {scanned_queued})"
        );
        assert_eq!(
            pending_count + queued_count,
            self.by_id.len() + self.expiring_nonce_txs.len(),
            "Pending ({}) + queued ({}) != total transactions (by_id: {} + expiring: {})",
            pending_count,
            queued_count,
            self.by_id.len(),
            self.expiring_nonce_txs.len()
        );

        // Verify quota compliance - counts don't exceed limits
        assert!(
            pending_count <= self.config.pending_limit.max_txs,
            "pending_count {} exceeds limit {}",
            pending_count,
            self.config.pending_limit.max_txs
        );
        assert!(
            queued_count <= self.config.queued_limit.max_txs,
            "queued_count {} exceeds limit {}",
            queued_count,
            self.config.queued_limit.max_txs
        );

        // Verify expiring nonce txs integrity
        for (hash, pending_tx) in &self.expiring_nonce_txs {
            let tx_hash = *pending_tx.transaction.hash();
            assert!(
                self.by_hash.contains_key(&tx_hash),
                "Expiring nonce tx {tx_hash:?} not in by_hash (expiring hash {hash:?})"
            );
            assert!(
                self.expiring_nonce_eviction_order
                    .iter()
                    .any(|key| key.expiring_hash() == *hash
                        && key.submission_id() == pending_tx.submission_id),
                "Expiring nonce tx {tx_hash:?} not in expiring_nonce_eviction_order"
            );
            assert!(
                pending_tx.transaction.transaction.is_expiring_nonce(),
                "Transaction in expiring_nonce_txs is not an expiring nonce tx"
            );
        }

        for key in &self.expiring_nonce_eviction_order {
            let expiring_hash = key.expiring_hash();
            let Some(pending_tx) = self.expiring_nonce_txs.get(&expiring_hash) else {
                panic!("Expiring nonce eviction key {expiring_hash:?} not in expiring_nonce_txs");
            };
            assert_eq!(
                key.submission_id(),
                pending_tx.submission_id,
                "Expiring nonce eviction key {expiring_hash:?} has mismatched submission id"
            );
            assert_eq!(
                key.transaction.hash(),
                pending_tx.transaction.hash(),
                "Expiring nonce eviction key {expiring_hash:?} has mismatched transaction hash"
            );
        }
    }
}

/// Default maximum number of transactions per sender in the AA 2D pool.
///
/// This limit prevents a single sender from monopolizing pool capacity.
pub const DEFAULT_MAX_TXS_PER_SENDER: usize = 16;

/// Settings for the [`AA2dPoolConfig`]
#[derive(Debug, Clone)]
pub struct AA2dPoolConfig {
    /// Price bump (in %) for the transaction pool underpriced check.
    pub price_bump_config: PriceBumpConfig,
    /// Maximum number of pending (executable) transactions
    pub pending_limit: SubPoolLimit,
    /// Maximum number of queued (non-executable) transactions
    pub queued_limit: SubPoolLimit,
    /// Maximum number of transactions per sender.
    ///
    /// Prevents a single sender from monopolizing pool capacity (DoS protection).
    pub max_txs_per_sender: usize,
}

impl Default for AA2dPoolConfig {
    fn default() -> Self {
        Self {
            price_bump_config: PriceBumpConfig::default(),
            pending_limit: SubPoolLimit::default(),
            queued_limit: SubPoolLimit::default(),
            max_txs_per_sender: DEFAULT_MAX_TXS_PER_SENDER,
        }
    }
}

#[derive(Debug, Clone)]
struct AA2dStoredTransaction {
    submission_id: u64,
    transaction: Arc<ValidPoolTransaction<TempoPooledTransaction>>,
}

impl AA2dStoredTransaction {
    fn new(
        submission_id: u64,
        transaction: Arc<ValidPoolTransaction<TempoPooledTransaction>>,
    ) -> Self {
        Self {
            submission_id,
            transaction,
        }
    }

    fn clone_into_pending(&self, base_fee: u64) -> PendingTransaction<TxOrdering> {
        PendingTransaction {
            submission_id: self.submission_id,
            priority: TempoTipOrdering::default().priority(&self.transaction.transaction, base_fee),
            transaction: self.transaction.clone(),
        }
    }
}

#[derive(Debug)]
struct AA2dInternalTransaction {
    /// Keeps track of the transaction without an ordering priority.
    inner: AA2dStoredTransaction,
    /// Whether this transaction is pending/executable.
    ///
    /// If it's not pending, it is queued.
    ///
    /// Uses `AtomicBool` so we can mutate this flag without removing/reinserting
    /// the transaction from the eviction set. This allows a single eviction set for
    /// all transactions, with pending/queued filtering done at eviction time.
    is_pending: AtomicBool,
}

impl AA2dInternalTransaction {
    /// Returns whether this transaction is pending/executable.
    fn is_pending(&self) -> bool {
        self.is_pending.load(Ordering::Relaxed)
    }

    /// Sets the pending status of this transaction, returning the previous value.
    fn set_pending(&self, pending: bool) -> bool {
        self.is_pending.swap(pending, Ordering::Relaxed)
    }
}

/// Minimal ordering key for eviction set lookups.
///
/// [`EvictionKey`] and [`ExpiringNonceEvictionKey`] borrow as this type so BTreeSet
/// lookup APIs like `remove`, `contains`, and `get` do not need to construct a full
/// key with a cloned transaction.
#[derive(Debug, Clone, PartialEq, Eq)]
struct EvictionOrderKey {
    priority: Priority<u64>,
    submission_id: u64,
}

impl EvictionOrderKey {
    fn new(priority: Priority<u64>, submission_id: u64) -> Self {
        Self {
            priority,
            submission_id,
        }
    }
}

impl Ord for EvictionOrderKey {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        self.priority
            .cmp(&other.priority)
            .then_with(|| other.submission_id.cmp(&self.submission_id))
    }
}

impl PartialOrd for EvictionOrderKey {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        Some(self.cmp(other))
    }
}

/// Ordering key for `AA2dPool::expiring_nonce_eviction_order`.
///
/// This mirrors `EvictionKey` for expiring nonce transactions, which are not
/// stored in `by_id` and therefore cannot use `AA2dTransactionId`. The key
/// carries the transaction so `BestAA2dTransactions` can clone only the ordered
/// index and pop expiring nonce transactions directly from it.
///
/// Order:
/// 1. Priority ascending (lowest priority evicted first)
/// 2. Submission ID descending (newer transactions evicted first among equal priority)
///
/// `submission_id` is unique for live entries.
#[derive(Debug, Clone)]
struct ExpiringNonceEvictionKey {
    order: EvictionOrderKey,
    transaction: Arc<ValidPoolTransaction<TempoPooledTransaction>>,
}

impl ExpiringNonceEvictionKey {
    fn from_pending_with_base_fee(tx: &AA2dStoredTransaction, base_fee: u64) -> Self {
        Self {
            order: EvictionOrderKey::new(
                TempoTipOrdering::default().priority(&tx.transaction.transaction, base_fee),
                tx.submission_id,
            ),
            transaction: tx.transaction.clone(),
        }
    }

    fn from_pending_owned(tx: PendingTransaction<TxOrdering>) -> Self {
        Self {
            order: EvictionOrderKey::new(tx.priority, tx.submission_id),
            transaction: tx.transaction,
        }
    }

    fn into_transaction(self) -> PendingTransaction<TxOrdering> {
        PendingTransaction {
            submission_id: self.order.submission_id,
            priority: self.order.priority,
            transaction: self.transaction,
        }
    }

    fn priority(&self) -> &Priority<u64> {
        &self.order.priority
    }

    fn submission_id(&self) -> u64 {
        self.order.submission_id
    }

    fn expiring_hash(&self) -> B256 {
        self.transaction
            .transaction
            .precomputed_expiring_nonce_hash()
    }
}

impl Borrow<EvictionOrderKey> for ExpiringNonceEvictionKey {
    fn borrow(&self) -> &EvictionOrderKey {
        &self.order
    }
}

impl PartialEq for ExpiringNonceEvictionKey {
    fn eq(&self, other: &Self) -> bool {
        self.order == other.order
    }
}

impl Eq for ExpiringNonceEvictionKey {}

impl Ord for ExpiringNonceEvictionKey {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        self.order.cmp(&other.order)
    }
}

impl PartialOrd for ExpiringNonceEvictionKey {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        Some(self.cmp(other))
    }
}

/// Key for ordering transactions by eviction priority.
///
/// Orders by:
/// 1. Priority ascending (lowest priority evicted first)
/// 2. Submission ID descending (newer transactions evicted first among same priority)
///
/// This is the inverse of the execution order (where highest priority, oldest submission wins).
/// Newer transactions are evicted first to preserve older transactions that have been waiting longer.
#[derive(Debug, Clone)]
struct EvictionKey {
    /// The wrapped transaction, used to read live pending/queued status.
    tx: Arc<AA2dInternalTransaction>,
    /// The transaction's unique identifier (cached for lookup during eviction).
    /// We cache this because deriving it from the transaction requires
    /// `aa_transaction_id()` which returns an Option and does more work.
    tx_id: AA2dTransactionId,
    /// Priority and submission ID snapshot used for eviction ordering.
    order: EvictionOrderKey,
}

impl EvictionKey {
    fn with_base_fee(
        tx: Arc<AA2dInternalTransaction>,
        tx_id: AA2dTransactionId,
        base_fee: u64,
    ) -> Self {
        let priority =
            TempoTipOrdering::default().priority(&tx.inner.transaction.transaction, base_fee);
        let submission_id = tx.inner.submission_id;
        Self {
            tx,
            tx_id,
            order: EvictionOrderKey::new(priority, submission_id),
        }
    }

    /// Returns the transaction's priority.
    fn priority(&self) -> &Priority<u64> {
        &self.order.priority
    }

    /// Returns the submission ID.
    fn submission_id(&self) -> u64 {
        self.order.submission_id
    }

    /// Returns whether this transaction is pending.
    fn is_pending(&self) -> bool {
        self.tx.is_pending()
    }
}

impl Borrow<EvictionOrderKey> for EvictionKey {
    fn borrow(&self) -> &EvictionOrderKey {
        &self.order
    }
}

impl PartialEq for EvictionKey {
    fn eq(&self, other: &Self) -> bool {
        self.submission_id() == other.submission_id()
    }
}

impl Eq for EvictionKey {}

impl Ord for EvictionKey {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        // Lower priority first (evict lowest priority)
        self.priority()
            .cmp(other.priority())
            // Then newer submission first (evict newer transactions among same priority)
            // This preserves older transactions that have been waiting longer
            .then_with(|| other.submission_id().cmp(&self.submission_id()))
    }
}

impl PartialOrd for EvictionKey {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        Some(self.cmp(other))
    }
}

/// Maximum number of new transactions to drain from the channel per `next()` call.
const MAX_NEW_TRANSACTIONS_PER_BATCH: usize = 16;

/// Determines how a newly received transaction should be handled based on its priority
/// relative to transactions already yielded by the iterator.
enum IncomingAA2dTransaction {
    /// Priority ≤ last yielded — safe to add to both `by_id` and `independent`.
    Process(PendingTransaction<TxOrdering>),
    /// Priority > last yielded — add only to `by_id` for nonce chain lookups, not `independent`.
    Stash(PendingTransaction<TxOrdering>),
}

enum PoppedAA2dTransaction {
    Regular(AA2dTransactionId, PendingTransaction<TxOrdering>),
    Expiring(PendingTransaction<TxOrdering>),
}

/// A snapshot of the sub-pool containing all executable transactions.
#[derive(Debug)]
pub(crate) struct BestAA2dTransactions {
    /// Regular 2D nonce independent pending transactions sorted by their priority.
    ///
    /// Expiring nonce transactions are not included here because they are keyed
    /// by expiring nonce hash, not `AA2dTransactionId`.
    independent: BTreeSet<PendingTransaction<TxOrdering>>,
    /// Regular 2D nonce pending transactions grouped by their unique identifier.
    ///
    /// Expiring nonce transactions are not stored in `by_id`; they are tracked
    /// separately by `expiring_nonce_order`.
    by_id: HashMap<AA2dTransactionId, AA2dStoredTransaction>,
    /// Expiring nonce pending transactions in eviction order. The best
    /// transaction is at the back of the set, and the key carries the pending
    /// transaction so this snapshot does not clone the pool's expiring hash map.
    expiring_nonce_order: BTreeSet<ExpiringNonceEvictionKey>,

    /// There might be the case where a yielded transactions is invalid, this will track it.
    invalid: HashSet<AASequenceId>,
    /// Live feed of new pending transactions arriving after this iterator was created.
    new_transaction_receiver: Option<broadcast::Receiver<AA2dStoredTransaction>>,
    /// Priority of the most recently yielded transaction, used to maintain ordering invariant.
    last_priority: Option<Priority<u64>>,
    /// Base fee used to filter and prioritize this block-building snapshot.
    base_fee: u64,
}

impl BestAA2dTransactions {
    /// Removes the best regular transaction from the set.
    fn pop_best_regular(&mut self) -> Option<(AA2dTransactionId, PendingTransaction<TxOrdering>)> {
        let tx = self.independent.pop_last()?;
        let id = tx
            .transaction
            .transaction
            .aa_transaction_id()
            .expect("Transaction in AA2D pool must be an AA transaction with valid nonce key");
        self.by_id.remove(&id);
        Some((id, tx))
    }

    /// Removes the best expiring nonce transaction from the set.
    fn pop_best_expiring_nonce(&mut self) -> Option<PendingTransaction<TxOrdering>> {
        let key = self.expiring_nonce_order.pop_last()?;
        Some(key.into_transaction())
    }

    /// Removes the best regular or expiring nonce transaction.
    fn pop_best(&mut self) -> Option<PoppedAA2dTransaction> {
        match (self.independent.last(), self.expiring_nonce_order.last()) {
            (Some(regular), Some(expiring)) => {
                if regular
                    .priority
                    .cmp(expiring.priority())
                    .then_with(|| expiring.submission_id().cmp(&regular.submission_id))
                    .is_ge()
                {
                    let (id, tx) = self.pop_best_regular()?;
                    Some(PoppedAA2dTransaction::Regular(id, tx))
                } else {
                    self.pop_best_expiring_nonce()
                        .map(PoppedAA2dTransaction::Expiring)
                }
            }
            (Some(_), None) => {
                let (id, tx) = self.pop_best_regular()?;
                Some(PoppedAA2dTransaction::Regular(id, tx))
            }
            (None, Some(_)) => self
                .pop_best_expiring_nonce()
                .map(PoppedAA2dTransaction::Expiring),
            (None, None) => None,
        }
    }

    /// Non-blocking read on the new pending transactions subscription channel.
    fn try_recv(&mut self) -> Option<IncomingAA2dTransaction> {
        loop {
            match self.new_transaction_receiver.as_mut()?.try_recv() {
                Ok(tx) => {
                    let priority = TempoTipOrdering::default()
                        .priority(&tx.transaction.transaction, self.base_fee);
                    let tx = PendingTransaction {
                        submission_id: tx.submission_id,
                        transaction: tx.transaction,
                        priority,
                    };
                    if let Some(last_priority) = &self.last_priority
                        && &tx.priority > last_priority
                    {
                        // Higher priority than what we already yielded — stash in `by_id`
                        // only (not `independent`) to preserve nonce chain lookups.
                        return Some(IncomingAA2dTransaction::Stash(tx));
                    }
                    return Some(IncomingAA2dTransaction::Process(tx));
                }
                Err(broadcast::error::TryRecvError::Lagged(_)) => {
                    // Buffer overflowed; self-corrects on next call.
                }
                Err(_) => return None,
            }
        }
    }

    /// Drains new pending transactions from the broadcast channel and inserts them.
    fn add_new_transactions(&mut self) {
        for _ in 0..MAX_NEW_TRANSACTIONS_PER_BATCH {
            if let Some(incoming) = self.try_recv() {
                let (tx, process) = match incoming {
                    IncomingAA2dTransaction::Process(tx) => (tx, true),
                    IncomingAA2dTransaction::Stash(tx) => (tx, false),
                };
                if tx.transaction.transaction.is_expiring_nonce() {
                    if process && can_pay_base_fee(&tx, self.base_fee) {
                        self.expiring_nonce_order
                            .insert(ExpiringNonceEvictionKey::from_pending_owned(tx));
                    }
                } else if let Some(id) = tx.transaction.transaction.aa_transaction_id() {
                    if process {
                        // Only mark as independent if no ancestor is already tracked
                        if !self.by_id.contains_key(&AA2dTransactionId::new(
                            id.seq_id,
                            id.nonce.saturating_sub(1),
                        )) || id.nonce == 0
                        {
                            self.independent.insert(tx.clone());
                        }
                    }
                    self.by_id.insert(
                        id,
                        AA2dStoredTransaction {
                            submission_id: tx.submission_id,
                            transaction: tx.transaction,
                        },
                    );
                }
            } else {
                break;
            }
        }
    }

    /// Returns the next best transaction with its priority.
    pub(crate) fn next_tx_and_priority(
        &mut self,
    ) -> Option<(
        Arc<ValidPoolTransaction<TempoPooledTransaction>>,
        Priority<u64>,
    )> {
        loop {
            self.add_new_transactions();
            let best = match self.pop_best()? {
                PoppedAA2dTransaction::Regular(id, best) => {
                    if self.invalid.contains(&id.seq_id) {
                        continue;
                    }
                    if !can_pay_base_fee(&best, self.base_fee) {
                        self.invalid.insert(id.seq_id);
                        continue;
                    }
                    // Advance transaction that just got unlocked, if any.
                    if let Some(unlocked) = self.by_id.get(&id.unlocks()) {
                        self.independent
                            .insert(unlocked.clone_into_pending(self.base_fee));
                    }
                    best
                }
                PoppedAA2dTransaction::Expiring(best) => {
                    if !can_pay_base_fee(&best, self.base_fee) {
                        continue;
                    }
                    best
                }
            };
            if self.new_transaction_receiver.is_some() {
                self.last_priority = Some(best.priority.clone());
            }
            return Some((best.transaction, best.priority));
        }
    }
}

fn can_pay_base_fee(tx: &PendingTransaction<TxOrdering>, base_fee: u64) -> bool {
    tx.transaction.transaction.max_fee_per_gas() >= u128::from(base_fee)
}

impl Iterator for BestAA2dTransactions {
    type Item = Arc<ValidPoolTransaction<TempoPooledTransaction>>;

    fn next(&mut self) -> Option<Self::Item> {
        self.next_tx_and_priority().map(|(tx, _)| tx)
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        (
            0,
            self.by_id
                .len()
                .checked_add(self.expiring_nonce_order.len()),
        )
    }
}

impl BestTransactions for BestAA2dTransactions {
    fn mark_invalid(&mut self, transaction: &Self::Item, _kind: InvalidPoolTransactionError) {
        // Skip invalidation for expiring nonce transactions - they are independent
        // and should not block other expiring nonce txs from the same sender
        if transaction.transaction.is_expiring_nonce() {
            return;
        }

        if let Some(id) = transaction.transaction.aa_transaction_id() {
            self.invalid.insert(id.seq_id);
        }
    }

    fn no_updates(&mut self) {
        self.new_transaction_receiver.take();
        self.last_priority.take();
    }

    fn set_skip_blobs(&mut self, _skip_blobs: bool) {}
}

/// Key for identifying a unique sender sequence in 2D nonce system.
///
/// This combines the sender address with its nonce key, which
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Ord, PartialOrd)]
pub struct AASequenceId {
    /// The sender address.
    pub address: Address,
    /// The nonce key for 2D nonce transactions.
    pub nonce_key: U256,
}

impl AASequenceId {
    /// Creates a new instance with the address and nonce key.
    pub const fn new(address: Address, nonce_key: U256) -> Self {
        Self { address, nonce_key }
    }

    const fn start_bound(self) -> std::ops::Bound<AA2dTransactionId> {
        std::ops::Bound::Included(AA2dTransactionId::new(self, 0))
    }

    /// Returns a range of transactions for this sequence.
    const fn range(self) -> std::ops::RangeInclusive<AA2dTransactionId> {
        AA2dTransactionId::new(self, 0)..=AA2dTransactionId::new(self, u64::MAX)
    }
}

/// Unique identifier for an AA transaction.
///
/// Identified by its sender, nonce key and nonce for that nonce key.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Ord, PartialOrd)]
pub struct AA2dTransactionId {
    /// Uniquely identifies the accounts nonce key sequence
    pub(crate) seq_id: AASequenceId,
    /// The nonce in that sequence
    pub(crate) nonce: u64,
}

impl AA2dTransactionId {
    /// Creates a new identifier.
    pub(crate) const fn new(seq_id: AASequenceId, nonce: u64) -> Self {
        Self { seq_id, nonce }
    }

    /// Returns the next transaction in the sequence.
    pub(crate) fn unlocks(&self) -> Self {
        Self::new(self.seq_id, self.nonce.saturating_add(1))
    }

    /// Returns the nonce key sequence of this transaction.
    pub fn seq_id(&self) -> &AASequenceId {
        &self.seq_id
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::test_utils::{TxBuilder, wrap_valid_tx};
    use alloy_eips::eip2930::AccessList;
    use alloy_primitives::{Address, Bytes, Signature, TxKind, U256};
    use reth_primitives_traits::Recovered;
    use reth_transaction_pool::PoolTransaction;
    use std::collections::HashSet;
    use tempo_chainspec::{hardfork::TempoHardfork, spec::TEMPO_T1_BASE_FEE};
    use tempo_primitives::{
        TempoTxEnvelope,
        transaction::{
            TempoTransaction,
            tempo_transaction::Call,
            tt_signature::{PrimitiveSignature, TempoSignature},
            tt_signed::AASigned,
        },
    };

    #[test_case::test_case(U256::ZERO)]
    #[test_case::test_case(U256::random())]
    fn insert_pending(nonce_key: U256) {
        let mut pool = AA2dPool::default();

        // Set up a sender with a tracked nonce key
        let sender = Address::random();

        // Create a transaction with nonce_key=1, nonce=0 (should be pending)
        let tx = TxBuilder::aa(sender).nonce_key(nonce_key).build();
        let valid_tx = wrap_valid_tx(tx, TransactionOrigin::Local);

        // Add the transaction to the pool
        let result = pool.add_transaction(Arc::new(valid_tx), 0, TempoHardfork::T1);

        // Should be added as pending
        assert!(result.is_ok(), "Transaction should be added successfully");
        let added = result.unwrap();
        assert!(
            matches!(added, AddedTransaction::Pending(_)),
            "Transaction should be pending, got: {added:?}"
        );

        // Verify pool state
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count, 1, "Should have 1 pending transaction");
        assert_eq!(queued_count, 0, "Should have 0 queued transactions");

        pool.assert_invariants();
    }

    #[test_case::test_case(U256::ZERO)]
    #[test_case::test_case(U256::random())]
    fn insert_with_nonce_gap_then_fill(nonce_key: U256) {
        let mut pool = AA2dPool::default();

        // Set up a sender with a tracked nonce key
        let sender = Address::random();

        // Step 1: Insert transaction with nonce=1 (creates a gap, should be queued)
        let tx1 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(1).build();
        let valid_tx1 = wrap_valid_tx(tx1, TransactionOrigin::Local);
        let tx1_hash = *valid_tx1.hash();

        let result1 = pool.add_transaction(Arc::new(valid_tx1), 0, TempoHardfork::T1);

        // Should be queued due to nonce gap
        assert!(
            result1.is_ok(),
            "Transaction 1 should be added successfully"
        );
        let added1 = result1.unwrap();
        assert!(
            matches!(
                added1,
                AddedTransaction::Parked {
                    subpool: SubPool::Queued,
                    ..
                }
            ),
            "Transaction 1 should be queued due to nonce gap, got: {added1:?}"
        );

        // Verify pool state after first insert
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count, 0, "Should have 0 pending transactions");
        assert_eq!(queued_count, 1, "Should have 1 queued transaction");

        // Verify tx1 is NOT in independent set
        let seq_id = AASequenceId::new(sender, nonce_key);
        let tx1_id = AA2dTransactionId::new(seq_id, 1);
        assert!(
            !pool.independent_transactions.contains_key(&tx1_id.seq_id),
            "Transaction 1 should not be in independent set yet"
        );

        pool.assert_invariants();

        // Step 2: Insert transaction with nonce=0 (fills the gap)
        let tx0 = TxBuilder::aa(sender).nonce_key(nonce_key).build();
        let valid_tx0 = wrap_valid_tx(tx0, TransactionOrigin::Local);
        let tx0_hash = *valid_tx0.hash();

        let result0 = pool.add_transaction(Arc::new(valid_tx0), 0, TempoHardfork::T1);

        // Should be pending and promote tx1
        assert!(
            result0.is_ok(),
            "Transaction 0 should be added successfully"
        );
        let added0 = result0.unwrap();

        // Verify it's pending and promoted tx1
        match added0 {
            AddedTransaction::Pending(ref pending) => {
                assert_eq!(pending.transaction.hash(), &tx0_hash, "Should be tx0");
                assert_eq!(
                    pending.promoted.len(),
                    1,
                    "Should have promoted 1 transaction"
                );
                assert_eq!(
                    pending.promoted[0].hash(),
                    &tx1_hash,
                    "Should have promoted tx1"
                );
            }
            _ => panic!("Transaction 0 should be pending, got: {added0:?}"),
        }

        // Verify pool state after filling the gap
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count, 2, "Should have 2 pending transactions");
        assert_eq!(queued_count, 0, "Should have 0 queued transactions");

        // Verify both transactions are now pending
        let tx0_id = AA2dTransactionId::new(seq_id, 0);
        assert!(
            pool.by_id.get(&tx0_id).unwrap().is_pending(),
            "Transaction 0 should be pending"
        );
        assert!(
            pool.by_id.get(&tx1_id).unwrap().is_pending(),
            "Transaction 1 should be pending after promotion"
        );

        // Verify tx0 (at on-chain nonce) is in independent set
        assert!(
            pool.independent_transactions.contains_key(&tx0_id.seq_id),
            "Transaction 0 should be in independent set (at on-chain nonce)"
        );

        // Verify the independent transaction for this sequence is tx0, not tx1
        let independent_tx = pool.independent_transactions.get(&seq_id).unwrap();
        assert_eq!(
            independent_tx.transaction.hash(),
            &tx0_hash,
            "Independent transaction should be tx0, not tx1"
        );

        pool.assert_invariants();
    }

    #[test_case::test_case(U256::ZERO)]
    #[test_case::test_case(U256::random())]
    fn replace_pending_transaction(nonce_key: U256) {
        let mut pool = AA2dPool::default();

        // Set up a sender with a tracked nonce key
        let sender = Address::random();

        // Step 1: Insert initial pending transaction with lower gas price
        let tx_low = TxBuilder::aa(sender)
            .nonce_key(nonce_key)
            .max_priority_fee(1_000_000_000)
            .max_fee(2_000_000_000)
            .build();
        let valid_tx_low = wrap_valid_tx(tx_low, TransactionOrigin::Local);
        let tx_low_hash = *valid_tx_low.hash();

        let result_low = pool.add_transaction(Arc::new(valid_tx_low), 0, TempoHardfork::T1);

        // Should be pending (at on-chain nonce)
        assert!(
            result_low.is_ok(),
            "Initial transaction should be added successfully"
        );
        let added_low = result_low.unwrap();
        assert!(
            matches!(added_low, AddedTransaction::Pending(_)),
            "Initial transaction should be pending"
        );

        // Verify initial state
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count, 1, "Should have 1 pending transaction");
        assert_eq!(queued_count, 0, "Should have 0 queued transactions");

        // Verify tx_low is in independent set
        let seq_id = AASequenceId::new(sender, nonce_key);
        let tx_id = AA2dTransactionId::new(seq_id, 0);
        assert!(
            pool.independent_transactions.contains_key(&tx_id.seq_id),
            "Initial transaction should be in independent set"
        );

        // Verify the transaction in independent set is tx_low
        let independent_tx = pool.independent_transactions.get(&tx_id.seq_id).unwrap();
        assert_eq!(
            independent_tx.transaction.hash(),
            &tx_low_hash,
            "Independent set should contain tx_low"
        );

        pool.assert_invariants();

        // Step 2: Replace with higher gas price transaction
        // Price bump needs to be at least 10% higher (default price bump config)
        let tx_high = TxBuilder::aa(sender)
            .nonce_key(nonce_key)
            .max_priority_fee(1_200_000_000)
            .max_fee(2_400_000_000)
            .build();
        let valid_tx_high = wrap_valid_tx(tx_high, TransactionOrigin::Local);
        let tx_high_hash = *valid_tx_high.hash();

        let result_high = pool.add_transaction(Arc::new(valid_tx_high), 0, TempoHardfork::T1);

        // Should successfully replace
        assert!(
            result_high.is_ok(),
            "Replacement transaction should be added successfully"
        );
        let added_high = result_high.unwrap();

        // Verify it's pending and replaced the old transaction
        match added_high {
            AddedTransaction::Pending(ref pending) => {
                assert_eq!(
                    pending.transaction.hash(),
                    &tx_high_hash,
                    "Should be tx_high"
                );
                assert!(
                    pending.replaced.is_some(),
                    "Should have replaced a transaction"
                );
                assert_eq!(
                    pending.replaced.as_ref().unwrap().hash(),
                    &tx_low_hash,
                    "Should have replaced tx_low"
                );
            }
            _ => panic!("Replacement transaction should be pending, got: {added_high:?}"),
        }

        // Verify pool state - still 1 pending, 0 queued
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(
            pending_count, 1,
            "Should still have 1 pending transaction after replacement"
        );
        assert_eq!(queued_count, 0, "Should still have 0 queued transactions");

        // Verify old transaction is no longer in the pool
        assert!(
            !pool.contains(&tx_low_hash),
            "Old transaction should be removed from pool"
        );

        // Verify new transaction is in the pool
        assert!(
            pool.contains(&tx_high_hash),
            "New transaction should be in pool"
        );

        // Verify independent set is updated with new transaction
        assert!(
            pool.independent_transactions.contains_key(&tx_id.seq_id),
            "Transaction ID should still be in independent set"
        );

        let independent_tx_after = pool.independent_transactions.get(&tx_id.seq_id).unwrap();
        assert_eq!(
            independent_tx_after.transaction.hash(),
            &tx_high_hash,
            "Independent set should now contain tx_high (not tx_low)"
        );

        // Verify the transaction in by_id is the new one
        let tx_in_pool = pool.by_id.get(&tx_id).unwrap();
        assert_eq!(
            tx_in_pool.inner.transaction.hash(),
            &tx_high_hash,
            "Transaction in by_id should be tx_high"
        );
        assert!(tx_in_pool.is_pending(), "Transaction should be pending");

        pool.assert_invariants();
    }

    #[test_case::test_case(U256::ZERO)]
    #[test_case::test_case(U256::random())]
    fn on_chain_nonce_update_with_gaps(nonce_key: U256) {
        let mut pool = AA2dPool::default();

        // Set up a sender with a tracked nonce key
        let sender = Address::random();

        // Insert transactions with nonces: 0, 1, 3, 4, 6
        // Expected initial state:
        // - 0, 1: pending (consecutive from on-chain nonce 0)
        // - 3, 4, 6: queued (gaps at nonce 2 and 5)
        let tx0 = TxBuilder::aa(sender).nonce_key(nonce_key).build();
        let tx1 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(1).build();
        let tx3 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(3).build();
        let tx4 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(4).build();
        let tx6 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(6).build();

        let valid_tx0 = wrap_valid_tx(tx0, TransactionOrigin::Local);
        let valid_tx1 = wrap_valid_tx(tx1, TransactionOrigin::Local);
        let valid_tx3 = wrap_valid_tx(tx3, TransactionOrigin::Local);
        let valid_tx4 = wrap_valid_tx(tx4, TransactionOrigin::Local);
        let valid_tx6 = wrap_valid_tx(tx6, TransactionOrigin::Local);

        let tx0_hash = *valid_tx0.hash();
        let tx1_hash = *valid_tx1.hash();
        let tx3_hash = *valid_tx3.hash();
        let tx4_hash = *valid_tx4.hash();
        let tx6_hash = *valid_tx6.hash();

        // Add all transactions
        pool.add_transaction(Arc::new(valid_tx0), 0, TempoHardfork::T1)
            .unwrap();
        pool.add_transaction(Arc::new(valid_tx1), 0, TempoHardfork::T1)
            .unwrap();
        pool.add_transaction(Arc::new(valid_tx3), 0, TempoHardfork::T1)
            .unwrap();
        pool.add_transaction(Arc::new(valid_tx4), 0, TempoHardfork::T1)
            .unwrap();
        pool.add_transaction(Arc::new(valid_tx6), 0, TempoHardfork::T1)
            .unwrap();

        // Verify initial state
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(
            pending_count, 2,
            "Should have 2 pending transactions (0, 1)"
        );
        assert_eq!(
            queued_count, 3,
            "Should have 3 queued transactions (3, 4, 6)"
        );

        // Verify tx0 is in independent set
        let seq_id = AASequenceId::new(sender, nonce_key);
        let tx0_id = AA2dTransactionId::new(seq_id, 0);
        assert!(
            pool.independent_transactions.contains_key(&tx0_id.seq_id),
            "Transaction 0 should be in independent set"
        );

        pool.assert_invariants();

        // Step 1: Simulate mining block with nonces 0 and 1
        // New on-chain nonce becomes 2
        let mut on_chain_ids = HashMap::default();
        on_chain_ids.insert(seq_id, 2u64);

        let (promoted, mined) = pool.on_nonce_changes(on_chain_ids);

        // Verify mined transactions
        assert_eq!(mined.len(), 2, "Should have mined 2 transactions (0, 1)");
        let mined_hashes: HashSet<_> = mined.iter().map(|tx| tx.hash()).collect();
        assert!(
            mined_hashes.contains(&&tx0_hash),
            "Transaction 0 should be mined"
        );
        assert!(
            mined_hashes.contains(&&tx1_hash),
            "Transaction 1 should be mined"
        );

        // No transactions should be promoted (there's a gap at nonce 2)
        assert_eq!(
            promoted.len(),
            0,
            "No transactions should be promoted (gap at nonce 2)"
        );

        // Verify pool state after mining
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(
            pending_count, 0,
            "Should have 0 pending transactions (gap at nonce 2)"
        );
        assert_eq!(
            queued_count, 3,
            "Should have 3 queued transactions (3, 4, 6)"
        );

        // Verify mined transactions are removed
        assert!(!pool.contains(&tx0_hash), "Transaction 0 should be removed");
        assert!(!pool.contains(&tx1_hash), "Transaction 1 should be removed");

        // Verify remaining transactions are still in pool
        assert!(pool.contains(&tx3_hash), "Transaction 3 should remain");
        assert!(pool.contains(&tx4_hash), "Transaction 4 should remain");
        assert!(pool.contains(&tx6_hash), "Transaction 6 should remain");

        // Verify all remaining transactions are queued (not pending)
        let tx3_id = AA2dTransactionId::new(seq_id, 3);
        let tx4_id = AA2dTransactionId::new(seq_id, 4);
        let tx6_id = AA2dTransactionId::new(seq_id, 6);

        assert!(
            !pool.by_id.get(&tx3_id).unwrap().is_pending(),
            "Transaction 3 should be queued (gap at nonce 2)"
        );
        assert!(
            !pool.by_id.get(&tx4_id).unwrap().is_pending(),
            "Transaction 4 should be queued"
        );
        assert!(
            !pool.by_id.get(&tx6_id).unwrap().is_pending(),
            "Transaction 6 should be queued"
        );

        // Verify independent set is empty (no transaction at on-chain nonce)
        assert!(
            pool.independent_transactions.is_empty(),
            "Independent set should be empty (gap at on-chain nonce 2)"
        );

        pool.assert_invariants();

        // Step 2: Simulate mining block with nonce 2
        // New on-chain nonce becomes 3
        let mut on_chain_ids = HashMap::default();
        on_chain_ids.insert(seq_id, 3u64);

        let (promoted, mined) = pool.on_nonce_changes(on_chain_ids);

        // No transactions should be mined (nonce 2 was never in pool)
        assert_eq!(
            mined.len(),
            0,
            "No transactions should be mined (nonce 2 was never in pool)"
        );

        // Transactions 3 and 4 should be promoted
        assert_eq!(promoted.len(), 2, "Transactions 3 and 4 should be promoted");
        let promoted_hashes: HashSet<_> = promoted.iter().map(|tx| tx.hash()).collect();
        assert!(
            promoted_hashes.contains(&&tx3_hash),
            "Transaction 3 should be promoted"
        );
        assert!(
            promoted_hashes.contains(&&tx4_hash),
            "Transaction 4 should be promoted"
        );

        // Verify pool state after second update
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(
            pending_count, 2,
            "Should have 2 pending transactions (3, 4)"
        );
        assert_eq!(queued_count, 1, "Should have 1 queued transaction (6)");

        // Verify transactions 3 and 4 are now pending
        assert!(
            pool.by_id.get(&tx3_id).unwrap().is_pending(),
            "Transaction 3 should be pending"
        );
        assert!(
            pool.by_id.get(&tx4_id).unwrap().is_pending(),
            "Transaction 4 should be pending"
        );

        // Verify transaction 6 is still queued
        assert!(
            !pool.by_id.get(&tx6_id).unwrap().is_pending(),
            "Transaction 6 should still be queued (gap at nonce 5)"
        );

        // Verify transaction 3 is the independent transaction (at on-chain nonce)
        assert!(
            pool.independent_transactions.contains_key(&tx3_id.seq_id),
            "Transaction 3 should be in independent set (at on-chain nonce 3)"
        );

        // Verify the independent transaction is tx3 specifically, not tx4 or tx6
        let independent_tx = pool.independent_transactions.get(&seq_id).unwrap();
        assert_eq!(
            independent_tx.transaction.hash(),
            &tx3_hash,
            "Independent transaction should be tx3 (nonce 3), not tx4 or tx6"
        );

        pool.assert_invariants();
    }

    #[test_case::test_case(U256::ZERO)]
    #[test_case::test_case(U256::random())]
    fn reject_outdated_transaction(nonce_key: U256) {
        let mut pool = AA2dPool::default();

        // Set up a sender with a tracked nonce key
        let sender = Address::random();

        // Create a transaction with nonce 3 (outdated)
        let tx = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(3).build();
        let valid_tx = wrap_valid_tx(tx, TransactionOrigin::Local);

        // Try to insert it and specify the on-chain nonce 5, making it outdated
        let result = pool.add_transaction(Arc::new(valid_tx), 5, TempoHardfork::T1);

        // Should fail with nonce error
        assert!(result.is_err(), "Should reject outdated transaction");

        let err = result.unwrap_err();
        assert!(
            matches!(
                err.kind,
                PoolErrorKind::InvalidTransaction(InvalidPoolTransactionError::Consensus(
                    InvalidTransactionError::NonceNotConsistent { tx: 3, state: 5 }
                ))
            ),
            "Should fail with NonceNotConsistent error, got: {:?}",
            err.kind
        );

        // Pool should remain empty
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count, 0, "Pool should be empty");
        assert_eq!(queued_count, 0, "Pool should be empty");

        pool.assert_invariants();
    }

    #[test_case::test_case(U256::ZERO)]
    #[test_case::test_case(U256::random())]
    fn replace_with_insufficient_price_bump(nonce_key: U256) {
        let mut pool = AA2dPool::default();

        // Set up a sender
        let sender = Address::random();

        // Insert initial transaction
        let tx_low = TxBuilder::aa(sender)
            .nonce_key(nonce_key)
            .max_priority_fee(1_000_000_000)
            .max_fee(2_000_000_000)
            .build();
        let valid_tx_low = wrap_valid_tx(tx_low, TransactionOrigin::Local);

        pool.add_transaction(Arc::new(valid_tx_low), 0, TempoHardfork::T1)
            .unwrap();

        // Try to replace with only 5% price bump (default requires 10%)
        let tx_insufficient = TxBuilder::aa(sender)
            .nonce_key(nonce_key)
            .max_priority_fee(1_050_000_000)
            .max_fee(2_100_000_000)
            .build();
        let valid_tx_insufficient = wrap_valid_tx(tx_insufficient, TransactionOrigin::Local);

        let result = pool.add_transaction(Arc::new(valid_tx_insufficient), 0, TempoHardfork::T1);

        // Should fail with ReplacementUnderpriced
        assert!(
            result.is_err(),
            "Should reject replacement with insufficient price bump"
        );
        let err = result.unwrap_err();
        assert!(
            matches!(err.kind, PoolErrorKind::ReplacementUnderpriced),
            "Should fail with ReplacementUnderpriced, got: {:?}",
            err.kind
        );

        pool.assert_invariants();
    }

    #[test_case::test_case(U256::ZERO)]
    #[test_case::test_case(U256::random())]
    fn fill_gap_in_middle(nonce_key: U256) {
        let mut pool = AA2dPool::default();

        let sender = Address::random();

        // Insert transactions: 0, 1, 3, 4 (gap at 2)
        let tx0 = TxBuilder::aa(sender).nonce_key(nonce_key).build();
        let tx1 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(1).build();
        let tx3 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(3).build();
        let tx4 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(4).build();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx3, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx4, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        // Verify initial state: 0, 1 pending | 3, 4 queued
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count, 2, "Should have 2 pending (0, 1)");
        assert_eq!(queued_count, 2, "Should have 2 queued (3, 4)");

        // Fill the gap with nonce 2
        let tx2 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(2).build();
        let valid_tx2 = wrap_valid_tx(tx2, TransactionOrigin::Local);

        let result = pool.add_transaction(Arc::new(valid_tx2), 0, TempoHardfork::T1);
        assert!(result.is_ok(), "Should successfully add tx2");

        // Verify tx3 and tx4 were promoted
        match result.unwrap() {
            AddedTransaction::Pending(ref pending) => {
                assert_eq!(pending.promoted.len(), 2, "Should promote tx3 and tx4");
            }
            _ => panic!("tx2 should be added as pending"),
        }

        // Verify all transactions are now pending
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count, 5, "All 5 transactions should be pending");
        assert_eq!(queued_count, 0, "No transactions should be queued");

        // Verify tx0 is in independent set
        let seq_id = AASequenceId::new(sender, nonce_key);
        let tx0_id = AA2dTransactionId::new(seq_id, 0);
        assert!(
            pool.independent_transactions.contains_key(&tx0_id.seq_id),
            "tx0 should be in independent set"
        );

        pool.assert_invariants();
    }

    #[test_case::test_case(U256::ZERO)]
    #[test_case::test_case(U256::random())]
    fn remove_pending_transaction(nonce_key: U256) {
        let mut pool = AA2dPool::default();

        let sender = Address::random();

        // Insert consecutive transactions: 0, 1, 2
        let tx0 = TxBuilder::aa(sender).nonce_key(nonce_key).build();
        let tx1 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(1).build();
        let tx2 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(2).build();

        let valid_tx0 = wrap_valid_tx(tx0, TransactionOrigin::Local);
        let valid_tx1 = wrap_valid_tx(tx1, TransactionOrigin::Local);
        let valid_tx2 = wrap_valid_tx(tx2, TransactionOrigin::Local);

        let tx1_hash = *valid_tx1.hash();

        pool.add_transaction(Arc::new(valid_tx0), 0, TempoHardfork::T1)
            .unwrap();
        pool.add_transaction(Arc::new(valid_tx1), 0, TempoHardfork::T1)
            .unwrap();
        pool.add_transaction(Arc::new(valid_tx2), 0, TempoHardfork::T1)
            .unwrap();

        // All should be pending
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count, 3, "All 3 should be pending");
        assert_eq!(queued_count, 0, "None should be queued");

        let seq_id = AASequenceId::new(sender, nonce_key);
        let tx1_id = AA2dTransactionId::new(seq_id, 1);
        let tx2_id = AA2dTransactionId::new(seq_id, 2);

        // Verify tx2 is pending before removal
        assert!(
            pool.by_id.get(&tx2_id).unwrap().is_pending(),
            "tx2 should be pending before removal"
        );

        // Remove tx1 (creates a gap)
        let removed = pool.remove_transactions([&tx1_hash].into_iter());
        assert_eq!(removed.len(), 1, "Should remove tx1");

        // Verify tx1 is removed from pool
        assert!(!pool.by_id.contains_key(&tx1_id), "tx1 should be removed");
        assert!(!pool.contains(&tx1_hash), "tx1 should be removed");

        // Verify tx0 and tx2 remain
        assert_eq!(pool.by_id.len(), 2, "Should have 2 transactions left");

        // Verify tx2 is now demoted to queued since tx1 removal creates a gap
        assert!(
            !pool.by_id.get(&tx2_id).unwrap().is_pending(),
            "tx2 should be demoted to queued after tx1 removal creates a gap"
        );

        // Verify counts: tx0 is pending, tx2 is queued
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count, 1, "Only tx0 should be pending");
        assert_eq!(queued_count, 1, "tx2 should be queued");

        pool.assert_invariants();
    }

    #[test_case::test_case(U256::ZERO, U256::random())]
    #[test_case::test_case(U256::random(), U256::ZERO)]
    #[test_case::test_case(U256::random(), U256::random())]
    fn multiple_senders_independent_set(nonce_key_a: U256, nonce_key_b: U256) {
        let mut pool = AA2dPool::default();

        // Set up two senders with different nonce keys
        let sender_a = Address::random();
        let sender_b = Address::random();

        // Insert transactions for both senders
        // Sender A: [0, 1]
        let tx_a0 = TxBuilder::aa(sender_a).nonce_key(nonce_key_a).build();
        let tx_a1 = TxBuilder::aa(sender_a)
            .nonce_key(nonce_key_a)
            .nonce(1)
            .build();

        // Sender B: [0, 1]
        let tx_b0 = TxBuilder::aa(sender_b).nonce_key(nonce_key_b).build();
        let tx_b1 = TxBuilder::aa(sender_b)
            .nonce_key(nonce_key_b)
            .nonce(1)
            .build();

        let valid_tx_a0 = wrap_valid_tx(tx_a0, TransactionOrigin::Local);
        let valid_tx_a1 = wrap_valid_tx(tx_a1, TransactionOrigin::Local);
        let valid_tx_b0 = wrap_valid_tx(tx_b0, TransactionOrigin::Local);
        let valid_tx_b1 = wrap_valid_tx(tx_b1, TransactionOrigin::Local);

        let tx_a0_hash = *valid_tx_a0.hash();

        pool.add_transaction(Arc::new(valid_tx_a0), 0, TempoHardfork::T1)
            .unwrap();
        pool.add_transaction(Arc::new(valid_tx_a1), 0, TempoHardfork::T1)
            .unwrap();
        pool.add_transaction(Arc::new(valid_tx_b0), 0, TempoHardfork::T1)
            .unwrap();
        pool.add_transaction(Arc::new(valid_tx_b1), 0, TempoHardfork::T1)
            .unwrap();

        // Both senders' tx0 should be in independent set
        let sender_a_id = AASequenceId::new(sender_a, nonce_key_a);
        let sender_b_id = AASequenceId::new(sender_b, nonce_key_b);
        let tx_a0_id = AA2dTransactionId::new(sender_a_id, 0);
        let tx_b0_id = AA2dTransactionId::new(sender_b_id, 0);

        assert_eq!(
            pool.independent_transactions.len(),
            2,
            "Should have 2 independent transactions"
        );
        assert!(
            pool.independent_transactions.contains_key(&tx_a0_id.seq_id),
            "Sender A's tx0 should be independent"
        );
        assert!(
            pool.independent_transactions.contains_key(&tx_b0_id.seq_id),
            "Sender B's tx0 should be independent"
        );

        // All 4 transactions should be pending
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count, 4, "All 4 transactions should be pending");
        assert_eq!(queued_count, 0, "No transactions should be queued");

        // Simulate mining sender A's tx0
        let mut on_chain_ids = HashMap::default();
        on_chain_ids.insert(sender_a_id, 1u64);

        let (promoted, mined) = pool.on_nonce_changes(on_chain_ids);

        // Only sender A's tx0 should be mined
        assert_eq!(mined.len(), 1, "Only sender A's tx0 should be mined");
        assert_eq!(mined[0].hash(), &tx_a0_hash, "Should mine tx_a0");

        // No transactions should be promoted (tx_a1 was already pending)
        assert_eq!(
            promoted.len(),
            0,
            "No transactions should be promoted (tx_a1 was already pending)"
        );

        // Verify independent set now has A's tx1 and B's tx0
        let tx_a1_id = AA2dTransactionId::new(sender_a_id, 1);
        assert_eq!(
            pool.independent_transactions.len(),
            2,
            "Should still have 2 independent transactions"
        );
        assert!(
            pool.independent_transactions.contains_key(&tx_a1_id.seq_id),
            "Sender A's tx1 should now be independent"
        );
        assert!(
            pool.independent_transactions.contains_key(&tx_b0_id.seq_id),
            "Sender B's tx0 should still be independent"
        );

        pool.assert_invariants();
    }

    #[test_case::test_case(U256::ZERO)]
    #[test_case::test_case(U256::random())]
    fn concurrent_replacements_same_nonce(nonce_key: U256) {
        let mut pool = AA2dPool::default();
        let sender = Address::random();
        let seq_id = AASequenceId {
            address: sender,
            nonce_key,
        };

        // Insert initial transaction at nonce 0 with gas prices 1_000_000_000, 2_000_000_000
        let tx0 = TxBuilder::aa(sender)
            .nonce_key(nonce_key)
            .max_priority_fee(1_000_000_000)
            .max_fee(2_000_000_000)
            .build();
        let tx0_hash = *tx0.hash();
        let valid_tx0 = wrap_valid_tx(tx0, TransactionOrigin::Local);
        let result = pool.add_transaction(Arc::new(valid_tx0), 0, TempoHardfork::T1);
        assert!(result.is_ok());
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count + queued_count, 1);

        // Try to replace with slightly higher gas (1_050_000_000, 2_100_000_000 = ~5% bump) - should fail (< 10% bump)
        let tx0_replacement1 = TxBuilder::aa(sender)
            .nonce_key(nonce_key)
            .max_priority_fee(1_050_000_000)
            .max_fee(2_100_000_000)
            .build();
        let valid_tx1 = wrap_valid_tx(tx0_replacement1, TransactionOrigin::Local);
        let result = pool.add_transaction(Arc::new(valid_tx1), 0, TempoHardfork::T1);
        assert!(result.is_err(), "Should reject insufficient price bump");
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count + queued_count, 1);
        assert!(
            pool.contains(&tx0_hash),
            "Original tx should still be present"
        );

        // Replace with sufficient bump (1_100_000_000, 2_200_000_000 = 10% bump)
        let tx0_replacement2 = TxBuilder::aa(sender)
            .nonce_key(nonce_key)
            .max_priority_fee(1_100_000_000)
            .max_fee(2_200_000_000)
            .build();
        let tx0_replacement2_hash = *tx0_replacement2.hash();
        let valid_tx2 = wrap_valid_tx(tx0_replacement2, TransactionOrigin::Local);
        let result = pool.add_transaction(Arc::new(valid_tx2), 0, TempoHardfork::T1);
        assert!(result.is_ok(), "Should accept 10% price bump");
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count + queued_count, 1, "Pool size should remain 1");
        assert!(!pool.contains(&tx0_hash), "Old tx should be removed");
        assert!(
            pool.contains(&tx0_replacement2_hash),
            "New tx should be present"
        );

        // Try to replace with even higher gas (1_500_000_000, 3_000_000_000 = ~36% bump over original)
        let tx0_replacement3 = TxBuilder::aa(sender)
            .nonce_key(nonce_key)
            .max_priority_fee(1_500_000_000)
            .max_fee(3_000_000_000)
            .build();
        let tx0_replacement3_hash = *tx0_replacement3.hash();
        let valid_tx3 = wrap_valid_tx(tx0_replacement3, TransactionOrigin::Local);
        let result = pool.add_transaction(Arc::new(valid_tx3), 0, TempoHardfork::T1);
        assert!(result.is_ok(), "Should accept higher price bump");
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count + queued_count, 1);
        assert!(
            !pool.contains(&tx0_replacement2_hash),
            "Previous tx should be removed"
        );
        assert!(
            pool.contains(&tx0_replacement3_hash),
            "Highest priority tx should win"
        );

        // Verify independent set has the final replacement
        let tx0_id = AA2dTransactionId::new(seq_id, 0);
        assert!(pool.independent_transactions.contains_key(&tx0_id.seq_id));

        pool.assert_invariants();
    }

    #[test_case::test_case(U256::ZERO)]
    #[test_case::test_case(U256::random())]
    fn long_gap_chain(nonce_key: U256) {
        let mut pool = AA2dPool::default();
        let sender = Address::random();
        let seq_id = AASequenceId {
            address: sender,
            nonce_key,
        };

        // Insert transactions with large gaps: [0, 5, 10, 15]
        let tx0 = TxBuilder::aa(sender).nonce_key(nonce_key).build();
        let tx5 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(5).build();
        let tx10 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(10).build();
        let tx15 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(15).build();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx5, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx10, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx15, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count + queued_count, 4);

        // Only tx0 should be pending, rest should be queued
        let tx0_id = AA2dTransactionId::new(seq_id, 0);
        assert!(pool.by_id.get(&tx0_id).unwrap().is_pending());
        assert!(
            !pool
                .by_id
                .get(&AA2dTransactionId::new(seq_id, 5))
                .unwrap()
                .is_pending()
        );
        assert!(
            !pool
                .by_id
                .get(&AA2dTransactionId::new(seq_id, 10))
                .unwrap()
                .is_pending()
        );
        assert!(
            !pool
                .by_id
                .get(&AA2dTransactionId::new(seq_id, 15))
                .unwrap()
                .is_pending()
        );
        assert_eq!(pool.independent_transactions.len(), 1);

        // Fill gap [1,2,3,4]
        for nonce in 1..=4 {
            let tx = TxBuilder::aa(sender)
                .nonce_key(nonce_key)
                .nonce(nonce)
                .build();
            pool.add_transaction(
                Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            )
            .unwrap();
        }

        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count + queued_count, 8);

        // Now [0,1,2,3,4,5] should be pending
        for nonce in 0..=5 {
            let id = AA2dTransactionId::new(seq_id, nonce);
            assert!(
                pool.by_id.get(&id).unwrap().is_pending(),
                "Nonce {nonce} should be pending"
            );
        }
        // [10, 15] should still be queued
        assert!(
            !pool
                .by_id
                .get(&AA2dTransactionId::new(seq_id, 10))
                .unwrap()
                .is_pending()
        );
        assert!(
            !pool
                .by_id
                .get(&AA2dTransactionId::new(seq_id, 15))
                .unwrap()
                .is_pending()
        );

        // Fill gap [6,7,8,9]
        for nonce in 6..=9 {
            let tx = TxBuilder::aa(sender)
                .nonce_key(nonce_key)
                .nonce(nonce)
                .build();
            pool.add_transaction(
                Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            )
            .unwrap();
        }

        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count + queued_count, 12);

        // Now [0..=10] should be pending
        for nonce in 0..=10 {
            let id = AA2dTransactionId::new(seq_id, nonce);
            assert!(
                pool.by_id.get(&id).unwrap().is_pending(),
                "Nonce {nonce} should be pending"
            );
        }
        // Only [15] should be queued
        assert!(
            !pool
                .by_id
                .get(&AA2dTransactionId::new(seq_id, 15))
                .unwrap()
                .is_pending()
        );

        // Fill final gap [11,12,13,14]
        for nonce in 11..=14 {
            let tx = TxBuilder::aa(sender)
                .nonce_key(nonce_key)
                .nonce(nonce)
                .build();
            pool.add_transaction(
                Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            )
            .unwrap();
        }

        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count + queued_count, 16);

        // All should be pending now
        for nonce in 0..=15 {
            let id = AA2dTransactionId::new(seq_id, nonce);
            assert!(
                pool.by_id.get(&id).unwrap().is_pending(),
                "Nonce {nonce} should be pending"
            );
        }

        pool.assert_invariants();
    }

    #[test_case::test_case(U256::ZERO)]
    #[test_case::test_case(U256::random())]
    fn remove_from_middle_of_chain(nonce_key: U256) {
        let mut pool = AA2dPool::default();
        let sender = Address::random();
        let seq_id = AASequenceId {
            address: sender,
            nonce_key,
        };

        // Insert continuous sequence [0,1,2,3,4]
        for nonce in 0..=4 {
            let tx = TxBuilder::aa(sender)
                .nonce_key(nonce_key)
                .nonce(nonce)
                .build();
            pool.add_transaction(
                Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            )
            .unwrap();
        }

        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count + queued_count, 5);

        // All should be pending
        for nonce in 0..=4 {
            assert!(
                pool.by_id
                    .get(&AA2dTransactionId::new(seq_id, nonce))
                    .unwrap()
                    .is_pending()
            );
        }

        // Remove nonce 2 from the middle
        let tx2_id = AA2dTransactionId::new(seq_id, 2);
        let tx2_hash = *pool.by_id.get(&tx2_id).unwrap().inner.transaction.hash();
        let removed = pool.remove_transactions([&tx2_hash].into_iter());
        assert_eq!(removed.len(), 1, "Should remove transaction");

        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count + queued_count, 4);

        // Transaction 2 should be gone
        assert!(!pool.by_id.contains_key(&tx2_id));

        // Note: Current implementation doesn't automatically re-scan after removal
        // So we verify that the removal succeeded but don't expect automatic gap detection
        // Transactions [0,1,3,4] remain in their current state

        pool.assert_invariants();
    }

    #[test_case::test_case(U256::ZERO)]
    #[test_case::test_case(U256::random())]
    fn independent_set_after_multiple_promotions(nonce_key: U256) {
        let mut pool = AA2dPool::default();
        let sender = Address::random();
        let seq_id = AASequenceId {
            address: sender,
            nonce_key,
        };

        // Start with gaps: insert [0, 2, 4]
        let tx0 = TxBuilder::aa(sender).nonce_key(nonce_key).build();
        let tx2 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(2).build();
        let tx4 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(4).build();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx2, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx4, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        // Only tx0 should be in independent set
        assert_eq!(pool.independent_transactions.len(), 1);
        assert!(pool.independent_transactions.contains_key(&seq_id));

        // Verify initial state: tx0 pending, tx2 and tx4 queued
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count, 1);
        assert_eq!(queued_count, 2);

        // Fill first gap: insert [1]
        let tx1 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(1).build();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        // Now [0, 1, 2] should be pending, tx4 still queued
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count, 3);
        assert_eq!(queued_count, 1);

        // Still only tx0 in independent set
        assert_eq!(pool.independent_transactions.len(), 1);
        assert!(pool.independent_transactions.contains_key(&seq_id));

        // Fill second gap: insert [3]
        let tx3 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(3).build();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx3, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        // Now all [0,1,2,3,4] should be pending
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count, 5);
        assert_eq!(queued_count, 0);

        // Simulate mining [0,1]
        let mut on_chain_ids = HashMap::default();
        on_chain_ids.insert(seq_id, 2u64);
        let (promoted, mined) = pool.on_nonce_changes(on_chain_ids);

        // Should have mined [0,1], no promotions (already pending)
        assert_eq!(mined.len(), 2);
        assert_eq!(promoted.len(), 0);

        // Now tx2 should be in independent set
        assert_eq!(pool.independent_transactions.len(), 1);
        assert!(pool.independent_transactions.contains_key(&seq_id));

        // Verify [2,3,4] remain in pool
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count + queued_count, 3);

        pool.assert_invariants();
    }

    #[test]
    fn stress_test_many_senders() {
        let mut pool = AA2dPool::default();
        const NUM_SENDERS: usize = 100;
        const TXS_PER_SENDER: u64 = 5;

        // Create 100 senders, each with 5 transactions
        let mut senders = Vec::new();
        for i in 0..NUM_SENDERS {
            let sender = Address::from_word(B256::from(U256::from(i)));
            let nonce_key = U256::from(i);
            senders.push((sender, nonce_key));

            // Insert transactions [0,1,2,3,4] for each sender
            for nonce in 0..TXS_PER_SENDER {
                let tx = TxBuilder::aa(sender)
                    .nonce_key(nonce_key)
                    .nonce(nonce)
                    .build();
                pool.add_transaction(
                    Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
                    0,
                    TempoHardfork::T1,
                )
                .unwrap();
            }
        }

        // Verify pool size
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(
            pending_count + queued_count,
            NUM_SENDERS * TXS_PER_SENDER as usize
        );

        // Each sender should have all transactions pending
        for (sender, nonce_key) in &senders {
            let seq_id = AASequenceId {
                address: *sender,
                nonce_key: *nonce_key,
            };
            for nonce in 0..TXS_PER_SENDER {
                let id = AA2dTransactionId::new(seq_id, nonce);
                assert!(pool.by_id.get(&id).unwrap().is_pending());
            }
        }

        // Independent set should have exactly NUM_SENDERS transactions (one per sender at nonce 0)
        assert_eq!(pool.independent_transactions.len(), NUM_SENDERS);
        for (sender, nonce_key) in &senders {
            let seq_id = AASequenceId {
                address: *sender,
                nonce_key: *nonce_key,
            };
            let tx0_id = AA2dTransactionId::new(seq_id, 0);
            assert!(
                pool.independent_transactions.contains_key(&tx0_id.seq_id),
                "Sender {sender:?} should have tx0 in independent set"
            );
        }

        // Simulate mining first transaction for each sender
        let mut on_chain_ids = HashMap::default();
        for (sender, nonce_key) in &senders {
            let seq_id = AASequenceId {
                address: *sender,
                nonce_key: *nonce_key,
            };
            on_chain_ids.insert(seq_id, 1u64);
        }

        let (promoted, mined) = pool.on_nonce_changes(on_chain_ids);

        // Should have mined NUM_SENDERS transactions
        assert_eq!(mined.len(), NUM_SENDERS);
        // No promotions - transactions [1,2,3,4] were already pending
        assert_eq!(promoted.len(), 0);

        // Pool size should be reduced
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(
            pending_count + queued_count,
            NUM_SENDERS * (TXS_PER_SENDER - 1) as usize
        );

        // Independent set should still have NUM_SENDERS transactions (now at nonce 1)
        assert_eq!(pool.independent_transactions.len(), NUM_SENDERS);
        for (sender, nonce_key) in &senders {
            let seq_id = AASequenceId {
                address: *sender,
                nonce_key: *nonce_key,
            };
            let tx1_id = AA2dTransactionId::new(seq_id, 1);
            assert!(
                pool.independent_transactions.contains_key(&tx1_id.seq_id),
                "Sender {sender:?} should have tx1 in independent set"
            );
        }

        pool.assert_invariants();
    }

    #[test_case::test_case(U256::ZERO)]
    #[test_case::test_case(U256::random())]
    fn on_chain_nonce_update_to_queued_tx_with_gaps(nonce_key: U256) {
        let mut pool = AA2dPool::default();
        let sender = Address::random();
        let seq_id = AASequenceId {
            address: sender,
            nonce_key,
        };

        // Start with gaps: insert [0, 3, 5]
        // This creates: tx0 (pending), tx3 (queued), tx5 (queued)
        let tx0 = TxBuilder::aa(sender).nonce_key(nonce_key).build();
        let tx3 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(3).build();
        let tx5 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(5).build();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx3, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx5, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        // Only tx0 should be in independent set
        assert_eq!(pool.independent_transactions.len(), 1);
        assert!(pool.independent_transactions.contains_key(&seq_id));

        // Verify initial state: tx0 pending, tx3 and tx5 queued
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count, 1, "Only tx0 should be pending");
        assert_eq!(queued_count, 2, "tx3 and tx5 should be queued");

        // Fill gaps to get [0, 1, 2, 3, 5]
        let tx1 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(1).build();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let tx2 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(2).build();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx2, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        // Now [0,1,2,3] should be pending, tx5 still queued
        let (pending_count, queued_count) = pool.pending_and_queued_txn_count();
        assert_eq!(pending_count, 4, "Transactions [0,1,2,3] should be pending");
        assert_eq!(queued_count, 1, "tx5 should still be queued");

        // Still only tx0 in independent set (at on-chain nonce 0)
        assert_eq!(pool.independent_transactions.len(), 1);
        assert!(pool.independent_transactions.contains_key(&seq_id));

        let mut on_chain_ids = HashMap::default();
        on_chain_ids.insert(seq_id, 3u64);
        let (_promoted, mined) = pool.on_nonce_changes(on_chain_ids);

        // Should have mined [0,1,2]
        assert_eq!(mined.len(), 3, "Should mine transactions [0,1,2]");

        // tx3 was already pending, so no promotions expected
        // After mining, tx3 should be in independent set
        assert_eq!(
            pool.independent_transactions.len(),
            1,
            "Should have one independent transaction"
        );
        let key = AA2dTransactionId::new(seq_id, 3);
        assert!(
            pool.independent_transactions.contains_key(&key.seq_id),
            "tx3 should be in independent set"
        );

        // Verify remaining pool state
        let (_pending_count, _queued_count) = pool.pending_and_queued_txn_count();
        // Should have tx3 (pending at on-chain nonce) and tx5 (queued due to gap at 4)

        pool.assert_invariants();

        // Now insert tx4 to fill the gap between tx3 and tx5
        // This is where the original test failure occurred
        let tx4 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(4).build();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx4, TransactionOrigin::Local)),
            3,
            TempoHardfork::T1,
        )
        .unwrap();

        // After inserting tx4, we should have [3, 4, 5] all in the pool
        let (_pending_count_after, _queued_count_after) = pool.pending_and_queued_txn_count();
        pool.assert_invariants();
    }

    #[test]
    fn append_pooled_transaction_elements_respects_limit() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();
        let nonce_key = U256::from(1);

        // Add 3 transactions with consecutive nonces
        let tx0 = TxBuilder::aa(sender).nonce_key(nonce_key).build();
        let tx0_hash = *tx0.hash();
        let tx0_len = tx0.encoded_length();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let tx1 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(1).build();
        let tx1_hash = *tx1.hash();
        let tx1_len = tx1.encoded_length();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let tx2 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(2).build();
        let tx2_hash = *tx2.hash();
        let tx2_len = tx2.encoded_length();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx2, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        // Test with no limit - should return all 3 transactions
        let mut accumulated = 0;
        let mut elements = Vec::new();
        pool.append_pooled_transaction_elements(
            &[tx0_hash, tx1_hash, tx2_hash],
            GetPooledTransactionLimit::None,
            &mut accumulated,
            &mut elements,
        );
        assert_eq!(elements.len(), 3, "Should return all 3 transactions");
        assert_eq!(
            accumulated,
            tx0_len + tx1_len + tx2_len,
            "Should accumulate all sizes"
        );

        // Test with a soft limit - stops after exceeding (not at) the limit
        // A limit of tx0_len - 1 means we stop after tx0 is added (since tx0_len > limit)
        let mut accumulated = 0;
        let mut elements = Vec::new();
        pool.append_pooled_transaction_elements(
            &[tx0_hash, tx1_hash, tx2_hash],
            GetPooledTransactionLimit::ResponseSizeSoftLimit(tx0_len - 1),
            &mut accumulated,
            &mut elements,
        );
        assert_eq!(
            elements.len(),
            1,
            "Should stop after first tx exceeds limit"
        );
        assert_eq!(accumulated, tx0_len, "Should accumulate first tx size");

        // Test with limit that allows exactly 2 transactions before exceeding
        // A limit of tx0_len + tx1_len - 1 means we stop after tx1 is added
        let mut accumulated = 0;
        let mut elements = Vec::new();
        pool.append_pooled_transaction_elements(
            &[tx0_hash, tx1_hash, tx2_hash],
            GetPooledTransactionLimit::ResponseSizeSoftLimit(tx0_len + tx1_len - 1),
            &mut accumulated,
            &mut elements,
        );
        assert_eq!(
            elements.len(),
            2,
            "Should stop after second tx exceeds limit"
        );
        assert_eq!(
            accumulated,
            tx0_len + tx1_len,
            "Should accumulate first two tx sizes"
        );

        // Test with pre-accumulated size that causes immediate stop after first tx
        let mut accumulated = tx0_len;
        let mut elements = Vec::new();
        pool.append_pooled_transaction_elements(
            &[tx1_hash, tx2_hash],
            GetPooledTransactionLimit::ResponseSizeSoftLimit(tx0_len + tx1_len - 1),
            &mut accumulated,
            &mut elements,
        );
        assert_eq!(
            elements.len(),
            1,
            "Should return 1 transaction when pre-accumulated size causes early stop"
        );
        assert_eq!(
            accumulated,
            tx0_len + tx1_len,
            "Should add to pre-accumulated size"
        );
    }
    // ============================================
    // Helper function tests
    // ============================================

    #[test]
    fn test_2d_pool_helpers() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();
        let tx = TxBuilder::aa(sender).build();
        let tx_hash = *tx.hash();

        assert!(!pool.contains(&tx_hash));
        assert!(pool.get(&tx_hash).is_none());

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        assert!(pool.contains(&tx_hash));
        let retrieved = pool.get(&tx_hash);
        assert!(retrieved.is_some());
        assert_eq!(retrieved.unwrap().hash(), &tx_hash);
    }

    #[test]
    fn test_pool_get_all() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();

        let tx0 = TxBuilder::aa(sender).build();
        let tx1 = TxBuilder::aa(sender).nonce(1).build();
        let tx0_hash = *tx0.hash();
        let tx1_hash = *tx1.hash();
        let fake_hash = alloy_primitives::B256::random();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let hashes = [tx0_hash, tx1_hash, fake_hash];
        let results = pool.get_all(hashes.iter());

        assert_eq!(results.len(), 2); // Only the two real transactions
    }

    #[test]
    fn test_pool_senders_iter() {
        let mut pool = AA2dPool::default();
        let sender1 = Address::random();
        let sender2 = Address::random();

        let tx1 = TxBuilder::aa(sender1).build();
        let tx2 = TxBuilder::aa(sender2).nonce_key(U256::from(1)).build();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx2, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let senders: Vec<_> = pool.senders_iter().collect();
        assert_eq!(senders.len(), 2);
        assert!(senders.contains(&&sender1));
        assert!(senders.contains(&&sender2));
    }

    #[test]
    fn test_pool_pending_and_queued_transactions() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();

        // Pending: tx0, tx1, tx2 (consecutive nonces starting from on-chain nonce 0)
        let tx0 = TxBuilder::aa(sender).build();
        let tx1 = TxBuilder::aa(sender).nonce(1).build();
        let tx2 = TxBuilder::aa(sender).nonce(2).build();
        let tx0_hash = *tx0.hash();
        let tx1_hash = *tx1.hash();
        let tx2_hash = *tx2.hash();

        // Queued: tx5, tx6, tx7 (gap after tx2)
        let tx5 = TxBuilder::aa(sender).nonce(5).build();
        let tx6 = TxBuilder::aa(sender).nonce(6).build();
        let tx7 = TxBuilder::aa(sender).nonce(7).build();
        let tx5_hash = *tx5.hash();
        let tx6_hash = *tx6.hash();
        let tx7_hash = *tx7.hash();

        for tx in [tx0, tx1, tx2, tx5, tx6, tx7] {
            pool.add_transaction(
                Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            )
            .unwrap();
        }

        let pending: Vec<_> = pool.pending_transactions().collect();
        assert_eq!(pending.len(), 3);
        let pending_hashes: HashSet<_> = pending.iter().map(|tx| *tx.hash()).collect();
        assert!(pending_hashes.contains(&tx0_hash));
        assert!(pending_hashes.contains(&tx1_hash));
        assert!(pending_hashes.contains(&tx2_hash));

        let queued: Vec<_> = pool.queued_transactions().collect();
        assert_eq!(queued.len(), 3);
        let queued_hashes: HashSet<_> = queued.iter().map(|tx| *tx.hash()).collect();
        assert!(queued_hashes.contains(&tx5_hash));
        assert!(queued_hashes.contains(&tx6_hash));
        assert!(queued_hashes.contains(&tx7_hash));
    }

    #[test]
    fn test_append_all_transactions() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();
        let expiring_sender = Address::random();

        let tx0 = TxBuilder::aa(sender).build();
        let tx2 = TxBuilder::aa(sender).nonce(2).build();
        let expiring_tx = TxBuilder::aa(expiring_sender).nonce_key(U256::MAX).build();

        for tx in [tx0, tx2, expiring_tx] {
            pool.add_transaction(
                Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            )
            .unwrap();
        }

        let expected_pending: HashSet<_> =
            pool.pending_transactions().map(|tx| *tx.hash()).collect();
        let expected_queued: HashSet<_> = pool.queued_transactions().map(|tx| *tx.hash()).collect();

        let mut transactions = AllPoolTransactions::default();
        pool.append_all_transactions(&mut transactions);

        let pending_hashes: HashSet<_> = transactions.pending.iter().map(|tx| *tx.hash()).collect();
        let queued_hashes: HashSet<_> = transactions.queued.iter().map(|tx| *tx.hash()).collect();

        assert_eq!(pending_hashes, expected_pending);
        assert_eq!(queued_hashes, expected_queued);
    }

    #[test]
    fn test_pool_get_transactions_by_sender_iter() {
        let mut pool = AA2dPool::default();
        let sender1 = Address::random();
        let sender2 = Address::random();

        let tx1 = TxBuilder::aa(sender1).nonce_key(U256::ZERO).build();
        let tx2 = TxBuilder::aa(sender2).nonce_key(U256::from(1)).build();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx2, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let sender1_txs: Vec<_> = pool.get_transactions_by_sender_iter(sender1).collect();
        assert_eq!(sender1_txs.len(), 1);
        assert_eq!(sender1_txs[0].sender(), sender1);

        let sender2_txs: Vec<_> = pool.get_transactions_by_sender_iter(sender2).collect();
        assert_eq!(sender2_txs.len(), 1);
        assert_eq!(sender2_txs[0].sender(), sender2);
    }

    #[test]
    fn test_pool_get_transactions_by_origin_iter() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();

        let tx0 = TxBuilder::aa(sender).nonce_key(U256::ZERO).build();
        let tx1 = TxBuilder::aa(sender).nonce_key(U256::ZERO).nonce(1).build();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::External)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let local_txs: Vec<_> = pool
            .get_transactions_by_origin_iter(TransactionOrigin::Local)
            .collect();
        assert_eq!(local_txs.len(), 1);

        let external_txs: Vec<_> = pool
            .get_transactions_by_origin_iter(TransactionOrigin::External)
            .collect();
        assert_eq!(external_txs.len(), 1);
    }

    #[test]
    fn test_pool_get_pending_transactions_by_origin_iter() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();

        let tx0 = TxBuilder::aa(sender).nonce_key(U256::ZERO).build();
        let tx2 = TxBuilder::aa(sender).nonce_key(U256::ZERO).nonce(2).build(); // Queued due to gap

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx2, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let pending_local: Vec<_> = pool
            .get_pending_transactions_by_origin_iter(TransactionOrigin::Local)
            .collect();
        assert_eq!(pending_local.len(), 1); // Only tx0 is pending
    }

    #[test]
    fn test_pool_all_transaction_hashes_iter() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();

        let tx0 = TxBuilder::aa(sender).nonce_key(U256::ZERO).build();
        let tx1 = TxBuilder::aa(sender).nonce_key(U256::ZERO).nonce(1).build();
        let tx0_hash = *tx0.hash();
        let tx1_hash = *tx1.hash();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let hashes: Vec<_> = pool.all_transaction_hashes_iter().collect();
        assert_eq!(hashes.len(), 2);
        assert!(hashes.contains(&tx0_hash));
        assert!(hashes.contains(&tx1_hash));
    }

    #[test]
    fn test_pool_pooled_transactions_hashes_iter() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();

        let tx0 = TxBuilder::aa(sender).nonce_key(U256::ZERO).build();
        let tx1 = TxBuilder::aa(sender).nonce_key(U256::ZERO).nonce(1).build();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let hashes: Vec<_> = pool.pooled_transactions_hashes_iter().collect();
        assert_eq!(hashes.len(), 2);
    }

    #[test]
    fn test_pool_pooled_transactions_iter() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();

        let tx0 = TxBuilder::aa(sender).nonce_key(U256::ZERO).build();
        let tx1 = TxBuilder::aa(sender).nonce_key(U256::ZERO).nonce(1).build();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let txs: Vec<_> = pool.pooled_transactions_iter().collect();
        assert_eq!(txs.len(), 2);
    }

    // ============================================
    // BestAA2dTransactions tests
    // ============================================

    #[test]
    fn test_best_transactions_iterator() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();

        let tx0 = TxBuilder::aa(sender).nonce_key(U256::ZERO).build();
        let tx1 = TxBuilder::aa(sender).nonce_key(U256::ZERO).nonce(1).build();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let mut best = pool.best_transactions();

        // Should iterate through pending transactions
        let first = best.next();
        assert!(first.is_some());

        let second = best.next();
        assert!(second.is_some());

        let third = best.next();
        assert!(third.is_none());
    }

    #[test]
    fn test_best_transactions_size_hint_counts_snapshot() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();
        let expiring_sender = Address::random();

        let tx0 = TxBuilder::aa(sender).nonce_key(U256::ZERO).build();
        let tx1 = TxBuilder::aa(sender).nonce_key(U256::ZERO).nonce(1).build();
        let expiring_tx = TxBuilder::aa(expiring_sender).nonce_key(U256::MAX).build();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(expiring_tx, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let mut best = pool.best_transactions();

        assert_eq!(best.size_hint(), (0, Some(3)));
        assert!(best.next().is_some());
        assert_eq!(best.size_hint(), (0, Some(2)));
        assert!(best.next().is_some());
        assert_eq!(best.size_hint(), (0, Some(1)));
        assert!(best.next().is_some());
        assert_eq!(best.size_hint(), (0, Some(0)));
    }

    #[test]
    fn test_best_transactions_size_hint_ignores_unread_new_transactions() {
        let mut pool = AA2dPool::default();
        let snapshot_sender = Address::random();
        let incoming_sender = Address::random();

        let snapshot_tx = TxBuilder::aa(snapshot_sender).nonce_key(U256::ZERO).build();
        let incoming_tx = TxBuilder::aa(incoming_sender)
            .nonce_key(U256::from(1))
            .build();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(snapshot_tx, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let best = pool.best_transactions();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(incoming_tx, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        assert_eq!(best.size_hint(), (0, Some(1)));
    }

    #[test]
    fn test_best_transactions_mark_invalid() {
        use reth_primitives_traits::transaction::error::InvalidTransactionError;

        let mut pool = AA2dPool::default();
        let sender = Address::random();

        let tx0 = TxBuilder::aa(sender).nonce_key(U256::ZERO).build();
        let tx1 = TxBuilder::aa(sender).nonce_key(U256::ZERO).nonce(1).build();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let mut best = pool.best_transactions();

        let first = best.next().unwrap();

        // Mark it invalid
        let error = reth_transaction_pool::error::InvalidPoolTransactionError::Consensus(
            InvalidTransactionError::TxTypeNotSupported,
        );
        best.mark_invalid(&first, error);

        // The sequence should be in the invalid set, so next tx from same sender should be skipped
        // But since we already consumed tx0, we'd get tx1 next - but the sequence is now invalid
    }

    #[test]
    fn test_best_transactions_expiring_nonce_independent() {
        // Expiring nonce transactions (nonce_key == U256::MAX) are always independent
        // and should not trigger unlock logic for dependent transactions
        let mut pool = AA2dPool::default();
        let sender = Address::random();

        // Add expiring nonce transaction
        let tx = TxBuilder::aa(sender).nonce_key(U256::MAX).nonce(0).build();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let mut best = pool.best_transactions();

        // Should return the transaction
        let first = best.next();
        assert!(first.is_some());

        // No more transactions
        assert!(best.next().is_none());
    }

    // ============================================
    // Remove transactions tests
    // ============================================

    #[test]
    fn test_remove_transactions_by_sender() {
        let mut pool = AA2dPool::default();
        let sender1 = Address::random();
        let sender2 = Address::random();

        let tx1 = TxBuilder::aa(sender1).nonce_key(U256::ZERO).build();
        let tx2 = TxBuilder::aa(sender2).nonce_key(U256::from(1)).build();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx2, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let removed = pool.remove_transactions_by_sender(sender1);
        assert_eq!(removed.len(), 1);
        assert_eq!(removed[0].sender(), sender1);

        // sender1's tx should be gone, sender2's should remain
        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending + queued, 1);

        pool.assert_invariants();
    }

    #[test]
    fn test_remove_transactions_and_descendants() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();

        let tx0 = TxBuilder::aa(sender).nonce_key(U256::ZERO).build();
        let tx1 = TxBuilder::aa(sender).nonce_key(U256::ZERO).nonce(1).build();
        let tx2 = TxBuilder::aa(sender).nonce_key(U256::ZERO).nonce(2).build();
        let tx0_hash = *tx0.hash();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx2, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        // Remove tx0 and its descendants (tx1, tx2)
        let removed = pool.remove_transactions_and_descendants([&tx0_hash].into_iter());
        assert_eq!(removed.len(), 3);

        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending + queued, 0);

        pool.assert_invariants();
    }

    // ============================================
    // AASequenceId and AA2dTransactionId tests
    // ============================================

    #[test]
    fn test_aa_sequence_id_equality() {
        let addr = Address::random();
        let nonce_key = U256::from(42);

        let id1 = AASequenceId::new(addr, nonce_key);
        let id2 = AASequenceId::new(addr, nonce_key);
        let id3 = AASequenceId::new(Address::random(), nonce_key);

        assert_eq!(id1, id2);
        assert_ne!(id1, id3);
    }

    #[test]
    fn test_aa2d_transaction_id_unlocks() {
        let addr = Address::random();
        let seq_id = AASequenceId::new(addr, U256::ZERO);
        let tx_id = AA2dTransactionId::new(seq_id, 5);

        let next_id = tx_id.unlocks();
        assert_eq!(next_id.seq_id, seq_id);
        assert_eq!(next_id.nonce, 6);
    }

    #[test]
    fn test_aa2d_transaction_id_ordering() {
        let addr = Address::random();
        let seq_id = AASequenceId::new(addr, U256::ZERO);

        let id1 = AA2dTransactionId::new(seq_id, 1);
        let id2 = AA2dTransactionId::new(seq_id, 2);

        assert!(id1 < id2);
    }

    // ============================================
    // Edge case tests
    // ============================================

    #[test]
    fn test_nonce_overflow_at_u64_max() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();
        let nonce_key = U256::ZERO;

        let tx = TxBuilder::aa(sender)
            .nonce_key(nonce_key)
            .nonce(u64::MAX)
            .build();
        let valid_tx = wrap_valid_tx(tx, TransactionOrigin::Local);

        let result = pool.add_transaction(Arc::new(valid_tx), u64::MAX, TempoHardfork::T1);
        assert!(result.is_ok());

        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 1);
        assert_eq!(queued, 0);

        let seq_id = AASequenceId::new(sender, nonce_key);
        let tx_id = AA2dTransactionId::new(seq_id, u64::MAX);
        let unlocked = tx_id.unlocks();
        assert_eq!(
            unlocked.nonce,
            u64::MAX,
            "saturating_add should not overflow"
        );

        pool.assert_invariants();
    }

    #[test]
    fn test_nonce_near_max_with_gap() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();
        let nonce_key = U256::ZERO;

        let tx_max = TxBuilder::aa(sender)
            .nonce_key(nonce_key)
            .nonce(u64::MAX)
            .build();
        let tx_max_minus_1 = TxBuilder::aa(sender)
            .nonce_key(nonce_key)
            .nonce(u64::MAX - 1)
            .build();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx_max, TransactionOrigin::Local)),
            u64::MAX - 1,
            TempoHardfork::T1,
        )
        .unwrap();

        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 0, "tx at u64::MAX should be queued (gap exists)");
        assert_eq!(queued, 1);

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx_max_minus_1, TransactionOrigin::Local)),
            u64::MAX - 1,
            TempoHardfork::T1,
        )
        .unwrap();

        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 2, "both should now be pending");
        assert_eq!(queued, 0);

        pool.assert_invariants();
    }

    #[test]
    fn test_empty_pool_operations() {
        let pool = AA2dPool::default();

        assert_eq!(pool.pending_and_queued_txn_count(), (0, 0));
        assert!(pool.get(&B256::random()).is_none());
        assert!(!pool.contains(&B256::random()));
        assert_eq!(pool.senders_iter().count(), 0);
        assert_eq!(pool.pending_transactions().count(), 0);
        assert_eq!(pool.queued_transactions().count(), 0);
        assert_eq!(pool.all_transaction_hashes_iter().count(), 0);
        assert_eq!(pool.pooled_transactions_hashes_iter().count(), 0);
        assert_eq!(pool.pooled_transactions_iter().count(), 0);

        let mut best = pool.best_transactions();
        assert!(best.next().is_none());
    }

    #[test]
    fn test_empty_pool_remove_operations() {
        let mut pool = AA2dPool::default();
        let random_hash = B256::random();
        let random_sender = Address::random();

        let removed = pool.remove_transactions([&random_hash].into_iter());
        assert!(removed.is_empty());

        let removed = pool.remove_transactions_by_sender(random_sender);
        assert!(removed.is_empty());

        let removed = pool.remove_transactions_and_descendants([&random_hash].into_iter());
        assert!(removed.is_empty());

        pool.assert_invariants();
    }

    #[test]
    fn test_empty_pool_on_nonce_changes() {
        let mut pool = AA2dPool::default();

        let mut changes = HashMap::default();
        changes.insert(AASequenceId::new(Address::random(), U256::ZERO), 5u64);

        let (promoted, mined) = pool.on_nonce_changes(changes);
        assert!(promoted.is_empty());
        assert!(mined.is_empty());

        pool.assert_invariants();
    }

    // ============================================
    // Error path tests
    // ============================================

    #[test]
    fn test_add_already_imported_transaction() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();

        let tx = TxBuilder::aa(sender).nonce_key(U256::ZERO).build();
        let tx_hash = *tx.hash();
        let valid_tx = Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local));

        pool.add_transaction(valid_tx.clone(), 0, TempoHardfork::T1)
            .unwrap();

        let result = pool.add_transaction(valid_tx, 0, TempoHardfork::T1);
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert_eq!(err.hash, tx_hash);
        assert!(
            matches!(err.kind, PoolErrorKind::AlreadyImported),
            "Expected AlreadyImported, got {:?}",
            err.kind
        );

        pool.assert_invariants();
    }

    #[test]
    fn test_add_outdated_nonce_transaction() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();

        let tx = TxBuilder::aa(sender).nonce_key(U256::ZERO).nonce(5).build();
        let tx_hash = *tx.hash();
        let valid_tx = Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local));

        let result = pool.add_transaction(valid_tx, 10, TempoHardfork::T1);
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert_eq!(err.hash, tx_hash);
        assert!(
            matches!(
                err.kind,
                PoolErrorKind::InvalidTransaction(InvalidPoolTransactionError::Consensus(
                    InvalidTransactionError::NonceNotConsistent { tx: 5, state: 10 }
                ))
            ),
            "Expected NonceNotConsistent, got {:?}",
            err.kind
        );

        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending + queued, 0);
    }

    #[test]
    fn test_replacement_underpriced() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();

        let tx1 = TxBuilder::aa(sender)
            .nonce_key(U256::ZERO)
            .max_priority_fee(1_000_000_000)
            .max_fee(2_000_000_000)
            .build();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let tx2 = TxBuilder::aa(sender)
            .nonce_key(U256::ZERO)
            .max_priority_fee(1_000_000_001)
            .max_fee(2_000_000_001)
            .build();
        let tx2_hash = *tx2.hash();
        let result = pool.add_transaction(
            Arc::new(wrap_valid_tx(tx2, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        );

        assert!(result.is_err());
        let err = result.unwrap_err();
        assert_eq!(err.hash, tx2_hash);
        assert!(
            matches!(err.kind, PoolErrorKind::ReplacementUnderpriced),
            "Expected ReplacementUnderpriced, got {:?}",
            err.kind
        );

        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending + queued, 1);

        pool.assert_invariants();
    }

    // ============================================
    // Boundary tests (max_txs limit and discard)
    // ============================================

    #[test]
    fn test_discard_at_max_txs_limit() {
        let config = AA2dPoolConfig {
            price_bump_config: PriceBumpConfig::default(),
            pending_limit: SubPoolLimit {
                max_txs: 3,
                max_size: usize::MAX,
            },
            queued_limit: SubPoolLimit {
                max_txs: 10000,
                max_size: usize::MAX,
            },
            max_txs_per_sender: DEFAULT_MAX_TXS_PER_SENDER,
        };
        let mut pool = AA2dPool::new(config);

        for i in 0..5usize {
            let sender = Address::from_word(B256::from(U256::from(i)));
            let tx = TxBuilder::aa(sender).nonce_key(U256::from(i)).build();
            let result = pool.add_transaction(
                Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            );
            assert!(result.is_ok());
        }

        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending + queued, 3, "Pool should be capped at max_txs=3");
        assert_eq!(pending, 3, "All remaining transactions should be pending");

        pool.assert_invariants();
    }

    #[test]
    fn test_discard_removes_lowest_priority_same_priority_uses_submission_order() {
        let config = AA2dPoolConfig {
            price_bump_config: PriceBumpConfig::default(),
            pending_limit: SubPoolLimit {
                max_txs: 2,
                max_size: usize::MAX,
            },
            queued_limit: SubPoolLimit {
                max_txs: 10000,
                max_size: usize::MAX,
            },
            max_txs_per_sender: DEFAULT_MAX_TXS_PER_SENDER,
        };
        let mut pool = AA2dPool::new(config);
        let sender = Address::random();

        // All transactions have the same priority, so the tiebreaker is submission order.
        // The most recently submitted (tx2) should be evicted first.
        let tx0 = TxBuilder::aa(sender).nonce_key(U256::ZERO).build();
        let tx1 = TxBuilder::aa(sender).nonce_key(U256::ZERO).nonce(1).build();
        let tx2 = TxBuilder::aa(sender).nonce_key(U256::ZERO).nonce(2).build();
        let tx0_hash = *tx0.hash();
        let tx1_hash = *tx1.hash();
        let tx2_hash = *tx2.hash();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        let result = pool.add_transaction(
            Arc::new(wrap_valid_tx(tx2, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        );
        assert!(result.is_ok());

        let added = result.unwrap();
        if let AddedTransaction::Pending(pending) = added {
            assert!(
                !pending.discarded.is_empty(),
                "Should have discarded transactions"
            );
            assert_eq!(
                pending.discarded[0].hash(),
                &tx2_hash,
                "tx2 (last submitted, lowest priority tiebreaker) should be discarded"
            );
        } else {
            panic!("Expected Pending result");
        }

        assert!(pool.contains(&tx0_hash));
        assert!(pool.contains(&tx1_hash));
        assert!(!pool.contains(&tx2_hash));

        pool.assert_invariants();
    }

    /// Tests that queued transactions (with nonce gaps) also respect the max_txs limit.
    #[test]
    fn test_discard_enforced_for_queued_transactions() {
        let config = AA2dPoolConfig {
            price_bump_config: PriceBumpConfig::default(),
            pending_limit: SubPoolLimit {
                max_txs: 2,
                max_size: usize::MAX,
            },
            queued_limit: SubPoolLimit {
                max_txs: 2,
                max_size: usize::MAX,
            },
            max_txs_per_sender: DEFAULT_MAX_TXS_PER_SENDER,
        };
        let mut pool = AA2dPool::new(config);

        // Add 5 transactions each with a LARGE nonce gap so they are all queued
        for i in 0..5usize {
            let sender = Address::from_word(B256::from(U256::from(i)));
            let tx = TxBuilder::aa(sender)
                .nonce_key(U256::from(i))
                .nonce(1000)
                .build();
            let result = pool.add_transaction(
                Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            );
            assert!(result.is_ok(), "Transaction {i} should be added");
        }

        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(
            pending + queued,
            2,
            "Pool should be capped at max_txs=2, but has {pending} pending + {queued} queued",
        );

        pool.assert_invariants();
    }

    /// Verifies queued transactions respect their own limit independently
    #[test]
    fn test_queued_limit_enforced_separately() {
        let config = AA2dPoolConfig {
            price_bump_config: PriceBumpConfig::default(),
            pending_limit: SubPoolLimit {
                max_txs: 10,
                max_size: usize::MAX,
            },
            queued_limit: SubPoolLimit {
                max_txs: 3,
                max_size: usize::MAX,
            },
            max_txs_per_sender: DEFAULT_MAX_TXS_PER_SENDER,
        };
        let mut pool = AA2dPool::new(config);

        // Add 5 queued transactions (far-future nonces)
        for i in 0..5usize {
            let sender = Address::from_word(B256::from(U256::from(i)));
            let tx = TxBuilder::aa(sender)
                .nonce_key(U256::from(i))
                .nonce(1000)
                .build();
            let _ = pool.add_transaction(
                Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            );
        }

        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(queued, 3, "Queued should be capped at 3");
        assert_eq!(pending, 0, "No pending transactions");
        pool.assert_invariants();
    }

    /// Verifies pending transactions respect their own limit independently
    #[test]
    fn test_pending_limit_enforced_separately() {
        let config = AA2dPoolConfig {
            price_bump_config: PriceBumpConfig::default(),
            pending_limit: SubPoolLimit {
                max_txs: 3,
                max_size: usize::MAX,
            },
            queued_limit: SubPoolLimit {
                max_txs: 10,
                max_size: usize::MAX,
            },
            max_txs_per_sender: DEFAULT_MAX_TXS_PER_SENDER,
        };
        let mut pool = AA2dPool::new(config);

        // Add 5 pending transactions (nonce=0, different senders)
        for i in 0..5usize {
            let sender = Address::from_word(B256::from(U256::from(i)));
            let tx = TxBuilder::aa(sender).nonce_key(U256::from(i)).build();
            let _ = pool.add_transaction(
                Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            );
        }

        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 3, "Pending should be capped at 3");
        assert_eq!(queued, 0, "No queued transactions");
        pool.assert_invariants();
    }

    /// Verifies queued spam cannot evict pending transactions
    #[test]
    fn test_queued_eviction_does_not_affect_pending() {
        let config = AA2dPoolConfig {
            price_bump_config: PriceBumpConfig::default(),
            pending_limit: SubPoolLimit {
                max_txs: 5,
                max_size: usize::MAX,
            },
            queued_limit: SubPoolLimit {
                max_txs: 2,
                max_size: usize::MAX,
            },
            max_txs_per_sender: DEFAULT_MAX_TXS_PER_SENDER,
        };
        let mut pool = AA2dPool::new(config);

        // First add 3 pending transactions
        let mut pending_hashes = Vec::new();
        for i in 0..3usize {
            let sender = Address::from_word(B256::from(U256::from(i)));
            let tx = TxBuilder::aa(sender).nonce_key(U256::from(i)).build();
            let hash = *tx.hash();
            pending_hashes.push(hash);
            let _ = pool.add_transaction(
                Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            );
        }

        // Now flood with 10 queued transactions
        for i in 100..110usize {
            let sender = Address::from_word(B256::from(U256::from(i)));
            let tx = TxBuilder::aa(sender)
                .nonce_key(U256::from(i))
                .nonce(1000)
                .build();
            let _ = pool.add_transaction(
                Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            );
        }

        // All pending should still be there
        for hash in &pending_hashes {
            assert!(
                pool.contains(hash),
                "Pending tx should not be evicted by queued spam"
            );
        }

        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 3, "All 3 pending should remain");
        assert_eq!(queued, 2, "Queued capped at 2");
        pool.assert_invariants();
    }

    /// Tests that eviction is based on priority, not address ordering.
    /// This prevents DoS attacks where adversaries use vanity addresses with leading zeroes.
    #[test]
    fn test_discard_evicts_low_priority_over_vanity_address() {
        let config = AA2dPoolConfig {
            price_bump_config: PriceBumpConfig::default(),
            pending_limit: SubPoolLimit {
                max_txs: 2,
                max_size: usize::MAX,
            },
            queued_limit: SubPoolLimit {
                max_txs: 10,
                max_size: usize::MAX,
            },
            max_txs_per_sender: DEFAULT_MAX_TXS_PER_SENDER,
        };
        let mut pool = AA2dPool::new(config);

        // Vanity address with leading zeroes (would sort first lexicographically)
        let vanity_sender = Address::from_word(B256::from_slice(&[0u8; 32])); // 0x0000...0000
        // Normal address (would sort later lexicographically)
        let normal_sender = Address::from_word(B256::from_slice(&[0xff; 32])); // 0xffff...ffff

        // max_fee must be > TEMPO_T1_BASE_FEE (20 gwei) for priority calculation to work
        // effective_tip = min(max_fee - base_fee, max_priority_fee)
        let high_max_fee = 30_000_000_000u128; // 30 gwei, above 20 gwei base fee

        // Add vanity address tx with HIGH priority (should be kept despite sorting first lexicographically)
        // effective_tip = min(30 gwei - 20 gwei, 5 gwei) = 5 gwei
        let high_priority_tx = TxBuilder::aa(vanity_sender)
            .nonce_key(U256::ZERO)
            .max_fee(high_max_fee)
            .max_priority_fee(5_000_000_000) // 5 gwei priority
            .build();
        let high_priority_hash = *high_priority_tx.hash();

        // Add normal address tx with LOW priority (should be evicted)
        // effective_tip = min(30 gwei - 20 gwei, 1 wei) = 1 wei
        let low_priority_tx = TxBuilder::aa(normal_sender)
            .nonce_key(U256::ZERO)
            .max_fee(high_max_fee)
            .max_priority_fee(1) // Very low priority
            .build();
        let low_priority_hash = *low_priority_tx.hash();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(high_priority_tx, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(low_priority_tx, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        // Add a third tx that triggers eviction
        // effective_tip = min(30 gwei - 20 gwei, 3 gwei) = 3 gwei (medium)
        let trigger_tx = TxBuilder::aa(Address::random())
            .nonce_key(U256::from(1))
            .max_fee(high_max_fee)
            .max_priority_fee(3_000_000_000) // 3 gwei - medium priority
            .build();
        let trigger_hash = *trigger_tx.hash();

        let result = pool.add_transaction(
            Arc::new(wrap_valid_tx(trigger_tx, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        );
        assert!(result.is_ok());

        let added = result.unwrap();
        if let AddedTransaction::Pending(pending) = added {
            assert!(
                !pending.discarded.is_empty(),
                "Should have discarded transactions"
            );
            // The low priority tx (normal address) should be evicted, NOT the high priority vanity address
            assert_eq!(
                pending.discarded[0].hash(),
                &low_priority_hash,
                "Low priority tx should be evicted, not the high-priority vanity address tx"
            );
        } else {
            panic!("Expected Pending result");
        }

        // Verify: high priority vanity address tx should be kept, low priority normal address tx should be evicted
        assert!(
            pool.contains(&high_priority_hash),
            "High priority vanity address tx should be kept"
        );
        assert!(
            !pool.contains(&low_priority_hash),
            "Low priority tx should be evicted"
        );
        assert!(pool.contains(&trigger_hash), "Trigger tx should be kept");

        pool.assert_invariants();
    }

    /// Tests that a sender cannot exceed the per-sender transaction limit.
    #[test]
    fn test_per_sender_limit_rejects_excess_transactions() {
        let config = AA2dPoolConfig {
            price_bump_config: PriceBumpConfig::default(),
            pending_limit: SubPoolLimit {
                max_txs: 1000,
                max_size: usize::MAX,
            },
            queued_limit: SubPoolLimit {
                max_txs: 1000,
                max_size: usize::MAX,
            },
            max_txs_per_sender: 3,
        };
        let mut pool = AA2dPool::new(config);
        let sender = Address::random();

        // Add transactions up to the limit
        for nonce in 0..3u64 {
            let tx = TxBuilder::aa(sender)
                .nonce_key(U256::ZERO)
                .nonce(nonce)
                .build();
            let result = pool.add_transaction(
                Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            );
            assert!(result.is_ok(), "Transaction {nonce} should be accepted");
        }

        // The 4th transaction from the same sender should be rejected
        let tx = TxBuilder::aa(sender).nonce_key(U256::ZERO).nonce(3).build();
        let result = pool.add_transaction(
            Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        );
        assert!(result.is_err(), "4th transaction should be rejected");
        let err = result.unwrap_err();
        assert!(
            matches!(err.kind, PoolErrorKind::SpammerExceededCapacity(_)),
            "Error should be SpammerExceededCapacity, got {:?}",
            err.kind
        );

        // A different sender should still be able to add transactions
        let other_sender = Address::random();
        let tx = TxBuilder::aa(other_sender).nonce_key(U256::ZERO).build();
        let result = pool.add_transaction(
            Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        );
        assert!(result.is_ok(), "Different sender should be accepted");

        pool.assert_invariants();
    }

    /// Tests that replacing a transaction doesn't count against the per-sender limit.
    #[test]
    fn test_per_sender_limit_allows_replacement() {
        let config = AA2dPoolConfig {
            price_bump_config: PriceBumpConfig::default(),
            pending_limit: SubPoolLimit {
                max_txs: 1000,
                max_size: usize::MAX,
            },
            queued_limit: SubPoolLimit {
                max_txs: 1000,
                max_size: usize::MAX,
            },
            max_txs_per_sender: 2,
        };
        let mut pool = AA2dPool::new(config);
        let sender = Address::random();

        // Add 2 transactions to reach the limit
        for nonce in 0..2u64 {
            let tx = TxBuilder::aa(sender)
                .nonce_key(U256::ZERO)
                .nonce(nonce)
                .build();
            pool.add_transaction(
                Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            )
            .unwrap();
        }

        // Replace the first transaction with a higher fee (should succeed)
        let replacement_tx = TxBuilder::aa(sender)
            .nonce_key(U256::ZERO)
            .nonce(0)
            .max_fee(100_000_000_000) // Higher fee to pass replacement check
            .max_priority_fee(50_000_000_000)
            .build();
        let result = pool.add_transaction(
            Arc::new(wrap_valid_tx(replacement_tx, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        );
        assert!(
            result.is_ok(),
            "Replacement should be allowed even at limit"
        );

        pool.assert_invariants();
    }

    /// Tests that removing a transaction frees up a slot for the sender.
    #[test]
    fn test_per_sender_limit_freed_after_removal() {
        let config = AA2dPoolConfig {
            price_bump_config: PriceBumpConfig::default(),
            pending_limit: SubPoolLimit {
                max_txs: 1000,
                max_size: usize::MAX,
            },
            queued_limit: SubPoolLimit {
                max_txs: 1000,
                max_size: usize::MAX,
            },
            max_txs_per_sender: 2,
        };
        let mut pool = AA2dPool::new(config);
        let sender = Address::random();

        // Add 2 transactions to reach the limit
        let tx1 = TxBuilder::aa(sender).nonce_key(U256::ZERO).nonce(0).build();
        let tx1_hash = *tx1.hash();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let tx2 = TxBuilder::aa(sender).nonce_key(U256::ZERO).nonce(1).build();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx2, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        // 3rd should fail
        let tx3 = TxBuilder::aa(sender).nonce_key(U256::ZERO).nonce(2).build();
        let result = pool.add_transaction(
            Arc::new(wrap_valid_tx(tx3.clone(), TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        );
        assert!(result.is_err(), "3rd should be rejected at limit");

        // Remove the first transaction
        pool.remove_transactions(std::iter::once(&tx1_hash));

        // Now adding the 3rd should succeed
        let result = pool.add_transaction(
            Arc::new(wrap_valid_tx(tx3, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        );
        assert!(result.is_ok(), "3rd should succeed after removal");

        pool.assert_invariants();
    }

    /// Tests that expiring nonce transactions also respect per-sender limits.
    #[test]
    fn test_per_sender_limit_includes_expiring_nonce_txs() {
        let config = AA2dPoolConfig {
            price_bump_config: PriceBumpConfig::default(),
            pending_limit: SubPoolLimit {
                max_txs: 1000,
                max_size: usize::MAX,
            },
            queued_limit: SubPoolLimit {
                max_txs: 1000,
                max_size: usize::MAX,
            },
            max_txs_per_sender: 2,
        };
        let mut pool = AA2dPool::new(config);
        let sender = Address::random();

        // Add one regular 2D nonce tx
        let tx1 = TxBuilder::aa(sender).nonce_key(U256::ZERO).nonce(0).build();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        // Add one expiring nonce tx (nonce_key = U256::MAX)
        let tx2 = TxBuilder::aa(sender).nonce_key(U256::MAX).nonce(0).build();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx2, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        // The 3rd transaction (either type) should be rejected
        let tx3 = TxBuilder::aa(sender)
            .nonce_key(U256::from(1))
            .nonce(0)
            .build();
        let result = pool.add_transaction(
            Arc::new(wrap_valid_tx(tx3, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        );
        assert!(
            result.is_err(),
            "3rd tx should be rejected due to per-sender limit"
        );

        pool.assert_invariants();
    }

    // ============================================
    // Improved BestTransactions tests
    // ============================================

    #[test]
    fn test_best_transactions_mark_invalid_skips_sequence() {
        use reth_primitives_traits::transaction::error::InvalidTransactionError;

        let mut pool = AA2dPool::default();
        let sender1 = Address::random();
        let sender2 = Address::random();

        let tx1_0 = TxBuilder::aa(sender1).nonce_key(U256::ZERO).build();
        let tx1_1 = TxBuilder::aa(sender1)
            .nonce_key(U256::ZERO)
            .nonce(1)
            .build();
        let tx2_0 = TxBuilder::aa(sender2).nonce_key(U256::from(1)).build();

        let tx1_0_hash = *tx1_0.hash();
        let tx2_0_hash = *tx2_0.hash();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1_0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1_1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx2_0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let mut best = pool.best_transactions();

        let first = best.next().unwrap();
        let first_hash = *first.hash();

        let error =
            InvalidPoolTransactionError::Consensus(InvalidTransactionError::TxTypeNotSupported);
        best.mark_invalid(&first, error);

        let mut remaining_hashes = HashSet::new();
        for tx in best {
            remaining_hashes.insert(*tx.hash());
        }

        if first_hash == tx1_0_hash {
            assert!(
                !remaining_hashes.contains(&tx1_0_hash),
                "tx1_0 was consumed"
            );
            assert!(
                remaining_hashes.contains(&tx2_0_hash),
                "tx2_0 should still be yielded"
            );
        } else {
            assert!(
                remaining_hashes.contains(&tx1_0_hash) || remaining_hashes.contains(&tx2_0_hash),
                "At least one other independent tx should be yielded"
            );
        }
    }

    #[test]
    fn test_best_transactions_order_by_priority() {
        let mut pool = AA2dPool::default();

        let sender1 = Address::random();
        let sender2 = Address::random();

        let low_priority = TxBuilder::aa(sender1)
            .nonce_key(U256::ZERO)
            .max_priority_fee(1_000_000)
            .max_fee(2_000_000)
            .build();
        let high_priority = TxBuilder::aa(sender2)
            .nonce_key(U256::from(1))
            .max_priority_fee(10_000_000_000)
            .max_fee(20_000_000_000)
            .build();
        let high_priority_hash = *high_priority.hash();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(low_priority, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(high_priority, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let mut best = pool.best_transactions();
        let first = best.next().unwrap();

        assert_eq!(
            first.hash(),
            &high_priority_hash,
            "Higher priority transaction should come first"
        );
    }

    fn priority_flip_pool(block_base_fee: u64) -> (AA2dPool, B256, B256) {
        priority_flip_pool_with_config(block_base_fee, AA2dPoolConfig::default())
    }

    fn priority_flip_pool_with_config(
        block_base_fee: u64,
        config: AA2dPoolConfig,
    ) -> (AA2dPool, B256, B256) {
        let mut pool = AA2dPool::new(config);
        pool.set_base_fee(TEMPO_T1_BASE_FEE);

        let high_at_insert_low_at_block = TxBuilder::aa(Address::random())
            .nonce_key(U256::from(1))
            .max_priority_fee(10_000_000_000)
            .max_fee(u128::from(block_base_fee) + 1)
            .build();
        let high_at_insert_low_at_block_hash = *high_at_insert_low_at_block.hash();

        let low_at_insert_high_at_block = TxBuilder::aa(Address::random())
            .nonce_key(U256::from(2))
            .max_priority_fee(5_000_000_000)
            .max_fee(u128::from(block_base_fee) + 5_000_000_000)
            .build();
        let low_at_insert_high_at_block_hash = *low_at_insert_high_at_block.hash();

        for tx in [high_at_insert_low_at_block, low_at_insert_high_at_block] {
            pool.add_transaction(
                Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            )
            .unwrap();
        }

        (
            pool,
            low_at_insert_high_at_block_hash,
            high_at_insert_low_at_block_hash,
        )
    }

    #[test]
    fn test_best_transactions_with_base_fee_reprioritizes_regular_transactions() {
        let block_base_fee = TEMPO_T1_BASE_FEE + 10_000_000_000;
        let (pool, expected_first, expected_second) = priority_flip_pool(block_base_fee);

        let hashes = pool
            .best_transactions_with_base_fee(block_base_fee)
            .map(|tx| *tx.hash())
            .collect::<Vec<_>>();

        assert_eq!(hashes, vec![expected_first, expected_second]);
    }

    #[test]
    fn test_discard_reprices_eviction_priorities() {
        let block_base_fee = TEMPO_T1_BASE_FEE + 10_000_000_000;
        let (mut pool, expected_kept, expected_evicted) = priority_flip_pool_with_config(
            block_base_fee,
            AA2dPoolConfig {
                pending_limit: SubPoolLimit {
                    max_txs: 2,
                    max_size: usize::MAX,
                },
                queued_limit: SubPoolLimit {
                    max_txs: 10,
                    max_size: usize::MAX,
                },
                ..Default::default()
            },
        );
        pool.set_base_fee(block_base_fee);
        let trigger = TxBuilder::aa(Address::random())
            .nonce_key(U256::from(3))
            .max_priority_fee(10_000_000_000)
            .max_fee(u128::from(block_base_fee) + 10_000_000_000)
            .build();

        let result = pool
            .add_transaction(
                Arc::new(wrap_valid_tx(trigger, TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            )
            .unwrap();

        let AddedTransaction::Pending(pending) = result else {
            panic!("expected pending transaction")
        };
        assert_eq!(pending.discarded[0].hash(), &expected_evicted);
        assert!(!pool.contains(&expected_evicted));
        assert!(pool.contains(&expected_kept));
    }

    #[test]
    fn test_best_transactions_with_base_fee_filters_underpriced_regular_sequence() {
        let mut pool = AA2dPool::default();
        let block_base_fee = TEMPO_T1_BASE_FEE + 10_000_000_000;
        let sequence_sender = Address::random();

        let underpriced_parent = TxBuilder::aa(sequence_sender)
            .nonce_key(U256::from(1))
            .max_fee(u128::from(block_base_fee - 1))
            .max_priority_fee(1_000_000_000)
            .build();
        let valid_child = TxBuilder::aa(sequence_sender)
            .nonce_key(U256::from(1))
            .nonce(1)
            .max_fee(u128::from(block_base_fee) + 10_000_000_000)
            .max_priority_fee(10_000_000_000)
            .build();
        let valid_independent = TxBuilder::aa(Address::random())
            .nonce_key(U256::from(2))
            .max_fee(u128::from(block_base_fee) + 1_000_000_000)
            .max_priority_fee(1_000_000_000)
            .build();
        let valid_independent_hash = *valid_independent.hash();

        for tx in [underpriced_parent, valid_child, valid_independent] {
            pool.add_transaction(
                Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            )
            .unwrap();
        }

        let hashes = pool
            .best_transactions_with_base_fee(block_base_fee)
            .map(|tx| *tx.hash())
            .collect::<Vec<_>>();

        assert_eq!(hashes, vec![valid_independent_hash]);
    }

    #[test]
    fn test_best_transactions_with_base_fee_filters_underpriced_expiring_nonce() {
        let mut pool = AA2dPool::default();
        let block_base_fee = TEMPO_T1_BASE_FEE + 10_000_000_000;

        let underpriced = TxBuilder::aa(Address::random())
            .nonce_key(U256::MAX)
            .max_fee(u128::from(block_base_fee - 1))
            .max_priority_fee(1_000_000_000)
            .build();
        let valid = TxBuilder::aa(Address::random())
            .nonce_key(U256::MAX)
            .max_fee(u128::from(block_base_fee) + 1_000_000_000)
            .max_priority_fee(1_000_000_000)
            .build();
        let valid_hash = *valid.hash();

        for tx in [underpriced, valid] {
            pool.add_transaction(
                Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            )
            .unwrap();
        }

        let hashes = pool
            .best_transactions_with_base_fee(block_base_fee)
            .map(|tx| *tx.hash())
            .collect::<Vec<_>>();

        assert_eq!(hashes, vec![valid_hash]);
    }

    #[test]
    fn test_best_transactions_merges_regular_and_expiring_by_priority() {
        let mut pool = AA2dPool::default();
        let max_fee = 30_000_000_000u128;

        let regular_low = TxBuilder::aa(Address::random())
            .nonce_key(U256::from(1))
            .max_fee(max_fee)
            .max_priority_fee(1_000_000_000)
            .build();
        let regular_low_hash = *regular_low.hash();

        let expiring_high = TxBuilder::aa(Address::random())
            .nonce_key(U256::MAX)
            .max_fee(max_fee)
            .max_priority_fee(5_000_000_000)
            .build();
        let expiring_high_hash = *expiring_high.hash();

        let regular_mid = TxBuilder::aa(Address::random())
            .nonce_key(U256::from(2))
            .max_fee(max_fee)
            .max_priority_fee(3_000_000_000)
            .build();
        let regular_mid_hash = *regular_mid.hash();

        let expiring_low = TxBuilder::aa(Address::random())
            .nonce_key(U256::MAX)
            .max_fee(max_fee)
            .max_priority_fee(2_000_000_000)
            .build();
        let expiring_low_hash = *expiring_low.hash();

        for tx in [regular_low, expiring_high, regular_mid, expiring_low] {
            pool.add_transaction(
                Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            )
            .unwrap();
        }
        pool.assert_invariants();

        let hashes = pool
            .best_transactions()
            .map(|tx| *tx.hash())
            .collect::<Vec<_>>();

        assert_eq!(
            hashes,
            vec![
                expiring_high_hash,
                regular_mid_hash,
                expiring_low_hash,
                regular_low_hash,
            ]
        );
    }

    #[test]
    fn test_best_transactions_merges_regular_and_expiring_by_submission_id() {
        let max_fee = 30_000_000_000u128;

        let mut expiring_older_pool = AA2dPool::default();
        let expiring_older = TxBuilder::aa(Address::random())
            .nonce_key(U256::MAX)
            .max_fee(max_fee)
            .max_priority_fee(1_000_000_000)
            .build();
        let expiring_older_hash = *expiring_older.hash();
        let regular_newer = TxBuilder::aa(Address::random())
            .nonce_key(U256::from(1))
            .max_fee(max_fee)
            .max_priority_fee(1_000_000_000)
            .build();

        for tx in [expiring_older, regular_newer] {
            expiring_older_pool
                .add_transaction(
                    Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
                    0,
                    TempoHardfork::T1,
                )
                .unwrap();
        }

        assert_eq!(
            expiring_older_pool
                .best_transactions()
                .next()
                .unwrap()
                .hash(),
            &expiring_older_hash
        );

        let mut regular_older_pool = AA2dPool::default();
        let regular_older = TxBuilder::aa(Address::random())
            .nonce_key(U256::from(1))
            .max_fee(max_fee)
            .max_priority_fee(1_000_000_000)
            .build();
        let regular_older_hash = *regular_older.hash();
        let expiring_newer = TxBuilder::aa(Address::random())
            .nonce_key(U256::MAX)
            .max_fee(max_fee)
            .max_priority_fee(1_000_000_000)
            .build();

        for tx in [regular_older, expiring_newer] {
            regular_older_pool
                .add_transaction(
                    Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
                    0,
                    TempoHardfork::T1,
                )
                .unwrap();
        }

        assert_eq!(
            regular_older_pool
                .best_transactions()
                .next()
                .unwrap()
                .hash(),
            &regular_older_hash
        );
    }

    // ============================================
    // on_state_updates tests
    // ============================================

    #[test]
    fn on_state_updates_clears_scratch_buffers_without_nonce_state() {
        let mut pool = AA2dPool::default();
        pool.state_update_nonce_changes
            .insert(AASequenceId::new(Address::random(), U256::from(1)), 1);
        pool.state_update_included_expiring_nonce_hashes
            .push(B256::random());

        let state = AddressMap::default();
        let (promoted, mined) = pool.on_state_updates(&state);

        assert!(promoted.is_empty());
        assert!(mined.is_empty());
        assert!(pool.state_update_nonce_changes.is_empty());
        assert!(pool.state_update_included_expiring_nonce_hashes.is_empty());
    }

    #[test]
    fn test_on_state_updates_with_nonce_precompile_slot() {
        use revm::database::{AccountStatus, BundleAccount, states::StorageSlot};

        let mut pool = AA2dPool::default();
        let sender = Address::random();
        let nonce_key = U256::from(1);

        let tx0 = TxBuilder::aa(sender).nonce_key(nonce_key).build();
        let tx1 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(1).build();
        let tx2 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(2).build();
        let nonce_slot = tx0
            .nonce_key_slot()
            .expect("2D nonce tx should have nonce key slot");

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx2, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 3);
        assert_eq!(queued, 0);

        let mut storage = HashMap::default();
        storage.insert(
            nonce_slot,
            StorageSlot::new_changed(U256::ZERO, U256::from(2u64)),
        );
        let mut state = AddressMap::default();
        state.insert(
            NONCE_PRECOMPILE_ADDRESS,
            BundleAccount::new(None, None, storage, AccountStatus::Changed),
        );

        let (promoted, mined) = pool.on_state_updates(&state);

        assert!(promoted.is_empty(), "tx2 was already pending");
        assert_eq!(mined.len(), 2, "tx0 and tx1 should be mined");

        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 1, "Only tx2 should remain pending");
        assert_eq!(queued, 0);

        pool.assert_invariants();
        assert!(pool.state_update_nonce_changes.is_empty());
        assert!(pool.state_update_included_expiring_nonce_hashes.is_empty());
    }

    #[test]
    fn test_on_state_updates_creates_gap_demotion() {
        use revm::database::{AccountStatus, BundleAccount, states::StorageSlot};

        let mut pool = AA2dPool::default();
        let sender = Address::random();
        let nonce_key = U256::from(1);

        let tx0 = TxBuilder::aa(sender).nonce_key(nonce_key).build();
        let tx1 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(1).build();
        let tx3 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(3).build();
        let nonce_slot = tx0
            .nonce_key_slot()
            .expect("2D nonce tx should have nonce key slot");

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx3, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 2);
        assert_eq!(queued, 1);

        let mut storage = HashMap::default();
        storage.insert(
            nonce_slot,
            StorageSlot::new_changed(U256::ZERO, U256::from(2u64)),
        );
        let mut state = AddressMap::default();
        state.insert(
            NONCE_PRECOMPILE_ADDRESS,
            BundleAccount::new(None, None, storage, AccountStatus::Changed),
        );

        let (promoted, mined) = pool.on_state_updates(&state);

        assert_eq!(mined.len(), 2, "tx0 and tx1 should be mined");
        assert!(promoted.is_empty());

        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 0, "tx3 should still be queued (gap at nonce 2)");
        assert_eq!(queued, 1);

        pool.assert_invariants();
    }

    #[test]
    fn test_on_nonce_changes_promotes_queued_transactions() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();
        let nonce_key = U256::ZERO;
        let seq_id = AASequenceId::new(sender, nonce_key);

        let tx2 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(2).build();
        let tx3 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(3).build();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx2.clone(), TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx3, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 0);
        assert_eq!(queued, 2);

        let mut changes = HashMap::default();
        changes.insert(seq_id, 2u64);

        let (promoted, mined) = pool.on_nonce_changes(changes);

        assert!(
            mined.is_empty(),
            "No transactions to mine (on-chain nonce jumped)"
        );
        assert_eq!(promoted.len(), 2, "tx2 and tx3 should be promoted");
        assert!(promoted.iter().any(|t| t.hash() == tx2.hash()));

        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 2);
        assert_eq!(queued, 0);

        pool.assert_invariants();
    }

    // ============================================
    // Interleaved inserts across sequence IDs
    // ============================================

    #[test]
    fn test_interleaved_inserts_multiple_nonce_keys() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();

        let key_a = U256::ZERO;
        let key_b = U256::from(1);

        let tx_a0 = TxBuilder::aa(sender).nonce_key(key_a).build();
        let tx_b0 = TxBuilder::aa(sender).nonce_key(key_b).build();
        let tx_a1 = TxBuilder::aa(sender).nonce_key(key_a).nonce(1).build();
        let tx_b2 = TxBuilder::aa(sender).nonce_key(key_b).nonce(2).build();
        let tx_b1 = TxBuilder::aa(sender).nonce_key(key_b).nonce(1).build();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx_a0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx_b0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx_a1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx_b2, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx_b1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 5, "All transactions should be pending");
        assert_eq!(queued, 0);

        assert_eq!(
            pool.independent_transactions.len(),
            2,
            "Two nonce keys = two independent txs"
        );

        pool.assert_invariants();
    }

    #[test]
    fn test_same_sender_different_nonce_keys_independent() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();

        let key_a = U256::from(100);
        let key_b = U256::from(200);

        let tx_a5 = TxBuilder::aa(sender).nonce_key(key_a).nonce(5).build();
        let tx_b0 = TxBuilder::aa(sender).nonce_key(key_b).build();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx_a5, TransactionOrigin::Local)),
            5,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx_b0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 2);
        assert_eq!(queued, 0);

        assert_eq!(pool.independent_transactions.len(), 2);

        pool.assert_invariants();
    }

    /// Test reorg handling when on-chain nonce decreases.
    ///
    /// When a reorg occurs, the canonical nonce can decrease. If no transaction
    /// exists at the new on-chain nonce, `independent_transactions` must be cleared.
    #[test_case::test_case(U256::ZERO)]
    #[test_case::test_case(U256::random())]
    fn reorg_nonce_decrease_clears_stale_independent_transaction(nonce_key: U256) {
        let mut pool = AA2dPool::default();
        let sender = Address::random();
        let seq_id = AASequenceId::new(sender, nonce_key);

        // Step 1: Add txs with nonces [3, 4, 5], starting with on_chain_nonce=3
        let tx3 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(3).build();
        let tx4 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(4).build();
        let tx5 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(5).build();
        let tx5_hash = *tx5.hash();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx3, TransactionOrigin::Local)),
            3,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx4, TransactionOrigin::Local)),
            3,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx5, TransactionOrigin::Local)),
            3,
            TempoHardfork::T1,
        )
        .unwrap();

        // Verify initial state: all 3 txs pending, tx3 is independent
        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 3, "All transactions should be pending");
        assert_eq!(queued, 0);
        assert_eq!(pool.independent_transactions.len(), 1);
        assert_eq!(
            pool.independent_transactions
                .get(&seq_id)
                .unwrap()
                .transaction
                .nonce(),
            3,
            "tx3 should be independent initially"
        );
        pool.assert_invariants();

        // Step 2: Simulate mining of tx3 and tx4, on_chain_nonce becomes 5
        let mut on_chain_ids = HashMap::default();
        on_chain_ids.insert(seq_id, 5u64);
        let (promoted, mined) = pool.on_nonce_changes(on_chain_ids);

        assert_eq!(mined.len(), 2, "tx3 and tx4 should be mined");
        assert!(promoted.is_empty(), "No promotions expected");

        // Now tx5 should be the only tx in pool and be independent
        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 1, "Only tx5 should remain pending");
        assert_eq!(queued, 0);
        assert_eq!(pool.independent_transactions.len(), 1);
        assert_eq!(
            pool.independent_transactions
                .get(&seq_id)
                .unwrap()
                .transaction
                .hash(),
            &tx5_hash,
            "tx5 should be independent after mining"
        );
        pool.assert_invariants();

        // Step 3: Simulate reorg - nonce decreases back to 3
        let mut on_chain_ids = HashMap::default();
        on_chain_ids.insert(seq_id, 3u64);
        let (promoted, mined) = pool.on_nonce_changes(on_chain_ids);

        // No transactions should be mined (tx5.nonce=5 >= on_chain_nonce=3)
        assert!(mined.is_empty(), "No transactions should be mined");
        // No promotions expected
        assert!(promoted.is_empty(), "No promotions expected");

        // tx5 should still be in the pool but is now QUEUED (gap at nonces 3, 4)
        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 0, "tx5 should not be pending (nonce gap)");
        assert_eq!(queued, 1, "tx5 should be queued");

        // No tx at on_chain_nonce=3, so independent_transactions should be cleared
        assert!(
            !pool.independent_transactions.contains_key(&seq_id),
            "independent_transactions should not contain stale entry after reorg"
        );

        pool.assert_invariants();
    }

    /// Simulates the full reorg flow as handled by reth's maintain_transaction_pool:
    ///
    /// 1. Add txs [3, 4, 5] → all pending
    /// 2. Mine tx3 and tx4 via on_nonce_changes(nonce=5) → tx5 remains pending
    /// 3. Reorg reverts the block: reth re-injects orphaned tx3 and tx4 via add_transaction
    ///    with the correct on_chain_nonce=3 (read from the new tip's state).
    ///
    /// This verifies that add_transaction's rescan from on_chain_nonce correctly
    /// reclassifies all transactions as pending without needing an explicit nonce reset.
    #[test_case::test_case(U256::ZERO)]
    #[test_case::test_case(U256::random())]
    fn reorg_reinjection_via_add_transaction_restores_pending_state(nonce_key: U256) {
        let mut pool = AA2dPool::default();
        let sender = Address::random();
        let seq_id = AASequenceId::new(sender, nonce_key);

        // Step 1: Add txs with nonces [3, 4, 5], on_chain_nonce=3
        let tx3 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(3).build();
        let tx4 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(4).build();
        let tx5 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(5).build();
        let tx3_hash = *tx3.hash();
        let tx4_hash = *tx4.hash();
        let tx5_hash = *tx5.hash();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx3.clone(), TransactionOrigin::Local)),
            3,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx4.clone(), TransactionOrigin::Local)),
            3,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx5, TransactionOrigin::Local)),
            3,
            TempoHardfork::T1,
        )
        .unwrap();

        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 3);
        assert_eq!(queued, 0);
        pool.assert_invariants();

        // Step 2: Mine tx3 and tx4 (on_chain_nonce becomes 5)
        let mut nonce_changes = HashMap::default();
        nonce_changes.insert(seq_id, 5u64);
        let (_promoted, mined) = pool.on_nonce_changes(nonce_changes);
        assert_eq!(mined.len(), 2);

        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 1, "only tx5 should remain pending");
        assert_eq!(queued, 0);
        pool.assert_invariants();

        // Step 3: Simulate reorg — reth re-injects orphaned tx3 and tx4 via add_transaction
        // with the correct on_chain_nonce=3 (reverted state).
        // This is exactly what reth's maintain_transaction_pool does after a reorg.
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx3, TransactionOrigin::External)),
            3,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx4, TransactionOrigin::External)),
            3,
            TempoHardfork::T1,
        )
        .unwrap();

        // All 3 txs should be pending again — add_transaction rescans from on_chain_nonce
        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 3, "all txs should be pending after re-injection");
        assert_eq!(queued, 0);

        // tx3 should be independent (at on_chain_nonce)
        assert_eq!(
            pool.independent_transactions
                .get(&seq_id)
                .unwrap()
                .transaction
                .nonce(),
            3,
        );

        // All txs should be in the pool
        assert!(pool.contains(&tx3_hash));
        assert!(pool.contains(&tx4_hash));
        assert!(pool.contains(&tx5_hash));

        pool.assert_invariants();
    }

    /// Test that gap demotion marks ALL subsequent transactions as non-pending.
    ///
    /// When a transaction is removed creating a gap, all transactions after the gap
    /// should be marked as queued (is_pending=false), not just the first one.
    #[test_case::test_case(U256::ZERO)]
    #[test_case::test_case(U256::random())]
    fn gap_demotion_marks_all_subsequent_transactions_as_queued(nonce_key: U256) {
        let mut pool = AA2dPool::default();
        let sender = Address::random();
        let seq_id = AASequenceId::new(sender, nonce_key);

        // Step 1: Add txs with nonces [5, 6, 7, 8], on_chain_nonce=5
        let tx5 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(5).build();
        let tx6 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(6).build();
        let tx7 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(7).build();
        let tx8 = TxBuilder::aa(sender).nonce_key(nonce_key).nonce(8).build();
        let tx6_hash = *tx6.hash();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx5, TransactionOrigin::Local)),
            5,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx6, TransactionOrigin::Local)),
            5,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx7, TransactionOrigin::Local)),
            5,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx8, TransactionOrigin::Local)),
            5,
            TempoHardfork::T1,
        )
        .unwrap();

        // Verify initial state: all 4 txs pending
        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 4, "All transactions should be pending initially");
        assert_eq!(queued, 0);
        assert_eq!(pool.independent_transactions.len(), 1);
        pool.assert_invariants();

        // Step 2: Remove tx6 to create a gap at nonce 6
        // Pool now has: [5, _, 7, 8] where _ is the gap
        let removed = pool.remove_transactions(std::iter::once(&tx6_hash));
        assert_eq!(removed.len(), 1, "Should remove exactly tx6");

        // Step 3: Trigger nonce change processing to re-evaluate pending status
        // The on-chain nonce is still 5
        let mut on_chain_ids = HashMap::default();
        on_chain_ids.insert(seq_id, 5u64);
        let (promoted, mined) = pool.on_nonce_changes(on_chain_ids);

        assert!(mined.is_empty(), "No transactions should be mined");
        assert!(promoted.is_empty(), "No promotions expected");

        // Step 4: Verify that tx7 AND tx8 are both queued (not pending)
        // BUG: Current code only marks tx7 as non-pending, tx8 incorrectly stays pending
        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(
            pending, 1,
            "Only tx5 should be pending (tx7 and tx8 are after the gap)"
        );
        assert_eq!(
            queued, 2,
            "tx7 and tx8 should both be queued due to gap at nonce 6"
        );

        pool.assert_invariants();
    }

    #[test]
    fn expiring_nonce_tx_increments_pending_count() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();

        // Create an expiring nonce transaction (nonce_key = U256::MAX)
        let tx = TxBuilder::aa(sender).nonce_key(U256::MAX).build();
        let valid_tx = wrap_valid_tx(tx, TransactionOrigin::Local);

        // Add the expiring nonce transaction
        let result = pool.add_transaction(Arc::new(valid_tx), 0, TempoHardfork::T1);
        assert!(result.is_ok(), "Transaction should be added successfully");
        assert!(
            matches!(result.unwrap(), AddedTransaction::Pending(_)),
            "Expiring nonce transaction should be pending"
        );

        // Verify counts - expiring nonce txs should increment pending_count
        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 1, "Should have 1 pending transaction");
        assert_eq!(queued, 0, "Should have 0 queued transactions");

        // This will fail if pending_count wasn't incremented
        pool.assert_invariants();
    }

    #[test]
    fn expiring_nonce_tx_dedup_uses_expiring_nonce_hash() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();
        let call_to = Address::random();
        let fee_token = Address::random();
        let calls = vec![Call {
            to: TxKind::Call(call_to),
            value: U256::ZERO,
            input: Bytes::new(),
        }];

        let build_tx = |fee_payer_signature: Signature| {
            let tx = TempoTransaction {
                chain_id: 1,
                max_priority_fee_per_gas: 1_000_000_000,
                max_fee_per_gas: 2_000_000_000,
                gas_limit: 1_000_000,
                calls: calls.clone(),
                nonce_key: U256::MAX,
                nonce: 0,
                fee_token: Some(fee_token),
                fee_payer_signature: Some(fee_payer_signature),
                valid_after: None,
                valid_before: Some(core::num::NonZeroU64::new(123).unwrap()),
                access_list: AccessList::default(),
                tempo_authorization_list: Vec::new(),
                key_authorization: None,
            };

            let signature = TempoSignature::Primitive(PrimitiveSignature::Secp256k1(
                Signature::test_signature(),
            ));
            let aa_signed = AASigned::new_unhashed(tx, signature);
            let envelope: TempoTxEnvelope = aa_signed.into();
            let recovered = Recovered::new_unchecked(envelope, sender);
            TempoPooledTransaction::new(recovered)
        };

        let tx1 = build_tx(Signature::new(U256::from(1), U256::from(2), false));
        let tx2 = build_tx(Signature::new(U256::from(3), U256::from(4), false));

        assert_ne!(tx1.hash(), tx2.hash(), "tx hashes must differ");
        let expiring_hash_1 = tx1
            .expiring_nonce_hash()
            .expect("expiring nonce tx must be AA");
        let expiring_hash_2 = tx2
            .expiring_nonce_hash()
            .expect("expiring nonce tx must be AA");
        assert_eq!(
            expiring_hash_1, expiring_hash_2,
            "expiring nonce hashes must match"
        );

        let tx1_hash = *tx1.hash();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let tx2_hash = *tx2.hash();
        let result = pool.add_transaction(
            Arc::new(wrap_valid_tx(tx2, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        );
        assert!(result.is_err(), "Expected AlreadyImported error");
        let err = result.unwrap_err();
        assert_eq!(err.hash, tx2_hash);
        assert!(
            matches!(err.kind, PoolErrorKind::AlreadyImported),
            "Expected AlreadyImported, got {:?}",
            err.kind
        );

        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 1, "Expected 1 pending transaction");
        assert_eq!(queued, 0, "Expected 0 queued transactions");
        assert!(pool.by_hash.contains_key(&tx1_hash));
        assert_eq!(pool.expiring_nonce_txs.len(), 1);
        assert_expiring_eviction_index_len(&pool, 1);
        pool.assert_invariants();
    }

    /// Verifies that removing an expiring nonce tx by hash correctly cleans up
    /// both `expiring_nonce_txs` and `by_hash`.
    #[test]
    fn remove_included_expiring_nonce_tx_uses_correct_key() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();
        let fee_token = Address::random();
        let calls = vec![Call {
            to: TxKind::Call(Address::random()),
            value: U256::ZERO,
            input: Bytes::new(),
        }];

        let tx = TempoTransaction {
            chain_id: 1,
            max_priority_fee_per_gas: 1_000_000_000,
            max_fee_per_gas: 2_000_000_000,
            gas_limit: 1_000_000,
            calls,
            nonce_key: U256::MAX,
            nonce: 0,
            fee_token: Some(fee_token),
            fee_payer_signature: Some(Signature::new(U256::from(1), U256::from(2), false)),
            valid_before: Some(core::num::NonZeroU64::new(123).unwrap()),
            access_list: AccessList::default(),
            tempo_authorization_list: Vec::new(),
            key_authorization: None,
            valid_after: None,
        };

        let signature =
            TempoSignature::Primitive(PrimitiveSignature::Secp256k1(Signature::test_signature()));
        let aa_signed = AASigned::new_unhashed(tx, signature);
        let envelope: TempoTxEnvelope = aa_signed.into();
        let recovered = Recovered::new_unchecked(envelope, sender);
        let pooled = TempoPooledTransaction::new(recovered);

        let tx_hash = *pooled.hash();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(pooled, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        assert_eq!(pool.expiring_nonce_txs.len(), 1);
        assert_expiring_eviction_index_len(&pool, 1);
        assert!(pool.by_hash.contains_key(&tx_hash));
        pool.assert_invariants();

        // Simulate block mining: remove by tx_hash
        let removed = pool.remove_transactions(std::iter::once(&tx_hash));
        assert_eq!(removed.len(), 1, "should remove the tx by its tx_hash");
        assert_eq!(*removed[0].hash(), tx_hash);

        // Both maps must be empty
        assert!(
            pool.expiring_nonce_txs.is_empty(),
            "expiring_nonce_txs not cleaned up"
        );
        assert!(
            pool.expiring_nonce_eviction_order.is_empty(),
            "expiring_nonce_eviction_order not cleaned up"
        );
        assert!(
            !pool.by_hash.contains_key(&tx_hash),
            "by_hash not cleaned up"
        );

        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 0);
        assert_eq!(queued, 0);
        pool.assert_invariants();
    }

    #[test]
    fn on_state_updates_removes_included_expiring_nonce_from_eviction_index() {
        use revm::database::{AccountStatus, BundleAccount, states::StorageSlot};

        let mut pool = AA2dPool::default();
        let sender = Address::random();

        let tx = TxBuilder::aa(sender)
            .nonce_key(U256::MAX)
            .valid_before(123)
            .max_fee(30_000_000_000)
            .build();
        let tx_hash = *tx.hash();
        let expiring_hash = tx
            .expiring_nonce_hash()
            .expect("expiring nonce tx must have expiring hash");
        let slot = tx
            .expiring_nonce_slot()
            .expect("expiring nonce tx must have storage slot");

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        assert_expiring_eviction_index_len(&pool, 1);
        assert_expiring_eviction_index_contains(&pool, expiring_hash);

        let mut storage = HashMap::default();
        storage.insert(
            slot,
            StorageSlot::new_changed(U256::ZERO, U256::from(123u64)),
        );
        let mut state = AddressMap::default();
        state.insert(
            NONCE_PRECOMPILE_ADDRESS,
            BundleAccount::new(None, None, storage, AccountStatus::Changed),
        );

        let (promoted, mined) = pool.on_state_updates(&state);

        assert!(promoted.is_empty());
        assert_eq!(mined.len(), 1);
        assert_eq!(mined[0].hash(), &tx_hash);
        assert!(!pool.contains(&tx_hash));
        assert!(pool.expiring_nonce_txs.is_empty());
        assert!(pool.slot_to_expiring_nonce_hash.is_empty());
        assert_expiring_eviction_index_len(&pool, 0);
        pool.assert_invariants();
        assert!(pool.state_update_nonce_changes.is_empty());
        assert!(pool.state_update_included_expiring_nonce_hashes.is_empty());
    }

    /// Pool with pending limit of 2 for eviction tests.
    fn eviction_test_pool() -> AA2dPool {
        AA2dPool::new(AA2dPoolConfig {
            pending_limit: SubPoolLimit {
                max_txs: 2,
                max_size: usize::MAX,
            },
            queued_limit: SubPoolLimit {
                max_txs: 10,
                max_size: usize::MAX,
            },
            ..Default::default()
        })
    }

    fn assert_expiring_eviction_index_len(pool: &AA2dPool, len: usize) {
        assert_eq!(pool.expiring_nonce_txs.len(), len);
        assert_eq!(pool.expiring_nonce_eviction_order.len(), len);
        pool.assert_invariants();
    }

    fn assert_expiring_eviction_index_contains(pool: &AA2dPool, expiring_hash: B256) {
        assert!(
            pool.expiring_nonce_eviction_order
                .iter()
                .any(|key| key.expiring_hash() == expiring_hash),
            "expiring_nonce_eviction_order should contain {expiring_hash:?}"
        );
    }

    fn assert_expiring_eviction_index_missing(pool: &AA2dPool, expiring_hash: B256) {
        assert!(
            pool.expiring_nonce_eviction_order
                .iter()
                .all(|key| key.expiring_hash() != expiring_hash),
            "expiring_nonce_eviction_order should not contain {expiring_hash:?}"
        );
    }

    #[test]
    fn eviction_same_priority_evicts_newer() {
        // Direction 1: newer expiring tx evicted over older 2D txs
        let mut pool = eviction_test_pool();
        let sender = Address::random();

        let tx1 = TxBuilder::aa(sender)
            .nonce_key(U256::from(1))
            .nonce(0)
            .build();
        let tx2 = TxBuilder::aa(sender)
            .nonce_key(U256::from(2))
            .nonce(0)
            .build();
        let tx_exp = TxBuilder::aa(sender).nonce_key(U256::MAX).build();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1.clone(), TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx2.clone(), TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        let result = pool
            .add_transaction(
                Arc::new(wrap_valid_tx(tx_exp.clone(), TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            )
            .unwrap();

        let AddedTransaction::Pending(pending) = result else {
            panic!("expected pending")
        };
        assert_eq!(pending.discarded[0].hash(), tx_exp.hash());
        assert!(pool.contains(tx1.hash()));
        assert!(pool.contains(tx2.hash()));
        assert!(!pool.contains(tx_exp.hash()));
        pool.assert_invariants();

        // Test opposite direction where newer 2D tx evicted over older expiring tx
        let mut pool = eviction_test_pool();
        let sender = Address::random();

        let tx_exp = TxBuilder::aa(sender).nonce_key(U256::MAX).build();
        let tx2 = TxBuilder::aa(sender)
            .nonce_key(U256::from(1))
            .nonce(0)
            .build();
        let tx3 = TxBuilder::aa(sender)
            .nonce_key(U256::from(2))
            .nonce(0)
            .build();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx_exp.clone(), TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx2.clone(), TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        let result = pool
            .add_transaction(
                Arc::new(wrap_valid_tx(tx3.clone(), TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            )
            .unwrap();

        let AddedTransaction::Pending(pending) = result else {
            panic!("expected pending")
        };
        assert_eq!(pending.discarded[0].hash(), tx3.hash());
        assert!(pool.contains(tx_exp.hash()));
        assert!(pool.contains(tx2.hash()));
        assert!(!pool.contains(tx3.hash()));
        pool.assert_invariants();
    }

    #[test]
    fn eviction_lower_priority_expiring_evicted() {
        let mut pool = eviction_test_pool();
        let sender = Address::random();

        // Expiring nonce tx added first but with lower priority
        let tx_exp = TxBuilder::aa(sender)
            .nonce_key(U256::MAX)
            .max_priority_fee(100)
            .max_fee(200)
            .build();
        let tx2 = TxBuilder::aa(sender)
            .nonce_key(U256::from(1))
            .nonce(0)
            .build();
        let tx3 = TxBuilder::aa(sender)
            .nonce_key(U256::from(2))
            .nonce(0)
            .build();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx_exp.clone(), TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx2, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        let result = pool
            .add_transaction(
                Arc::new(wrap_valid_tx(tx3.clone(), TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            )
            .unwrap();

        // Lower-priority expiring tx evicted even though it was added first
        let AddedTransaction::Pending(pending) = result else {
            panic!("expected pending")
        };
        assert_eq!(pending.discarded[0].hash(), tx_exp.hash());
        assert!(!pool.contains(tx_exp.hash()));
        assert!(pool.contains(tx3.hash()));
        pool.assert_invariants();
    }

    #[test]
    fn eviction_lower_priority_2d_evicted() {
        let mut pool = eviction_test_pool();
        let sender = Address::random();

        // 2D tx with low priority added first
        let tx_low = TxBuilder::aa(sender)
            .nonce_key(U256::from(1))
            .nonce(0)
            .max_priority_fee(100)
            .max_fee(200)
            .build();
        let tx_exp = TxBuilder::aa(sender).nonce_key(U256::MAX).build();
        let tx3 = TxBuilder::aa(sender)
            .nonce_key(U256::from(2))
            .nonce(0)
            .build();

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx_low.clone(), TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx_exp.clone(), TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        let result = pool
            .add_transaction(
                Arc::new(wrap_valid_tx(tx3, TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            )
            .unwrap();

        // Lower-priority 2D tx evicted even though expiring nonce tx is newer
        let AddedTransaction::Pending(pending) = result else {
            panic!("expected pending")
        };
        assert_eq!(pending.discarded[0].hash(), tx_low.hash());
        assert!(!pool.contains(tx_low.hash()));
        assert!(pool.contains(tx_exp.hash()));
        pool.assert_invariants();
    }

    #[test]
    fn expiring_nonce_eviction_order_evicts_lowest_priority() {
        let mut pool = eviction_test_pool();

        let tx_low = TxBuilder::aa(Address::random())
            .nonce_key(U256::MAX)
            .max_priority_fee(1_000_000_000)
            .max_fee(30_000_000_000)
            .build();
        let tx_high = TxBuilder::aa(Address::random())
            .nonce_key(U256::MAX)
            .max_priority_fee(3_000_000_000)
            .max_fee(30_000_000_000)
            .build();
        let tx_mid = TxBuilder::aa(Address::random())
            .nonce_key(U256::MAX)
            .max_priority_fee(2_000_000_000)
            .max_fee(30_000_000_000)
            .build();

        let low_expiring_hash = tx_low
            .expiring_nonce_hash()
            .expect("expiring nonce tx must have expiring hash");
        let mid_expiring_hash = tx_mid
            .expiring_nonce_hash()
            .expect("expiring nonce tx must have expiring hash");
        let high_expiring_hash = tx_high
            .expiring_nonce_hash()
            .expect("expiring nonce tx must have expiring hash");

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx_low.clone(), TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx_high.clone(), TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        let result = pool
            .add_transaction(
                Arc::new(wrap_valid_tx(tx_mid.clone(), TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            )
            .unwrap();

        let AddedTransaction::Pending(pending) = result else {
            panic!("expected pending")
        };
        assert_eq!(pending.discarded.len(), 1);
        assert_eq!(pending.discarded[0].hash(), tx_low.hash());
        assert!(!pool.contains(tx_low.hash()));
        assert!(pool.contains(tx_mid.hash()));
        assert!(pool.contains(tx_high.hash()));
        assert_expiring_eviction_index_len(&pool, 2);
        assert_expiring_eviction_index_missing(&pool, low_expiring_hash);
        assert_expiring_eviction_index_contains(&pool, mid_expiring_hash);
        assert_expiring_eviction_index_contains(&pool, high_expiring_hash);
    }

    #[test]
    fn expiring_nonce_eviction_order_evicts_newer_same_priority() {
        let mut pool = eviction_test_pool();

        let tx_old_1 = TxBuilder::aa(Address::random())
            .nonce_key(U256::MAX)
            .max_fee(30_000_000_000)
            .build();
        let tx_old_2 = TxBuilder::aa(Address::random())
            .nonce_key(U256::MAX)
            .max_fee(30_000_000_000)
            .build();
        let tx_new = TxBuilder::aa(Address::random())
            .nonce_key(U256::MAX)
            .max_fee(30_000_000_000)
            .build();

        let new_expiring_hash = tx_new
            .expiring_nonce_hash()
            .expect("expiring nonce tx must have expiring hash");

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx_old_1.clone(), TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx_old_2.clone(), TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();
        let result = pool
            .add_transaction(
                Arc::new(wrap_valid_tx(tx_new.clone(), TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            )
            .unwrap();

        let AddedTransaction::Pending(pending) = result else {
            panic!("expected pending")
        };
        assert_eq!(pending.discarded.len(), 1);
        assert_eq!(pending.discarded[0].hash(), tx_new.hash());
        assert!(pool.contains(tx_old_1.hash()));
        assert!(pool.contains(tx_old_2.hash()));
        assert!(!pool.contains(tx_new.hash()));
        assert_expiring_eviction_index_len(&pool, 2);
        assert_expiring_eviction_index_missing(&pool, new_expiring_hash);
    }

    #[test]
    fn expiring_nonce_tx_uses_separate_eviction_index() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();

        let tx = TxBuilder::aa(sender).nonce_key(U256::MAX).build();
        let expiring_hash = tx
            .expiring_nonce_hash()
            .expect("expiring nonce tx must have expiring hash");

        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        assert!(
            pool.by_eviction_order.is_empty(),
            "expiring nonce txs should not be inserted into by_eviction_order"
        );
        assert_expiring_eviction_index_len(&pool, 1);
        assert_expiring_eviction_index_contains(&pool, expiring_hash);
    }

    #[test]
    fn expiring_nonce_tx_subject_to_eviction() {
        // Create pool with very small pending limit
        let config = AA2dPoolConfig {
            pending_limit: SubPoolLimit {
                max_txs: 2,
                max_size: usize::MAX,
            },
            queued_limit: SubPoolLimit {
                max_txs: 10,
                max_size: usize::MAX,
            },
            ..Default::default()
        };
        let mut pool = AA2dPool::new(config);
        let sender = Address::random();

        // Add 3 expiring nonce transactions - should evict to maintain limit of 2
        for i in 0..3 {
            let tx = TxBuilder::aa(sender)
                .nonce_key(U256::MAX)
                .max_priority_fee(1_000_000_000 + i as u128 * 100_000_000)
                .max_fee(2_000_000_000 + i as u128 * 100_000_000)
                .build();
            let valid_tx = wrap_valid_tx(tx, TransactionOrigin::Local);
            let _ = pool.add_transaction(Arc::new(valid_tx), 0, TempoHardfork::T1);
        }

        // Should only have 2 transactions (evicted one to maintain limit)
        let (pending, queued) = pool.pending_and_queued_txn_count();
        assert!(
            pending <= 2,
            "Should have at most 2 pending transactions due to limit, got {pending}"
        );
        assert_eq!(queued, 0, "Should have 0 queued transactions");

        pool.assert_invariants();
    }

    #[test]
    fn remove_expiring_nonce_tx_decrements_pending_count() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();

        // Add two expiring nonce transactions
        let tx1 = TxBuilder::aa(sender)
            .nonce_key(U256::MAX)
            .max_priority_fee(1_000_000_000)
            .max_fee(2_000_000_000)
            .build();
        let valid_tx1 = wrap_valid_tx(tx1, TransactionOrigin::Local);
        let tx1_hash = *valid_tx1.hash();
        let tx1_expiring_hash = valid_tx1
            .transaction
            .expiring_nonce_hash()
            .expect("expiring nonce tx must have expiring hash");

        let tx2 = TxBuilder::aa(sender)
            .nonce_key(U256::MAX)
            .max_priority_fee(1_100_000_000)
            .max_fee(2_200_000_000)
            .build();
        let valid_tx2 = wrap_valid_tx(tx2, TransactionOrigin::Local);

        pool.add_transaction(Arc::new(valid_tx1), 0, TempoHardfork::T1)
            .unwrap();
        pool.add_transaction(Arc::new(valid_tx2), 0, TempoHardfork::T1)
            .unwrap();

        // Verify we have 2 pending
        let (pending, _) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 2, "Should have 2 pending transactions");
        assert_expiring_eviction_index_len(&pool, 2);

        // Remove one via hash
        let removed = pool.remove_transactions(std::iter::once(&tx1_hash));
        assert_eq!(removed.len(), 1, "Should remove exactly 1 transaction");

        // Verify pending count decremented
        let (pending, _) = pool.pending_and_queued_txn_count();
        assert_eq!(
            pending, 1,
            "Should have 1 pending transaction after removal"
        );
        assert_expiring_eviction_index_len(&pool, 1);
        assert_expiring_eviction_index_missing(&pool, tx1_expiring_hash);

        // This will fail if pending_count wasn't decremented
        pool.assert_invariants();
    }

    #[test]
    fn remove_expiring_nonce_tx_by_hash_updates_pending_count() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();

        let tx = TxBuilder::aa(sender)
            .nonce_key(U256::MAX)
            .max_priority_fee(1_000_000_000)
            .max_fee(2_000_000_000)
            .build();
        let valid_tx = wrap_valid_tx(tx, TransactionOrigin::Local);
        let tx_hash = *valid_tx.hash();
        let expiring_hash = valid_tx
            .transaction
            .expiring_nonce_hash()
            .expect("expiring nonce tx must have expiring hash");

        pool.add_transaction(Arc::new(valid_tx), 0, TempoHardfork::T1)
            .unwrap();

        let (pending, _) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 1);
        assert_expiring_eviction_index_len(&pool, 1);
        assert_expiring_eviction_index_contains(&pool, expiring_hash);

        // Remove via remove_transactions (uses remove_transaction_by_hash_no_demote)
        let removed = pool.remove_transactions(std::iter::once(&tx_hash));
        assert_eq!(removed.len(), 1);

        let (pending, _) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 0);
        assert_expiring_eviction_index_len(&pool, 0);
        assert_expiring_eviction_index_missing(&pool, expiring_hash);
    }

    #[test]
    fn remove_expiring_nonce_tx_by_sender_updates_pending_count() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();

        let tx1 = TxBuilder::aa(sender)
            .nonce_key(U256::MAX)
            .max_priority_fee(1_000_000_000)
            .max_fee(2_000_000_000)
            .build();
        let valid_tx1 = wrap_valid_tx(tx1, TransactionOrigin::Local);

        let tx2 = TxBuilder::aa(sender)
            .nonce_key(U256::MAX)
            .max_priority_fee(1_100_000_000)
            .max_fee(2_200_000_000)
            .build();
        let valid_tx2 = wrap_valid_tx(tx2, TransactionOrigin::Local);

        pool.add_transaction(Arc::new(valid_tx1), 0, TempoHardfork::T1)
            .unwrap();
        pool.add_transaction(Arc::new(valid_tx2), 0, TempoHardfork::T1)
            .unwrap();

        let (pending, _) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 2);
        assert_expiring_eviction_index_len(&pool, 2);

        // Remove via remove_transactions_by_sender
        let removed = pool.remove_transactions_by_sender(sender);
        assert_eq!(removed.len(), 2);

        let (pending, _) = pool.pending_and_queued_txn_count();
        assert_eq!(pending, 0);
        assert_expiring_eviction_index_len(&pool, 0);
    }

    #[test]
    fn test_rejected_2d_tx_does_not_leak_slot_entries() {
        let config = AA2dPoolConfig {
            price_bump_config: PriceBumpConfig::default(),
            pending_limit: SubPoolLimit {
                max_txs: 1000,
                max_size: usize::MAX,
            },
            queued_limit: SubPoolLimit {
                max_txs: 1000,
                max_size: usize::MAX,
            },
            max_txs_per_sender: 1,
        };
        let mut pool = AA2dPool::new(config);
        let sender = Address::random();

        let tx0 = TxBuilder::aa(sender)
            .nonce_key(U256::from(1))
            .nonce(0)
            .build();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        assert_eq!(pool.slot_to_seq_id.len(), 1);

        for i in 2..12u64 {
            let tx = TxBuilder::aa(sender)
                .nonce_key(U256::from(i))
                .nonce(0)
                .build();
            let result = pool.add_transaction(
                Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
                0,
                TempoHardfork::T1,
            );
            assert!(
                result.is_err(),
                "tx with nonce_key {i} should be rejected by sender limit"
            );
        }

        assert_eq!(
            pool.slot_to_seq_id.len(),
            1,
            "rejected txs with new nonce keys should not grow slot_to_seq_id"
        );
        pool.assert_invariants();
    }

    #[test_case::test_case(false ; "live updates")]
    #[test_case::test_case(true  ; "no updates")]
    fn best_transactions_live_new_tx(no_updates: bool) {
        let mut pool = AA2dPool::default();
        let sender = Address::random();

        // Add one tx before creating the iterator
        let tx0 = TxBuilder::aa(sender).nonce_key(U256::ZERO).build();
        let tx0_hash = *tx0.hash();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let mut best = pool.best_transactions();
        if no_updates {
            best.no_updates();
        }

        // Add a new tx from a different sender while iterator is active
        let sender2 = Address::random();
        let tx1 = TxBuilder::aa(sender2).nonce_key(U256::ZERO).build();
        let tx1_hash = *tx1.hash();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let mut yielded = HashSet::new();
        for tx in best {
            yielded.insert(*tx.hash());
        }

        assert!(
            yielded.contains(&tx0_hash),
            "should always yield pre-existing tx"
        );
        assert_eq!(
            yielded.contains(&tx1_hash),
            !no_updates,
            "new tx should only be yielded when live updates are enabled"
        );
    }

    #[test]
    fn best_transactions_live_promoted() {
        let mut pool = AA2dPool::default();
        let sender = Address::random();

        // Insert tx with nonce=1 (queued due to gap)
        let tx1 = TxBuilder::aa(sender).nonce_key(U256::ZERO).nonce(1).build();
        let tx1_hash = *tx1.hash();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        // Create iterator — snapshot is empty (tx1 is queued)
        let mut best = pool.best_transactions();
        assert!(best.next().is_none(), "no pending txs yet");

        // Fill the gap with nonce=0, promoting tx1
        let tx0 = TxBuilder::aa(sender).nonce_key(U256::ZERO).nonce(0).build();
        let tx0_hash = *tx0.hash();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let mut yielded = HashSet::new();
        for tx in best {
            yielded.insert(*tx.hash());
        }

        assert_eq!(yielded.len(), 2, "should yield both tx0 and promoted tx1");
        assert!(yielded.contains(&tx0_hash));
        assert!(yielded.contains(&tx1_hash));
    }

    #[test]
    fn best_transactions_live_gapped_unblock_higher_fee_not_promoted() {
        // Scenario: tx at nonce=1 is queued (gap). A new tx arrives at nonce=0 that fills the
        // gap but has higher priority than the last yielded tx. The gap-filler should be stashed
        // (not added to `independent`) so neither nonce=0 nor nonce=1 gets yielded.
        let mut pool = AA2dPool::default();

        let sender_low = Address::random();
        let sender_gapped = Address::random();

        // Add a low-priority tx from sender_low so the iterator has something to yield first.
        // max_fee must exceed the T1 base fee (20 gwei) so that effective_tip > 0.
        let tx_low = TxBuilder::aa(sender_low)
            .nonce_key(U256::ZERO)
            .max_priority_fee(1_000_000_000)
            .max_fee(30_000_000_000)
            .build();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx_low, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        // Add a gapped tx (nonce=1) for sender_gapped — this will be queued.
        let tx_n1 = TxBuilder::aa(sender_gapped)
            .nonce_key(U256::ZERO)
            .nonce(1)
            .max_priority_fee(2_000_000_000)
            .max_fee(30_000_000_000)
            .build();
        let tx_n1_hash = *tx_n1.hash();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx_n1, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        // Create iterator and yield the low-priority tx to set `last_priority`.
        let mut best = pool.best_transactions();
        let first = best.next();
        assert!(first.is_some(), "should yield the low-priority tx");

        // Now fill the gap with nonce=0 that has HIGHER priority than the already-yielded tx.
        let tx_n0 = TxBuilder::aa(sender_gapped)
            .nonce_key(U256::ZERO)
            .nonce(0)
            .max_priority_fee(2_000_000_000)
            .max_fee(30_000_000_000)
            .build();
        let tx_n0_hash = *tx_n0.hash();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx_n0, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        // Neither nonce=0 nor nonce=1 should be yielded because nonce=0's priority is higher
        // than what was already yielded, so it gets stashed rather than added to `independent`.
        let remaining: Vec<_> = best.map(|tx| *tx.hash()).collect();
        assert!(
            !remaining.contains(&tx_n0_hash),
            "gap-filler with higher fee must not be yielded"
        );
        assert!(
            !remaining.contains(&tx_n1_hash),
            "gapped tx must not be promoted when gap-filler is stashed"
        );
    }

    #[test]
    fn best_transactions_live_expiring_nonce() {
        let mut pool = AA2dPool::default();

        let mut best = pool.best_transactions();

        // Add expiring nonce tx while iterator is active
        let sender = Address::random();
        let tx = TxBuilder::aa(sender).nonce_key(U256::MAX).nonce(0).build();
        let tx_hash = *tx.hash();
        pool.add_transaction(
            Arc::new(wrap_valid_tx(tx, TransactionOrigin::Local)),
            0,
            TempoHardfork::T1,
        )
        .unwrap();

        let first = best.next();
        assert!(first.is_some(), "should yield the expiring nonce tx");
        assert_eq!(*first.unwrap().hash(), tx_hash);
        assert!(best.next().is_none());
    }
}