anvil/eth/pool/

mod.rs

//! # Transaction Pool implementation
//!
//! The transaction pool is responsible for managing a set of transactions that can be included in
//! upcoming blocks.
//!
//! The main task of the pool is to prepare an ordered list of transactions that are ready to be
//! included in a new block.
//!
//! Each imported block can affect the validity of transactions already in the pool.
//! The miner expects the most up-to-date transactions when attempting to create a new block.
//! After being included in a block, a transaction should be removed from the pool, this process is
//! called _pruning_ and due to separation of concerns is triggered externally.
//! The pool essentially performs following services:
//!   * import transactions
//!   * order transactions
//!   * provide ordered set of transactions that are ready for inclusion
//!   * prune transactions
//!
//! Each transaction in the pool contains markers that it _provides_ or _requires_. This property is
//! used to determine whether it can be included in a block (transaction is ready) or whether it
//! still _requires_ other transactions to be mined first (transaction is pending).
//! A transaction is associated with the nonce of the account it's sent from. A unique identifying
//! marker for a transaction is therefore the pair `(nonce + account)`. An incoming transaction with
//! a `nonce > nonce on chain` will _require_ `(nonce -1, account)` first, before it is ready to be
//! included in a block.
//!
//! This implementation is adapted from <https://github.com/paritytech/substrate/tree/master/client/transaction-pool>

use crate::{
    eth::{
        error::PoolError,
        fees::FeeManager,
        pool::transactions::{
            PendingPoolTransaction, PendingTransactions, PoolTransaction, ReadyTransactions,
            TransactionsIterator, TxMarker,
        },
    },
    mem::storage::MinedBlockOutcome,
};
use alloy_consensus::Transaction;
use alloy_primitives::{Address, TxHash};
use alloy_rpc_types::txpool::TxpoolStatus;
use anvil_core::eth::transaction::PendingTransaction;
use futures::channel::mpsc::{Receiver, Sender, channel};
use parking_lot::{Mutex, RwLock};
use std::{collections::VecDeque, fmt, sync::Arc};

#[cfg(feature = "base")]
use alloy_consensus::Typed2718;

pub mod transactions;

/// Transaction pool that performs validation.
pub struct Pool<T> {
    /// processes all pending transactions
    inner: RwLock<PoolInner<T>>,
    /// listeners for new ready transactions
    transaction_listener: Mutex<Vec<ReadyListener>>,
    /// source of the base fee that ready transactions are ranked at.
    fees: Option<FeeManager>,
}

struct ReadyListener {
    sender: Sender<TxHash>,
    close_on_full: bool,
}

/// An independent snapshot of a transaction pool.
#[derive(Debug)]
pub(crate) struct PoolSnapshot<T> {
    inner: PoolInner<T>,
}

impl<T> Default for Pool<T> {
    fn default() -> Self {
        Self {
            inner: RwLock::new(PoolInner::default()),
            transaction_listener: Default::default(),
            fees: None,
        }
    }
}

// == impl Pool ==

impl<T> Pool<T> {
    /// Creates a pool that ranks ready transactions at the base fee of the given fee manager.
    pub fn new(fees: FeeManager) -> Self {
        Self { fees: Some(fees), ..Default::default() }
    }

    /// Returns the base fee of the next block.
    fn base_fee(&self) -> u64 {
        self.fees.as_ref().map_or(0, FeeManager::base_fee)
    }

    /// Returns an independent snapshot of the pool.
    pub(crate) fn snapshot(&self) -> PoolSnapshot<T> {
        let pool = self.inner.read();
        PoolSnapshot {
            inner: PoolInner {
                ready_transactions: pool.ready_transactions.snapshot(),
                pending_transactions: pool.pending_transactions.snapshot(),
            },
        }
    }

    /// Restores the pool to a previous snapshot.
    pub(crate) fn restore(&self, snapshot: PoolSnapshot<T>) {
        let ready = {
            let mut pool = self.inner.write();
            *pool = snapshot.inner;
            pool.ready_transactions(self.base_fee()).map(|tx| tx.hash()).collect::<Vec<_>>()
        };
        for hash in ready {
            self.notify_listener(hash);
        }
    }

