Ensure payload envelope streamer always serves canonical envelopes after the split slot (#9085)

Co-Authored-By: Eitan Seri- Levi <eserilev@gmail.com>

Co-Authored-By: Eitan Seri-Levi <eserilev@ucsc.edu>

Author: eserilev

beacon_node/beacon_chain/src/canonical_head.rs

@@ -383,11 +383,24 @@ impl<T: BeaconChainTypes> CanonicalHead<T> {
         Ok((head, execution_status))
     }
 
-    // TODO(gloas) just a stub for now, implement this once we have fork choice.
-    /// Returns true if the payload for this block is canonical according to fork choice
-    /// Returns an error if the block root doesn't exist in fork choice.
-    pub fn block_has_canonical_payload(&self, _root: &Hash256) -> Result<bool, Error> {
-        Ok(true)
+    /// Returns `true` if the payload for this block is canonical (Full) according to fork choice.
+    pub fn block_has_canonical_payload(
+        &self,
+        root: &Hash256,
+        spec: &ChainSpec,
+    ) -> Result<bool, Error> {
+        let cached_head = self.cached_head();
+        let head_root = cached_head.head_block_root();
+        let head_payload_status = cached_head.head_payload_status();
+
+        if *root == head_root {
+            return Ok(head_payload_status == PayloadStatus::Full);
+        }
+
+        self.fork_choice_read_lock()
+            .get_canonical_payload_status(root, spec)
+            .map(|status| status == PayloadStatus::Full)
+            .map_err(Error::ForkChoiceError)
     }
 
     /// Returns a clone of `self.cached_head`.

beacon_node/beacon_chain/src/payload_envelope_streamer/beacon_chain_adapter.rs

@@ -37,6 +37,8 @@ impl<T: BeaconChainTypes> EnvelopeStreamerBeaconAdapter<T> {
         &self,
         root: &Hash256,
     ) -> Result<bool, BeaconChainError> {
-        self.chain.canonical_head.block_has_canonical_payload(root)
+        self.chain
+            .canonical_head
+            .block_has_canonical_payload(root, &self.chain.spec)
     }
 }

beacon_node/beacon_chain/src/payload_envelope_streamer/mod.rs

@@ -132,13 +132,8 @@ impl<T: BeaconChainTypes> PayloadEnvelopeStreamer<T> {
                                         results.push((*root, Ok(None)));
                                     }
                                 }
