Add payload support to BlockReplayer

beacon_node/beacon_chain/tests/rewards.rs

@@ -845,13 +845,14 @@ async fn check_all_base_rewards_for_subset(
                 .state_at_slot(Slot::new(slot - 1), StateSkipConfig::WithoutStateRoots)
                 .unwrap();
 
+            // TODO(gloas): handle payloads?
             let mut pre_state = BlockReplayer::<E, BlockReplayError, IntoIter<_, 0>>::new(
                 parent_state,
                 &harness.spec,
             )
             .no_signature_verification()
             .minimal_block_root_verification()
-            .apply_blocks(vec![], Some(block.slot()))
+            .apply_blocks(vec![], vec![], Some(block.slot()))
             .unwrap()
             .into_state();
 

beacon_node/beacon_chain/tests/store_tests.rs

@@ -697,6 +697,7 @@ 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());
@@ -724,7 +725,7 @@ async fn block_replayer_hooks() {
             post_block_slots.push(block.slot());
             Ok(())
         }))
-        .apply_blocks(blocks, None)
+        .apply_blocks(blocks, vec![], None)
         .unwrap()
         .into_state();
 

beacon_node/http_api/src/attestation_performance.rs

@@ -205,8 +205,9 @@ pub fn get_attestation_performance<T: BeaconChainTypes>(
             })
             .collect::<Result<Vec<_>, _>>()?;
 
+        // TODO(gloas): add payloads
         replayer = replayer
-            .apply_blocks(blocks, None)
+            .apply_blocks(blocks, vec![], None)
             .map_err(|e| custom_server_error(format!("{:?}", e)))?;
     }
 

beacon_node/http_api/src/block_packing_efficiency.rs

@@ -398,8 +398,9 @@ pub fn get_block_packing_efficiency<T: BeaconChainTypes>(
             })
             .collect::<Result<Vec<_>, _>>()?;
 
+        // TODO(gloas): add payloads
         replayer = replayer
-            .apply_blocks(blocks, None)
+            .apply_blocks(blocks, vec![], None)
             .map_err(|e: PackingEfficiencyError| custom_server_error(format!("{:?}", e)))?;
     }
 

beacon_node/http_api/src/block_rewards.rs

@@ -56,6 +56,7 @@ 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)?;
@@ -78,7 +79,7 @@ pub fn get_block_rewards<T: BeaconChainTypes>(
         )
         .no_signature_verification()
         .minimal_block_root_verification()
-        .apply_blocks(blocks, None)
+        .apply_blocks(blocks, vec![], None)
         .map_err(unhandled_error)?;
 
     if block_replayer.state_root_miss() {
@@ -138,11 +139,12 @@ pub fn compute_block_rewards<T: BeaconChainTypes>(
                     ))
                 })?;
 
+            // TODO(gloas): handle payloads?
             let block_replayer = BlockReplayer::new(parent_state, &chain.spec)
                 .no_signature_verification()
                 .state_root_iter([Ok((parent_block.state_root(), parent_block.slot()))].into_iter())
                 .minimal_block_root_verification()
-                .apply_blocks(vec![], Some(block.slot()))
+                .apply_blocks(vec![], vec![], Some(block.slot()))
                 .map_err(unhandled_error::<BeaconChainError>)?;
 
             if block_replayer.state_root_miss() {

beacon_node/http_api/src/sync_committee_rewards.rs

@@ -66,11 +66,12 @@ pub fn get_state_before_applying_block<T: BeaconChainTypes>(
         })
         .map_err(|e| custom_not_found(format!("Parent state is not available! {:?}", e)))?;
 
+    // TODO(gloas): handle payloads?
     let replayer = BlockReplayer::new(parent_state, &chain.spec)
         .no_signature_verification()
         .state_root_iter([Ok((parent_block.state_root(), parent_block.slot()))].into_iter())
         .minimal_block_root_verification()
-        .apply_blocks(vec![], Some(block.slot()))
+        .apply_blocks(vec![], vec![], Some(block.slot()))
         .map_err(unhandled_error::<BeaconChainError>)?;
 
     Ok(replayer.into_state())