    /// Returns an iterator that yields all transactions that are currently ready
    pub fn ready_transactions(&self) -> TransactionsIterator<T> {
        self.inner.read().ready_transactions(self.base_fee())
    }

    /// Returns all transactions that are not ready to be included in a block yet
    pub fn pending_transactions(&self) -> Vec<Arc<PoolTransaction<T>>> {
        self.inner.read().pending_transactions.transactions().collect()
    }

    /// Returns every ready and queued transaction.
    #[cfg(feature = "base")]
    pub fn all_transactions(&self) -> Vec<Arc<PoolTransaction<T>>> {
        let pool = self.inner.read();
        pool.pending_transactions
            .transactions()
            .chain(pool.ready_transactions(self.base_fee()))
            .collect()
    }

    /// Returns the number of tx that are ready and queued for further execution
    pub fn txpool_status(&self) -> TxpoolStatus {
        // Note: naming differs here compared to geth's `TxpoolStatus`
        let pending: u64 = self.inner.read().ready_transactions.len().try_into().unwrap_or(0);
        let queued: u64 = self.inner.read().pending_transactions.len().try_into().unwrap_or(0);
        TxpoolStatus { pending, queued }
    }

    /// Adds a new transaction listener to the pool that gets notified about every new ready
    /// transaction
    pub fn add_ready_listener(&self) -> Receiver<TxHash> {
        self.add_ready_listener_with_policy(false)
    }

    /// Adds a subscription listener that closes instead of dropping notifications on overflow.
    pub(crate) fn add_ready_subscription_listener(&self) -> Receiver<TxHash> {
        self.add_ready_listener_with_policy(true)
    }

    fn add_ready_listener_with_policy(&self, close_on_full: bool) -> Receiver<TxHash> {
        const TX_LISTENER_BUFFER_SIZE: usize = 2048;
        let (tx, rx) = channel(TX_LISTENER_BUFFER_SIZE);
        self.transaction_listener.lock().push(ReadyListener { sender: tx, close_on_full });
        rx
    }

    /// Returns true if this pool already contains the transaction
    pub fn contains(&self, tx_hash: &TxHash) -> bool {
        self.inner.read().contains(tx_hash)
    }

    /// Returns true if this pool contains a transaction from `sender` with `nonce`.
    pub fn contains_sender_nonce(&self, sender: Address, nonce: u64) -> bool
    where
        T: Transaction,
    {
        self.inner
            .read()
            .transactions_by_sender(sender)
            .any(|tx| tx.pending_transaction.nonce() == nonce)
    }

    /// Returns a transaction from `sender` that provides exactly `markers`.
    #[cfg(feature = "base")]
    pub fn transaction_with_markers(
        &self,
        sender: Address,
        markers: &[TxMarker],
    ) -> Option<Arc<PoolTransaction<T>>> {
        self.inner.read().transactions_by_sender(sender).find(|tx| tx.provides == markers)
    }

    /// Removes all transactions from the pool
    pub fn clear(&self) {
        let mut pool = self.inner.write();
        pool.clear();
    }

    /// Remove the given transactions from the pool
    pub fn remove_invalid(&self, tx_hashes: Vec<TxHash>) -> Vec<Arc<PoolTransaction<T>>> {
        self.inner.write().remove_invalid(tx_hashes)
    }

    /// Remove transactions by sender
    pub fn remove_transactions_by_address(&self, sender: Address) -> Vec<Arc<PoolTransaction<T>>> {
        self.inner.write().remove_transactions_by_address(sender)
    }

    /// Removes a single transaction from the pool
    ///
    /// This is similar to `[Pool::remove_invalid()]` but for a single transaction.
    ///
    /// **Note**: this will also drop any transaction that depend on the `tx`
    pub fn drop_transaction(&self, tx: TxHash) -> Option<Arc<PoolTransaction<T>>> {
        trace!(target: "txpool", "Dropping transaction: [{:?}]", tx);
        let removed = {
            let mut pool = self.inner.write();
            let mut removed = pool.ready_transactions.remove_with_markers(vec![tx], None);
            let invalidated = removed.iter().flat_map(|tx| tx.provides.iter().cloned());
            removed.extend(pool.pending_transactions.remove_with_dependents(vec![tx], invalidated));
            removed
        };
        trace!(target: "txpool", "Dropped transactions: {:?}", removed.iter().map(|tx| tx.hash()).collect::<Vec<_>>());

        if removed.is_empty() {
            None
        } else {
            removed.into_iter().find(|t| *t.pending_transaction.hash() == tx)
        }
    }

