Port eth1_test_rig to stable futures

tests/eth1_test_rig/Cargo.toml

@@ -6,8 +6,8 @@ edition = "2018"
 
 [dependencies]
 web3 = "0.8.0"
-tokio = "0.1.22"
+tokio = { version = "0.2", features = ["time"] }
-futures = "0.1.25"
+futures = { version = "0.3", features = ["compat"] }
 types = { path = "../../eth2/types"}
 serde_json = "1.0"
 deposit_contract = { path = "../../eth2/utils/deposit_contract"}

tests/eth1_test_rig/src/ganache.rs

@@ -1,4 +1,4 @@
-use futures::Future;
+use futures::compat::Future01CompatExt;
 use serde_json::json;
 use std::io::prelude::*;
 use std::io::BufReader;
@@ -98,28 +98,34 @@ impl GanacheInstance {
     }
 
     /// Increase the timestamp on future blocks by `increase_by` seconds.
-    pub fn increase_time(&self, increase_by: u64) -> impl Future<Item = (), Error = String> {
+    pub async fn increase_time(&self, increase_by: u64) -> Result<(), String> {
         self.web3
             .transport()
             .execute("evm_increaseTime", vec![json!(increase_by)])
+            .compat()
+            .await
             .map(|_json_value| ())
             .map_err(|e| format!("Failed to increase time on EVM (is this ganache?): {:?}", e))
     }
 
     /// Returns the current block number, as u64
-    pub fn block_number(&self) -> impl Future<Item = u64, Error = String> {
+    pub async fn block_number(&self) -> Result<u64, String> {
         self.web3
             .eth()
             .block_number()
+            .compat()
+            .await
             .map(|v| v.as_u64())
             .map_err(|e| format!("Failed to get block number: {:?}", e))
     }
 
     /// Mines a single block.
-    pub fn evm_mine(&self) -> impl Future<Item = (), Error = String> {
+    pub async fn evm_mine(&self) -> Result<(), String> {
         self.web3
             .transport()
             .execute("evm_mine", vec![])
+            .compat()
+            .await
             .map(|_| ())
             .map_err(|_| {
                 "utils should mine new block with evm_mine (only works with ganache-cli!)"

tests/eth1_test_rig/src/lib.rs

@@ -10,10 +10,10 @@ mod ganache;
 use deposit_contract::{
     encode_eth1_tx_data, testnet, ABI, BYTECODE, CONTRACT_DEPLOY_GAS, DEPOSIT_GAS,
 };
-use futures::{future, stream, Future, IntoFuture, Stream};
+use futures::compat::Future01CompatExt;
 use ganache::GanacheInstance;
-use std::time::{Duration, Instant};
+use std::time::Duration;
-use tokio::{runtime::Runtime, timer::Delay};
+use tokio::time::delay_for;
 use types::DepositData;
 use types::{test_utils::generate_deterministic_keypair, EthSpec, Hash256, Keypair, Signature};
 use web3::contract::{Contract, Options};
@@ -31,13 +31,14 @@ pub struct GanacheEth1Instance {
 }
 
 impl GanacheEth1Instance {
-    pub fn new() -> impl Future<Item = Self, Error = String> {
+    pub async fn new() -> Result<Self, String> {
-        GanacheInstance::new().into_future().and_then(|ganache| {
+        let ganache = GanacheInstance::new()?;
-            DepositContract::deploy(ganache.web3.clone(), 0, None).map(|deposit_contract| Self {
+        DepositContract::deploy(ganache.web3.clone(), 0, None)
+            .await
+            .map(|deposit_contract| Self {
                 ganache,
                 deposit_contract,
             })
-        })
     }
 
     pub fn endpoint(&self) -> String {
@@ -57,19 +58,19 @@ pub struct DepositContract {
 }
 
 impl DepositContract {
-    pub fn deploy(
+    pub async fn deploy(
         web3: Web3<Http>,
         confirmations: usize,
         password: Option<String>,
-    ) -> impl Future<Item = Self, Error = String> {
+    ) -> Result<Self, String> {
-        Self::deploy_bytecode(web3, confirmations, BYTECODE, ABI, password)
+        Self::deploy_bytecode(web3, confirmations, BYTECODE, ABI, password).await
     }
 
-    pub fn deploy_testnet(
+    pub async fn deploy_testnet(
         web3: Web3<Http>,
         confirmations: usize,
         password: Option<String>,
-    ) -> impl Future<Item = Self, Error = String> {
+    ) -> Result<Self, String> {
         Self::deploy_bytecode(
             web3,
             confirmations,
@@ -77,35 +78,33 @@ impl DepositContract {
             testnet::ABI,
             password,
         )
+        .await
     }
 
-    fn deploy_bytecode(
+    async fn deploy_bytecode(
         web3: Web3<Http>,
         confirmations: usize,
         bytecode: &[u8],
         abi: &[u8],
         password: Option<String>,
-    ) -> impl Future<Item = Self, Error = String> {
+    ) -> Result<Self, String> {
-        let web3_1 = web3.clone();
+        let address = deploy_deposit_contract(
-
-        deploy_deposit_contract(
             web3.clone(),
             confirmations,
             bytecode.to_vec(),
             abi.to_vec(),
             password,
         )
+        .await
         .map_err(|e| {
             format!(
                 "Failed to deploy contract: {}. Is scripts/ganache_tests_node.sh running?.",
                 e
             )
-        })
+        })?;
-        .and_then(move |address| {
+        Contract::from_json(web3.clone().eth(), address, ABI)
-            Contract::from_json(web3_1.eth(), address, ABI)
             .map_err(|e| format!("Failed to init contract: {:?}", e))
-        })
+            .map(move |contract| Self { contract, web3 })
-        .map(|contract| Self { contract, web3 })
     }
 
     /// The deposit contract's address in `0x00ab...` format.
@@ -136,7 +135,7 @@ impl DepositContract {
     /// Creates a random, valid deposit and submits it to the deposit contract.
     ///
     /// The keypairs are created randomly and destroyed.
-    pub fn deposit_random<E: EthSpec>(&self, runtime: &mut Runtime) -> Result<(), String> {
+    pub async fn deposit_random<E: EthSpec>(&self) -> Result<(), String> {
         let keypair = Keypair::random();
 
         let mut deposit = DepositData {
@@ -148,21 +147,21 @@ impl DepositContract {
 
         deposit.signature = deposit.create_signature(&keypair.sk, &E::default_spec());
 
-        self.deposit(runtime, deposit)
+        self.deposit(deposit).await
     }
 
     /// Perfoms a blocking deposit.
-    pub fn deposit(&self, runtime: &mut Runtime, deposit_data: DepositData) -> Result<(), String> {
+    pub async fn deposit(&self, deposit_data: DepositData) -> Result<(), String> {
-        runtime
+        self.deposit_async(deposit_data)
-            .block_on(self.deposit_async(deposit_data))
+            .await
             .map_err(|e| format!("Deposit failed: {:?}", e))
     }
 
-    pub fn deposit_deterministic_async<E: EthSpec>(
+    pub async fn deposit_deterministic_async<E: EthSpec>(
         &self,
         keypair_index: usize,
         amount: u64,
-    ) -> impl Future<Item = (), Error = String> {
+    ) -> Result<(), String> {
         let keypair = generate_deterministic_keypair(keypair_index);
 
         let mut deposit = DepositData {
@@ -174,20 +173,17 @@ impl DepositContract {
 
         deposit.signature = deposit.create_signature(&keypair.sk, &E::default_spec());
 
-        self.deposit_async(deposit)
+        self.deposit_async(deposit).await
     }
 
     /// Performs a non-blocking deposit.
-    pub fn deposit_async(
+    pub async fn deposit_async(&self, deposit_data: DepositData) -> Result<(), String> {
-        &self,
+        let from = self
-        deposit_data: DepositData,
+            .web3
-    ) -> impl Future<Item = (), Error = String> {
-        let contract = self.contract.clone();
-        let web3_1 = self.web3.clone();
-
-        self.web3
             .eth()
             .accounts()
+            .compat()
+            .await
             .map_err(|e| format!("Failed to get accounts: {:?}", e))
             .and_then(|accounts| {
                 accounts
@@ -216,31 +212,18 @@ impl DepositContract {
                 web3_1
                     .eth()
                     .send_transaction(tx_request)
-                    .map_err(|e| format!("Failed to call deposit fn: {:?}", e))
+                    .compa().await()
-            })
+                    .map_err(|e| format!("Failed to call deposit fn: {:?}", e))?;
-            .map(|_| ())
+                Ok())
     }
 
     /// Peforms many deposits, each preceded by a delay.
-    pub fn deposit_multiple(
+    pub async fn deposit_multiple(&self, deposits: Vec<DelayThenDeposit>) -> Result<(), String> {
-        &self,
+        for deposit in deposits.into_iter() {
-        deposits: Vec<DelayThenDeposit>,
+            delay_for(deposit.delay).await;
-    ) -> impl Future<Item = (), Error = String> {
+            self.deposit_async(deposit.deposit).await?;
-        let s = self.clone();
-        stream::unfold(deposits.into_iter(), move |mut deposit_iter| {
-            let s = s.clone();
-            match deposit_iter.next() {
-                Some(deposit) => Some(
-                    Delay::new(Instant::now() + deposit.delay)
-                        .map_err(|e| format!("Failed to execute delay: {:?}", e))
-                        .and_then(move |_| s.deposit_async(deposit.deposit))
-                        .map(move |yielded| (yielded, deposit_iter)),
-                ),
-                None => None,
         }
-        })
+        Ok(())
-        .collect()
-        .map(|_| ())
     }
 }
 
@@ -260,49 +243,44 @@ fn from_gwei(gwei: u64) -> U256 {
 
 /// Deploys the deposit contract to the given web3 instance using the account with index
 /// `DEPLOYER_ACCOUNTS_INDEX`.
-fn deploy_deposit_contract(
+async fn deploy_deposit_contract(
     web3: Web3<Http>,
     confirmations: usize,
     bytecode: Vec<u8>,
     abi: Vec<u8>,
     password_opt: Option<String>,
-) -> impl Future<Item = Address, Error = String> {
+) -> Result<Address, String> {
     let bytecode = String::from_utf8(bytecode).expect("bytecode must be valid utf8");
-    let web3_1 = web3.clone();
 
-    web3.eth()
+    let from_address = web3
+        .eth()
         .accounts()
+        .compat()
+        .await
         .map_err(|e| format!("Failed to get accounts: {:?}", e))
         .and_then(|accounts| {
             accounts
                 .get(DEPLOYER_ACCOUNTS_INDEX)
                 .cloned()
                 .ok_or_else(|| "Insufficient accounts for deployer".to_string())
-        })
+        })?;
-        .and_then(move |from_address| {
-            let future: Box<dyn Future<Item = Address, Error = String> + Send> =
-                if let Some(password) = password_opt {
-                    // Unlock for only a single transaction.
-                    let duration = None;
 
-                    let future = web3_1
+    let deploy_address = if let Some(password) = password_opt {
+        let result = web3
             .personal()
-                        .unlock_account(from_address, &password, duration)
+            .unlock_account(from_address, &password, None)
-                        .then(move |result| match result {
+            .compat()
-                            Ok(true) => Ok(from_address),
+            .await;
-                            Ok(false) => Err("Eth1 node refused to unlock account".to_string()),
+        match result {
-                            Err(e) => Err(format!("Eth1 unlock request failed: {:?}", e)),
+            Ok(true) => return Ok(from_address),
-                        });
+            Ok(false) => return Err("Eth1 node refused to unlock account".to_string()),
-
+            Err(e) => return Err(format!("Eth1 unlock request failed: {:?}", e)),
-                    Box::new(future)
+        };
     } else {
-                    Box::new(future::ok(from_address))
+        from_address
     };
 
-            future
+    let pending_contract = Contract::deploy(web3.eth(), &abi)
-        })
-        .and_then(move |deploy_address| {
-            Contract::deploy(web3.eth(), &abi)
         .map_err(|e| format!("Unable to build contract deployer: {:?}", e))?
         .confirmations(confirmations)
         .options(Options {
@@ -310,11 +288,11 @@ fn deploy_deposit_contract(
             ..Options::default()
         })
         .execute(bytecode, (), deploy_address)
-                .map_err(|e| format!("Failed to execute deployment: {:?}", e))
+        .map_err(|e| format!("Failed to execute deployment: {:?}", e))?;
-        })
+
-        .and_then(|pending_contract| {
     pending_contract
+        .compat()
+        .await
         .map(|contract| contract.address())
         .map_err(|e| format!("Unable to resolve pending contract: {:?}", e))
-        })
 }