beacon_node/store/src/hot_cold_store.rs

@@ -2548,8 +2548,9 @@ 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, Some(target_slot))
+            .apply_blocks(blocks, vec![], Some(target_slot))
             .map(|block_replayer| {
                 if have_state_root_iterator && block_replayer.state_root_miss() {
                     warn!(

beacon_node/store/src/reconstruct.rs

@@ -67,6 +67,7 @@ where
 
         state.build_caches(&self.spec)?;
 
+        // TODO(gloas): handle payload envelope replay
         process_results(block_root_iter, |iter| -> Result<(), Error> {
             let mut io_batch = vec![];
 

consensus/state_processing/src/block_replayer.rs

@@ -1,14 +1,19 @@
 use crate::{
     BlockProcessingError, BlockSignatureStrategy, ConsensusContext, SlotProcessingError,
-    VerifyBlockRoot, per_block_processing, per_epoch_processing::EpochProcessingSummary,
+    VerifyBlockRoot, VerifySignatures,
+    envelope_processing::{
+        EnvelopeProcessingError, VerifyStateRoot, process_execution_payload_envelope,
+    },
+    per_block_processing,
+    per_epoch_processing::EpochProcessingSummary,
     per_slot_processing,
 };
 use itertools::Itertools;
 use std::iter::Peekable;
 use std::marker::PhantomData;
 use types::{
-    BeaconState, BeaconStateError, BlindedPayload, ChainSpec, EthSpec, Hash256, SignedBeaconBlock,
+    BeaconState, BeaconStateError, BlindedPayload, ChainSpec, EthSpec, ExecutionBlockHash, Hash256,
-    Slot,
+    SignedBeaconBlock, SignedExecutionPayloadEnvelope, Slot,
 };
 
 pub type PreBlockHook<'a, E, Error> = Box<
@@ -24,7 +29,7 @@ pub type PostSlotHook<'a, E, Error> = Box<
 >;
 pub type StateRootIterDefault<Error> = std::iter::Empty<Result<(Hash256, Slot), Error>>;
 
-/// Efficiently apply blocks to a state while configuring various parameters.
+/// Efficiently apply blocks and payloads to a state while configuring various parameters.
 ///
 /// Usage follows a builder pattern.
 pub struct BlockReplayer<
@@ -41,6 +46,11 @@ pub struct BlockReplayer<
     post_block_hook: Option<PostBlockHook<'a, Spec, Error>>,
     pre_slot_hook: Option<PreSlotHook<'a, Spec, Error>>,
     post_slot_hook: Option<PostSlotHook<'a, Spec, Error>>,
+    /// Iterator over state roots for all *block* states.
+    ///
+    /// Pre-Gloas, this is all states. Post-Gloas, this is *just* the states corresponding to beacon
+    /// blocks. For states corresponding to payloads, we read the state root from the payload
+    /// envelope.
     pub(crate) state_root_iter: Option<Peekable<StateRootIter>>,
     state_root_miss: bool,
     _phantom: PhantomData<Error>,
@@ -50,7 +60,13 @@ pub struct BlockReplayer<
 pub enum BlockReplayError {
     SlotProcessing(SlotProcessingError),
     BlockProcessing(BlockProcessingError),
+    EnvelopeProcessing(EnvelopeProcessingError),
     BeaconState(BeaconStateError),
+    /// A payload envelope for this `slot` was required but not provided.
+    MissingPayloadEnvelope {
+        slot: Slot,
+        block_hash: ExecutionBlockHash,
+    },
 }
 
 impl From<SlotProcessingError> for BlockReplayError {
@@ -65,6 +81,12 @@ impl From<BlockProcessingError> for BlockReplayError {
     }
 }
 
+impl From<EnvelopeProcessingError> for BlockReplayError {
+    fn from(e: EnvelopeProcessingError) -> Self {
+        Self::EnvelopeProcessing(e)
+    }
+}
+
 impl From<BeaconStateError> for BlockReplayError {
     fn from(e: BeaconStateError) -> Self {
         Self::BeaconState(e)
@@ -215,8 +237,11 @@ where
     pub fn apply_blocks(
         mut self,
         blocks: Vec<SignedBeaconBlock<E, BlindedPayload<E>>>,
+        payload_envelopes: Vec<SignedExecutionPayloadEnvelope<E>>,
         target_slot: Option<Slot>,
     ) -> Result<Self, Error> {
+        let mut envelopes_iter = payload_envelopes.into_iter();
+
         for (i, block) in blocks.iter().enumerate() {
             // Allow one additional block at the start which is only used for its state root.
             if i == 0 && block.slot() <= self.state.slot() {
@@ -224,7 +249,74 @@ where
             }
 
             while self.state.slot() < block.slot() {
-                let state_root = self.get_state_root(&blocks, i)?;
+                let block_state_root = self.get_state_root(&blocks, i)?;
+
+                // Apply the payload for the *previous* block if the bid in the current block
+                // indicates that the parent is full.
+                // TODO(gloas): check this condition at the fork boundary.
+                let state_root = if self.state.slot() == self.state.latest_block_header().slot
+                    && block.fork_name_unchecked().gloas_enabled()
+                {
+                    let state_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 let Some(envelope) = envelopes_iter.next()
+                            && envelope.message.slot == self.state.slot()
+                        {
+                            // TODO(gloas): bulk signature verification could be relevant here?
+                            let verify_payload_signatures =
+                                if let BlockSignatureStrategy::NoVerification =
+                                    self.block_sig_strategy
+                                {
+                                    VerifySignatures::False
+                                } else {
+                                    VerifySignatures::True
+                                };
+                            // TODO(gloas): state root verif enabled during initial
+                            // prototyping/testing
+                            let verify_state_root = VerifyStateRoot::True;
+                            process_execution_payload_envelope(
+                                &mut self.state,
+                                Some(block_state_root),
+                                &envelope,
+                                verify_payload_signatures,
+                                verify_state_root,
+                                self.spec,
+                            )
+                            .map_err(BlockReplayError::from)?;
+
+                            // State root for next slot processing is now the envelope's state root.
+                            envelope.message.state_root
+                        } else {
+                            return Err(BlockReplayError::MissingPayloadEnvelope {
+                                slot: block.slot(),
+                                block_hash: state_block_hash,
+                            }
+                            .into());
+                        }
+                    } else {
+                        // Empty payload at this slot, the state root is unchanged from when the
+                        // beacon block was applied.
+                        block_state_root
+                    }
+                } else {
+                    // Pre-Gloas or at skipped slots post-Gloas, the state root of the parent state
+                    // is always the output from `self.get_state_root`.
+                    block_state_root
+                };
 
                 if let Some(ref mut pre_slot_hook) = self.pre_slot_hook {
                     pre_slot_hook(state_root, &mut self.state)?;
@@ -268,6 +360,8 @@ where
             }
         }
 
+        // TODO(gloas): apply last payload, but how to know if it *should* be applied?
+
         if let Some(target_slot) = target_slot {
             while self.state.slot() < target_slot {
                 let state_root = self.get_state_root(&blocks, blocks.len())?;

consensus/state_processing/src/envelope_processing.rs

@@ -241,8 +241,6 @@ pub fn process_execution_payload_envelope<E: EthSpec>(
     // TODO(gloas): newPayload happens here in the spec, ensure we wire that up correctly
 
     process_deposit_requests_post_gloas(state, &execution_requests.deposits, spec)?;
-
-    // TODO(gloas): gotta update these
     process_withdrawal_requests(state, &execution_requests.withdrawals, spec)?;
     process_consolidation_requests(state, &execution_requests.consolidations, spec)?;
 

consensus/state_processing/src/per_block_processing/tests.rs

@@ -1140,7 +1140,7 @@ async fn block_replayer_peeking_state_roots() {
     let block_replayer = BlockReplayer::new(parent_state, &harness.chain.spec)
         .state_root_iter(state_root_iter.into_iter())
         .no_signature_verification()
-        .apply_blocks(vec![target_block], None)
+        .apply_blocks(vec![target_block], vec![], None)
         .unwrap();
 
     assert_eq!(