Add tests for importing blocks on invalid parents (#3123)

## Issue Addressed

NA

## Proposed Changes

- Adds more checks to prevent importing blocks atop parent with invalid execution payloads.
- Adds a test for these conditions.

## Additional Info

NA

beacon_node/beacon_chain/src/block_verification.rs

@@ -1006,21 +1006,25 @@ impl<'a, T: BeaconChainTypes> FullyVerifiedBlock<'a, T> {
         parent: PreProcessingSnapshot<T::EthSpec>,
         chain: &BeaconChain<T>,
     ) -> Result<Self, BlockError<T::EthSpec>> {
-        // Reject any block if its parent is not known to fork choice.
-        //
-        // A block that is not in fork choice is either:
-        //
-        //  - Not yet imported: we should reject this block because we should only import a child
-        //  after its parent has been fully imported.
-        //  - Pre-finalized: if the parent block is _prior_ to finalization, we should ignore it
-        //  because it will revert finalization. Note that the finalized block is stored in fork
-        //  choice, so we will not reject any child of the finalized block (this is relevant during
-        //  genesis).
-        if !chain
-            .fork_choice
-            .read()
-            .contains_block(&block.parent_root())
-        {
+        if let Some(parent) = chain.fork_choice.read().get_block(&block.parent_root()) {
+            // Reject any block where the parent has an invalid payload. It's impossible for a valid
+            // block to descend from an invalid parent.
+            if parent.execution_status.is_invalid() {
+                return Err(BlockError::ParentExecutionPayloadInvalid {
+                    parent_root: block.parent_root(),
+                });
+            }
+        } else {
+            // Reject any block if its parent is not known to fork choice.
+            //
+            // A block that is not in fork choice is either:
+            //
+            //  - Not yet imported: we should reject this block because we should only import a child
+            //  after its parent has been fully imported.
+            //  - Pre-finalized: if the parent block is _prior_ to finalization, we should ignore it
+            //  because it will revert finalization. Note that the finalized block is stored in fork
+            //  choice, so we will not reject any child of the finalized block (this is relevant during
+            //  genesis).
             return Err(BlockError::ParentUnknown(Box::new(block)));
         }
 

beacon_node/beacon_chain/tests/payload_invalidation.rs

@@ -8,8 +8,10 @@ use beacon_chain::{
 use execution_layer::{
     json_structures::JsonPayloadAttributesV1, ExecutionLayer, PayloadAttributes,
 };
-use proto_array::ExecutionStatus;
+use fork_choice::{Error as ForkChoiceError, InvalidationOperation, PayloadVerificationStatus};
+use proto_array::{Error as ProtoArrayError, ExecutionStatus};
 use slot_clock::SlotClock;
+use std::time::Duration;
 use task_executor::ShutdownReason;
 use types::*;
 
@@ -233,6 +235,13 @@ impl InvalidPayloadRig {
 
         block_root
     }
+
+    fn invalidate_manually(&self, block_root: Hash256) {
+        self.harness
+            .chain
+            .process_invalid_execution_payload(&InvalidationOperation::InvalidateOne { block_root })
+            .unwrap();
+    }
 }
 
 /// Simple test of the different import types.
@@ -693,3 +702,66 @@ fn payload_preparation() {
     };
     assert_eq!(rig.previous_payload_attributes(), payload_attributes);
 }
+
+#[test]
+fn invalid_parent() {
+    let mut rig = InvalidPayloadRig::new();
+    rig.move_to_terminal_block();
+    rig.import_block(Payload::Valid); // Import a valid transition block.
+
+    // Import a syncing block atop the transition block (we'll call this the "parent block" since we
+    // build another block on it later).
+    let parent_root = rig.import_block(Payload::Syncing);
+    let parent_block = rig.harness.get_block(parent_root.into()).unwrap();
+    let parent_state = rig
+        .harness
+        .get_hot_state(parent_block.state_root().into())
+        .unwrap();
+
+    // Produce another block atop the parent, but don't import yet.
+    let slot = parent_block.slot() + 1;
+    rig.harness.set_current_slot(slot);
+    let (block, state) = rig.harness.make_block(parent_state, slot);
+    let block_root = block.canonical_root();
+    assert_eq!(block.parent_root(), parent_root);
+
+    // Invalidate the parent block.
+    rig.invalidate_manually(parent_root);
+    assert!(rig.execution_status(parent_root).is_invalid());
+
+    // Ensure the block built atop an invalid payload is invalid for gossip.
+    assert!(matches!(
+        rig.harness.chain.verify_block_for_gossip(block.clone()),
+        Err(BlockError::ParentExecutionPayloadInvalid { parent_root: invalid_root })
+        if invalid_root == parent_root
+    ));
+
+    // Ensure the block built atop an invalid payload is invalid for import.
+    assert!(matches!(
+        rig.harness.chain.process_block(block.clone()),
+        Err(BlockError::ParentExecutionPayloadInvalid { parent_root: invalid_root })
+        if invalid_root == parent_root
+    ));
+
+    // Ensure the block built atop an invalid payload cannot be imported to fork choice.
+    let (block, _block_signature) = block.deconstruct();
+    assert!(matches!(
+        rig.harness.chain.fork_choice.write().on_block(
+            slot,
+            &block,
+            block_root,
+            Duration::from_secs(0),
+            &state,
+            PayloadVerificationStatus::NotVerified,
+            &rig.harness.chain.spec
+        ),
+        Err(ForkChoiceError::ProtoArrayError(message))
+        if message.contains(&format!(
+            "{:?}",
+            ProtoArrayError::ParentExecutionStatusIsInvalid {
+                block_root,
+                parent_root
+            }
+        ))
+    ));
+}

consensus/proto_array/src/error.rs

@@ -44,6 +44,10 @@ pub enum Error {
     IrrelevantDescendant {
         block_root: Hash256,
     },
+    ParentExecutionStatusIsInvalid {
+        block_root: Hash256,
+        parent_root: Hash256,
+    },
 }
 
 #[derive(Clone, PartialEq, Debug)]

consensus/proto_array/src/proto_array.rs

@@ -315,6 +315,21 @@ impl ProtoArray {
             execution_status: block.execution_status,
         };
 
+        // If the parent has an invalid execution status, return an error before adding the block to
+        // `self`.
+        if let Some(parent_index) = node.parent {
+            let parent = self
+                .nodes
+                .get(parent_index)
+                .ok_or(Error::InvalidNodeIndex(parent_index))?;
+            if parent.execution_status.is_invalid() {
+                return Err(Error::ParentExecutionStatusIsInvalid {
+                    block_root: block.root,
+                    parent_root: parent.root,
+                });
+            }
+        }
+
         self.indices.insert(node.root, node_index);
         self.nodes.push(node.clone());