[Merge] Block validator duties when EL is not ready (#2672)

* Reject some HTTP endpoints when EL is not ready

* Restrict more endpoints

* Add watchdog task

* Change scheduling

* Update to new schedule

* Add "syncing" concept

* Remove RequireSynced

* Add is_merge_complete to head_info

* Cache latest_head in Engines

* Call consensus_forkchoiceUpdate on startup

Author: paulhauner

Cargo.lock

@@ -194,6 +194,7 @@ pub struct HeadInfo {
     pub genesis_time: u64,
     pub genesis_validators_root: Hash256,
     pub proposer_shuffling_decision_root: Hash256,
+    pub is_merge_complete: bool,
 }
 
 pub trait BeaconChainTypes: Send + Sync + 'static {
@@ -204,6 +205,19 @@ pub trait BeaconChainTypes: Send + Sync + 'static {
     type EthSpec: types::EthSpec;
 }
 
+/// Indicates the status of the `ExecutionLayer`.
+#[derive(Debug, PartialEq)]
+pub enum ExecutionLayerStatus {
+    /// The execution layer is synced and reachable.
+    Ready,
+    /// The execution layer either syncing or unreachable.
+    NotReady,
+    /// The execution layer is required, but has not been enabled. This is a configuration error.
+    Missing,
+    /// The execution layer is not yet required, therefore the status is irrelevant.
+    NotRequired,
+}
+
 pub type BeaconForkChoice<T> = ForkChoice<
     BeaconForkChoiceStore<
         <T as BeaconChainTypes>::EthSpec,
@@ -1001,6 +1015,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 genesis_time: head.beacon_state.genesis_time(),
                 genesis_validators_root: head.beacon_state.genesis_validators_root(),
                 proposer_shuffling_decision_root,
+                is_merge_complete: is_merge_complete(&head.beacon_state),
             })
         })
     }
@@ -3405,6 +3420,39 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             .map_err(Error::ExecutionForkChoiceUpdateFailed)
     }
 
+    /// Indicates the status of the execution layer.
+    pub async fn execution_layer_status(&self) -> Result<ExecutionLayerStatus, BeaconChainError> {
+        let epoch = self.epoch()?;
+        if self.spec.merge_fork_epoch.map_or(true, |fork| epoch < fork) {
+            return Ok(ExecutionLayerStatus::NotRequired);
+        }
+
+        if let Some(execution_layer) = &self.execution_layer {
+            if execution_layer.is_synced().await {
+                Ok(ExecutionLayerStatus::Ready)
+            } else {
+                Ok(ExecutionLayerStatus::NotReady)
+            }
+        } else {
+            // This branch is slightly more restrictive than what is minimally required.
+            //
+            // It is possible for a node without an execution layer (EL) to follow the chain
+            // *after* the merge fork and *before* the terminal execution block, as long as
+            // that node is not required to produce blocks.
+            //
+            // However, here we say that all nodes *must* have an EL as soon as the merge fork
+            // happens. We do this because it's very difficult to determine that the terminal
+            // block has been met if we don't already have an EL. As far as we know, the
+            // terminal execution block might already exist and we've been rejecting it since
+            // we don't have an EL to verify it.
+            //
+            // I think it is very reasonable to say that the beacon chain expects all BNs to
+            // be paired with an EL node by the time the merge fork epoch is reached. So, we
+            // enforce that here.
+            Ok(ExecutionLayerStatus::Missing)
+        }
+    }
+
     /// This function takes a configured weak subjectivity `Checkpoint` and the latest finalized `Checkpoint`.
     /// If the weak subjectivity checkpoint and finalized checkpoint share the same epoch, we compare
     /// roots. If we the weak subjectivity checkpoint is from an older epoch, we iterate back through

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -36,7 +36,8 @@ mod validator_pubkey_cache;
 
 pub use self::beacon_chain::{
     AttestationProcessingOutcome, BeaconChain, BeaconChainTypes, BeaconStore, ChainSegmentResult,
-    ForkChoiceError, HeadInfo, StateSkipConfig, WhenSlotSkipped, MAXIMUM_GOSSIP_CLOCK_DISPARITY,
+    ExecutionLayerStatus, ForkChoiceError, HeadInfo, StateSkipConfig, WhenSlotSkipped,
+    MAXIMUM_GOSSIP_CLOCK_DISPARITY,
 };
 pub use self::beacon_snapshot::BeaconSnapshot;
 pub use self::chain_config::ChainConfig;

beacon_node/beacon_chain/src/lib.rs

@@ -647,8 +647,55 @@ where
 
         if let Some(beacon_chain) = self.beacon_chain.as_ref() {
             let state_advance_context = runtime_context.service_context("state_advance".into());
-            let log = state_advance_context.log().clone();
-            spawn_state_advance_timer(state_advance_context.executor, beacon_chain.clone(), log);
+            let state_advance_log = state_advance_context.log().clone();
+            spawn_state_advance_timer(
+                state_advance_context.executor,
+                beacon_chain.clone(),
+                state_advance_log,
+            );
+
+            if let Some(execution_layer) = beacon_chain.execution_layer.as_ref() {
+                let store = beacon_chain.store.clone();
+                let inner_execution_layer = execution_layer.clone();
+
+                let head = beacon_chain
+                    .head_info()
+                    .map_err(|e| format!("Unable to read beacon chain head: {:?}", e))?;
+
+                // Issue the head to the execution engine on startup. This ensures it can start
+                // syncing.
+                if head.is_merge_complete {
+                    let result = runtime_context
+                        .executor
+                        .runtime()
+                        .upgrade()
+                        .ok_or_else(|| "Cannot update engine head, shutting down".to_string())?
+                        .block_on(async move {
+                            BeaconChain::<
+                                Witness<TSlotClock, TEth1Backend, TEthSpec, THotStore, TColdStore>,
+                            >::update_execution_engine_forkchoice(
+                                inner_execution_layer,
+                                store,
+                                head.finalized_checkpoint.root,
+                                head.block_root,
+                            )
+                            .await
+                        });
+
+                    // No need to exit early if setting the head fails. It will be set again if/when the
+                    // node comes online.
+                    if let Err(e) = result {
+                        warn!(
+                            log,
+                            "Failed to update head on execution engines";
+                            "error" => ?e
+                        );
+                    }
+                }
+
+                // Spawn a routine that tracks the status of the execution engines.
+                execution_layer.spawn_watchdog_routine(beacon_chain.slot_clock.clone());
+            }
         }
 
         Ok(Client {

beacon_node/client/src/builder.rs

@@ -28,3 +28,4 @@ exit-future = "0.2.0"
 tree_hash = { path = "../../consensus/tree_hash"}
 tree_hash_derive = { path = "../../consensus/tree_hash_derive"}
 parking_lot = "0.11.0"
+slot_clock = { path = "../../common/slot_clock" }