alloy_provider/fillers/

nonce.rs

use crate::{
    fillers::{FillerControlFlow, TxFiller},
    provider::SendableTx,
    Provider,
};
use alloy_network::{Network, TransactionBuilder};
use alloy_primitives::Address;
use alloy_transport::TransportResult;
use async_trait::async_trait;
use dashmap::DashMap;
use futures::lock::Mutex;
use std::sync::Arc;

/// A trait that determines the behavior of filling nonces.
#[cfg_attr(target_family = "wasm", async_trait(?Send))]
#[cfg_attr(not(target_family = "wasm"), async_trait)]
pub trait NonceManager: Clone + Send + Sync + std::fmt::Debug {
    /// Get the next nonce for the given account.
    async fn get_next_nonce<P, N>(&self, provider: &P, address: Address) -> TransportResult<u64>
    where
        P: Provider<N>,
        N: Network;
}

/// This [`NonceManager`] fetches the pending transaction count on every request and does not
/// reserve nonces locally. It makes more RPC calls than [`CachedNonceManager`] and is more
/// resilient to chain reorganizations, but concurrent transactions from the same address can
/// receive the same nonce.
#[derive(Clone, Debug, Default)]
#[non_exhaustive]
pub struct SimpleNonceManager;

#[cfg_attr(target_family = "wasm", async_trait(?Send))]
#[cfg_attr(not(target_family = "wasm"), async_trait)]
impl NonceManager for SimpleNonceManager {
    async fn get_next_nonce<P, N>(&self, provider: &P, address: Address) -> TransportResult<u64>
    where
        P: Provider<N>,
        N: Network,
    {
        provider.get_transaction_count(address).pending().await
    }
}

/// Cached nonce manager
///
/// This [`NonceManager`] fetches the pending transaction count the first time it sees an account,
/// then reserves subsequent nonces locally as transactions are filled. Clones share the same
/// per-address state, so they can allocate distinct nonces to concurrent requests.
///
/// The local nonce advances before broadcast. A failed submission, a chain reorganization, or a
/// transaction sent through another manager can therefore leave it out of sync with the node; it
/// does not resynchronize automatically.
///
/// There is also an alternative implementation [`SimpleNonceManager`] that does not store the
/// transaction count locally.
#[derive(Clone, Debug, Default)]
pub struct CachedNonceManager {
    nonces: Arc<DashMap<Address, Arc<Mutex<u64>>>>,
}

#[cfg_attr(target_family = "wasm", async_trait(?Send))]
#[cfg_attr(not(target_family = "wasm"), async_trait)]
impl NonceManager for CachedNonceManager {
    async fn get_next_nonce<P, N>(&self, provider: &P, address: Address) -> TransportResult<u64>
    where
        P: Provider<N>,
        N: Network,
    {
        // Use `u64::MAX` as a sentinel value to indicate that the nonce has not been fetched yet.
        const NONE: u64 = u64::MAX;

        // Locks dashmap internally for a short duration to clone the `Arc`.
        // We also don't want to hold the dashmap lock through the await point below.
        let nonce = {
            let rm = self.nonces.entry(address).or_insert_with(|| Arc::new(Mutex::new(NONE)));
            Arc::clone(rm.value())
        };

        let mut nonce = nonce.lock().await;
        let new_nonce = if *nonce == NONE {
            // Initialize the nonce if we haven't seen this account before.
            trace!(%address, "fetching nonce");
            provider.get_transaction_count(address).pending().await?
        } else {
            trace!(%address, current_nonce = *nonce, "incrementing nonce");
            *nonce + 1
        };
        *nonce = new_nonce;
        Ok(new_nonce)
    }
}

