Merge branch 'nfd-2' into fc-refactor

beacon_node/beacon_chain/src/snapshot_cache.rs

@@ -254,12 +254,11 @@ impl<T: EthSpec> SnapshotCache<T> {
             .position(|snapshot| snapshot.beacon_block_root == block_root)
             .map(|i| {
                 if let Some(cache) = self.snapshots.get(i) {
-                    if block_slot > cache.beacon_block.slot() + 1 {
-                        return (cache.clone_as_pre_state(), true);
-                    }
+                    // Avoid cloning the block during sync (when the `block_delay` is `None`).
                     if let Some(delay) = block_delay {
                         if delay >= MINIMUM_BLOCK_DELAY_FOR_CLONE
                             && delay <= Duration::from_secs(spec.seconds_per_slot) * 4
+                            || block_slot > cache.beacon_block.slot() + 1
                         {
                             return (cache.clone_as_pre_state(), true);
                         }

beacon_node/beacon_chain/src/validator_monitor.rs

@@ -231,6 +231,11 @@ impl MonitoredValidator {
             }
         }
     }
+
+    /// Ensure epoch summary is added to the summaries map
+    fn touch_epoch_summary(&self, epoch: Epoch) {
+        self.with_epoch_summary(epoch, |_| {});
+    }
 }
 
 /// Holds a collection of `MonitoredValidator` and is notified about a variety of events on the P2P
@@ -306,6 +311,7 @@ impl<T: EthSpec> ValidatorMonitor<T> {
         // Update metrics for individual validators.
         for monitored_validator in self.validators.values() {
             if let Some(i) = monitored_validator.index {
+                monitored_validator.touch_epoch_summary(current_epoch);
                 let i = i as usize;
                 let id = &monitored_validator.id;
 

beacon_node/client/src/builder.rs

@@ -276,6 +276,8 @@ where
                     BeaconNodeHttpClient::new(url, Timeouts::set_all(CHECKPOINT_SYNC_HTTP_TIMEOUT));
                 let slots_per_epoch = TEthSpec::slots_per_epoch();
 
+                debug!(context.log(), "Downloading finalized block");
+
                 // Find a suitable finalized block on an epoch boundary.
                 let mut block = remote
                     .get_beacon_blocks_ssz::<TEthSpec>(BlockId::Finalized, &spec)
@@ -290,6 +292,8 @@ where
                     })?
                     .ok_or("Finalized block missing from remote, it returned 404")?;
 
+                debug!(context.log(), "Downloaded finalized block");
+
                 let mut block_slot = block.slot();
 
                 while block.slot() % slots_per_epoch != 0 {
@@ -301,6 +305,12 @@ where
                         "block_slot" => block_slot,
                     );
 
+                    debug!(
+                        context.log(),
+                        "Searching for aligned checkpoint block";
+                        "block_slot" => block_slot
+                    );
+
                     if let Some(found_block) = remote
                         .get_beacon_blocks_ssz::<TEthSpec>(BlockId::Slot(block_slot), &spec)
                         .await
@@ -312,7 +322,19 @@ where
                     }
                 }
 
+                debug!(
+                    context.log(),
+                    "Downloaded aligned finalized block";
+                    "block_root" => ?block.canonical_root(),
+                    "block_slot" => block.slot(),
+                );
+
                 let state_root = block.state_root();
+                debug!(
+                    context.log(),
+                    "Downloading finalized state";
+                    "state_root" => ?state_root
+                );
                 let state = remote
                     .get_debug_beacon_states_ssz::<TEthSpec>(StateId::Root(state_root), &spec)
                     .await
@@ -326,6 +348,8 @@ where
                         format!("Checkpoint state missing from remote: {:?}", state_root)
                     })?;
 
+                debug!(context.log(), "Downloaded finalized state");
+
                 let genesis_state = BeaconState::from_ssz_bytes(&genesis_state_bytes, &spec)
                     .map_err(|e| format!("Unable to parse genesis state SSZ: {:?}", e))?;
 

beacon_node/eth1/src/service.rs

@@ -768,12 +768,32 @@ impl Service {
         *self.inner.remote_head_block.write() = Some(remote_head_block);
 
         let update_deposit_cache = async {
-            let outcome = self
+            let outcome_result = self
                 .update_deposit_cache(Some(new_block_numbers_deposit), &endpoints)
-                .await
-                .map_err(|e| {
-                    format!("Failed to update eth1 deposit cache: {:?}", process_err(e))
-                })?;
+                .await;
+
+            // Reset the `last_procesed block` to the last valid deposit's block number.
+            // This will ensure that the next batch of blocks fetched is immediately after
+            // the last cached valid deposit allowing us to recover from scenarios where
+            // the deposit cache gets corrupted due to invalid responses from eth1 nodes.
+            if let Err(Error::FailedToInsertDeposit(DepositCacheError::NonConsecutive {
+                log_index: _,
+                expected: _,
+            })) = &outcome_result
+            {
+                let mut deposit_cache = self.inner.deposit_cache.write();
+                debug!(
+                    self.log,
+                    "Resetting last processed block";
+                    "old_block_number" => deposit_cache.last_processed_block,
+                    "new_block_number" => deposit_cache.cache.latest_block_number(),
+                );
+                deposit_cache.last_processed_block = deposit_cache.cache.latest_block_number();
+            }
+
+            let outcome = outcome_result.map_err(|e| {
+                format!("Failed to update eth1 deposit cache: {:?}", process_err(e))
+            })?;
 
             trace!(
                 self.log,
@@ -1206,7 +1226,7 @@ impl Service {
                 "latest_block_age" => latest_block_mins,
                 "latest_block" => block_cache.highest_block_number(),
                 "total_cached_blocks" => block_cache.len(),
-                "new" => blocks_imported
+                "new" => %blocks_imported
             );
         } else {
             debug!(