Disallow attesting to optimistic head (#3140)

## Issue Addressed

NA

## Proposed Changes

Disallow the production of attestations and retrieval of unaggregated attestations when they reference an optimistic head. Add tests to this end.

I also moved `BeaconChain::produce_unaggregated_attestation_for_block` to the `BeaconChainHarness`. It was only being used during tests, so it's nice to stop pretending it's production code. I also needed something that could produce attestations to optimistic blocks in order to simulate scenarios where the justified checkpoint is determined invalid (if no one would attest to an optimistic block, we could never justify it and then flip it to invalid).

## Additional Info

- ~~Blocked on #3126~~

beacon_node/beacon_chain/tests/payload_invalidation.rs

@@ -14,6 +14,7 @@ use proto_array::{Error as ProtoArrayError, ExecutionStatus};
 use slot_clock::SlotClock;
 use std::time::Duration;
 use task_executor::ShutdownReason;
+use tree_hash::TreeHash;
 use types::*;
 
 const VALIDATOR_COUNT: usize = 32;
@@ -221,8 +222,8 @@ impl InvalidPayloadRig {
                 let execution_status = self.execution_status(root.into());
 
                 match is_valid {
-                    Payload::Syncing => assert!(execution_status.is_not_verified()),
-                    Payload::Valid => assert!(execution_status.is_valid()),
+                    Payload::Syncing => assert!(execution_status.is_optimistic()),
+                    Payload::Valid => assert!(execution_status.is_valid_and_post_bellatrix()),
                     Payload::Invalid { .. } => unreachable!(),
                 }
 
@@ -309,7 +310,7 @@ fn invalid_payload_invalidates_parent() {
         latest_valid_hash: Some(latest_valid_hash),
     });
 
-    assert!(rig.execution_status(roots[0]).is_valid());
+    assert!(rig.execution_status(roots[0]).is_valid_and_post_bellatrix());
     assert!(rig.execution_status(roots[1]).is_invalid());
     assert!(rig.execution_status(roots[2]).is_invalid());
 
@@ -397,9 +398,9 @@ fn pre_finalized_latest_valid_hash() {
         let slot = Slot::new(i);
         let root = rig.block_root_at_slot(slot).unwrap();
         if slot == 1 {
-            assert!(rig.execution_status(root).is_valid());
+            assert!(rig.execution_status(root).is_valid_and_post_bellatrix());
         } else {
-            assert!(rig.execution_status(root).is_not_verified());
+            assert!(rig.execution_status(root).is_optimistic());
         }
     }
 }
@@ -446,7 +447,7 @@ fn latest_valid_hash_will_validate() {
         } else if slot == 0 {
             assert!(execution_status.is_irrelevant())
         } else {
-            assert!(execution_status.is_valid())
+            assert!(execution_status.is_valid_and_post_bellatrix())
         }
     }
 }
@@ -484,9 +485,9 @@ fn latest_valid_hash_is_junk() {
         let slot = Slot::new(i);
         let root = rig.block_root_at_slot(slot).unwrap();
         if slot == 1 {
-            assert!(rig.execution_status(root).is_valid());
+            assert!(rig.execution_status(root).is_valid_and_post_bellatrix());
         } else {
-            assert!(rig.execution_status(root).is_not_verified());
+            assert!(rig.execution_status(root).is_optimistic());
         }
     }
 }