    /// Notifies listeners if the transaction was added to the ready queue.
    fn notify_ready(&self, tx: &AddedTransaction<T>) {
        if let AddedTransaction::Ready(ready) = tx {
            self.notify_listener(ready.hash);
            for promoted in ready.promoted.iter().copied() {
                self.notify_listener(promoted);
            }
        }
    }

    /// notifies all listeners about the transaction
    fn notify_listener(&self, hash: TxHash) {
        let mut listener = self.transaction_listener.lock();
        // this is basically a retain but with mut reference
        for n in (0..listener.len()).rev() {
            let mut listener_tx = listener.swap_remove(n);
            let retain = match listener_tx.sender.try_send(hash) {
                Ok(()) => true,
                Err(e) => {
                    if e.is_full() {
                        warn!(
                            target: "txpool",
                            "[{:?}] Failed to send tx notification because channel is full",
                            hash,
                        );
                        !listener_tx.close_on_full
                    } else {
                        false
                    }
                }
            };
            if retain {
                listener.push(listener_tx)
            }
        }
    }
}

impl<T: Clone> Pool<T> {
    /// Returns the _pending_ transaction for that `hash` if it exists in the mempool
    pub fn get_transaction(&self, hash: TxHash) -> Option<PendingTransaction<T>> {
        self.inner.read().get_transaction(hash)
    }
}

impl<T: Transaction> Pool<T> {
    /// Invoked when a set of transactions ([Self::ready_transactions()]) was executed.
    ///
    /// This will remove the transactions from the pool.
    ///
    /// Returns `true` if ready transactions left behind by the block can be included by mining
    /// again right away, e.g. because the block hit `max_transactions` or ran out of gas.
    pub fn on_mined_block(self: &Arc<Self>, outcome: MinedBlockOutcome<T>) -> bool {
        let MinedBlockOutcome { block_number, included, stale, invalid, not_yet_valid } = outcome;
        // Requiring txs to leave the pool keeps this retry from mining empty blocks for txs that
        // can never be included. Not-yet-valid txs and their dependents are retried by the delayed
        // re-notify.
        let made_progress = !included.is_empty() || !stale.is_empty() || !invalid.is_empty();
        let retry_ready = made_progress && not_yet_valid.is_empty();

        // remove invalid transactions from the pool
        self.remove_invalid(invalid.into_iter().map(|tx| tx.hash()).collect());

        // Prune mined and stale markers; both are satisfied by the resulting state.
        let res = self.prune_markers(
            block_number,
            included.into_iter().chain(stale).flat_map(|tx| tx.provides.clone()),
        );
        trace!(target: "txpool", "pruned transaction markers {:?}", res);

        // Re-notify the miner about not-yet-valid transactions so they'll be retried.
        // Delay by 1 second to let time advance before the next mining attempt.
        if !not_yet_valid.is_empty() {
            let tx_hashes: Vec<_> = not_yet_valid.iter().map(|tx| tx.hash()).collect();
            let pool = Arc::clone(self);
            tokio::spawn(async move {
                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
                for hash in tx_hashes {
                    trace!(target: "txpool", "re-notifying for not-yet-valid tx: {:?}", hash);
                    pool.notify_listener(hash);
                }
            });
        }

        retry_ready && !self.inner.read().ready_transactions.is_empty()
    }

    /// Removes ready transactions for the given iterator of identifying markers.
    ///
    /// For each marker we can remove transactions in the pool that either provide the marker
    /// directly or are a dependency of the transaction associated with that marker.
    pub fn prune_markers(
        &self,
        block_number: u64,
        markers: impl IntoIterator<Item = TxMarker>,
    ) -> PruneResult<T> {
        debug!(target: "txpool", ?block_number, "pruning transactions");
        let res = self.inner.write().prune_markers(markers);
        for tx in &res.promoted {
            self.notify_ready(tx);
        }
        res
    }