-                                Err(_) => {
-                                    results.push((
-                                        *root,
-                                        Err(BeaconChainError::EnvelopeStreamerError(
-                                            Error::BlockMissingFromForkChoice,
-                                        )),
-                                    ));
+                                Err(e) => {
+                                    results.push((*root, Err(e)));
                                 }
                             }
                         } else {

beacon_node/beacon_chain/src/payload_envelope_streamer/tests.rs

@@ -1,4 +1,5 @@
 use super::*;
+use crate::beacon_chain::ForkChoiceError;
 use crate::payload_envelope_streamer::beacon_chain_adapter::MockEnvelopeStreamerBeaconAdapter;
 use crate::test_utils::EphemeralHarnessType;
 use bls::{FixedBytesExtended, Signature};
@@ -279,15 +280,18 @@ async fn stream_envelopes_by_root() {
 }
 
 /// When `block_has_canonical_payload` returns an error, the streamer should
-/// yield `Err(EnvelopeStreamerError(BlockMissingFromForkChoice))` for those roots.
+/// propagate that error for those roots.
 #[tokio::test]
 async fn stream_envelopes_error() {
     let chain = build_chain(4, &[], &[], &[]);
     let (mut mock, _runtime) = mock_adapter();
     mock.expect_get_split_slot().return_const(Slot::new(0));
     mock_envelopes(&mut mock, &chain);
-    mock.expect_block_has_canonical_payload()
-        .returning(|_| Err(BeaconChainError::CanonicalHeadLockTimeout));
+    mock.expect_block_has_canonical_payload().returning(|_| {
+        Err(BeaconChainError::ForkChoiceError(
+            ForkChoiceError::DoesNotDescendFromFinalizedCheckpoint,
+        ))
+    });
 
     let streamer = PayloadEnvelopeStreamer::new(mock, EnvelopeRequestSource::ByRange);
     let mut stream = streamer.launch_stream(roots(&chain));
@@ -299,13 +303,8 @@ async fn stream_envelopes_error() {
             .unwrap_or_else(|| panic!("stream ended early at index {i}"));
         assert_eq!(root, entry.block_root, "root mismatch at index {i}");
         assert!(
-            matches!(
-                result.as_ref(),
-                Err(BeaconChainError::EnvelopeStreamerError(
-                    Error::BlockMissingFromForkChoice
-                ))
-            ),
-            "expected BlockMissingFromForkChoice error at index {i}, got {:?}",
+            result.as_ref().is_err(),
+            "expected error at index {i}, got {:?}",
             result
         );
     }

consensus/fork_choice/src/fork_choice.rs

@@ -78,6 +78,7 @@ pub enum Error<T> {
     UnrealizedVoteProcessing(state_processing::EpochProcessingError),
     ValidatorStatuses(BeaconStateError),
     ChainSpecError(String),
+    DoesNotDescendFromFinalizedCheckpoint,
 }
 
 impl<T> From<InvalidAttestation> for Error<T> {
@@ -1523,6 +1524,29 @@ where
         }
     }
 
+    /// Returns the canonical payload status of a block. See
+    /// `ProtoArrayForkChoice::get_canonical_payload_status`.
+    pub fn get_canonical_payload_status(
+        &self,
+        block_root: &Hash256,
+        spec: &ChainSpec,
+    ) -> Result<PayloadStatus, Error<T::Error>> {
+        if self.is_finalized_checkpoint_or_descendant(*block_root) {
+            let current_slot = self.fc_store.get_current_slot();
+            let proposer_boost_root = self.fc_store.proposer_boost_root();
+            self.proto_array
+                .get_canonical_payload_status::<E>(
+                    block_root,
+                    current_slot,
+                    proposer_boost_root,
+                    spec,
+                )
+                .map_err(Error::ProtoArrayError)
+        } else {
+            Err(Error::DoesNotDescendFromFinalizedCheckpoint)
+        }
+    }
+
     /// Returns the weight for the given block root.
     pub fn get_block_weight(&self, block_root: &Hash256) -> Option<u64> {
         self.proto_array.get_weight(block_root)

consensus/proto_array/src/fork_choice_test_definition.rs

@@ -4,6 +4,7 @@ mod gloas_payload;
 mod no_votes;
 mod votes;
 
+use crate::error::Error;
 use crate::proto_array_fork_choice::{Block, ExecutionStatus, PayloadStatus, ProtoArrayForkChoice};
 use crate::{InvalidationOperation, JustifiedBalances};
 use fixed_bytes::FixedBytesExtended;
@@ -30,6 +31,8 @@ pub enum Operation {
         justified_state_balances: Vec<u64>,
         expected_head: Hash256,
         current_slot: Slot,
+        // TODO(gloas): Make this non-optional. `find_head` always returns a `PayloadStatus`
+        // (Empty for pre-GLOAS), so every test should assert on it explicitly.
         #[serde(default)]
         expected_payload_status: Option<PayloadStatus>,
     },
@@ -61,6 +64,12 @@ pub enum Operation {
         block_root: Hash256,
         attestation_slot: Slot,
     },
+    ProcessGloasAttestation {
+        validator_index: usize,
+        block_root: Hash256,
+        attestation_slot: Slot,
+        payload_present: bool,
+    },
     ProcessPayloadAttestation {
         validator_index: usize,
         block_root: Hash256,
@@ -105,6 +114,16 @@ pub enum Operation {
         block_root: Hash256,
         expected: bool,
     },
+    AssertPayloadStatusByWeight {
+        block_root: Hash256,
+        expected_status: PayloadStatus,
+        /// Override `current_slot`. Defaults to the `current_slot` of the last `FindHead`.
+        #[serde(default)]
+        current_slot: Option<Slot>,
+        /// Override the proposer boost root. Defaults to `Hash256::zero()`.
+        #[serde(default)]
+        proposer_boost_root: Option<Hash256>,
+    },
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -149,6 +168,7 @@ impl ForkChoiceTestDefinition {
         )
         .expect("should create fork choice struct");
         let equivocating_indices = BTreeSet::new();
+        let mut last_current_slot = Slot::new(0);
 
         for (op_index, op) in self.operations.into_iter().enumerate() {
             match op.clone() {
@@ -189,6 +209,16 @@ impl ForkChoiceTestDefinition {
                             op_index, op
                         );
                     }
+                    assert_canonical_payload_status_matches_find_head(
+                        &fork_choice,
+                        &head,
+                        current_slot,
+                        Hash256::zero(),
+                        &spec,
+                        payload_status,
+                        op_index,
+                    );
+                    last_current_slot = current_slot;
                     check_bytes_round_trip(&fork_choice);
                 }
                 Operation::ProposerBoostFindHead {
@@ -201,7 +231,7 @@ impl ForkChoiceTestDefinition {
                     let justified_balances =
                         JustifiedBalances::from_effective_balances(justified_state_balances)
                             .unwrap();
-                    let (head, _payload_status) = fork_choice
+                    let (head, payload_status) = fork_choice
                         .find_head::<MainnetEthSpec>(
                             justified_checkpoint,
                             finalized_checkpoint,
@@ -220,6 +250,15 @@ impl ForkChoiceTestDefinition {
                         "Operation at index {} failed head check. Operation: {:?}",
                         op_index, op
                     );
+                    assert_canonical_payload_status_matches_find_head(
+                        &fork_choice,
+                        &head,
+                        Slot::new(0),
+                        proposer_boost_root,
+                        &spec,
+                        payload_status,
+                        op_index,
+                    );
                     check_bytes_round_trip(&fork_choice);
                 }
                 Operation::InvalidFindHead {
@@ -308,6 +347,27 @@ impl ForkChoiceTestDefinition {
                         });
                     check_bytes_round_trip(&fork_choice);
                 }
+                Operation::ProcessGloasAttestation {
+                    validator_index,
+                    block_root,
+                    attestation_slot,
+                    payload_present,
+                } => {
+                    fork_choice
+                        .process_attestation(
+                            validator_index,
+                            block_root,
+                            attestation_slot,
+                            payload_present,
+                        )
+                        .unwrap_or_else(|_| {
+                            panic!(
+                                "process_attestation op at index {} returned error",
+                                op_index
+                            )
+                        });
+                    check_bytes_round_trip(&fork_choice);
+                }
                 Operation::ProcessPayloadAttestation {
                     validator_index,
                     block_root,
@@ -522,6 +582,26 @@ impl ForkChoiceTestDefinition {
                         op_index
                     );
                 }
+                Operation::AssertPayloadStatusByWeight {
+                    block_root,
+                    expected_status,
+                    current_slot,
+                    proposer_boost_root,
+                } => {
+                    let actual = fork_choice
+                        .get_canonical_payload_status::<MainnetEthSpec>(
+                            &block_root,
+                            current_slot.unwrap_or(last_current_slot),
+                            proposer_boost_root.unwrap_or_else(Hash256::zero),
+                            &spec,
+                        )
+                        .unwrap();
+                    assert_eq!(
+                        actual, expected_status,
+                        "canonical payload status mismatch at op index {}",
+                        op_index
+                    );
+                }
             }
         }
     }
@@ -546,6 +626,37 @@ fn get_checkpoint(i: u64) -> Checkpoint {
     }
 }
 
