Fallback nodes for eth1 access (#1918)

## Issue Addressed

part of  #1883

## Proposed Changes

Adds a new cli argument `--eth1-endpoints` that can be used instead of `--eth1-endpoint` to specify a comma-separated list of endpoints. If the first endpoint returns an error for some request the other endpoints are tried in the given order.

## Additional Info

Currently if the first endpoint fails the fallbacks are used silently (except for `try_fallback_test_endpoint` that is used in `do_update` which logs a `WARN` for each endpoint that is not reachable). A question is if we should add more logs so that the user gets warned if his main endpoint is for example just slow and sometimes hits timeouts.

beacon_node/eth1/tests/test.rs

@@ -1,8 +1,8 @@
 #![cfg(test)]
 use environment::{Environment, EnvironmentBuilder};
 use eth1::http::{get_deposit_count, get_deposit_logs_in_range, get_deposit_root, Block, Log};
-use eth1::DepositCache;
 use eth1::{Config, Service};
+use eth1::{DepositCache, DEFAULT_CHAIN_ID, DEFAULT_NETWORK_ID};
 use eth1_test_rig::GanacheEth1Instance;
 use futures::compat::Future01CompatExt;
 use merkle_proof::verify_merkle_proof;
@@ -97,6 +97,10 @@ async fn get_block_number(web3: &Web3<Http>) -> u64 {
         .expect("should get block number")
 }
 
+async fn new_ganache_instance() -> Result<GanacheEth1Instance, String> {
+    GanacheEth1Instance::new(DEFAULT_NETWORK_ID.into(), DEFAULT_CHAIN_ID.into()).await
+}
+
 mod eth1_cache {
     use super::*;
     use types::{EthSpec, MainnetEthSpec};
@@ -106,7 +110,7 @@ mod eth1_cache {
         let log = null_logger();
 
         for follow_distance in 0..2 {
-            let eth1 = GanacheEth1Instance::new()
+            let eth1 = new_ganache_instance()
                 .await
                 .expect("should start eth1 environment");
             let deposit_contract = &eth1.deposit_contract;
@@ -116,7 +120,7 @@ mod eth1_cache {
 
             let service = Service::new(
                 Config {
-                    endpoint: eth1.endpoint(),
+                    endpoints: vec![eth1.endpoint()],
                     deposit_contract_address: deposit_contract.address(),
                     lowest_cached_block_number: initial_block_number,
                     follow_distance,
@@ -145,17 +149,19 @@ mod eth1_cache {
                     eth1.ganache.evm_mine().await.expect("should mine block");
                 }
 
+                let endpoints = service.init_endpoints();
+
                 service
-                    .update_deposit_cache(None)
+                    .update_deposit_cache(None, &endpoints)
                     .await
                     .expect("should update deposit cache");
                 service
-                    .update_block_cache(None)
+                    .update_block_cache(None, &endpoints)
                     .await
                     .expect("should update block cache");
 
                 service
-                    .update_block_cache(None)
+                    .update_block_cache(None, &endpoints)
                     .await
                     .expect("should update cache when nothing has changed");
 
@@ -181,7 +187,7 @@ mod eth1_cache {
     async fn big_skip() {
         let log = null_logger();
 
-        let eth1 = GanacheEth1Instance::new()
+        let eth1 = new_ganache_instance()
             .await
             .expect("should start eth1 environment");
         let deposit_contract = &eth1.deposit_contract;
@@ -191,7 +197,7 @@ mod eth1_cache {
 
         let service = Service::new(
             Config {
-                endpoint: eth1.endpoint(),
+                endpoints: vec![eth1.endpoint()],
                 deposit_contract_address: deposit_contract.address(),
                 lowest_cached_block_number: get_block_number(&web3).await,
                 follow_distance: 0,
@@ -208,12 +214,14 @@ mod eth1_cache {
             eth1.ganache.evm_mine().await.expect("should mine block")
         }
 
+        let endpoints = service.init_endpoints();
+
         service
-            .update_deposit_cache(None)
+            .update_deposit_cache(None, &endpoints)
             .await
             .expect("should update deposit cache");
         service
-            .update_block_cache(None)
+            .update_block_cache(None, &endpoints)
             .await
             .expect("should update block cache");
 
@@ -230,7 +238,7 @@ mod eth1_cache {
     async fn pruning() {
         let log = null_logger();
 
-        let eth1 = GanacheEth1Instance::new()
+        let eth1 = new_ganache_instance()
             .await
             .expect("should start eth1 environment");
         let deposit_contract = &eth1.deposit_contract;
@@ -240,7 +248,7 @@ mod eth1_cache {
 
         let service = Service::new(
             Config {
-                endpoint: eth1.endpoint(),
+                endpoints: vec![eth1.endpoint()],
                 deposit_contract_address: deposit_contract.address(),
                 lowest_cached_block_number: get_block_number(&web3).await,
                 follow_distance: 0,
@@ -255,12 +263,13 @@ mod eth1_cache {
             for _ in 0..cache_len / 2 {
                 eth1.ganache.evm_mine().await.expect("should mine block")
             }
+            let endpoints = service.init_endpoints();
             service
-                .update_deposit_cache(None)
+                .update_deposit_cache(None, &endpoints)
                 .await
                 .expect("should update deposit cache");
             service
-                .update_block_cache(None)
+                .update_block_cache(None, &endpoints)
                 .await
                 .expect("should update block cache");
         }
@@ -278,7 +287,7 @@ mod eth1_cache {
 
         let n = 16;
 
-        let eth1 = GanacheEth1Instance::new()
+        let eth1 = new_ganache_instance()
             .await
             .expect("should start eth1 environment");
         let deposit_contract = &eth1.deposit_contract;
@@ -286,7 +295,7 @@ mod eth1_cache {
 
         let service = Service::new(
             Config {
-                endpoint: eth1.endpoint(),
+                endpoints: vec![eth1.endpoint()],
                 deposit_contract_address: deposit_contract.address(),
                 lowest_cached_block_number: get_block_number(&web3).await,
                 follow_distance: 0,
@@ -299,14 +308,16 @@ mod eth1_cache {
         for _ in 0..n {
             eth1.ganache.evm_mine().await.expect("should mine block")
         }
+
+        let endpoints = service.init_endpoints();
         futures::try_join!(
-            service.update_deposit_cache(None),
-            service.update_deposit_cache(None)
+            service.update_deposit_cache(None, &endpoints),
+            service.update_deposit_cache(None, &endpoints)
         )
         .expect("should perform two simultaneous updates of deposit cache");
         futures::try_join!(
-            service.update_block_cache(None),
-            service.update_block_cache(None)
+            service.update_block_cache(None, &endpoints),
+            service.update_block_cache(None, &endpoints)
         )
         .expect("should perform two simultaneous updates of block cache");
 
@@ -323,7 +334,7 @@ mod deposit_tree {
 
         let n = 4;
 
-        let eth1 = GanacheEth1Instance::new()
+        let eth1 = new_ganache_instance()
             .await
             .expect("should start eth1 environment");
         let deposit_contract = &eth1.deposit_contract;
@@ -333,7 +344,7 @@ mod deposit_tree {
 
         let service = Service::new(
             Config {
-                endpoint: eth1.endpoint(),
+                endpoints: vec![eth1.endpoint()],
                 deposit_contract_address: deposit_contract.address(),
                 deposit_contract_deploy_block: start_block,
                 follow_distance: 0,
@@ -353,13 +364,15 @@ mod deposit_tree {
                     .expect("should perform a deposit");
             }
 
+            let endpoints = service.init_endpoints();
+
             service
-                .update_deposit_cache(None)
+                .update_deposit_cache(None, &endpoints)
                 .await
                 .expect("should perform update");
 
             service
-                .update_deposit_cache(None)
+                .update_deposit_cache(None, &endpoints)
                 .await
                 .expect("should perform update when nothing has changed");
 
@@ -398,7 +411,7 @@ mod deposit_tree {
 
         let n = 8;
 
-        let eth1 = GanacheEth1Instance::new()
+        let eth1 = new_ganache_instance()
             .await
             .expect("should start eth1 environment");
         let deposit_contract = &eth1.deposit_contract;
@@ -408,7 +421,7 @@ mod deposit_tree {
 
         let service = Service::new(
             Config {
-                endpoint: eth1.endpoint(),
+                endpoints: vec![eth1.endpoint()],
                 deposit_contract_address: deposit_contract.address(),
                 deposit_contract_deploy_block: start_block,
                 lowest_cached_block_number: start_block,
@@ -428,9 +441,10 @@ mod deposit_tree {
                 .expect("should perform a deposit");
         }
 
+        let endpoints = service.init_endpoints();
         futures::try_join!(
-            service.update_deposit_cache(None),
-            service.update_deposit_cache(None)
+            service.update_deposit_cache(None, &endpoints),
+            service.update_deposit_cache(None, &endpoints)
         )
         .expect("should perform two updates concurrently");
 
@@ -445,7 +459,7 @@ mod deposit_tree {
 
         let deposits: Vec<_> = (0..n).map(|_| random_deposit_data()).collect();
 
-        let eth1 = GanacheEth1Instance::new()
+        let eth1 = new_ganache_instance()
             .await
             .expect("should start eth1 environment");
         let deposit_contract = &eth1.deposit_contract;
@@ -552,7 +566,7 @@ mod http {
 
     #[tokio::test]
     async fn incrementing_deposits() {
-        let eth1 = GanacheEth1Instance::new()
+        let eth1 = new_ganache_instance()
             .await
             .expect("should start eth1 environment");
         let deposit_contract = &eth1.deposit_contract;
@@ -644,7 +658,7 @@ mod fast {
     async fn deposit_cache_query() {
         let log = null_logger();
 
-        let eth1 = GanacheEth1Instance::new()
+        let eth1 = new_ganache_instance()
             .await
             .expect("should start eth1 environment");
         let deposit_contract = &eth1.deposit_contract;
@@ -653,7 +667,7 @@ mod fast {
         let now = get_block_number(&web3).await;
         let service = Service::new(
             Config {
-                endpoint: eth1.endpoint(),
+                endpoints: vec![eth1.endpoint()],
                 deposit_contract_address: deposit_contract.address(),
                 deposit_contract_deploy_block: now,
                 lowest_cached_block_number: now,
@@ -675,8 +689,9 @@ mod fast {
             eth1.ganache.evm_mine().await.expect("should mine block");
         }
 
+        let endpoints = service.init_endpoints();
         service
-            .update_deposit_cache(None)
+            .update_deposit_cache(None, &endpoints)
             .await
             .expect("should perform update");
 
@@ -717,7 +732,7 @@ mod persist {
     async fn test_persist_caches() {
         let log = null_logger();
 
-        let eth1 = GanacheEth1Instance::new()
+        let eth1 = new_ganache_instance()
             .await
             .expect("should start eth1 environment");
         let deposit_contract = &eth1.deposit_contract;
@@ -725,7 +740,7 @@ mod persist {
 
         let now = get_block_number(&web3).await;
         let config = Config {
-            endpoint: eth1.endpoint(),
+            endpoints: vec![eth1.endpoint()],
             deposit_contract_address: deposit_contract.address(),
             deposit_contract_deploy_block: now,
             lowest_cached_block_number: now,
@@ -743,8 +758,9 @@ mod persist {
                 .expect("should perform a deposit");
         }
 
+        let endpoints = service.init_endpoints();
         service
-            .update_deposit_cache(None)
+            .update_deposit_cache(None, &endpoints)
             .await
             .expect("should perform update");
 
@@ -756,7 +772,7 @@ mod persist {
         let deposit_count = service.deposit_cache_len();
 
         service
-            .update_block_cache(None)
+            .update_block_cache(None, &endpoints)
             .await
             .expect("should perform update");
 
@@ -786,3 +802,273 @@ mod persist {
         );
     }
 }
+
+/// Tests for eth1 fallback
+mod fallbacks {
+    use super::*;
+    use tokio::time::delay_for;
+
+    #[tokio::test]
+    async fn test_fallback_when_offline() {
+        let log = null_logger();
+        let endpoint2 = new_ganache_instance()
+            .await
+            .expect("should start eth1 environment");
+        let deposit_contract = &endpoint2.deposit_contract;
+
+        let initial_block_number = get_block_number(&endpoint2.web3()).await;
+
+        // Create some blocks and then consume them, performing the test `rounds` times.
+        let new_blocks = 4;
+
+        for _ in 0..new_blocks {
+            endpoint2
+                .ganache
+                .evm_mine()
+                .await
+                .expect("should mine block");
+        }
+
+        let endpoint1 = endpoint2
+            .ganache
+            .fork()
+            .expect("should start eth1 environment");
+
+        //mine additional blocks on top of the original endpoint
+        for _ in 0..new_blocks {
+            endpoint2
+                .ganache
+                .evm_mine()
+                .await
+                .expect("should mine block");
+        }
+
+        let service = Service::new(
+            Config {
+                endpoints: vec![endpoint1.endpoint(), endpoint2.endpoint()],
+                deposit_contract_address: deposit_contract.address(),
+                lowest_cached_block_number: initial_block_number,
+                follow_distance: 0,
+                ..Config::default()
+            },
+            log.clone(),
+            MainnetEthSpec::default_spec(),
+        );
+
+        let endpoint1_block_number = get_block_number(&endpoint1.web3).await;
+        //the first call will only query endpoint1
+        service.update().await.expect("should update deposit cache");
+        assert_eq!(
+            service.deposits().read().last_processed_block.unwrap(),
+            endpoint1_block_number
+        );
+
+        drop(endpoint1);
+
+        let endpoint2_block_number = get_block_number(&endpoint2.web3()).await;
+        assert!(endpoint1_block_number < endpoint2_block_number);
+        //endpoint1 is offline => query will import blocks from endpoint2
+        service.update().await.expect("should update deposit cache");
+        assert_eq!(
+            service.deposits().read().last_processed_block.unwrap(),
+            endpoint2_block_number
+        );
+    }
+
+    #[tokio::test]
+    async fn test_fallback_when_wrong_network_id() {
+        let log = null_logger();
+        let correct_network_id: u64 = DEFAULT_NETWORK_ID.into();
+        let wrong_network_id = correct_network_id + 1;
+        let endpoint1 = GanacheEth1Instance::new(wrong_network_id, DEFAULT_CHAIN_ID.into())
+            .await
+            .expect("should start eth1 environment");
+        let endpoint2 = new_ganache_instance()
+            .await
+            .expect("should start eth1 environment");
+        let deposit_contract = &endpoint2.deposit_contract;
+
+        let initial_block_number = get_block_number(&endpoint2.web3()).await;
+
+        // Create some blocks and then consume them, performing the test `rounds` times.
+        let new_blocks = 4;
+
+        for _ in 0..new_blocks {
+            endpoint1
+                .ganache
+                .evm_mine()
+                .await
+                .expect("should mine block");
+            endpoint2
+                .ganache
+                .evm_mine()
+                .await
+                .expect("should mine block");
+        }
+
+        //additional blocks for endpoint1 to be able to distinguish
+        for _ in 0..new_blocks {
+            endpoint1
+                .ganache
+                .evm_mine()
+                .await
+                .expect("should mine block");
+        }
+
+        let service = Service::new(
+            Config {
+                endpoints: vec![endpoint2.endpoint(), endpoint1.endpoint()],
+                deposit_contract_address: deposit_contract.address(),
+                lowest_cached_block_number: initial_block_number,
+                follow_distance: 0,
+                ..Config::default()
+            },
+            log.clone(),
+            MainnetEthSpec::default_spec(),
+        );
+
+        let endpoint1_block_number = get_block_number(&endpoint1.web3()).await;
+        let endpoint2_block_number = get_block_number(&endpoint2.web3()).await;
+        assert!(endpoint2_block_number < endpoint1_block_number);
+        //the call will fallback to endpoint2
+        service.update().await.expect("should update deposit cache");
+        assert_eq!(
+            service.deposits().read().last_processed_block.unwrap(),
+            endpoint2_block_number
+        );
+    }
+
+    #[tokio::test]
+    async fn test_fallback_when_wrong_chain_id() {
+        let log = null_logger();
+        let correct_chain_id: u64 = DEFAULT_CHAIN_ID.into();
+        let wrong_chain_id = correct_chain_id + 1;
+        let endpoint1 = GanacheEth1Instance::new(DEFAULT_NETWORK_ID.into(), wrong_chain_id)
+            .await
+            .expect("should start eth1 environment");
+        let endpoint2 = new_ganache_instance()
+            .await
+            .expect("should start eth1 environment");
+        let deposit_contract = &endpoint2.deposit_contract;
+
+        let initial_block_number = get_block_number(&endpoint2.web3()).await;
+
+        // Create some blocks and then consume them, performing the test `rounds` times.
+        let new_blocks = 4;
+
+        for _ in 0..new_blocks {
+            endpoint1
+                .ganache
+                .evm_mine()
+                .await
+                .expect("should mine block");
+            endpoint2
+                .ganache
+                .evm_mine()
+                .await
+                .expect("should mine block");
+        }
+
+        //additional blocks for endpoint1 to be able to distinguish
+        for _ in 0..new_blocks {
+            endpoint1
+                .ganache
+                .evm_mine()
+                .await
+                .expect("should mine block");
+        }
+
+        let service = Service::new(
+            Config {
+                endpoints: vec![endpoint2.endpoint(), endpoint1.endpoint()],
+                deposit_contract_address: deposit_contract.address(),
+                lowest_cached_block_number: initial_block_number,
+                follow_distance: 0,
+                ..Config::default()
+            },
+            log.clone(),
+            MainnetEthSpec::default_spec(),
+        );
+
+        let endpoint1_block_number = get_block_number(&endpoint1.web3()).await;
+        let endpoint2_block_number = get_block_number(&endpoint2.web3()).await;
+        assert!(endpoint2_block_number < endpoint1_block_number);
+        //the call will fallback to endpoint2
+        service.update().await.expect("should update deposit cache");
+        assert_eq!(
+            service.deposits().read().last_processed_block.unwrap(),
+            endpoint2_block_number
+        );
+    }
+
+    #[tokio::test]
+    async fn test_fallback_when_node_far_behind() {
+        let log = null_logger();
+        let endpoint2 = new_ganache_instance()
+            .await
+            .expect("should start eth1 environment");
+        let deposit_contract = &endpoint2.deposit_contract;
+
+        let initial_block_number = get_block_number(&endpoint2.web3()).await;
+
+        // Create some blocks and then consume them, performing the test `rounds` times.
+        let new_blocks = 4;
+
+        for _ in 0..new_blocks {
+            endpoint2
+                .ganache
+                .evm_mine()
+                .await
+                .expect("should mine block");
+        }
+
+        let endpoint1 = endpoint2
+            .ganache
+            .fork()
+            .expect("should start eth1 environment");
+
+        let service = Service::new(
+            Config {
+                endpoints: vec![endpoint1.endpoint(), endpoint2.endpoint()],
+                deposit_contract_address: deposit_contract.address(),
+                lowest_cached_block_number: initial_block_number,
+                follow_distance: 0,
+                node_far_behind_seconds: 5,
+                ..Config::default()
+            },
+            log.clone(),
+            MainnetEthSpec::default_spec(),
+        );
+
+        let endpoint1_block_number = get_block_number(&endpoint1.web3).await;
+        //the first call will only query endpoint1
+        service.update().await.expect("should update deposit cache");
+        assert_eq!(
+            service.deposits().read().last_processed_block.unwrap(),
+            endpoint1_block_number
+        );
+
+        delay_for(Duration::from_secs(7)).await;
+
+        //both endpoints don't have recent blocks => should return error
+        assert!(service.update().await.is_err());
+
+        //produce some new blocks on endpoint2
+        for _ in 0..new_blocks {
+            endpoint2
+                .ganache
+                .evm_mine()
+                .await
+                .expect("should mine block");
+        }
+
+        let endpoint2_block_number = get_block_number(&endpoint2.web3()).await;
+
+        //endpoint1 is far behind + endpoint2 not => update will import blocks from endpoint2
+        service.update().await.expect("should update deposit cache");
+        assert_eq!(
+            service.deposits().read().last_processed_block.unwrap(),
+            endpoint2_block_number
+        );
+    }
+}