    /// Adds a new transaction to the pool
    pub fn add_transaction(
        &self,
        tx: PoolTransaction<T>,
    ) -> Result<AddedTransaction<T>, PoolError> {
        let added = self.inner.write().add_transaction(tx)?;
        self.notify_ready(&added);
        Ok(added)
    }
}

#[cfg(feature = "base")]
impl<T: Typed2718> Pool<T> {
    /// Removes every transaction with the given EIP-2718 type.
    pub fn clear_transaction_type(&self, tx_type: u8) -> Vec<Arc<PoolTransaction<T>>> {
        let hashes = {
            let pool = self.inner.read();
            pool.pending_transactions
                .transactions()
                .chain(pool.ready_transactions(self.base_fee()))
                .filter_map(|tx| {
                    (tx.pending_transaction.transaction.ty() == tx_type).then_some(tx.hash())
                })
                .collect()
        };
        self.remove_invalid(hashes)
    }
}

/// A Transaction Pool
///
/// Contains all transactions that are ready to be executed
#[derive(Debug)]
struct PoolInner<T> {
    ready_transactions: ReadyTransactions<T>,
    pending_transactions: PendingTransactions<T>,
}

impl<T> Default for PoolInner<T> {
    fn default() -> Self {
        Self { ready_transactions: Default::default(), pending_transactions: Default::default() }
    }
}

// == impl PoolInner ==

impl<T> PoolInner<T> {
    /// Returns an iterator over transactions that are ready.
    fn ready_transactions(&self, base_fee: u64) -> TransactionsIterator<T> {
        self.ready_transactions.get_transactions(base_fee)
    }

    /// Clears
    fn clear(&mut self) {
        self.ready_transactions.clear();
        self.pending_transactions.clear();
    }

    /// Returns an iterator over all transactions in the pool filtered by the sender
    pub fn transactions_by_sender(
        &self,
        sender: Address,
    ) -> impl Iterator<Item = Arc<PoolTransaction<T>>> + '_ {
        let pending_txs = self
            .pending_transactions
            .transactions()
            .filter(move |tx| tx.pending_transaction.sender().eq(&sender));

        let ready_txs = self
            .ready_transactions
            // Only membership matters here, so the ranking base fee is irrelevant.
            .get_transactions(0)
            .filter(move |tx| tx.pending_transaction.sender().eq(&sender));

        pending_txs.chain(ready_txs)
    }

    /// Returns true if this pool already contains the transaction
    fn contains(&self, tx_hash: &TxHash) -> bool {
        self.pending_transactions.contains(tx_hash) || self.ready_transactions.contains(tx_hash)
    }

    /// Remove the given transactions from the pool
    fn remove_invalid(&mut self, tx_hashes: Vec<TxHash>) -> Vec<Arc<PoolTransaction<T>>> {
        // early exit in case there is no invalid transactions.
        if tx_hashes.is_empty() {
            return vec![];
        }
        trace!(target: "txpool", "Removing invalid transactions: {:?}", tx_hashes);

        let mut removed = self.ready_transactions.remove_with_markers(tx_hashes.clone(), None);
        removed.extend(self.pending_transactions.remove(tx_hashes));

        trace!(target: "txpool", "Removed invalid transactions: {:?}", removed.iter().map(|tx| tx.hash()).collect::<Vec<_>>());

        removed
    }

    /// Remove transactions by sender address
    fn remove_transactions_by_address(&mut self, sender: Address) -> Vec<Arc<PoolTransaction<T>>> {
        let tx_hashes =
            self.transactions_by_sender(sender).map(move |tx| tx.hash()).collect::<Vec<TxHash>>();

        if tx_hashes.is_empty() {
            return vec![];
        }

        trace!(target: "txpool", "Removing transactions: {:?}", tx_hashes);

        let mut removed = self.ready_transactions.remove_with_markers(tx_hashes.clone(), None);
        removed.extend(self.pending_transactions.remove(tx_hashes));

        trace!(target: "txpool", "Removed transactions: {:?}", removed.iter().map(|tx| tx.hash()).collect::<Vec<_>>());

        removed
    }
}