/// A [`TxFiller`] that fills nonces on transactions. The behavior of filling nonces is determined
/// by the [`NonceManager`].
///
/// # Note
///
/// - If the transaction request does not have a sender set, this layer will not fill nonces.
/// - An explicit nonce is preserved and does not consult the manager.
/// - For concurrent sends from one address, reuse a provider or cloned [`CachedNonceManager`]. Two
///   independent cached managers can allocate conflicting nonces, while [`SimpleNonceManager`] does
///   not reserve nonces between concurrent requests.
///
/// # Example
///
/// ```
/// # use alloy_network::{Ethereum};
/// # use alloy_rpc_types_eth::TransactionRequest;
/// # use alloy_provider::{ProviderBuilder, RootProvider, Provider};
/// # use alloy_signer_local::PrivateKeySigner;
/// # async fn test(url: url::Url) -> Result<(), Box<dyn std::error::Error>> {
/// let pk: PrivateKeySigner = "0x...".parse()?;
/// let provider = ProviderBuilder::<_, _, Ethereum>::default()
///     .with_simple_nonce_management()
///     .wallet(pk)
///     .connect_http(url);
///
/// provider.send_transaction(TransactionRequest::default()).await;
/// # Ok(())
/// # }
/// ```
#[derive(Clone, Debug, Default)]
pub struct NonceFiller<M: NonceManager = CachedNonceManager> {
    nonce_manager: M,
}

impl<M: NonceManager> NonceFiller<M> {
    /// Creates a new [`NonceFiller`] with the specified [`NonceManager`].
    ///
    /// To instantiate with the [`SimpleNonceManager`], use [`NonceFiller::simple()`].
    ///
    /// To instantiate with the [`CachedNonceManager`], use [`NonceFiller::cached()`].
    pub const fn new(nonce_manager: M) -> Self {
        Self { nonce_manager }
    }

    /// Creates a new [`NonceFiller`] with the [`SimpleNonceManager`].
    ///
    /// [`SimpleNonceManager`] fetches the pending transaction count for every request, resulting in
    /// frequent RPC calls and no local nonce reservation.
    pub const fn simple() -> NonceFiller<SimpleNonceManager> {
        NonceFiller { nonce_manager: SimpleNonceManager }
    }

    /// Creates a new [`NonceFiller`] with the [`CachedNonceManager`].
    ///
    /// [`CachedNonceManager`] will fetch the transaction count for any new account it sees,
    /// store it locally, and increment the locally stored nonce as transactions are filled,
    /// reducing the number of RPC calls. Reservation happens before broadcast and also applies to
    /// direct [`FillProvider::fill`](crate::fillers::FillProvider::fill) calls.
    pub fn cached() -> NonceFiller<CachedNonceManager> {
        NonceFiller { nonce_manager: CachedNonceManager::default() }
    }

    /// Get a reference to the nonce manager.
    pub const fn nonce_manager(&self) -> &M {
        &self.nonce_manager
    }

    /// Get a mutable reference to the nonce manager.
    pub const fn nonce_manager_mut(&mut self) -> &mut M {
        &mut self.nonce_manager
    }
}

impl<M: NonceManager, N: Network> TxFiller<N> for NonceFiller<M> {
    type Fillable = u64;

    fn status(&self, tx: &<N as Network>::TransactionRequest) -> FillerControlFlow {
        if tx.nonce().is_some() {
            return FillerControlFlow::Finished;
        }
        if tx.from().is_none() {
            return FillerControlFlow::missing("NonceManager", vec!["from"]);
        }
        FillerControlFlow::Ready
    }

    fn fill_sync(&self, _tx: &mut SendableTx<N>) {}

    async fn prepare<P>(
        &self,
        provider: &P,
        tx: &N::TransactionRequest,
    ) -> TransportResult<Self::Fillable>
    where
        P: Provider<N>,
    {
        let from = tx.from().expect("checked by 'ready()'");
        self.nonce_manager.get_next_nonce(provider, from).await
    }