+/// Checks that `get_canonical_payload_status` agrees with the `payload_status`
+/// returned by `find_head` for the head block.
+fn assert_canonical_payload_status_matches_find_head(
+    fork_choice: &ProtoArrayForkChoice,
+    head: &Hash256,
+    current_slot: Slot,
+    proposer_boost_root: Hash256,
+    spec: &ChainSpec,
+    expected: PayloadStatus,
+    op_index: usize,
+) {
+    match fork_choice.get_canonical_payload_status::<MainnetEthSpec>(
+        head,
+        current_slot,
+        proposer_boost_root,
+        spec,
+    ) {
+        Ok(actual) => assert_eq!(
+            actual, expected,
+            "get_canonical_payload_status disagreed with find_head for head {:?} at op index {}",
+            head, op_index
+        ),
+        // Skip the check for pre-gloas nodes
+        Err(Error::InvalidNodeVariant { .. }) => {}
+        Err(e) => panic!(
+            "get_canonical_payload_status failed at op index {}: {:?}",
+            op_index, e
+        ),
+    }
+}
+
 fn check_bytes_round_trip(original: &ProtoArrayForkChoice) {
     let bytes = original.as_bytes();
     let decoded = ProtoArrayForkChoice::from_bytes(&bytes, original.balances.clone())