impl<T: Clone> PoolInner<T> {
    /// checks both pools for the matching transaction
    ///
    /// Returns `None` if the transaction does not exist in the pool
    fn get_transaction(&self, hash: TxHash) -> Option<PendingTransaction<T>> {
        if let Some(pending) = self.pending_transactions.get(&hash) {
            return Some(pending.transaction.pending_transaction.clone());
        }
        Some(
            self.ready_transactions.get(&hash)?.transaction.transaction.pending_transaction.clone(),
        )
    }
}

impl<T: Transaction> PoolInner<T> {
    fn add_transaction(
        &mut self,
        tx: PoolTransaction<T>,
    ) -> Result<AddedTransaction<T>, PoolError> {
        if self.contains(&tx.hash()) {
            debug!(target: "txpool", "[{:?}] Already imported", tx.hash());
            return Err(PoolError::AlreadyImported(tx.hash()));
        }

        let tx = PendingPoolTransaction::new(tx, self.ready_transactions.provided_markers());
        trace!(target: "txpool", "[{:?}] ready={}", tx.transaction.hash(), tx.is_ready());

        // If all markers are not satisfied import to future
        if !tx.is_ready() {
            let hash = tx.transaction.hash();
            self.pending_transactions.add_transaction(tx)?;
            return Ok(AddedTransaction::Pending { hash });
        }
        self.add_ready_transaction(tx)
    }

    /// Adds the transaction to the ready queue
    fn add_ready_transaction(
        &mut self,
        tx: PendingPoolTransaction<T>,
    ) -> Result<AddedTransaction<T>, PoolError> {
        let hash = tx.transaction.hash();
        trace!(target: "txpool", "adding ready transaction [{:?}]", hash);
        let mut ready = ReadyTransaction::new(hash);

        let mut tx_queue = VecDeque::from([tx]);
        // tracks whether we're processing the given `tx`
        let mut is_new_tx = true;

        // take first transaction from the list
        while let Some(current_tx) = tx_queue.pop_front() {
            // also add the transaction that the current transaction unlocks
            tx_queue.extend(
                self.pending_transactions.mark_and_unlock(&current_tx.transaction.provides),
            );

            let current_hash = current_tx.transaction.hash();
            // try to add the transaction to the ready pool
            match self.ready_transactions.add_transaction(current_tx) {
                Ok(replaced_transactions) => {
                    if !is_new_tx {
                        ready.promoted.push(current_hash);
                    }
                    // tx removed from ready pool
                    ready.removed.extend(replaced_transactions);
                }
                Err(err) => {
                    // failed to add transaction
                    if is_new_tx {
                        debug!(target: "txpool", "[{:?}] Failed to add tx: {:?}", current_hash,
        err);
                        return Err(err);
                    }
                    ready.discarded.push(current_hash);
                }
            }
            is_new_tx = false;
        }

        // check for a cycle where importing a transaction resulted in pending transactions to be
        // added while removing current transaction. in which case we move this transaction back to
        // the pending queue
        if ready.removed.iter().any(|tx| *tx.hash() == hash) {
            self.ready_transactions.clear_transactions(&ready.promoted);
            return Err(PoolError::CyclicTransaction);
        }

        Ok(AddedTransaction::Ready(ready))
    }

    /// Prunes the transactions that provide the given markers
    ///
    /// This will effectively remove those transactions that satisfy the markers and transactions
    /// from the pending queue might get promoted to if the markers unlock them.
    pub fn prune_markers(&mut self, markers: impl IntoIterator<Item = TxMarker>) -> PruneResult<T> {
        let mut imports = vec![];
        let mut pruned = vec![];

        for marker in markers {
            // mark as satisfied and store the transactions that got unlocked
            imports.extend(self.pending_transactions.mark_and_unlock(Some(&marker)));
            // prune transactions
            pruned.extend(self.ready_transactions.prune_tags(marker.clone()));
        }

        let mut promoted = vec![];
        let mut failed = vec![];
        for tx in imports {
            let hash = tx.transaction.hash();
            match self.add_ready_transaction(tx) {
                Ok(res) => promoted.push(res),
                Err(e) => {
                    warn!(target: "txpool", "Failed to promote tx [{:?}] : {:?}", hash, e);
                    failed.push(hash)
                }
            }
        }

        PruneResult { pruned, failed, promoted }
    }
}