    async fn fill(
        &self,
        nonce: Self::Fillable,
        mut tx: SendableTx<N>,
    ) -> TransportResult<SendableTx<N>> {
        if let Some(builder) = tx.as_mut_builder() {
            builder.set_nonce(nonce);
        }
        Ok(tx)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{ProviderBuilder, WalletProvider};
    use alloy_consensus::Transaction;
    use alloy_primitives::{address, U256};
    use alloy_rpc_types_eth::TransactionRequest;

    async fn check_nonces<P, N, M>(
        filler: &NonceFiller<M>,
        provider: &P,
        address: Address,
        start: u64,
    ) where
        P: Provider<N>,
        N: Network,
        M: NonceManager,
    {
        for i in start..start + 5 {
            let nonce = filler.nonce_manager.get_next_nonce(&provider, address).await.unwrap();
            assert_eq!(nonce, i);
        }
    }

    #[tokio::test]
    async fn smoke_test() {
        let filler = NonceFiller::<CachedNonceManager>::default();
        let provider = ProviderBuilder::new().connect_anvil();
        let address = Address::ZERO;
        check_nonces(&filler, &provider, address, 0).await;

        #[cfg(feature = "anvil-api")]
        {
            use crate::ext::AnvilApi;
            filler.nonce_manager.nonces.clear();
            provider.anvil_set_nonce(address, 69).await.unwrap();
            check_nonces(&filler, &provider, address, 69).await;
        }
    }

    #[tokio::test]
    async fn concurrency() {
        let filler = Arc::new(NonceFiller::<CachedNonceManager>::default());
        let provider = Arc::new(ProviderBuilder::new().connect_anvil());
        let address = Address::ZERO;
        let tasks = (0..5)
            .map(|_| {
                let filler = Arc::clone(&filler);
                let provider = Arc::clone(&provider);
                tokio::spawn(async move {
                    filler.nonce_manager.get_next_nonce(&provider, address).await
                })
            })
            .collect::<Vec<_>>();

        let mut ns = Vec::new();
        for task in tasks {
            ns.push(task.await.unwrap().unwrap());
        }
        ns.sort_unstable();
        assert_eq!(ns, (0..5).collect::<Vec<_>>());

        assert_eq!(filler.nonce_manager.nonces.len(), 1);
        assert_eq!(*filler.nonce_manager.nonces.get(&address).unwrap().value().lock().await, 4);
    }

    #[tokio::test]
    async fn no_nonce_if_sender_unset() {
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .with_cached_nonce_management()
            .connect_anvil();

        let tx = TransactionRequest {
            value: Some(U256::from(100)),
            to: Some(address!("d8dA6BF26964aF9D7eEd9e03E53415D37aA96045").into()),
            gas_price: Some(20e9 as u128),
            gas: Some(21000),
            ..Default::default()
        };

        // errors because signer layer expects nonce to be set, which it is not
        assert!(provider.send_transaction(tx).await.is_err());
    }

    #[tokio::test]
    async fn increments_nonce() {
        let provider = ProviderBuilder::new()
            .disable_recommended_fillers()
            .with_cached_nonce_management()
            .connect_anvil_with_wallet();

        let from = provider.default_signer_address();
        let tx = TransactionRequest {
            from: Some(from),
            value: Some(U256::from(100)),
            to: Some(address!("d8dA6BF26964aF9D7eEd9e03E53415D37aA96045").into()),
            gas_price: Some(20e9 as u128),
            gas: Some(21000),
            ..Default::default()
        };

        let pending = provider.send_transaction(tx.clone()).await.unwrap();
        let tx_hash = pending.watch().await.unwrap();
        let mined_tx = provider
            .get_transaction_by_hash(tx_hash)
            .await
            .expect("failed to fetch tx")
            .expect("tx not included");
        assert_eq!(mined_tx.nonce(), 0);

        let pending = provider.send_transaction(tx).await.unwrap();
        let tx_hash = pending.watch().await.unwrap();
        let mined_tx = provider
            .get_transaction_by_hash(tx_hash)
            .await
            .expect("fail to fetch tx")
            .expect("tx didn't finalize");
        assert_eq!(mined_tx.nonce(), 1);
    }

    #[tokio::test]
    async fn cloned_managers() {
        let cnm1 = CachedNonceManager::default();
        let cnm2 = cnm1.clone();

        let provider = ProviderBuilder::new().connect_anvil();
        let address = Address::ZERO;

        assert_eq!(cnm1.get_next_nonce(&provider, address).await.unwrap(), 0);
        assert_eq!(cnm2.get_next_nonce(&provider, address).await.unwrap(), 1);
        assert_eq!(cnm1.get_next_nonce(&provider, address).await.unwrap(), 2);
        assert_eq!(cnm2.get_next_nonce(&provider, address).await.unwrap(), 3);
    }
}