@@ -556,7 +557,7 @@ fn invalidates_all_descendants() {
         let execution_status = rig.execution_status(root);
         if slot <= latest_valid_slot {
             // Blocks prior to the latest valid hash are valid.
-            assert!(execution_status.is_valid());
+            assert!(execution_status.is_valid_and_post_bellatrix());
         } else {
             // Blocks after the latest valid hash are invalid.
             assert!(execution_status.is_invalid());
@@ -607,7 +608,7 @@ fn switches_heads() {
     assert_eq!(rig.head_info().block_root, fork_block_root);
 
     // The fork block has not yet been validated.
-    assert!(rig.execution_status(fork_block_root).is_not_verified());
+    assert!(rig.execution_status(fork_block_root).is_optimistic());
 
     for root in blocks {
         let slot = rig.harness.chain.get_block(&root).unwrap().unwrap().slot();
@@ -620,7 +621,7 @@ fn switches_heads() {
         let execution_status = rig.execution_status(root);
         if slot <= latest_valid_slot {
             // Blocks prior to the latest valid hash are valid.
-            assert!(execution_status.is_valid());
+            assert!(execution_status.is_valid_and_post_bellatrix());
         } else {
             // Blocks after the latest valid hash are invalid.
             assert!(execution_status.is_invalid());
@@ -691,13 +692,13 @@ fn manually_validate_child() {
     let parent = rig.import_block(Payload::Syncing);
     let child = rig.import_block(Payload::Syncing);
 
-    assert!(rig.execution_status(parent).is_not_verified());
-    assert!(rig.execution_status(child).is_not_verified());
+    assert!(rig.execution_status(parent).is_optimistic());
+    assert!(rig.execution_status(child).is_optimistic());
 
     rig.validate_manually(child);
 
-    assert!(rig.execution_status(parent).is_valid());
-    assert!(rig.execution_status(child).is_valid());
+    assert!(rig.execution_status(parent).is_valid_and_post_bellatrix());
+    assert!(rig.execution_status(child).is_valid_and_post_bellatrix());
 }
 
 #[test]
@@ -709,13 +710,13 @@ fn manually_validate_parent() {
     let parent = rig.import_block(Payload::Syncing);
     let child = rig.import_block(Payload::Syncing);
 
-    assert!(rig.execution_status(parent).is_not_verified());
-    assert!(rig.execution_status(child).is_not_verified());
+    assert!(rig.execution_status(parent).is_optimistic());
+    assert!(rig.execution_status(child).is_optimistic());
 
     rig.validate_manually(parent);
 
-    assert!(rig.execution_status(parent).is_valid());
-    assert!(rig.execution_status(child).is_not_verified());
+    assert!(rig.execution_status(parent).is_valid_and_post_bellatrix());
+    assert!(rig.execution_status(child).is_optimistic());
 }
 
 #[test]
@@ -819,7 +820,7 @@ fn invalid_parent() {
             block_root,
             Duration::from_secs(0),
             &state,
-            PayloadVerificationStatus::NotVerified,
+            PayloadVerificationStatus::Optimistic,
             &rig.harness.chain.spec
         ),
         Err(ForkChoiceError::ProtoArrayError(message))
@@ -885,3 +886,107 @@ fn payload_preparation_before_transition_block() {
     assert_eq!(payload_attributes.suggested_fee_recipient, fee_recipient);
     assert_eq!(fork_choice_state.head_block_hash, latest_block_hash);
 }
+
+#[test]
+fn attesting_to_optimistic_head() {
+    let mut rig = InvalidPayloadRig::new();
+    rig.move_to_terminal_block();
+    rig.import_block(Payload::Valid); // Import a valid transition block.
+
+    let root = rig.import_block(Payload::Syncing);
+
+    let head = rig.harness.chain.head().unwrap();
+    let slot = head.beacon_block.slot();
+    assert_eq!(
+        head.beacon_block_root, root,
+        "the head should be the latest imported block"
+    );
+    assert!(
+        rig.execution_status(root).is_optimistic(),
+        "the head should be optimistic"
+    );
+
+    /*
+     * Define an attestation for use during testing. It doesn't have a valid signature, but that's
+     * not necessary here.
+     */
+
+    let attestation = {
+        let mut attestation = rig
+            .harness
+            .chain
+            .produce_unaggregated_attestation(Slot::new(0), 0)
+            .unwrap();
+
+        attestation.aggregation_bits.set(0, true).unwrap();
+        attestation.data.slot = slot;
+        attestation.data.beacon_block_root = root;
+
+        rig.harness
+            .chain
+            .naive_aggregation_pool
+            .write()
+            .insert(&attestation)
+            .unwrap();
+
+        attestation
+    };
+
+    /*
+     * Define some closures to produce attestations.
+     */
+
+    let produce_unaggregated = || rig.harness.chain.produce_unaggregated_attestation(slot, 0);
+
+    let get_aggregated = || {
+        rig.harness
+            .chain
+            .get_aggregated_attestation(&attestation.data)
+    };
+
+    let get_aggregated_by_slot_and_root = || {
+        rig.harness
+            .chain
+            .get_aggregated_attestation_by_slot_and_root(
+                attestation.data.slot,
+                &attestation.data.tree_hash_root(),
+            )
+    };
+
+    /*
+     * Ensure attestation production fails with an optimistic head.
+     */
+
+    macro_rules! assert_head_block_not_fully_verified {
+        ($func: expr) => {
+            assert!(matches!(
+                $func,
+                Err(BeaconChainError::HeadBlockNotFullyVerified {
+                    beacon_block_root,
+                    execution_status
+                })
+                if beacon_block_root == root && matches!(execution_status, ExecutionStatus::Optimistic(_))
+            ));
+        }
+    }
+
+    assert_head_block_not_fully_verified!(produce_unaggregated());
+    assert_head_block_not_fully_verified!(get_aggregated());
+    assert_head_block_not_fully_verified!(get_aggregated_by_slot_and_root());
+
+    /*
+     * Ensure attestation production succeeds once the head is verified.
+     *
+     * This is effectively a control for the previous tests.
+     */
+
+    rig.validate_manually(root);
+    assert!(
+        rig.execution_status(root).is_valid_and_post_bellatrix(),
+        "the head should no longer be optimistic"
+    );
+
+    produce_unaggregated().unwrap();
+    get_aggregated().unwrap();
+    get_aggregated_by_slot_and_root().unwrap();
+}