/// Represents the outcome of a prune
pub struct PruneResult<T> {
    /// a list of added transactions that a pruned marker satisfied
    pub promoted: Vec<AddedTransaction<T>>,
    /// all transactions that  failed to be promoted and now are discarded
    pub failed: Vec<TxHash>,
    /// all transactions that were pruned from the ready pool
    pub pruned: Vec<Arc<PoolTransaction<T>>>,
}

impl<T> fmt::Debug for PruneResult<T> {
    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(fmt, "PruneResult {{ ")?;
        write!(
            fmt,
            "promoted: {:?}, ",
            self.promoted.iter().map(|tx| *tx.hash()).collect::<Vec<_>>()
        )?;
        write!(fmt, "failed: {:?}, ", self.failed)?;
        write!(
            fmt,
            "pruned: {:?}, ",
            self.pruned.iter().map(|tx| *tx.pending_transaction.hash()).collect::<Vec<_>>()
        )?;
        write!(fmt, "}}")?;
        Ok(())
    }
}

#[derive(Clone, Debug)]
pub struct ReadyTransaction<T> {
    /// the hash of the submitted transaction
    hash: TxHash,
    /// transactions promoted to the ready queue
    promoted: Vec<TxHash>,
    /// transaction that failed and became discarded
    discarded: Vec<TxHash>,
    /// Transactions removed from the Ready pool
    removed: Vec<Arc<PoolTransaction<T>>>,
}

impl<T> ReadyTransaction<T> {
    pub fn new(hash: TxHash) -> Self {
        Self {
            hash,
            promoted: Default::default(),
            discarded: Default::default(),
            removed: Default::default(),
        }
    }
}

#[derive(Clone, Debug)]
pub enum AddedTransaction<T> {
    /// transaction was successfully added and being processed
    Ready(ReadyTransaction<T>),
    /// Transaction was successfully added but not yet queued for processing
    Pending {
        /// the hash of the submitted transaction
        hash: TxHash,
    },
}

impl<T> AddedTransaction<T> {
    pub const fn hash(&self) -> &TxHash {
        match self {
            Self::Ready(tx) => &tx.hash,
            Self::Pending { hash } => hash,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use alloy_primitives::U256;
    use futures::StreamExt;
    use std::task::{Context, Poll, Waker};

    #[test]
    fn subscription_listener_closes_on_overflow_without_closing_ordinary_listener() {
        let pool = Pool::<()>::default();
        let mut ordinary = pool.add_ready_listener();
        let mut subscription = pool.add_ready_subscription_listener();
        let hashes = (0..4_096).map(|index| TxHash::from(U256::from(index))).collect::<Vec<_>>();

        // Overflow both listeners without draining either.
        for &hash in &hashes {
            pool.notify_listener(hash);
        }

        let mut cx = Context::from_waker(Waker::noop());
        let mut expected = Vec::new();
        while let Poll::Ready(Some(hash)) = ordinary.poll_next_unpin(&mut cx) {
            expected.push(hash);
        }
        assert!(!expected.is_empty());
        assert!(expected.len() < hashes.len());
        assert_eq!(expected, hashes[..expected.len()]);
        assert_eq!(ordinary.poll_next_unpin(&mut cx), Poll::Pending);

        let mut received = Vec::new();
        while let Poll::Ready(Some(hash)) = subscription.poll_next_unpin(&mut cx) {
            received.push(hash);
        }
        assert_eq!(received, expected);
        assert_eq!(subscription.poll_next_unpin(&mut cx), Poll::Ready(None));

        let hash = TxHash::from(U256::MAX);
        pool.notify_listener(hash);
        assert_eq!(ordinary.poll_next_unpin(&mut cx), Poll::Ready(Some(hash)));
    }
}