chore: add printlns, update nonce manager to match ethers main

ethers-middleware/Cargo.toml

@@ -27,6 +27,7 @@ serde = { version = "1.0.124", default-features = false, features = ["derive"] }
 thiserror = { version = "1.0", default-features = false }
 futures-util = { version = "^0.3" }
 futures-locks = { version = "0.7", default-features = false }
+futures-channel.workspace = "0.3.28"
 tracing = { version = "0.1.37", default-features = false }
 tracing-futures = { version = "0.2.5", default-features = false }
 

ethers-middleware/src/gas_escalator/mod.rs

@@ -160,11 +160,13 @@ where
     ) -> Result<PendingTransaction<'_, M::Provider>, GasEscalatorError<M>> {
         let tx = tx.into();
 
+        println!("GAS_ESCALATOR: Sending transaction: {:?}", tx);
         let pending_tx = self
             .inner
             .send_transaction(tx.clone(), block)
             .await
             .map_err(GasEscalatorError::MiddlewareError)?;
+        println!("GAS_ESCALATOR: Sent transaction: {:?}", pending_tx);
 
         let tx = match tx {
             TypedTransaction::Legacy(inner) => inner,
@@ -309,6 +311,7 @@ impl<M, E> EscalationTask<M, E> {
                             match self.inner.send_transaction(replacement_tx.clone(), priority).await {
                                 Ok(new_tx_hash) => {
                                     let new_tx_hash = *new_tx_hash;
+                                    println!("escalated gas price");
                                     tracing::debug!(
                                         old_tx_hash = ?old_tx_hash,
                                         new_tx_hash = ?new_tx_hash,

ethers-middleware/src/nonce_manager.rs

@@ -1,20 +1,18 @@
 use async_trait::async_trait;
 use ethers_core::types::{transaction::eip2718::TypedTransaction, *};
 use ethers_providers::{FromErr, Middleware, PendingTransaction};
-use std::{
-    fmt::Debug,
-    sync::atomic::{AtomicBool, AtomicU64, Ordering},
-};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use thiserror::Error;
 
 #[derive(Debug)]
 /// Middleware used for calculating nonces locally, useful for signing multiple
 /// consecutive transactions without waiting for them to hit the mempool
 pub struct NonceManagerMiddleware<M> {
+    inner: M,
+    init_guard: futures_locks::Mutex<()>,
     initialized: AtomicBool,
     nonce: AtomicU64,
     address: Address,
-    inner: M,
 }
 
 impl<M> NonceManagerMiddleware<M>
@@ -24,7 +22,13 @@ where
     /// Instantiates the nonce manager with a 0 nonce. The `address` should be the
     /// address which you'll be sending transactions from
     pub fn new(inner: M, address: Address) -> Self {
-        Self { initialized: false.into(), nonce: 0.into(), inner, address }
+        Self {
+            inner,
+            init_guard: Default::default(),
+            initialized: Default::default(),
+            nonce: Default::default(),
+            address,
+        }
     }
 
     /// Returns the next nonce to be used
@@ -37,19 +41,29 @@ where
         &self,
         block: Option<BlockId>,
     ) -> Result<U256, NonceManagerError<M>> {
-        // initialize the nonce the first time the manager is called
-        if !self.initialized.load(Ordering::SeqCst) {
-            let nonce = self
-                .inner
-                .get_transaction_count(self.address, block)
-                .await
-                .map_err(FromErr::from)?;
-            self.nonce.store(nonce.as_u64(), Ordering::SeqCst);
-            self.initialized.store(true, Ordering::SeqCst);
+        if self.initialized.load(Ordering::SeqCst) {
+            // return current nonce
+            return Ok(self.nonce.load(Ordering::SeqCst).into());
         }
-        // return current nonce
-        Ok(self.nonce.load(Ordering::SeqCst).into())
-    }
+
+        let _guard = self.init_guard.lock().await;
+
+        // do this again in case multiple tasks enter this codepath
+        if self.initialized.load(Ordering::SeqCst) {
+            // return current nonce
+            return Ok(self.nonce.load(Ordering::SeqCst).into());
+        }
+
+        // initialize the nonce the first time the manager is called
+        let nonce = self
+            .inner
+            .get_transaction_count(self.address, block)
+            .await
+            .map_err(NonceManagerError::MiddlewareError)?;
+        self.nonce.store(nonce.as_u64(), Ordering::SeqCst);
+        self.initialized.store(true, Ordering::SeqCst);
+        Ok(nonce)
+    } // guard dropped here
 
     async fn get_transaction_count_with_manager(
         &self,
@@ -61,7 +75,7 @@ where
                 .inner
                 .get_transaction_count(self.address, block)
                 .await
-                .map_err(FromErr::from)?;
+                .map_err(NonceManagerError::MiddlewareError)?;
             self.nonce.store(nonce.as_u64(), Ordering::SeqCst);
             self.initialized.store(true, Ordering::SeqCst);
         }
@@ -78,12 +92,6 @@ pub enum NonceManagerError<M: Middleware> {
     MiddlewareError(M::Error),
 }
 
-impl<M: Middleware> FromErr<M::Error> for NonceManagerError<M> {
-    fn from(src: M::Error) -> Self {
-        NonceManagerError::MiddlewareError(src)
-    }
-}
-
 #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
 #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
 impl<M> Middleware for NonceManagerMiddleware<M>
@@ -107,7 +115,11 @@ where
             tx.set_nonce(self.get_transaction_count_with_manager(block).await?);
         }
 
-        Ok(self.inner().fill_transaction(tx, block).await.map_err(FromErr::from)?)
+        Ok(self
+            .inner()
+            .fill_transaction(tx, block)
+            .await
+            .map_err(NonceManagerError::MiddlewareError)?)
     }
 
     /// Signs and broadcasts the transaction. The optional parameter `block` can be passed so that
@@ -133,12 +145,22 @@ where
                     // was a nonce mismatch
                     self.nonce.store(nonce.as_u64(), Ordering::SeqCst);
                     tx.set_nonce(nonce);
-                    self.inner.send_transaction(tx, block).await.map_err(FromErr::from)
+                    self.inner
+                        .send_transaction(tx, block)
+                        .await
+                        .map_err(NonceManagerError::MiddlewareError)
                 } else {
                     // propagate the error otherwise
-                    Err(FromErr::from(err))
+                    Err(NonceManagerError::MiddlewareError(err))
                 }
             }
         }
     }
 }
