Connect up DB replay_blocks/load_blocks

beacon_node/beacon_chain/src/fork_revert.rs

@@ -159,7 +159,8 @@ pub fn reset_fork_choice_to_finalization<E: EthSpec, Hot: ItemStore<E>, Cold: It
     // Replay blocks from finalized checkpoint back to head.
     // We do not replay attestations presently, relying on the absence of other blocks
     // to guarantee `head_block_root` as the head.
-    let blocks = store
+    // TODO(gloas): this code doesn't work anyway, could just delete all of it
+    let (blocks, _envelopes) = store
         .load_blocks_to_replay(finalized_slot + 1, head_state.slot(), head_block_root)
         .map_err(|e| format!("Error loading blocks to replay for fork choice: {:?}", e))?;
 

beacon_node/beacon_chain/tests/store_tests.rs

@@ -688,7 +688,7 @@ async fn block_replayer_hooks() {
         .add_attested_blocks_at_slots(state.clone(), state_root, &block_slots, &all_validators)
         .await;
 
-    let blocks = store
+    let (blocks, envelopes) = store
         .load_blocks_to_replay(Slot::new(0), max_slot, end_block_root.into())
         .unwrap();
 
@@ -697,7 +697,6 @@ async fn block_replayer_hooks() {
     let mut pre_block_slots = vec![];
     let mut post_block_slots = vec![];
 
-    // TODO(gloas): handle payloads?
     let mut replay_state = BlockReplayer::<MinimalEthSpec>::new(state, &chain.spec)
         .pre_slot_hook(Box::new(|_, state| {
             pre_slots.push(state.slot());
@@ -725,7 +724,7 @@ async fn block_replayer_hooks() {
             post_block_slots.push(block.slot());
             Ok(())
         }))
-        .apply_blocks(blocks, vec![], None)
+        .apply_blocks(blocks, envelopes, None)
         .unwrap()
         .into_state();
 

beacon_node/http_api/src/block_rewards.rs

@@ -32,7 +32,7 @@ pub fn get_block_rewards<T: BeaconChainTypes>(
         .map_err(unhandled_error)?
         .ok_or_else(|| custom_bad_request(format!("block at end slot {} unknown", end_slot)))?;
 
-    let blocks = chain
+    let (blocks, envelopes) = chain
         .store
         .load_blocks_to_replay(start_slot, end_slot, end_block_root)
         .map_err(|e| unhandled_error(BeaconChainError::from(e)))?;
@@ -56,7 +56,6 @@ pub fn get_block_rewards<T: BeaconChainTypes>(
     let mut reward_cache = Default::default();
     let mut block_rewards = Vec::with_capacity(blocks.len());
 
-    // TODO(gloas): handle payloads
     let block_replayer = BlockReplayer::new(state, &chain.spec)
         .pre_block_hook(Box::new(|state, block| {
             state.build_all_committee_caches(&chain.spec)?;
@@ -79,7 +78,7 @@ pub fn get_block_rewards<T: BeaconChainTypes>(
         )
         .no_signature_verification()
         .minimal_block_root_verification()
-        .apply_blocks(blocks, vec![], None)
+        .apply_blocks(blocks, envelopes, None)
         .map_err(unhandled_error)?;
 
     if block_replayer.state_root_miss() {

beacon_node/store/src/hot_cold_store.rs

@@ -186,6 +186,7 @@ pub enum HotColdDBError {
     MissingHotHDiff(Hash256),
     MissingHDiff(Slot),
     MissingExecutionPayload(Hash256),
+    MissingExecutionPayloadEnvelope(Hash256),
     MissingFullBlockExecutionPayloadPruned(Hash256, Slot),
     MissingAnchorInfo,
     MissingFrozenBlockSlot(Hash256),
@@ -2020,7 +2021,8 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
             return Ok(base_state);
         }
 
-        let blocks = self.load_blocks_to_replay(base_state.slot(), slot, latest_block_root)?;
+        let (blocks, envelopes) =
+            self.load_blocks_to_replay(base_state.slot(), slot, latest_block_root)?;
         let _t = metrics::start_timer(&metrics::STORE_BEACON_REPLAY_HOT_BLOCKS_TIME);
 
         // If replaying blocks, and `update_cache` is true, also cache the epoch boundary
@@ -2053,6 +2055,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
         self.replay_blocks(
             base_state,
             blocks,
+            envelopes,
             slot,
             no_state_root_iter(),
             Some(Box::new(state_cache_hook)),
@@ -2357,6 +2360,8 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
         }
 
         let blocks = self.load_cold_blocks(base_state.slot() + 1, slot)?;
+        // TODO(gloas): load payload envelopes
+        let envelopes = vec![];
 
         // Include state root for base state as it is required by block processing to not
         // have to hash the state.
@@ -2365,7 +2370,14 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
             self.forwards_state_roots_iterator_until(base_state.slot(), slot, || {
                 Err(Error::StateShouldNotBeRequired(slot))
             })?;
-        let state = self.replay_blocks(base_state, blocks, slot, Some(state_root_iter), None)?;
+        let state = self.replay_blocks(
+            base_state,
+            blocks,
+            envelopes,
+            slot,
+            Some(state_root_iter),
+            None,
+        )?;
         debug!(
             target_slot = %slot,
             replay_time_ms = metrics::stop_timer_with_duration(replay_timer).as_millis(),
@@ -2480,18 +2492,31 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
         })?
     }
 
-    /// Load the blocks between `start_slot` and `end_slot` by backtracking from `end_block_hash`.
+    /// Load the blocks & envelopes between `start_slot` and `end_slot` by backtracking from
+    /// `end_block_root`.
     ///
     /// Blocks are returned in slot-ascending order, suitable for replaying on a state with slot
     /// equal to `start_slot`, to reach a state with slot equal to `end_slot`.
+    ///
+    /// Payloads are also returned in slot-ascending order, but only payloads forming part of
+    /// the chain are loaded (payloads for EMPTY slots are omitted). Prior to Gloas, an empty
+    /// vec of payloads will be returned.
+    // TODO(gloas): handle last payload
+    #[allow(clippy::type_complexity)]
     pub fn load_blocks_to_replay(
         &self,
         start_slot: Slot,
         end_slot: Slot,
-        end_block_hash: Hash256,
+        end_block_root: Hash256,
-    ) -> Result<Vec<SignedBeaconBlock<E, BlindedPayload<E>>>, Error> {
+    ) -> Result<
+        (
+            Vec<SignedBlindedBeaconBlock<E>>,
+            Vec<SignedExecutionPayloadEnvelope<E>>,
+        ),
+        Error,
+    > {
         let _t = metrics::start_timer(&metrics::STORE_BEACON_LOAD_HOT_BLOCKS_TIME);
-        let mut blocks = ParentRootBlockIterator::new(self, end_block_hash)
+        let mut blocks = ParentRootBlockIterator::new(self, end_block_root)
             .map(|result| result.map(|(_, block)| block))
             // Include the block at the end slot (if any), it needs to be
             // replayed in order to construct the canonical state at `end_slot`.
@@ -2518,7 +2543,35 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
             })
             .collect::<Result<Vec<_>, _>>()?;
         blocks.reverse();
-        Ok(blocks)
+
+        // If Gloas is not enabled for any slots in the range, just return `blocks`.
+        if !self.spec.fork_name_at_slot::<E>(start_slot).gloas_enabled()
+            && !self.spec.fork_name_at_slot::<E>(end_slot).gloas_enabled()
+        {
+            return Ok((blocks, vec![]));
+        }
+
+        // Load envelopes.
+        let mut envelopes = vec![];
+
+        for (block, next_block) in blocks.iter().tuple_windows() {
+            if block.fork_name_unchecked().gloas_enabled() {
+                // Check next block to see if this block's payload is canonical on this chain.
+                let block_hash = block.payload_bid_block_hash()?;
+                if !next_block.is_parent_block_full(block_hash) {
+                    // No payload at this slot (empty), nothing to load.
+                    continue;
+                }
+                // Using `parent_root` avoids computation.
+                let block_root = next_block.parent_root();
+                let envelope = self
+                    .get_payload_envelope(&block_root)?
+                    .ok_or(HotColdDBError::MissingExecutionPayloadEnvelope(block_root))?;
+                envelopes.push(envelope);
+            }
+        }
+
+        Ok((blocks, envelopes))
     }
 
     /// Replay `blocks` on top of `state` until `target_slot` is reached.
@@ -2528,7 +2581,8 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
     pub fn replay_blocks(
         &self,
         state: BeaconState<E>,
-        blocks: Vec<SignedBeaconBlock<E, BlindedPayload<E>>>,
+        blocks: Vec<SignedBlindedBeaconBlock<E>>,
+        envelopes: Vec<SignedExecutionPayloadEnvelope<E>>,
         target_slot: Slot,
         state_root_iter: Option<impl Iterator<Item = Result<(Hash256, Slot), Error>>>,
         pre_slot_hook: Option<PreSlotHook<E, Error>>,
@@ -2548,9 +2602,8 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
             block_replayer = block_replayer.pre_slot_hook(pre_slot_hook);
         }
 
-        // TODO(gloas): plumb through payloads here
         block_replayer
-            .apply_blocks(blocks, vec![], Some(target_slot))
+            .apply_blocks(blocks, envelopes, Some(target_slot))
             .map(|block_replayer| {
                 if have_state_root_iterator && block_replayer.state_root_miss() {
                     warn!(

consensus/state_processing/src/block_replayer.rs

@@ -257,22 +257,15 @@ where
                 let state_root = if self.state.slot() == self.state.latest_block_header().slot
                     && block.fork_name_unchecked().gloas_enabled()
                 {
-                    let state_block_hash = self
+                    let latest_bid_block_hash = self
                         .state
                         .latest_execution_payload_bid()
                         .map_err(BlockReplayError::from)?
                         .block_hash;
-                    let parent_block_hash = block
-                        .message()
-                        .body()
-                        .signed_execution_payload_bid()
-                        .map_err(BlockReplayError::from)?
-                        .message
-                        .parent_block_hash;
 
                     // Similar to `is_parent_block_full`, but reading the block hash from the
                     // not-yet-applied `block`.
-                    if state_block_hash == parent_block_hash {
+                    if block.is_parent_block_full(latest_bid_block_hash) {
                         if let Some(envelope) = envelopes_iter.next()
                             && envelope.message.slot == self.state.slot()
                         {
@@ -303,7 +296,7 @@ where
                         } else {
                             return Err(BlockReplayError::MissingPayloadEnvelope {
                                 slot: block.slot(),
-                                block_hash: state_block_hash,
+                                block_hash: latest_bid_block_hash,
                             }
                             .into());
                         }

consensus/types/src/block/signed_beacon_block.rs

@@ -14,6 +14,7 @@ use tree_hash::TreeHash;
 use tree_hash_derive::TreeHash;
 
 use crate::{
+    ExecutionBlockHash,
     block::{
         BLOB_KZG_COMMITMENTS_INDEX, BeaconBlock, BeaconBlockAltair, BeaconBlockBase,
         BeaconBlockBellatrix, BeaconBlockBodyBellatrix, BeaconBlockBodyCapella,
@@ -365,6 +366,32 @@ impl<E: EthSpec, Payload: AbstractExecPayload<E>> SignedBeaconBlock<E, Payload>
 
         format_kzg_commitments(commitments.as_ref())
     }
+
+    /// Convenience accessor for the block's bid's `block_hash`.
+    ///
+    /// This method returns an error prior to Gloas.
+    pub fn payload_bid_block_hash(&self) -> Result<ExecutionBlockHash, BeaconStateError> {
+        self.message()
+            .body()
+            .signed_execution_payload_bid()
+            .map(|bid| bid.message.block_hash)
+    }
+
+    /// Check if the `parent_hash` in this block's `signed_payload_bid` matches `block_hash`.
+    ///
+    /// This function is useful post-Gloas for determining if the parent block is full, *without*
+    /// necessarily needing access to a beacon state. The passed in `parent_block_hash` MUST be the
+    /// `block_hash` from the parent beacon block's bid. If the parent beacon state is available
+    /// this can alternatively be fetched from `state.latest_payload_bid`.
+    ///
+    /// This function returns `false` for all blocks prior to Gloas.
+    pub fn is_parent_block_full(&self, parent_block_hash: ExecutionBlockHash) -> bool {
+        let Ok(signed_payload_bid) = self.message().body().signed_execution_payload_bid() else {
+            // Prior to Gloas.
+            return false;
+        };
+        signed_payload_bid.message.parent_block_hash == parent_block_hash
+    }
 }
 
 // We can convert pre-Bellatrix blocks without payloads into blocks with payloads.