test: fix duplicate tx flakes by rotating through a list of wallets (#451)

* test: fix duplicate tx flakes by rotating through a list of wallets

* fix: make typed_txs test more robust

* fix: ensure wallets are loaded in a thread-safe way

without fetch_add, it's possible that 2 loads happen for the same id before the `store` mutex is acquired'

Cargo.lock

@@ -953,6 +953,7 @@ dependencies = [
  "futures-util",
  "hex",
  "instant",
+ "once_cell",
  "rand 0.8.4",
  "reqwest",
  "serde",

ethers-middleware/Cargo.toml

@@ -41,6 +41,7 @@ tokio = { version = "1.5" }
 hex = { version = "0.4.3", default-features = false, features = ["std"] }
 rand = { version = "0.8.4", default-features = false }
 ethers-providers = { version = "^0.5.0", path = "../ethers-providers", default-features = false, features = ["ws", "rustls"] }
+once_cell = "1.8.0"
 
 [target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies]
 tokio = { version = "1.5", default-features = false, features = ["rt", "macros", "time"] }

ethers-middleware/tests/signer.rs

@@ -1,9 +1,22 @@
-use ethers_providers::{Http, Middleware, Provider};
+use ethers_providers::{Http, JsonRpcClient, Middleware, Provider};
 
-use ethers_core::types::TransactionRequest;
+use ethers_core::{
+    types::{BlockNumber, TransactionRequest},
+    utils::parse_units,
+};
 use ethers_middleware::signer::SignerMiddleware;
-use ethers_signers::{LocalWallet, Signer};
-use std::{convert::TryFrom, time::Duration};
+use ethers_signers::{coins_bip39::English, LocalWallet, MnemonicBuilder, Signer};
+use once_cell::sync::Lazy;
+use std::{convert::TryFrom, sync::atomic::AtomicU8, time::Duration};
+
+static WALLETS: Lazy<TestWallets> = Lazy::new(|| {
+    TestWallets {
+        mnemonic: MnemonicBuilder::default()
+            // Please don't drain this :)
+            .phrase("impose air often almost medal sudden finish quote dwarf devote theme layer"),
+        next: Default::default(),
+    }
+});
 
 #[tokio::test]
 #[cfg(not(feature = "celo"))]
@@ -51,19 +64,14 @@ async fn pending_txs_with_confirmations_testnet() {
             .unwrap()
             .interval(Duration::from_millis(3000));
     let chain_id = provider.get_chainid().await.unwrap();
-    let wallet = "59c37cb6b16fa2de30675f034c8008f890f4b2696c729d6267946d29736d73e4"
-        .parse::<LocalWallet>()
-        .unwrap()
-        .with_chain_id(chain_id.as_u64());
+    let wallet = WALLETS.next().with_chain_id(chain_id.as_u64());
     let address = wallet.address();
     let provider = SignerMiddleware::new(provider, wallet);
     generic_pending_txs_test(provider, address).await;
 }
 
 #[cfg(not(feature = "celo"))]
-use ethers_core::types::{
-    transaction::eip2718::TypedTransaction, Address, Eip1559TransactionRequest,
-};
+use ethers_core::types::{Address, Eip1559TransactionRequest};
 
 // different keys to avoid nonce errors
 #[tokio::test]
@@ -75,10 +83,7 @@ async fn websocket_pending_txs_with_confirmations_testnet() {
             .unwrap()
             .interval(Duration::from_millis(3000));
     let chain_id = provider.get_chainid().await.unwrap();
-    let wallet = "ff7f80c6e9941865266ed1f481263d780169f1d98269c51167d20c630a5fdc8a"
-        .parse::<LocalWallet>()
-        .unwrap()
-        .with_chain_id(chain_id.as_u64());
+    let wallet = WALLETS.next().with_chain_id(chain_id.as_u64());
     let address = wallet.address();
     let provider = SignerMiddleware::new(provider, wallet);
     generic_pending_txs_test(provider, address).await;
@@ -89,7 +94,7 @@ async fn generic_pending_txs_test<M: Middleware>(provider: M, who: Address) {
     let tx = TransactionRequest::new().to(who).from(who);
     let pending_tx = provider.send_transaction(tx, None).await.unwrap();
     let tx_hash = *pending_tx;
-    let receipt = pending_tx.confirmations(3).await.unwrap().unwrap();
+    let receipt = pending_tx.confirmations(1).await.unwrap().unwrap();
     // got the correct receipt
     assert_eq!(receipt.transaction_hash, tx_hash);
 }
@@ -102,21 +107,22 @@ async fn typed_txs() {
             .unwrap();
 
     let chain_id = provider.get_chainid().await.unwrap();
-    let wallet = "87203087aed9246e0b2417e248752a1a0df4fdaf65085c11a2b48087ba036b41"
-        .parse::<LocalWallet>()
-        .unwrap()
-        .with_chain_id(chain_id.as_u64());
+    let wallet = WALLETS.next().with_chain_id(chain_id.as_u64());
     let address = wallet.address();
+    // our wallet
     let provider = SignerMiddleware::new(provider, wallet);
 
-    async fn check_tx<M: Middleware>(provider: &M, tx: TypedTransaction, expected: u64) {
-        let receipt = provider
-            .send_transaction(tx, None)
-            .await
-            .unwrap()
-            .await
-            .unwrap()
-            .unwrap();
+    // Uncomment the below and run this test to re-fund the wallets if they get drained.
+    // Would be ideal if we'd have a way to do this automatically, but this should be
+    // happening rarely enough that it doesn't matter.
+    // WALLETS.fund(provider.provider(), 10u32).await;
+
+    async fn check_tx<P: JsonRpcClient + Clone>(
+        pending_tx: ethers_providers::PendingTransaction<'_, P>,
+        expected: u64,
+    ) {
+        let provider = pending_tx.provider();
+        let receipt = pending_tx.await.unwrap().unwrap();
         let tx = provider
             .get_transaction(receipt.transaction_hash)
             .await
@@ -126,21 +132,39 @@ async fn typed_txs() {
         assert_eq!(tx.transaction_type, Some(expected.into()));
     }
 
-    let tx: TypedTransaction = TransactionRequest::new().from(address).to(address).into();
-    check_tx(&provider, tx, 0).await;
-
-    let tx: TypedTransaction = TransactionRequest::new()
+    let mut nonce = provider.get_transaction_count(address, None).await.unwrap();
+    let tx = TransactionRequest::new()
         .from(address)
         .to(address)
-        .with_access_list(vec![])
-        .into();
-    check_tx(&provider, tx, 1).await;
+        .nonce(nonce);
+    nonce += 1.into();
+    let tx1 = provider
+        .send_transaction(tx.clone(), Some(BlockNumber::Pending.into()))
+        .await
+        .unwrap();
 
-    let tx: TypedTransaction = Eip1559TransactionRequest::new()
+    let tx = tx
+        .clone()
+        .nonce(nonce)
         .from(address)
         .to(address)
-        .into();
-    check_tx(&provider, tx, 2).await;
+        .with_access_list(vec![]);
+    nonce += 1.into();
+    let tx2 = provider
+        .send_transaction(tx, Some(BlockNumber::Pending.into()))
+        .await
+        .unwrap();
+
+    let tx = Eip1559TransactionRequest::new()
+        .from(address)
+        .to(address)
+        .nonce(nonce);
+    let tx3 = provider
+        .send_transaction(tx, Some(BlockNumber::Pending.into()))
+        .await
+        .unwrap();
+
+    futures_util::join!(check_tx(tx1, 0), check_tx(tx2, 1), check_tx(tx3, 2),);
 }
 
 #[tokio::test]
@@ -276,7 +300,6 @@ async fn deploy_and_call_contract() {
         .parse::<LocalWallet>()
         .unwrap()
         .with_chain_id(chain_id);
-
     let client = SignerMiddleware::new(provider, wallet);
     let client = Arc::new(client);
 
@@ -300,3 +323,62 @@ async fn deploy_and_call_contract() {
     let value: U256 = contract.method("value", ()).unwrap().call().await.unwrap();
     assert_eq!(value, 1.into());
 }
+
+#[derive(Debug, Default)]
+struct TestWallets {
+    mnemonic: MnemonicBuilder<English>,
+    next: AtomicU8,
+}
+
+impl TestWallets {
+    /// Helper for funding the wallets with an instantiated provider
+    #[allow(unused)]
+    pub async fn fund<T: JsonRpcClient, U: Into<u32>>(&self, provider: &Provider<T>, n: U) {
+        let addrs = (0..n.into())
+            .map(|i| self.get(i).address())
+            .collect::<Vec<_>>();
+        // hardcoded funder address private key, rinkeby
+        let signer = "39aa18eeb5d12c071e5f19d8e9375a872e90cb1f2fa640384ffd8800a2f3e8f1"
+            .parse::<LocalWallet>()
+            .unwrap()
+            .with_chain_id(provider.get_chainid().await.unwrap().as_u64());
+        let provider = SignerMiddleware::new(provider, signer);
+        let addr = provider.address();
+
+        let mut nonce = provider.get_transaction_count(addr, None).await.unwrap();
+        let mut pending_txs = Vec::new();
+        for addr in addrs {
+            println!("Funding wallet {:?}", addr);
+            let tx = TransactionRequest::new()
+                .nonce(nonce)
+                .to(addr)
+                // 0.1 eth per wallet
+                .value(parse_units("1", 18).unwrap());
+            pending_txs.push(
+                provider
+                    .send_transaction(tx, Some(BlockNumber::Pending.into()))
+                    .await
+                    .unwrap(),
+            );
+            nonce += 1.into();
+        }
+
+        futures_util::future::join_all(pending_txs).await;
+    }
+
+    pub fn next(&self) -> LocalWallet {
+        let idx = self.next.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+        let wallet = self.get(idx);
+        // println!("Got wallet {:?}", wallet.address());
+        wallet
+    }
+
+    pub fn get<T: Into<u32>>(&self, idx: T) -> LocalWallet {
+        self.mnemonic
+            .clone()
+            .index(idx)
+            .expect("index not found")
+            .build()
+            .expect("cannot build wallet")
+    }
+}

ethers-providers/src/pending_transaction.rs

@@ -48,6 +48,14 @@ impl<'a, P: JsonRpcClient> PendingTransaction<'a, P> {
         }
     }
 
+    /// Returns the Provider associated with the pending transaction
+    pub fn provider(&self) -> Provider<P>
+    where
+        P: Clone,
+    {
+        self.provider.clone()
+    }
+
     /// Sets the number of confirmations for the pending transaction to resolve
     /// to a receipt
     pub fn confirmations(mut self, confs: usize) -> Self {