+
+// Boilerplate
+impl<M: Middleware> FromErr<M::Error> for NonceManagerError<M> {
+    fn from(src: M::Error) -> NonceManagerError<M> {
+        NonceManagerError::MiddlewareError(src)
+    }
+}

ethers-middleware/tests/gas_escalator.rs

@@ -4,14 +4,16 @@ use std::convert::TryFrom;
 use ethers_core::{types::*, utils::Anvil};
 use ethers_middleware::{
     gas_escalator::{Frequency, GasEscalatorMiddleware, GeometricGasPrice},
-    SignerMiddleware,
+    NonceManagerMiddleware, SignerMiddleware,
 };
 use ethers_providers::{Http, Middleware, Provider};
 use ethers_signers::{LocalWallet, Signer};
+use instant::Duration;
+use tokio::time::sleep;
 
 #[tokio::test]
 async fn gas_escalator_live() {
-    let anvil = Anvil::new().block_time(2u64).spawn();
+    let anvil = Anvil::new().port(8545u16).block_time(2u64).spawn();
     let chain_id = anvil.chain_id();
     let provider = Provider::<Http>::try_from(anvil.endpoint()).unwrap();
 
@@ -21,21 +23,40 @@ async fn gas_escalator_live() {
     let address = wallet.address();
     let provider = SignerMiddleware::new(provider, wallet);
 
+    // wrap with nonce manager
+    // let nonce_manager_provider = NonceManagerMiddleware::new(provider, address);
+
     // wrap with escalator
     let escalator = GeometricGasPrice::new(5.0, 1u64, Some(2_000_000_000_000u64));
     let provider = GasEscalatorMiddleware::new(provider, escalator, Frequency::Duration(300));
 
     let nonce = provider.get_transaction_count(address, None).await.unwrap();
     // 1 gwei default base fee
     let gas_price = U256::from(1_000_000_000_u64);
-    // 125_000_000_000
-    let tx = TransactionRequest::pay(Address::zero(), 1u64)
-        .gas_price(gas_price)
-        .nonce(nonce)
-        .chain_id(chain_id);
+    // 125000000000
+    let tx = TransactionRequest::pay(Address::zero(), 1u64).gas_price(gas_price).nonce(nonce);
+    // .chain_id(chain_id);
 
-    // regardless of whether we get a receipt here, if gas is escalated by sending a new tx, this receipt won't be useful,
-    // since a tx with a different hash will end up replacing
-    let _pending = provider.send_transaction(tx, None).await.expect("could not send").await;
-    // checking the logs shows that the gas price is indeed being escalated twice
+    eprintln!("sending");
+    let pending = provider.send_transaction(tx, None).await.expect("could not send");
+    eprintln!("waiting");
+    let receipt = pending.await;
+    //  match pending.await {
+    //     Ok(receipt) => receipt.expect("dropped"),
+    //     Err(e) => {
+    //         eprintln!("reverted: {:?}", e);
+    //         panic!()
+    //     }
+    // };
+    // assert_eq!(receipt.from, address);
+    // assert_eq!(receipt.to, Some(Address::zero()));
+    println!("done escalating");
+    sleep(Duration::from_secs(3)).await;
+    // assert!(receipt.effective_gas_price.unwrap() > gas_price * 2, "{receipt:?}");
+    println!(
+        "receipt gas price: , hardcoded_gas_price: {}, receipt: {:?}",
+        // receipt.effective_gas_price.unwrap(),
+        gas_price,
+        receipt
+    );
 }

ethers-middleware/tests/nonce_manager.rs

@@ -2,23 +2,22 @@
 #[tokio::test]
 #[cfg(not(feature = "celo"))]
 async fn nonce_manager() {
-    use ethers_core::types::*;
-    use ethers_middleware::{nonce_manager::NonceManagerMiddleware, signer::SignerMiddleware};
-    use ethers_providers::Middleware;
-    use ethers_signers::{LocalWallet, Signer};
-    use std::time::Duration;
+    use std::convert::TryFrom;
 
-    let provider = ethers_providers::GOERLI.provider().interval(Duration::from_millis(2000u64));
-    let chain_id = provider.get_chainid().await.unwrap().as_u64();
+    use ethers_core::{
+        types::{BlockNumber, TransactionRequest},
+        utils::Anvil,
+    };
+    use ethers_middleware::NonceManagerMiddleware;
+    use ethers_providers::{Http, Middleware, Provider};
 
-    let wallet = std::env::var("GOERLI_PRIVATE_KEY")
-        .unwrap()
-        .parse::<LocalWallet>()
-        .unwrap()
-        .with_chain_id(chain_id);
-    let address = wallet.address();
+    let anvil = Anvil::new().port(8545u16).block_time(2u64).spawn();
+    let chain_id = anvil.chain_id();
+    println!("anvil endpoint: {}", anvil.endpoint());
+    let provider = Provider::<Http>::try_from(anvil.endpoint()).unwrap();
 
-    let provider = SignerMiddleware::new(provider, wallet);
+    let address = anvil.addresses()[0];
+    let to = anvil.addresses()[1];
 
     // the nonce manager must be over the Client so that it overrides the nonce
     // before the client gets it
@@ -34,22 +33,16 @@ async fn nonce_manager() {
     let mut tx_hashes = Vec::with_capacity(num_tx);
     for _ in 0..num_tx {
         let tx = provider
-            .send_transaction(
-                Eip1559TransactionRequest::new().to(address).value(100u64).chain_id(chain_id),
-                None,
-            )
+            .send_transaction(TransactionRequest::new().from(address).to(to).value(100u64), None)
             .await
             .unwrap();
         tx_hashes.push(*tx);
     }
 
-    // sleep a bit to ensure there's no flakiness in the test
-    std::thread::sleep(std::time::Duration::new(5, 0));
-
     let mut nonces = Vec::with_capacity(num_tx);
     for tx_hash in tx_hashes {
         nonces.push(provider.get_transaction(tx_hash).await.unwrap().unwrap().nonce.as_u64());
     }
 
-    assert_eq!(nonces, (nonce..nonce + (num_tx as u64)).collect::<Vec<_>>())
+    assert_eq!(nonces, (nonce..nonce + num_tx as u64).collect::<Vec<_>>());
 }