Merge pull request #8625 from ethDreamer/gloas-envelope-processing-merge-unstable

Gloas envelope processing merge unstable

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -1144,6 +1144,23 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         }
     }
 
+    /// Returns the full block at the given root, if it's available in the database.
+    ///
+    /// Should always return a full block for pre-merge and post-gloas blocks.
+    /// An
+    pub fn get_full_block(
+        &self,
+        block_root: &Hash256,
+    ) -> Result<Option<SignedBeaconBlock<T::EthSpec>>, Error> {
+        match self.store.try_get_full_block(block_root)? {
+            Some(DatabaseBlock::Full(block)) => Ok(Some(block)),
+            Some(DatabaseBlock::Blinded(_)) => {
+                Err(Error::ExecutionPayloadMissingFromDatabase(*block_root))
+            }
+            None => Ok(None),
+        }
+    }
+
     /// Returns the block at the given root, if any.
     ///
     /// ## Errors
@@ -4824,7 +4841,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         let proposal_epoch = proposal_slot.epoch(T::EthSpec::slots_per_epoch());
         if head_state.current_epoch() == proposal_epoch {
             return get_expected_withdrawals(&unadvanced_state, &self.spec)
-                .map(|(withdrawals, _)| withdrawals)
+                .map(|(withdrawals, _, _)| withdrawals)
                 .map_err(Error::PrepareProposerFailed);
         }
 
@@ -4842,7 +4859,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             &self.spec,
         )?;
         get_expected_withdrawals(&advanced_state, &self.spec)
-            .map(|(withdrawals, _)| withdrawals)
+            .map(|(withdrawals, _, _)| withdrawals)
             .map_err(Error::PrepareProposerFailed)
     }
 

beacon_node/beacon_chain/src/block_verification.rs

@@ -709,7 +709,8 @@ pub struct SignatureVerifiedBlock<T: BeaconChainTypes> {
 }
 
 /// Used to await the result of executing payload with an EE.
-type PayloadVerificationHandle = JoinHandle<Option<Result<PayloadVerificationOutcome, BlockError>>>;
+pub type PayloadVerificationHandle =
+    JoinHandle<Option<Result<PayloadVerificationOutcome, BlockError>>>;
 
 /// A wrapper around a `SignedBeaconBlock` that indicates that this block is fully verified and
 /// ready to import into the `BeaconChain`. The validation includes:

beacon_node/beacon_chain/src/envelope_verification.rs

@@ -0,0 +1,445 @@
+//! The incremental processing steps (e.g., signatures verified but not the state transition) is
+//! represented as a sequence of wrapper-types around the block. There is a linear progression of
+//! types, starting at a `SignedBeaconBlock` and finishing with a `Fully VerifiedBlock` (see
+//! diagram below).
+//!
+//! ```ignore
+//!            START
+//!              |
+//!              ▼
+//! SignedExecutionPayloadEnvelope
+//!              |
+//!              |---------------
+//!              |              |
+//!              |              ▼
+//!              |    GossipVerifiedEnvelope
+//!              |              |
+//!              |---------------
+//!              |
+//!              ▼
+//!  ExecutionPendingEnvelope
+//!              |
+//!            await
+//!              |
+//!              ▼
+//!             END
+//!
+//! ```
+
+use crate::NotifyExecutionLayer;
+use crate::block_verification::{PayloadVerificationHandle, PayloadVerificationOutcome};
+use crate::envelope_verification_types::{EnvelopeImportData, MaybeAvailableEnvelope};
+use crate::execution_payload::PayloadNotifier;
+use crate::{BeaconChain, BeaconChainError, BeaconChainTypes};
+use educe::Educe;
+use slot_clock::SlotClock;
+use state_processing::envelope_processing::{EnvelopeProcessingError, envelope_processing};
+use state_processing::{BlockProcessingError, VerifySignatures};
+use std::sync::Arc;
+use tracing::{debug, instrument};
+use types::{
+    BeaconState, BeaconStateError, EthSpec, ExecutionBlockHash, Hash256, SignedBeaconBlock,
+    SignedExecutionPayloadEnvelope, Slot,
+};
+
+#[derive(Debug, Clone)]
+pub enum EnvelopeError {
+    /// The envelope's block root is unknown.
+    BlockRootUnknown {
+        block_root: Hash256,
+    },
+    /// The signature is invalid.
+    BadSignature,
+    /// The builder index doesn't match the committed bid
+    BuilderIndexMismatch {
+        committed_bid: u64,
+        envelope: u64,
+    },
+    // The slot doesn't match the parent block
+    SlotMismatch {
+        parent_block: Slot,
+        envelope: Slot,
+    },
+    // The validator index is unknown
+    UnknownValidator {
+        builder_index: u64,
+    },
+    // The block hash doesn't match the committed bid
+    BlockHashMismatch {
+        committed_bid: ExecutionBlockHash,
+        envelope: ExecutionBlockHash,
+    },
+    // Some Beacon Chain Error
+    BeaconChainError(Arc<BeaconChainError>),
+    // Some Beacon State error
+    BeaconStateError(BeaconStateError),
+    // Some BlockProcessingError (for electra operations)
+    BlockProcessingError(BlockProcessingError),
+    // Some EnvelopeProcessingError
+    EnvelopeProcessingError(EnvelopeProcessingError),
+}
+
+impl From<BeaconChainError> for EnvelopeError {
+    fn from(e: BeaconChainError) -> Self {
+        EnvelopeError::BeaconChainError(Arc::new(e))
+    }
+}
+
+impl From<BeaconStateError> for EnvelopeError {
+    fn from(e: BeaconStateError) -> Self {
+        EnvelopeError::BeaconStateError(e)
+    }
+}
+
+/// Pull errors up from EnvelopeProcessingError to EnvelopeError
+impl From<EnvelopeProcessingError> for EnvelopeError {
+    fn from(e: EnvelopeProcessingError) -> Self {
+        match e {
+            EnvelopeProcessingError::BadSignature => EnvelopeError::BadSignature,
+            EnvelopeProcessingError::BeaconStateError(e) => EnvelopeError::BeaconStateError(e),
+            EnvelopeProcessingError::BlockHashMismatch {
+                committed_bid,
+                envelope,
+            } => EnvelopeError::BlockHashMismatch {
+                committed_bid,
+                envelope,
+            },
+            EnvelopeProcessingError::BlockProcessingError(e) => {
+                EnvelopeError::BlockProcessingError(e)
+            }
+            e => EnvelopeError::EnvelopeProcessingError(e),
+        }
+    }
+}
+
+/// This snapshot is to be used for verifying a envelope of the block.
+#[derive(Debug, Clone)]
+pub struct EnvelopeProcessingSnapshot<E: EthSpec> {
+    /// This state is equivalent to the `self.beacon_block.state_root()` before applying the envelope.
+    pub pre_state: BeaconState<E>,
+    pub state_root: Hash256,
+    pub beacon_block_root: Hash256,
+}
+
+#[allow(clippy::type_complexity)]
+#[instrument(skip_all, level = "debug", fields(beacon_block_root = %envelope.beacon_block_root()))]
+fn load_snapshot<T: BeaconChainTypes>(
+    envelope: &SignedExecutionPayloadEnvelope<T::EthSpec>,
+    chain: &BeaconChain<T>,
+) -> Result<EnvelopeProcessingSnapshot<T::EthSpec>, EnvelopeError> {
+    // 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).
+
+    let beacon_block_root = envelope.beacon_block_root();
+    if !chain
+        .canonical_head
+        .fork_choice_read_lock()
+        .contains_block(&beacon_block_root)
+    {
+        return Err(EnvelopeError::BlockRootUnknown {
+            block_root: beacon_block_root,
+        });
+    }
+
+    let fork_choice_read_lock = chain.canonical_head.fork_choice_read_lock();
+    let Some(proto_beacon_block) = fork_choice_read_lock.get_block(&beacon_block_root) else {
+        return Err(EnvelopeError::BlockRootUnknown {
+            block_root: beacon_block_root,
+        });
+    };
+    drop(fork_choice_read_lock);
+
+    // TODO(EIP-7732): add metrics here
+
+    // Load the parent block's state from the database, returning an error if it is not found.
+    // It is an error because if we know the parent block we should also know the parent state.
+    // Retrieve any state that is advanced through to at most `block.slot()`: this is
+    // particularly important if `block` descends from the finalized/split block, but at a slot
+    // prior to the finalized slot (which is invalid and inaccessible in our DB schema).
+    let (parent_state_root, state) = chain
+        .store
+        // TODO(EIP-7732): the state doesn't need to be advanced here because we're applying an envelope
+        //                 but this function does use a lot of caches that could be more efficient. Is there
+        //                 a better way to do this?
+        .get_advanced_hot_state(
+            beacon_block_root,
+            proto_beacon_block.slot,
+            proto_beacon_block.state_root,
+        )
+        .map_err(|e| EnvelopeError::BeaconChainError(Arc::new(e.into())))?
+        .ok_or_else(|| {
+            BeaconChainError::DBInconsistent(format!(
+                "Missing state for parent block {beacon_block_root:?}",
+            ))
+        })?;
+
+    if state.slot() == proto_beacon_block.slot {
+        // Sanity check.
+        if parent_state_root != proto_beacon_block.state_root {
+            return Err(BeaconChainError::DBInconsistent(format!(
+                "Parent state at slot {} has the wrong state root: {:?} != {:?}",
+                state.slot(),
+                parent_state_root,
+                proto_beacon_block.state_root,
+            ))
+            .into());
+        }
+    }
+
+    Ok(EnvelopeProcessingSnapshot {
+        pre_state: state,
+        state_root: parent_state_root,
+        beacon_block_root,
+    })
+}
+
+/// A wrapper around a `SignedExecutionPayloadEnvelope` that indicates it has been approved for re-gossiping on
+/// the p2p network.
+#[derive(Educe)]
+#[educe(Debug(bound = "T: BeaconChainTypes"))]
+pub struct GossipVerifiedEnvelope<T: BeaconChainTypes> {
+    pub signed_envelope: Arc<SignedExecutionPayloadEnvelope<T::EthSpec>>,
+    pub parent_block: Arc<SignedBeaconBlock<T::EthSpec>>,
+    pub parent: Option<Box<EnvelopeProcessingSnapshot<T::EthSpec>>>,
+}
+
+impl<T: BeaconChainTypes> GossipVerifiedEnvelope<T> {
+    pub fn new(
+        signed_envelope: Arc<SignedExecutionPayloadEnvelope<T::EthSpec>>,
+        chain: &BeaconChain<T>,
+    ) -> Result<Self, EnvelopeError> {
+        let envelope = &signed_envelope.message;
+        let payload = &envelope.payload;
+        let beacon_block_root = envelope.beacon_block_root;
+
+        // check that we've seen the parent block of this envelope and that it passes validation
+        // TODO(EIP-7732): this check would fail if the block didn't pass validation right?
+        let fork_choice_read_lock = chain.canonical_head.fork_choice_read_lock();
+        let Some(parent_proto_block) = fork_choice_read_lock.get_block(&beacon_block_root) else {
+            return Err(EnvelopeError::BlockRootUnknown {
+                block_root: beacon_block_root,
+            });
+        };
+        drop(fork_choice_read_lock);
+
+        // TODO(EIP-7732): check that we haven't seen another valid `SignedExecutionPayloadEnvelope`
+        //                 for this block root from this builder - envelope status table check
+
+        // TODO(EIP-7732): this should probably be obtained from the ProtoBlock instead of the DB
+        //                 but this means the ProtoBlock needs to include something like the ExecutionBid
+        //                 will need to answer this question later.
+        let parent_block = chain
+            .get_full_block(&beacon_block_root)?
+            .ok_or_else(|| {
+                EnvelopeError::from(BeaconChainError::MissingBeaconBlock(beacon_block_root))
+            })
+            .map(Arc::new)?;
+        let execution_bid = &parent_block
+            .message()
+            .body()
+            .signed_execution_payload_bid()?
+            .message;
+
+        // TODO(EIP-7732): Gossip rules for the beacon block contain the following:
+        // https://github.com/ethereum/consensus-specs/blob/master/specs/phase0/p2p-interface.md#beacon_block
+        // [IGNORE] The block is not from a future slot (with a MAXIMUM_GOSSIP_CLOCK_DISPARITY allowance)
+        // [IGNORE] The block is from a slot greater than the latest finalized slot
+        // should these kinds of checks be included for envelopes as well?
+
+        // check that the slot of the envelope matches the slot of the parent block
+        if envelope.slot != parent_block.slot() {
+            return Err(EnvelopeError::SlotMismatch {
+                parent_block: parent_block.slot(),
+                envelope: envelope.slot,
+            });
+        }
+
+        // builder index matches committed bid
+        if envelope.builder_index != execution_bid.builder_index {
+            return Err(EnvelopeError::BuilderIndexMismatch {
+                committed_bid: execution_bid.builder_index,
+                envelope: envelope.builder_index,
+            });
+        }
+
+        // the block hash should match the block hash of the execution bid
+        if payload.block_hash != execution_bid.block_hash {
+            return Err(EnvelopeError::BlockHashMismatch {
+                committed_bid: execution_bid.block_hash,
+                envelope: payload.block_hash,
+            });
+        }
+
+        // TODO(EIP-7732): check these assumptions.. exactly what the most efficient way to verify the signatures
+        //                 in this case isn't clear. There are questions about the proposer cache, the pubkey cache,
+        //                 and so on.
+
+        // get the fork from the cache so we can verify the signature
+        let block_slot = envelope.slot;
+        let block_epoch = block_slot.epoch(T::EthSpec::slots_per_epoch());
+        let proposer_shuffling_decision_block =
+            parent_proto_block.proposer_shuffling_root_for_child_block(block_epoch, &chain.spec);
+        let mut opt_parent = None;
+        let envelope_ref = signed_envelope.as_ref();
+        let proposer = chain.with_proposer_cache::<_, EnvelopeError>(
+            proposer_shuffling_decision_block,
+            block_epoch,
+            |proposers| proposers.get_slot::<T::EthSpec>(block_slot),
+            || {
+                debug!(
+                    %beacon_block_root,
+                    block_hash = %envelope_ref.block_hash(),
+                    "Proposer shuffling cache miss for envelope verification"
+                );
+                // The proposer index was *not* cached and we must load the parent in order to
+                // determine the proposer index.
+                let snapshot = load_snapshot(envelope_ref, chain)?;
+                opt_parent = Some(Box::new(snapshot.clone()));
+                Ok((snapshot.state_root, snapshot.pre_state))
+            },
+        )?;
+        let fork = proposer.fork;
+
+        let signature_is_valid = {
+            let pubkey_cache = chain.validator_pubkey_cache.read();
+            let builder_pubkey = pubkey_cache
+                .get(envelope.builder_index as usize)
+                .ok_or_else(|| EnvelopeError::UnknownValidator {
+                    builder_index: envelope.builder_index,
+                })?;
+            signed_envelope.verify_signature(
+                builder_pubkey,
+                &fork,
+                chain.genesis_validators_root,
+                &chain.spec,
+            )
+        };
+
+        if !signature_is_valid {
+            return Err(EnvelopeError::BadSignature);
+        }
+
+        Ok(Self {
+            signed_envelope,
+            parent_block,
+            parent: opt_parent,
+        })
+    }
+
+    pub fn envelope_cloned(&self) -> Arc<SignedExecutionPayloadEnvelope<T::EthSpec>> {
+        self.signed_envelope.clone()
+    }
+}
+
+pub trait IntoExecutionPendingEnvelope<T: BeaconChainTypes>: Sized {
+    fn into_execution_pending_envelope(
+        self,
+        chain: &Arc<BeaconChain<T>>,
+        notify_execution_layer: NotifyExecutionLayer,
+    ) -> Result<ExecutionPendingEnvelope<T>, EnvelopeError>;
+}
+
+pub struct ExecutionPendingEnvelope<T: BeaconChainTypes> {
+    pub signed_envelope: MaybeAvailableEnvelope<T::EthSpec>,
+    pub import_data: EnvelopeImportData<T::EthSpec>,
+    pub payload_verification_handle: PayloadVerificationHandle,
+}
+
+impl<T: BeaconChainTypes> IntoExecutionPendingEnvelope<T> for GossipVerifiedEnvelope<T> {
+    fn into_execution_pending_envelope(
+        self,
+        chain: &Arc<BeaconChain<T>>,
+        notify_execution_layer: NotifyExecutionLayer,
+    ) -> Result<ExecutionPendingEnvelope<T>, EnvelopeError> {
+        let signed_envelope = self.signed_envelope;
+        let envelope = &signed_envelope.message;
+        let payload = &envelope.payload;
+
+        // Verify the execution payload is valid
+        let payload_notifier =
+            PayloadNotifier::from_envelope(chain.clone(), envelope, notify_execution_layer)?;
+        let block_root = envelope.beacon_block_root;
+        let slot = self.parent_block.slot();
+
+        let payload_verification_future = async move {
+            let chain = payload_notifier.chain.clone();
+            // TODO:(gloas): timing metrics
+            if let Some(started_execution) = chain.slot_clock.now_duration() {
+                chain.block_times_cache.write().set_time_started_execution(
+                    block_root,
+                    slot,
+                    started_execution,
+                );
+            }
+
+            let payload_verification_status = payload_notifier.notify_new_payload().await?;
+            Ok(PayloadVerificationOutcome {
+                payload_verification_status,
+                // This fork is after the merge so it'll never be the merge transition block
+                is_valid_merge_transition_block: false,
+            })
+        };
+        // Spawn the payload verification future as a new task, but don't wait for it to complete.
+        // The `payload_verification_future` will be awaited later to ensure verification completed
+        // successfully.
+        let payload_verification_handle = chain
+            .task_executor
+            .spawn_handle(
+                payload_verification_future,
+                "execution_payload_verification",
+            )
+            .ok_or(BeaconChainError::RuntimeShutdown)?;
+
+        let parent = if let Some(snapshot) = self.parent {
+            *snapshot
+        } else {
+            load_snapshot(signed_envelope.as_ref(), chain)?
+        };
+        let mut state = parent.pre_state;
+
+        // All the state modifications are done in envelope_processing
+        envelope_processing(
+            &mut state,
+            Some(parent.state_root),
+            &signed_envelope,
+            // verify signature already done for GossipVerifiedEnvelope
+            VerifySignatures::False,
+            &chain.spec,
+        )?;
+
+        Ok(ExecutionPendingEnvelope {
+            signed_envelope: MaybeAvailableEnvelope::AvailabilityPending {
+                block_hash: payload.block_hash,
+                envelope: signed_envelope,
+            },
+            import_data: EnvelopeImportData {
+                block_root,
+                parent_block: self.parent_block,
+                post_state: Box::new(state),
+            },
+            payload_verification_handle,
+        })
+    }
+}
+
+impl<T: BeaconChainTypes> IntoExecutionPendingEnvelope<T>
+    for Arc<SignedExecutionPayloadEnvelope<T::EthSpec>>
+{
+    fn into_execution_pending_envelope(
+        self,
+        chain: &Arc<BeaconChain<T>>,
+        notify_execution_layer: NotifyExecutionLayer,
+    ) -> Result<ExecutionPendingEnvelope<T>, EnvelopeError> {
+        // TODO(EIP-7732): figure out how this should be refactored..
+        GossipVerifiedEnvelope::new(self, chain)?
+            .into_execution_pending_envelope(chain, notify_execution_layer)
+    }
+}

beacon_node/beacon_chain/src/envelope_verification_types.rs

@@ -0,0 +1,30 @@
+use std::sync::Arc;
+use types::{
+    BeaconState, ChainSpec, DataColumnSidecarList, EthSpec, ExecutionBlockHash, Hash256,
+    SignedBeaconBlock, SignedExecutionPayloadEnvelope,
+};
+
+#[derive(PartialEq)]
+pub struct EnvelopeImportData<E: EthSpec> {
+    pub block_root: Hash256,
+    pub parent_block: Arc<SignedBeaconBlock<E>>,
+    pub post_state: Box<BeaconState<E>>,
+}
+
+#[derive(Debug)]
+#[allow(dead_code)]
+pub struct AvailableEnvelope<E: EthSpec> {
+    block_hash: ExecutionBlockHash,
+    envelope: Arc<SignedExecutionPayloadEnvelope<E>>,
+    columns: DataColumnSidecarList<E>,
+    /// Timestamp at which this block first became available (UNIX timestamp, time since 1970).
+    columns_available_timestamp: Option<std::time::Duration>,
+    pub spec: Arc<ChainSpec>,
+}
+pub enum MaybeAvailableEnvelope<E: EthSpec> {
+    Available(AvailableEnvelope<E>),
+    AvailabilityPending {
+        block_hash: ExecutionBlockHash,
+        envelope: Arc<SignedExecutionPayloadEnvelope<E>>,
+    },
+}

beacon_node/beacon_chain/src/errors.rs

@@ -149,6 +149,7 @@ pub enum BeaconChainError {
     EngineGetCapabilititesFailed(Box<execution_layer::Error>),
     ExecutionLayerGetBlockByNumberFailed(Box<execution_layer::Error>),
     ExecutionLayerGetBlockByHashFailed(Box<execution_layer::Error>),
+    ExecutionPayloadMissingFromDatabase(Hash256),
     BlockHashMissingFromExecutionLayer(ExecutionBlockHash),
     InconsistentPayloadReconstructed {
         slot: Slot,

beacon_node/beacon_chain/src/execution_payload.rs

@@ -9,7 +9,7 @@
 
 use crate::{
     BeaconChain, BeaconChainError, BeaconChainTypes, BlockError, BlockProductionError,
-    ExecutionPayloadError,
+    EnvelopeError, ExecutionPayloadError,
 };
 use execution_layer::{
     BlockProposalContents, BlockProposalContentsType, BuilderParams, NewPayloadRequest,
@@ -108,6 +108,16 @@ impl<T: BeaconChainTypes> PayloadNotifier<T> {
         })
     }
 
+    pub fn from_envelope(
+        _chain: Arc<BeaconChain<T>>,
+        _envelope: &ExecutionPayloadEnvelope<T::EthSpec>,
+        _notify_execution_layer: NotifyExecutionLayer,
+    ) -> Result<Self, EnvelopeError> {
+        todo!(
+            "this isn't a real method but something like this will be needed after refactoring this a bit"
+        );
+    }
+
     pub async fn notify_new_payload(self) -> Result<PayloadVerificationStatus, BlockError> {
         if let Some(precomputed_status) = self.payload_verification_status {
             Ok(precomputed_status)

beacon_node/beacon_chain/src/lib.rs

@@ -21,6 +21,8 @@ pub mod custody_context;
 pub mod data_availability_checker;
 pub mod data_column_verification;
 mod early_attester_cache;
+pub mod envelope_verification;
+pub mod envelope_verification_types;
 mod errors;
 pub mod events;
 pub mod execution_payload;
@@ -85,6 +87,7 @@ pub use block_verification_types::AvailabilityPendingExecutedBlock;
 pub use block_verification_types::ExecutedBlock;
 pub use canonical_head::{CachedHead, CanonicalHead, CanonicalHeadRwLock};
 pub use custody_context::CustodyContext;
+pub use envelope_verification::{EnvelopeError, GossipVerifiedEnvelope};
 pub use events::ServerSentEventHandler;
 pub use execution_layer::EngineState;
 pub use execution_payload::NotifyExecutionLayer;

beacon_node/beacon_chain/src/test_utils.rs

@@ -478,6 +478,7 @@ where
             execution_endpoint: url,
             secret_file: None,
             suggested_fee_recipient: Some(Address::repeat_byte(42)),
+            bypass_new_payload_cache: true,
             ..Default::default()
         };
         let execution_layer =

beacon_node/beacon_chain/tests/store_tests.rs

@@ -1412,7 +1412,7 @@ async fn proposer_shuffling_changing_with_lookahead() {
 
     let consolidation_request: ConsolidationRequest = ConsolidationRequest {
         source_address: validator_to_topup
-            .get_execution_withdrawal_address(spec)
+            .get_execution_withdrawal_address(spec, ForkName::Fulu)
             .unwrap(),
         source_pubkey: validator_to_topup.pubkey,
         target_pubkey: validator_to_topup.pubkey,
@@ -1491,7 +1491,7 @@ async fn proposer_shuffling_changing_with_lookahead() {
     let validator = current_epoch_state
         .get_validator(validator_to_topup_index)
         .unwrap();
-    assert!(validator.has_compounding_withdrawal_credential(spec));
+    assert!(validator.has_compounding_withdrawal_credential(spec, ForkName::Fulu));
     assert_eq!(validator.effective_balance, 95_000_000_000);
 
     // The shuffling for the current epoch from `prev_epoch_state` should match the shuffling

beacon_node/execution_layer/src/lib.rs

@@ -39,7 +39,7 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 use strum::AsRefStr;
 use task_executor::TaskExecutor;
 use tokio::{
-    sync::{Mutex, MutexGuard, RwLock},
+    sync::{Mutex, MutexGuard, RwLock, broadcast},
     time::sleep,
 };
 use tokio_stream::wrappers::WatchStream;
@@ -138,15 +138,15 @@ impl<E: EthSpec> TryFrom<BuilderBid<E>> for ProvenancedPayload<BlockProposalCont
     }
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub enum Error {
     NoEngine,
     NoPayloadBuilder,
-    ApiError(ApiError),
-    Builder(builder_client::Error),
+    ApiError(Arc<ApiError>),
+    Builder(Arc<builder_client::Error>),
     NoHeaderFromBuilder,
     CannotProduceHeader,
-    EngineError(Box<EngineError>),
+    EngineError(Arc<EngineError>),
     NotSynced,
     ShuttingDown,
     FeeRecipientUnspecified,
@@ -184,7 +184,7 @@ impl From<BeaconStateError> for Error {
 
 impl From<ApiError> for Error {
     fn from(e: ApiError) -> Self {
-        Error::ApiError(e)
+        Error::ApiError(Arc::new(e))
     }
 }
 
@@ -193,12 +193,18 @@ impl From<EngineError> for Error {
         match e {
             // This removes an unnecessary layer of indirection.
             // TODO (mark): consider refactoring these error enums
-            EngineError::Api { error } => Error::ApiError(error),
-            _ => Error::EngineError(Box::new(e)),
+            EngineError::Api { error } => Error::ApiError(Arc::new(error)),
+            _ => Error::EngineError(Arc::new(e)),
         }
     }
 }
 
+impl From<builder_client::Error> for Error {
+    fn from(e: builder_client::Error) -> Self {
+        Error::Builder(Arc::new(e))
+    }
+}
+
 pub enum BlockProposalContentsType<E: EthSpec> {
     Full(BlockProposalContents<E, FullPayload<E>>),
     Blinded(BlockProposalContents<E, BlindedPayload<E>>),
@@ -425,6 +431,108 @@ pub enum SubmitBlindedBlockResponse<E: EthSpec> {
 
 type PayloadContentsRefTuple<'a, E> = (ExecutionPayloadRef<'a, E>, Option<&'a BlobsBundle<E>>);
 
+/// Cache for deduplicating new payload requests.
+///
+/// Handles both in-flight requests and recently completed requests to avoid
+/// duplicate network calls to the execution engine.
+struct NewPayloadCache {
+    inner: Mutex<NewPayloadCacheInner>,
+}
+
+struct NewPayloadCacheInner {
+    /// In-flight requests mapped by block hash
+    in_flight: HashMap<ExecutionBlockHash, broadcast::Sender<Result<PayloadStatus, Error>>>,
+    /// Recently completed requests with their completion time
+    completed: LruCache<ExecutionBlockHash, (Instant, Result<PayloadStatus, Error>)>,
+}
+
+impl NewPayloadCache {
+    /// Cache TTL for completed requests (12 seconds)
+    const COMPLETED_TTL: Duration = Duration::from_secs(12);
+    /// Maximum number of completed requests to cache
+    const COMPLETED_CACHE_SIZE: NonZeroUsize = new_non_zero_usize(32);
+
+    fn new() -> Self {
+        Self {
+            inner: Mutex::new(NewPayloadCacheInner {
+                in_flight: HashMap::new(),
+                completed: LruCache::new(Self::COMPLETED_CACHE_SIZE),
+            }),
+        }
+    }
+
+    /// Get cached result or execute the provided function.
+    ///
+    /// Returns a future that resolves to the payload status. Handles:
+    /// 1. Returning cached completed results (if not expired)
+    /// 2. Joining in-flight requests
+    /// 3. Executing new requests and caching results
+    async fn get_or_execute<F, Fut>(
+        &self,
+        block_hash: ExecutionBlockHash,
+        execute_fn: F,
+    ) -> Result<PayloadStatus, Error>
+    where
+        F: FnOnce() -> Fut,
+        Fut: Future<Output = Result<PayloadStatus, Error>>,
+    {
+        let now = Instant::now();
+
+        // Single lock acquisition to handle all cases
+        let mut cache = self.inner.lock().await;
+
+        // 1. Check completed cache first
+        if let Some((timestamp, result)) = cache.completed.get(&block_hash) {
+            if now.duration_since(*timestamp) < Self::COMPLETED_TTL {
+                return result.clone();
+            } else {
+                // Entry expired, remove it
+                cache.completed.pop(&block_hash);
+            }
+        }
+
+        // 2. Check in-flight requests
+        if let Some(sender) = cache.in_flight.get(&block_hash) {
+            let mut receiver = sender.subscribe();
+            drop(cache); // Release lock early
+
+            match receiver.recv().await {
+                Ok(result) => return result,
+                Err(_) => {
+                    // Sender was dropped, fall through to execute new request
+                    error!(
+                        "NewPayloadCache: Sender was dropped for block hash {}. This shouldn't happen.",
+                        block_hash
+                    );
+                    // just call the execute_fn again
+                    return execute_fn().await;
+                }
+            }
+        }
+
+        // 3. Start new request
+        let (sender, _receiver) = broadcast::channel(1);
+        cache.in_flight.insert(block_hash, sender.clone());
+        drop(cache); // Release lock for execution
+
+        // Execute the function
+        let result = execute_fn().await;
+
+        // Update cache with result
+        let mut cache = self.inner.lock().await;
+        cache.in_flight.remove(&block_hash);
+        cache
+            .completed
+            .put(block_hash, (Instant::now(), result.clone()));
+        drop(cache);
+
+        // Broadcast result to any waiting receivers
+        let _ = sender.send(result.clone());
+
+        result
+    }
+}
+
 struct Inner<E: EthSpec> {
     engine: Arc<Engine>,
     builder: ArcSwapOption<BuilderHttpClient>,
@@ -440,6 +548,10 @@ struct Inner<E: EthSpec> {
     /// This is used *only* in the informational sync status endpoint, so that a VC using this
     /// node can prefer another node with a healthier EL.
     last_new_payload_errored: RwLock<bool>,
+    /// Cache for deduplicating `notify_new_payload` requests.
+    ///
+    /// Handles both in-flight requests and recently completed requests.
+    new_payload_cache: Option<NewPayloadCache>,
 }
 
 #[derive(Debug, Default, Clone, Serialize, Deserialize)]
@@ -467,6 +579,8 @@ pub struct Config {
     /// Default directory for the jwt secret if not provided through cli.
     pub default_datadir: PathBuf,
     pub execution_timeout_multiplier: Option<u32>,
+    /// Whether to bypass the new payload cache (useful for testing).
+    pub bypass_new_payload_cache: bool,
 }
 
 /// Provides access to one execution engine and provides a neat interface for consumption by the
@@ -491,6 +605,7 @@ impl<E: EthSpec> ExecutionLayer<E> {
             jwt_version,
             default_datadir,
             execution_timeout_multiplier,
+            bypass_new_payload_cache,
         } = config;
 
         let execution_url = url.ok_or(Error::NoEngine)?;
@@ -530,11 +645,17 @@ impl<E: EthSpec> ExecutionLayer<E> {
         let engine: Engine = {
             let auth = Auth::new(jwt_key, jwt_id, jwt_version);
             debug!(endpoint = %execution_url, jwt_path = ?secret_file.as_path(),"Loaded execution endpoint");
-            let api = HttpJsonRpc::new_with_auth(execution_url, auth, execution_timeout_multiplier)
-                .map_err(Error::ApiError)?;
+            let api =
+                HttpJsonRpc::new_with_auth(execution_url, auth, execution_timeout_multiplier)?;
             Engine::new(api, executor.clone())
         };
 
+        let new_payload_cache = if !bypass_new_payload_cache {
+            Some(NewPayloadCache::new())
+        } else {
+            None
+        };
+
         let inner = Inner {
             engine: Arc::new(engine),
             builder: ArcSwapOption::empty(),
@@ -546,6 +667,7 @@ impl<E: EthSpec> ExecutionLayer<E> {
             executor,
             payload_cache: PayloadCache::default(),
             last_new_payload_errored: RwLock::new(false),
+            new_payload_cache,
         };
 
         let el = Self {
@@ -589,7 +711,7 @@ impl<E: EthSpec> ExecutionLayer<E> {
             builder_header_timeout,
             disable_ssz,
         )
-        .map_err(Error::Builder)?;
+        .map_err(Into::<Error>::into)?;
         info!(
             ?builder_url,
             local_user_agent = builder_client.get_user_agent(),
@@ -1356,15 +1478,37 @@ impl<E: EthSpec> ExecutionLayer<E> {
                 Ok(GetPayloadResponseType::Full(payload_response))
             })
             .await
-            .map_err(Box::new)
-            .map_err(Error::EngineError)
+            .map_err(Into::into)
     }
 
     /// Maps to the `engine_newPayload` JSON-RPC call.
+    ///
+    /// Deduplicates concurrent requests with the same block hash - if multiple threads
+    /// call this function with the same block hash simultaneously, only one request
+    /// is sent to the execution engine, but all threads receive the same response.
+    /// Also caches recent results for a short time to avoid duplicate requests.
     /// TODO(EIP-7732) figure out how and why Mark relaxed new_payload_request param's typ to NewPayloadRequest<E>
     pub async fn notify_new_payload(
         &self,
         new_payload_request: NewPayloadRequest<'_, E>,
+    ) -> Result<PayloadStatus, Error> {
+        let block_hash = new_payload_request.block_hash();
+
+        if let Some(new_payload_cache) = &self.inner.new_payload_cache {
+            new_payload_cache
+                .get_or_execute(block_hash, || {
+                    self.notify_new_payload_impl(new_payload_request)
+                })
+                .await
+        } else {
+            self.notify_new_payload_impl(new_payload_request).await
+        }
+    }
+
+    /// Internal implementation of notify_new_payload without deduplication logic.
+    async fn notify_new_payload_impl(
+        &self,
+        new_payload_request: NewPayloadRequest<'_, E>,
     ) -> Result<PayloadStatus, Error> {
         let _timer = metrics::start_timer_vec(
             &metrics::EXECUTION_LAYER_REQUEST_TIMES,
@@ -1398,9 +1542,7 @@ impl<E: EthSpec> ExecutionLayer<E> {
         }
         *self.inner.last_new_payload_errored.write().await = result.is_err();
 
-        process_payload_status(block_hash, result)
-            .map_err(Box::new)
-            .map_err(Error::EngineError)
+        process_payload_status(block_hash, result).map_err(Into::into)
     }
 
     /// Update engine sync status.
@@ -1536,8 +1678,7 @@ impl<E: EthSpec> ExecutionLayer<E> {
             head_block_hash,
             result.map(|response| response.payload_status),
         )
-        .map_err(Box::new)
-        .map_err(Error::EngineError)
+        .map_err(Into::into)
     }
 
     /// Returns the execution engine capabilities resulting from a call to
@@ -1629,9 +1770,7 @@ impl<E: EthSpec> ExecutionLayer<E> {
                 }
                 Ok(block.map(|b| b.block_hash))
             })
-            .await
-            .map_err(Box::new)
-            .map_err(Error::EngineError)?;
+            .await?;
 
         if let Some(hash) = &hash_opt {
             info!(
@@ -1741,8 +1880,7 @@ impl<E: EthSpec> ExecutionLayer<E> {
                 Ok(None)
             })
             .await
-            .map_err(Box::new)
-            .map_err(Error::EngineError)
+            .map_err(Into::into)
     }
 
     /// This function should remain internal.
@@ -1793,8 +1931,7 @@ impl<E: EthSpec> ExecutionLayer<E> {
                 engine.api.get_payload_bodies_by_hash_v1(hashes).await
             })
             .await
-            .map_err(Box::new)
-            .map_err(Error::EngineError)
+            .map_err(Into::into)
     }
 
     pub async fn get_payload_bodies_by_range(
@@ -1811,8 +1948,7 @@ impl<E: EthSpec> ExecutionLayer<E> {
                     .await
             })
             .await
-            .map_err(Box::new)
-            .map_err(Error::EngineError)
+            .map_err(Into::into)
     }
 
     /// Fetch a full payload from the execution node.
@@ -1874,8 +2010,7 @@ impl<E: EthSpec> ExecutionLayer<E> {
             self.engine()
                 .request(|engine| async move { engine.api.get_blobs_v1(query).await })
                 .await
-                .map_err(Box::new)
-                .map_err(Error::EngineError)
+                .map_err(Into::into)
         } else {
             Err(Error::GetBlobsNotSupported)
         }
@@ -1891,8 +2026,7 @@ impl<E: EthSpec> ExecutionLayer<E> {
             self.engine()
                 .request(|engine| async move { engine.api.get_blobs_v2(query).await })
                 .await
-                .map_err(Box::new)
-                .map_err(Error::EngineError)
+                .map_err(Into::into)
         } else {
             Err(Error::GetBlobsNotSupported)
         }
@@ -1905,8 +2039,7 @@ impl<E: EthSpec> ExecutionLayer<E> {
         self.engine()
             .request(|engine| async move { engine.api.get_block_by_number(query).await })
             .await
-            .map_err(Box::new)
-            .map_err(Error::EngineError)
+            .map_err(Into::into)
     }
 
     pub async fn propose_blinded_beacon_block(
@@ -1955,12 +2088,12 @@ impl<E: EthSpec> ExecutionLayer<E> {
                         builder
                             .post_builder_blinded_blocks_v1_ssz(block)
                             .await
-                            .map_err(Error::Builder)
+                            .map_err(Into::into)
                     } else {
                         builder
                             .post_builder_blinded_blocks_v1(block)
                             .await
-                            .map_err(Error::Builder)
+                            .map_err(Into::into)
                             .map(|d| d.data)
                     }
                 })
@@ -2025,12 +2158,12 @@ impl<E: EthSpec> ExecutionLayer<E> {
                     builder
                         .post_builder_blinded_blocks_v2_ssz(block)
                         .await
-                        .map_err(Error::Builder)
+                        .map_err(Into::into)
                 } else {
                     builder
                         .post_builder_blinded_blocks_v2(block)
                         .await
-                        .map_err(Error::Builder)
+                        .map_err(Into::into)
                 }
             })
             .await;

beacon_node/execution_layer/src/test_utils/mock_execution_layer.rs

@@ -77,6 +77,7 @@ impl<E: EthSpec> MockExecutionLayer<E> {
             execution_endpoint: Some(url),
             secret_file: Some(path),
             suggested_fee_recipient: Some(Address::repeat_byte(42)),
+            bypass_new_payload_cache: true,
             ..Default::default()
         };
         let el = ExecutionLayer::from_config(config, executor.clone()).unwrap();

beacon_node/execution_layer/src/versioned_hashes.rs

@@ -3,7 +3,7 @@ use alloy_rlp::Decodable;
 use typenum::Unsigned;
 use types::{EthSpec, ExecutionPayloadRef, Hash256, VersionedHash};
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub enum Error {
     DecodingTransaction(String),
     LengthMismatch { expected: usize, found: usize },

beacon_node/http_api/src/builder_states.rs

@@ -32,7 +32,7 @@ pub fn get_next_withdrawals<T: BeaconChainTypes>(
     }
 
     match get_expected_withdrawals(&state, &chain.spec) {
-        Ok((withdrawals, _)) => Ok(withdrawals),
+        Ok((withdrawals, _, _)) => Ok(withdrawals),
         Err(e) => Err(warp_utils::reject::custom_server_error(format!(
             "failed to get expected withdrawal: {:?}",
             e

beacon_node/operation_pool/src/bls_to_execution_changes.rs

@@ -113,16 +113,18 @@ impl<E: EthSpec> BlsToExecutionChanges<E> {
                 .validators()
                 .get(validator_index as usize)
                 .is_none_or(|validator| {
-                    let prune = validator.has_execution_withdrawal_credential(spec)
-                        && head_block
-                            .message()
-                            .body()
-                            .bls_to_execution_changes()
-                            .map_or(true, |recent_changes| {
-                                !recent_changes
-                                    .iter()
-                                    .any(|c| c.message.validator_index == validator_index)
-                            });
+                    let prune = validator.has_execution_withdrawal_credential(
+                        spec,
+                        head_state.fork_name_unchecked(),
+                    ) && head_block
+                        .message()
+                        .body()
+                        .bls_to_execution_changes()
+                        .map_or(true, |recent_changes| {
+                            !recent_changes
+                                .iter()
+                                .any(|c| c.message.validator_index == validator_index)
+                        });
                     if prune {
                         validator_indices_pruned.push(validator_index);
                     }

beacon_node/operation_pool/src/lib.rs

@@ -582,7 +582,12 @@ impl<E: EthSpec> OperationPool<E> {
                 address_change.signature_is_still_valid(&state.fork())
                     && state
                         .get_validator(address_change.as_inner().message.validator_index as usize)
-                        .is_ok_and(|validator| !validator.has_execution_withdrawal_credential(spec))
+                        .is_ok_and(|validator| {
+                            !validator.has_execution_withdrawal_credential(
+                                spec,
+                                state.fork_name_unchecked(),
+                            )
+                        })
             },
             |address_change| address_change.as_inner().clone(),
             E::MaxBlsToExecutionChanges::to_usize(),

beacon_node/store/src/consensus_context.rs

@@ -35,6 +35,8 @@ impl<E: EthSpec> OnDiskConsensusContext<E> {
             proposer_index,
             current_block_root,
             indexed_attestations,
+            indexed_payload_attestations: _,
+            // TODO(EIP-7732): add indexed_payload_attestations to the on-disk format.
         } = ctxt;
         OnDiskConsensusContext {
             slot,

beacon_node/store/src/partial_beacon_state.rs

@@ -0,0 +1,511 @@
+use crate::chunked_vector::{
+    BlockRootsChunked, HistoricalRoots, HistoricalSummaries, RandaoMixes, StateRootsChunked,
+    load_variable_list_from_db, load_vector_from_db,
+};
+use crate::{DBColumn, Error, KeyValueStore, KeyValueStoreOp};
+use milhouse::{List, Vector};
+use ssz::{BitVector, Decode, DecodeError, Encode};
+use ssz_derive::{Decode, Encode};
+use std::sync::Arc;
+use superstruct::superstruct;
+use types::historical_summary::HistoricalSummary;
+use types::*;
+
+/// DEPRECATED Lightweight variant of the `BeaconState` that is stored in the database.
+///
+/// Utilises lazy-loading from separate storage for its vector fields.
+///
+/// This can be deleted once schema versions prior to V22 are no longer supported.
+#[superstruct(
+    variants(Base, Altair, Bellatrix, Capella, Deneb, Electra, Fulu, Gloas),
+    variant_attributes(derive(Debug, PartialEq, Clone, Encode, Decode))
+)]
+#[derive(Debug, PartialEq, Clone, Encode)]
+#[ssz(enum_behaviour = "transparent")]
+pub struct PartialBeaconState<E>
+where
+    E: EthSpec,
+{
+    // Versioning
+    pub genesis_time: u64,
+    pub genesis_validators_root: Hash256,
+    #[superstruct(getter(copy))]
+    pub slot: Slot,
+    pub fork: Fork,
+
+    // History
+    pub latest_block_header: BeaconBlockHeader,
+
+    #[ssz(skip_serializing, skip_deserializing)]
+    pub block_roots: Option<Vector<Hash256, E::SlotsPerHistoricalRoot>>,
+    #[ssz(skip_serializing, skip_deserializing)]
+    pub state_roots: Option<Vector<Hash256, E::SlotsPerHistoricalRoot>>,
+
+    #[ssz(skip_serializing, skip_deserializing)]
+    pub historical_roots: Option<List<Hash256, E::HistoricalRootsLimit>>,
+
+    // Ethereum 1.0 chain data
+    pub eth1_data: Eth1Data,
+    pub eth1_data_votes: List<Eth1Data, E::SlotsPerEth1VotingPeriod>,
+    pub eth1_deposit_index: u64,
+
+    // Registry
+    pub validators: List<Validator, E::ValidatorRegistryLimit>,
+    pub balances: List<u64, E::ValidatorRegistryLimit>,
+
+    // Shuffling
+    /// Randao value from the current slot, for patching into the per-epoch randao vector.
+    pub latest_randao_value: Hash256,
+    #[ssz(skip_serializing, skip_deserializing)]
+    pub randao_mixes: Option<Vector<Hash256, E::EpochsPerHistoricalVector>>,
+
+    // Slashings
+    slashings: Vector<u64, E::EpochsPerSlashingsVector>,
+
+    // Attestations (genesis fork only)
+    #[superstruct(only(Base))]
+    pub previous_epoch_attestations: List<PendingAttestation<E>, E::MaxPendingAttestations>,
+    #[superstruct(only(Base))]
+    pub current_epoch_attestations: List<PendingAttestation<E>, E::MaxPendingAttestations>,
+
+    // Participation (Altair and later)
+    #[superstruct(only(Altair, Bellatrix, Capella, Deneb, Electra, Fulu, Gloas))]
+    pub previous_epoch_participation: List<ParticipationFlags, E::ValidatorRegistryLimit>,
+    #[superstruct(only(Altair, Bellatrix, Capella, Deneb, Electra, Fulu, Gloas))]
+    pub current_epoch_participation: List<ParticipationFlags, E::ValidatorRegistryLimit>,
+
+    // Finality
+    pub justification_bits: BitVector<E::JustificationBitsLength>,
+    pub previous_justified_checkpoint: Checkpoint,
+    pub current_justified_checkpoint: Checkpoint,
+    pub finalized_checkpoint: Checkpoint,
+
+    // Inactivity
+    #[superstruct(only(Altair, Bellatrix, Capella, Deneb, Electra, Fulu, Gloas))]
+    pub inactivity_scores: List<u64, E::ValidatorRegistryLimit>,
+
+    // Light-client sync committees
+    #[superstruct(only(Altair, Bellatrix, Capella, Deneb, Electra, Fulu, Gloas))]
+    pub current_sync_committee: Arc<SyncCommittee<E>>,
+    #[superstruct(only(Altair, Bellatrix, Capella, Deneb, Electra, Fulu, Gloas))]
+    pub next_sync_committee: Arc<SyncCommittee<E>>,
+
+    // Execution
+    #[superstruct(
+        only(Bellatrix),
+        partial_getter(rename = "latest_execution_payload_header_bellatrix")
+    )]
+    pub latest_execution_payload_header: ExecutionPayloadHeaderBellatrix<E>,
+    #[superstruct(
+        only(Capella),
+        partial_getter(rename = "latest_execution_payload_header_capella")
+    )]
+    pub latest_execution_payload_header: ExecutionPayloadHeaderCapella<E>,
+    #[superstruct(
+        only(Deneb),
+        partial_getter(rename = "latest_execution_payload_header_deneb")
+    )]
+    pub latest_execution_payload_header: ExecutionPayloadHeaderDeneb<E>,
+    #[superstruct(
+        only(Electra),
+        partial_getter(rename = "latest_execution_payload_header_electra")
+    )]
+    pub latest_execution_payload_header: ExecutionPayloadHeaderElectra<E>,
+    #[superstruct(
+        only(Fulu),
+        partial_getter(rename = "latest_execution_payload_header_fulu")
+    )]
+    pub latest_execution_payload_header: ExecutionPayloadHeaderFulu<E>,
+
+    #[superstruct(
+        only(Gloas),
+        partial_getter(rename = "latest_execution_payload_bid_gloas")
+    )]
+    pub latest_execution_payload_bid: ExecutionPayloadBid,
+
+    // Capella
+    #[superstruct(only(Capella, Deneb, Electra, Fulu, Gloas))]
+    pub next_withdrawal_index: u64,
+    #[superstruct(only(Capella, Deneb, Electra, Fulu, Gloas))]
+    pub next_withdrawal_validator_index: u64,
+
+    #[ssz(skip_serializing, skip_deserializing)]
+    #[superstruct(only(Capella, Deneb, Electra, Fulu, Gloas))]
+    pub historical_summaries: Option<List<HistoricalSummary, E::HistoricalRootsLimit>>,
+
+    // Electra
+    #[superstruct(only(Electra, Fulu, Gloas))]
+    pub deposit_requests_start_index: u64,
+    #[superstruct(only(Electra, Fulu, Gloas))]
+    pub deposit_balance_to_consume: u64,
+    #[superstruct(only(Electra, Fulu, Gloas))]
+    pub exit_balance_to_consume: u64,
+    #[superstruct(only(Electra, Fulu, Gloas))]
+    pub earliest_exit_epoch: Epoch,
+    #[superstruct(only(Electra, Fulu, Gloas))]
+    pub consolidation_balance_to_consume: u64,
+    #[superstruct(only(Electra, Fulu, Gloas))]
+    pub earliest_consolidation_epoch: Epoch,
+
+    #[superstruct(only(Electra, Fulu, Gloas))]
+    pub pending_deposits: List<PendingDeposit, E::PendingDepositsLimit>,
+    #[superstruct(only(Electra, Fulu, Gloas))]
+    pub pending_partial_withdrawals:
+        List<PendingPartialWithdrawal, E::PendingPartialWithdrawalsLimit>,
+    #[superstruct(only(Electra, Fulu, Gloas))]
+    pub pending_consolidations: List<PendingConsolidation, E::PendingConsolidationsLimit>,
+    #[superstruct(only(Fulu, Gloas))]
+    pub proposer_lookahead: Vector<u64, E::ProposerLookaheadSlots>,
+
+    // Gloas
+    #[superstruct(only(Gloas))]
+    pub execution_payload_availability: BitVector<E::SlotsPerHistoricalRoot>,
+
+    #[superstruct(only(Gloas))]
+    pub builder_pending_payments: Vector<BuilderPendingPayment, E::BuilderPendingPaymentsLimit>,
+
+    #[superstruct(only(Gloas))]
+    pub builder_pending_withdrawals:
+        List<BuilderPendingWithdrawal, E::BuilderPendingWithdrawalsLimit>,
+
+    #[superstruct(only(Gloas))]
+    pub latest_block_hash: ExecutionBlockHash,
+
+    #[superstruct(only(Gloas))]
+    pub latest_withdrawals_root: Hash256,
+}
+
+impl<E: EthSpec> PartialBeaconState<E> {
+    /// SSZ decode.
+    pub fn from_ssz_bytes(bytes: &[u8], spec: &ChainSpec) -> Result<Self, ssz::DecodeError> {
+        // Slot is after genesis_time (u64) and genesis_validators_root (Hash256).
+        let slot_offset = <u64 as Decode>::ssz_fixed_len() + <Hash256 as Decode>::ssz_fixed_len();
+        let slot_len = <Slot as Decode>::ssz_fixed_len();
+        let slot_bytes = bytes.get(slot_offset..slot_offset + slot_len).ok_or(
+            DecodeError::InvalidByteLength {
+                len: bytes.len(),
+                expected: slot_offset + slot_len,
+            },
+        )?;
+
+        let slot = Slot::from_ssz_bytes(slot_bytes)?;
+        let fork_at_slot = spec.fork_name_at_slot::<E>(slot);
+
+        Ok(map_fork_name!(
+            fork_at_slot,
+            Self,
+            <_>::from_ssz_bytes(bytes)?
+        ))
+    }
+
+    /// Prepare the partial state for storage in the KV database.
+    pub fn as_kv_store_op(&self, state_root: Hash256) -> KeyValueStoreOp {
+        KeyValueStoreOp::PutKeyValue(
+            DBColumn::BeaconState,
+            state_root.as_slice().to_vec(),
+            self.as_ssz_bytes(),
+        )
+    }
+
+    pub fn load_block_roots<S: KeyValueStore<E>>(
+        &mut self,
+        store: &S,
+        spec: &ChainSpec,
+    ) -> Result<(), Error> {
+        if self.block_roots().is_none() {
+            *self.block_roots_mut() = Some(load_vector_from_db::<BlockRootsChunked, E, _>(
+                store,
+                self.slot(),
+                spec,
+            )?);
+        }
+        Ok(())
+    }
+
+    pub fn load_state_roots<S: KeyValueStore<E>>(
+        &mut self,
+        store: &S,
+        spec: &ChainSpec,
+    ) -> Result<(), Error> {
+        if self.state_roots().is_none() {
+            *self.state_roots_mut() = Some(load_vector_from_db::<StateRootsChunked, E, _>(
+                store,
+                self.slot(),
+                spec,
+            )?);
+        }
+        Ok(())
+    }
+
+    pub fn load_historical_roots<S: KeyValueStore<E>>(
+        &mut self,
+        store: &S,
+        spec: &ChainSpec,
+    ) -> Result<(), Error> {
+        if self.historical_roots().is_none() {
+            *self.historical_roots_mut() = Some(
+                load_variable_list_from_db::<HistoricalRoots, E, _>(store, self.slot(), spec)?,
+            );
+        }
+        Ok(())
+    }
+
+    pub fn load_historical_summaries<S: KeyValueStore<E>>(
+        &mut self,
+        store: &S,
+        spec: &ChainSpec,
+    ) -> Result<(), Error> {
+        let slot = self.slot();
+        if let Ok(historical_summaries) = self.historical_summaries_mut()
+            && historical_summaries.is_none()
+        {
+            *historical_summaries = Some(load_variable_list_from_db::<HistoricalSummaries, E, _>(
+                store, slot, spec,
+            )?);
+        }
+        Ok(())
+    }
+
+    pub fn load_randao_mixes<S: KeyValueStore<E>>(
+        &mut self,
+        store: &S,
+        spec: &ChainSpec,
+    ) -> Result<(), Error> {
+        if self.randao_mixes().is_none() {
+            // Load the per-epoch values from the database
+            let mut randao_mixes =
+                load_vector_from_db::<RandaoMixes, E, _>(store, self.slot(), spec)?;
+
+            // Patch the value for the current slot into the index for the current epoch
+            let current_epoch = self.slot().epoch(E::slots_per_epoch());
+            let len = randao_mixes.len();
+            *randao_mixes
+                .get_mut(current_epoch.as_usize() % len)
+                .ok_or(Error::RandaoMixOutOfBounds)? = *self.latest_randao_value();
+
+            *self.randao_mixes_mut() = Some(randao_mixes)
+        }
+        Ok(())
+    }
+}
+
+/// Implement the conversion from PartialBeaconState -> BeaconState.
+macro_rules! impl_try_into_beacon_state {
+    ($inner:ident, $variant_name:ident, $struct_name:ident, [$($extra_fields:ident),*], [$($extra_opt_fields:ident),*]) => {
+        BeaconState::$variant_name($struct_name {
+            // Versioning
+            genesis_time: $inner.genesis_time,
+            genesis_validators_root: $inner.genesis_validators_root,
+            slot: $inner.slot,
+            fork: $inner.fork,
+
+            // History
+            latest_block_header: $inner.latest_block_header,
+            block_roots: unpack_field($inner.block_roots)?,
+            state_roots: unpack_field($inner.state_roots)?,
+            historical_roots: unpack_field($inner.historical_roots)?,
+
+            // Eth1
+            eth1_data: $inner.eth1_data,
+            eth1_data_votes: $inner.eth1_data_votes,
+            eth1_deposit_index: $inner.eth1_deposit_index,
+
+            // Validator registry
+            validators: $inner.validators,
+            balances: $inner.balances,
+
+            // Shuffling
+            randao_mixes: unpack_field($inner.randao_mixes)?,
+
+            // Slashings
+            slashings: $inner.slashings,
+
+            // Finality
+            justification_bits: $inner.justification_bits,
+            previous_justified_checkpoint: $inner.previous_justified_checkpoint,
+            current_justified_checkpoint: $inner.current_justified_checkpoint,
+            finalized_checkpoint: $inner.finalized_checkpoint,
+
+            // Caching
+            total_active_balance: <_>::default(),
+            progressive_balances_cache: <_>::default(),
+            committee_caches: <_>::default(),
+            pubkey_cache: <_>::default(),
+            exit_cache: <_>::default(),
+            slashings_cache: <_>::default(),
+            epoch_cache: <_>::default(),
+
+            // Variant-specific fields
+            $(
+                $extra_fields: $inner.$extra_fields
+            ),*,
+
+            // Variant-specific optional fields
+            $(
+                $extra_opt_fields: unpack_field($inner.$extra_opt_fields)?
+            ),*
+        })
+    }
+}
+
+fn unpack_field<T>(x: Option<T>) -> Result<T, Error> {
+    x.ok_or(Error::PartialBeaconStateError)
+}
+
+impl<E: EthSpec> TryInto<BeaconState<E>> for PartialBeaconState<E> {
+    type Error = Error;
+
+    fn try_into(self) -> Result<BeaconState<E>, Error> {
+        let state = match self {
+            PartialBeaconState::Base(inner) => impl_try_into_beacon_state!(
+                inner,
+                Base,
+                BeaconStateBase,
+                [previous_epoch_attestations, current_epoch_attestations],
+                []
+            ),
+            PartialBeaconState::Altair(inner) => impl_try_into_beacon_state!(
+                inner,
+                Altair,
+                BeaconStateAltair,
+                [
+                    previous_epoch_participation,
+                    current_epoch_participation,
+                    current_sync_committee,
+                    next_sync_committee,
+                    inactivity_scores
+                ],
+                []
+            ),
+            PartialBeaconState::Bellatrix(inner) => impl_try_into_beacon_state!(
+                inner,
+                Bellatrix,
+                BeaconStateBellatrix,
+                [
+                    previous_epoch_participation,
+                    current_epoch_participation,
+                    current_sync_committee,
+                    next_sync_committee,
+                    inactivity_scores,
+                    latest_execution_payload_header
+                ],
+                []
+            ),
+            PartialBeaconState::Capella(inner) => impl_try_into_beacon_state!(
+                inner,
+                Capella,
+                BeaconStateCapella,
+                [
+                    previous_epoch_participation,
+                    current_epoch_participation,
+                    current_sync_committee,
+                    next_sync_committee,
+                    inactivity_scores,
+                    latest_execution_payload_header,
+                    next_withdrawal_index,
+                    next_withdrawal_validator_index
+                ],
+                [historical_summaries]
+            ),
+            PartialBeaconState::Deneb(inner) => impl_try_into_beacon_state!(
+                inner,
+                Deneb,
+                BeaconStateDeneb,
+                [
+                    previous_epoch_participation,
+                    current_epoch_participation,
+                    current_sync_committee,
+                    next_sync_committee,
+                    inactivity_scores,
+                    latest_execution_payload_header,
+                    next_withdrawal_index,
+                    next_withdrawal_validator_index
+                ],
+                [historical_summaries]
+            ),
+            PartialBeaconState::Electra(inner) => impl_try_into_beacon_state!(
+                inner,
+                Electra,
+                BeaconStateElectra,
+                [
+                    previous_epoch_participation,
+                    current_epoch_participation,
+                    current_sync_committee,
+                    next_sync_committee,
+                    inactivity_scores,
+                    latest_execution_payload_header,
+                    next_withdrawal_index,
+                    next_withdrawal_validator_index,
+                    deposit_requests_start_index,
+                    deposit_balance_to_consume,
+                    exit_balance_to_consume,
+                    earliest_exit_epoch,
+                    consolidation_balance_to_consume,
+                    earliest_consolidation_epoch,
+                    pending_deposits,
+                    pending_partial_withdrawals,
+                    pending_consolidations
+                ],
+                [historical_summaries]
+            ),
+            PartialBeaconState::Fulu(inner) => impl_try_into_beacon_state!(
+                inner,
+                Fulu,
+                BeaconStateFulu,
+                [
+                    previous_epoch_participation,
+                    current_epoch_participation,
+                    current_sync_committee,
+                    next_sync_committee,
+                    inactivity_scores,
+                    latest_execution_payload_header,
+                    next_withdrawal_index,
+                    next_withdrawal_validator_index,
+                    deposit_requests_start_index,
+                    deposit_balance_to_consume,
+                    exit_balance_to_consume,
+                    earliest_exit_epoch,
+                    consolidation_balance_to_consume,
+                    earliest_consolidation_epoch,
+                    pending_deposits,
+                    pending_partial_withdrawals,
+                    pending_consolidations,
+                    proposer_lookahead
+                ],
+                [historical_summaries]
+            ),
+            PartialBeaconState::Gloas(inner) => impl_try_into_beacon_state!(
+                inner,
+                Gloas,
+                BeaconStateGloas,
+                [
+                    previous_epoch_participation,
+                    current_epoch_participation,
+                    current_sync_committee,
+                    next_sync_committee,
+                    inactivity_scores,
+                    latest_execution_payload_bid,
+                    next_withdrawal_index,
+                    next_withdrawal_validator_index,
+                    deposit_requests_start_index,
+                    deposit_balance_to_consume,
+                    exit_balance_to_consume,
+                    earliest_exit_epoch,
+                    consolidation_balance_to_consume,
+                    earliest_consolidation_epoch,
+                    pending_deposits,
+                    pending_partial_withdrawals,
+                    pending_consolidations,
+                    proposer_lookahead,
+                    execution_payload_availability,
+                    builder_pending_payments,
+                    builder_pending_withdrawals,
+                    latest_block_hash,
+                    latest_withdrawals_root
+                ],
+                [historical_summaries]
+            ),
+        };
+        Ok(state)
+    }
+}

consensus/state_processing/src/common/get_payload_attesting_indices.rs

@@ -0,0 +1,45 @@
+use crate::per_block_processing::errors::{
+    BlockOperationError, PayloadAttestationInvalid as Invalid,
+};
+use ssz_types::VariableList;
+use typenum::Unsigned;
+use types::*;
+
+pub fn get_indexed_payload_attestation<E: EthSpec>(
+    state: &BeaconState<E>,
+    slot: Slot,
+    payload_attestation: &PayloadAttestation<E>,
+    spec: &ChainSpec,
+) -> Result<IndexedPayloadAttestation<E>, BlockOperationError<Invalid>> {
+    let attesting_indices = get_payload_attesting_indices(state, slot, payload_attestation, spec)?;
+
+    Ok(IndexedPayloadAttestation {
+        attesting_indices: VariableList::new(attesting_indices)?,
+        data: payload_attestation.data.clone(),
+        signature: payload_attestation.signature.clone(),
+    })
+}
+
+pub fn get_payload_attesting_indices<E: EthSpec>(
+    state: &BeaconState<E>,
+    slot: Slot,
+    payload_attestation: &PayloadAttestation<E>,
+    spec: &ChainSpec,
+) -> Result<Vec<u64>, BeaconStateError> {
+    let ptc = state.get_ptc(slot, spec)?;
+
+    let bitlist = &payload_attestation.aggregation_bits;
+    if bitlist.len() != E::PTCSize::to_usize() {
+        return Err(BeaconStateError::InvalidBitfield);
+    }
+
+    let mut attesting_indices = Vec::<u64>::new();
+    for (i, index) in ptc.into_iter().enumerate() {
+        if let Ok(true) = bitlist.get(i) {
+            attesting_indices.push(index as u64);
+        }
+    }
+    attesting_indices.sort_unstable();
+
+    Ok(attesting_indices)
+}

consensus/state_processing/src/common/mod.rs

@@ -1,6 +1,7 @@
 mod deposit_data_tree;
 mod get_attestation_participation;
 mod get_attesting_indices;
+mod get_payload_attesting_indices;
 mod initiate_validator_exit;
 mod slash_validator;
 
@@ -13,6 +14,9 @@ pub use get_attestation_participation::get_attestation_participation_flag_indice
 pub use get_attesting_indices::{
     attesting_indices_base, attesting_indices_electra, get_attesting_indices_from_state,
 };
+pub use get_payload_attesting_indices::{
+    get_indexed_payload_attestation, get_payload_attesting_indices,
+};
 pub use initiate_validator_exit::initiate_validator_exit;
 pub use slash_validator::slash_validator;
 

consensus/state_processing/src/consensus_context.rs

@@ -1,11 +1,16 @@
 use crate::EpochCacheError;
-use crate::common::{attesting_indices_base, attesting_indices_electra};
-use crate::per_block_processing::errors::{AttestationInvalid, BlockOperationError};
+use crate::common::{
+    attesting_indices_base, attesting_indices_electra, get_indexed_payload_attestation,
+};
+use crate::per_block_processing::errors::{
+    AttestationInvalid, BlockOperationError, PayloadAttestationInvalid,
+};
 use std::collections::{HashMap, hash_map::Entry};
 use tree_hash::TreeHash;
 use types::{
     AbstractExecPayload, AttestationRef, BeaconState, BeaconStateError, ChainSpec, Epoch, EthSpec,
-    Hash256, IndexedAttestation, IndexedAttestationRef, SignedBeaconBlock, Slot,
+    Hash256, IndexedAttestation, IndexedAttestationRef, IndexedPayloadAttestation,
+    PayloadAttestation, SignedBeaconBlock, Slot,
 };
 
 #[derive(Debug, PartialEq, Clone)]
@@ -22,6 +27,8 @@ pub struct ConsensusContext<E: EthSpec> {
     pub current_block_root: Option<Hash256>,
     /// Cache of indexed attestations constructed during block processing.
     pub indexed_attestations: HashMap<Hash256, IndexedAttestation<E>>,
+    /// Cache of indexed payload attestations constructed during block processing.
+    pub indexed_payload_attestations: HashMap<Hash256, IndexedPayloadAttestation<E>>,
 }
 
 #[derive(Debug, PartialEq, Clone)]
@@ -55,6 +62,7 @@ impl<E: EthSpec> ConsensusContext<E> {
             proposer_index: None,
             current_block_root: None,
             indexed_attestations: HashMap::new(),
+            indexed_payload_attestations: HashMap::new(),
         }
     }
 
@@ -177,6 +185,25 @@ impl<E: EthSpec> ConsensusContext<E> {
         .map(|indexed_attestation| (*indexed_attestation).to_ref())
     }
 
+    pub fn get_indexed_payload_attestation<'a>(
+        &'a mut self,
+        state: &BeaconState<E>,
+        slot: Slot,
+        payload_attestation: &'a PayloadAttestation<E>,
+        spec: &ChainSpec,
+    ) -> Result<&'a IndexedPayloadAttestation<E>, BlockOperationError<PayloadAttestationInvalid>>
+    {
+        let key = payload_attestation.tree_hash_root();
+        match self.indexed_payload_attestations.entry(key) {
+            Entry::Occupied(occupied) => Ok(occupied.into_mut()),
+            Entry::Vacant(vacant) => {
+                let indexed_payload_attestation =
+                    get_indexed_payload_attestation(state, slot, payload_attestation, spec)?;
+                Ok(vacant.insert(indexed_payload_attestation))
+            }
+        }
+    }
+
     pub fn num_cached_indexed_attestations(&self) -> usize {
         self.indexed_attestations.len()
     }

consensus/state_processing/src/envelope_processing.rs

@@ -0,0 +1,284 @@
+use crate::BlockProcessingError;
+use crate::VerifySignatures;
+use crate::per_block_processing::compute_timestamp_at_slot;
+use crate::per_block_processing::process_operations::{
+    process_consolidation_requests, process_deposit_requests, process_withdrawal_requests,
+};
+use safe_arith::{ArithError, SafeArith};
+use tree_hash::TreeHash;
+use types::{
+    BeaconState, BeaconStateError, BuilderPendingPayment, ChainSpec, EthSpec, ExecutionBlockHash,
+    Hash256, SignedExecutionPayloadEnvelope, Slot,
+};
+
+// TODO(EIP-7732): don't use this redefinition..
+macro_rules! envelope_verify {
+    ($condition: expr, $result: expr) => {
+        if !$condition {
+            return Err($result);
+        }
+    };
+}
+
+#[derive(Debug, Clone)]
+pub enum EnvelopeProcessingError {
+    /// Bad Signature
+    BadSignature,
+    BeaconStateError(BeaconStateError),
+    BlockProcessingError(BlockProcessingError),
+    ArithError(ArithError),
+    /// Envelope doesn't match latest beacon block header
+    LatestBlockHeaderMismatch {
+        envelope_root: Hash256,
+        block_header_root: Hash256,
+    },
+    /// Envelope doesn't match latest beacon block slot
+    SlotMismatch {
+        envelope_slot: Slot,
+        parent_state_slot: Slot,
+    },
+    /// The withdrawals root doesn't match the state's latest withdrawals root
+    WithdrawalsRootMismatch {
+        state: Hash256,
+        envelope: Hash256,
+    },
+    // The gas limit doesn't match the committed bid
+    GasLimitMismatch {
+        committed_bid: u64,
+        envelope: u64,
+    },
+    // The block hash doesn't match the committed bid
+    BlockHashMismatch {
+        committed_bid: ExecutionBlockHash,
+        envelope: ExecutionBlockHash,
+    },
+    // The parent hash doesn't match the previous execution payload
+    ParentHashMismatch {
+        state: ExecutionBlockHash,
+        envelope: ExecutionBlockHash,
+    },
+    /// The blob KZG commitments root doesn't match the committed bid
+    BlobKzgCommitmentsRootMismatch {
+        committed_bid: Hash256,
+        envelope: Hash256,
+    },
+    // The previous randao didn't match the payload
+    PrevRandaoMismatch {
+        state: Hash256,
+        envelope: Hash256,
+    },
+    // The timestamp didn't match the payload
+    TimestampMismatch {
+        state: u64,
+        envelope: u64,
+    },
+    // Blob committments exceeded the maximum
+    BlobLimitExceeded {
+        max: usize,
+        envelope: usize,
+    },
+    // Invalid state root
+    InvalidStateRoot {
+        state: Hash256,
+        envelope: Hash256,
+    },
+    // BitFieldError
+    BitFieldError(ssz::BitfieldError),
+    // Some kind of error calculating the builder payment index
+    BuilderPaymentIndexOutOfBounds(usize),
+}
+
+impl From<BeaconStateError> for EnvelopeProcessingError {
+    fn from(e: BeaconStateError) -> Self {
+        EnvelopeProcessingError::BeaconStateError(e)
+    }
+}
+
+impl From<BlockProcessingError> for EnvelopeProcessingError {
+    fn from(e: BlockProcessingError) -> Self {
+        EnvelopeProcessingError::BlockProcessingError(e)
+    }
+}
+
+impl From<ArithError> for EnvelopeProcessingError {
+    fn from(e: ArithError) -> Self {
+        EnvelopeProcessingError::ArithError(e)
+    }
+}
+
+/// Processes a `SignedExecutionPayloadEnvelope`
+///
+/// This function does all the state modifications inside `process_execution_payload()`
+pub fn envelope_processing<E: EthSpec>(
+    state: &mut BeaconState<E>,
+    parent_state_root: Option<Hash256>,
+    signed_envelope: &SignedExecutionPayloadEnvelope<E>,
+    verify_signatures: VerifySignatures,
+    spec: &ChainSpec,
+) -> Result<(), EnvelopeProcessingError> {
+    if verify_signatures.is_true() {
+        // Verify Signed Envelope Signature
+        // TODO(EIP-7732): there is probably a more efficient way to do this..
+        if !signed_envelope.verify_signature_with_state(state, spec)? {
+            return Err(EnvelopeProcessingError::BadSignature);
+        }
+    }
+
+    let envelope = &signed_envelope.message;
+    let payload = &envelope.payload;
+    let execution_requests = &envelope.execution_requests;
+
+    // Cache latest block header state root
+    if state.latest_block_header().state_root == Hash256::default() {
+        let previous_state_root = parent_state_root
+            .map(Ok)
+            .unwrap_or_else(|| state.canonical_root())?;
+        state.latest_block_header_mut().state_root = previous_state_root;
+    }
+
+    // Verify consistency with the beacon block
+    envelope_verify!(
+        envelope.beacon_block_root == state.latest_block_header().tree_hash_root(),
+        EnvelopeProcessingError::LatestBlockHeaderMismatch {
+            envelope_root: envelope.beacon_block_root,
+            block_header_root: state.latest_block_header().tree_hash_root(),
+        }
+    );
+    envelope_verify!(
+        envelope.slot == state.slot(),
+        EnvelopeProcessingError::SlotMismatch {
+            envelope_slot: envelope.slot,
+            parent_state_slot: state.slot(),
+        }
+    );
+
+    // Verify consistency with the committed bid
+    let committed_bid = state.latest_execution_payload_bid()?;
+    // builder index match already verified
+    if committed_bid.blob_kzg_commitments_root != envelope.blob_kzg_commitments.tree_hash_root() {
+        return Err(EnvelopeProcessingError::BlobKzgCommitmentsRootMismatch {
+            committed_bid: committed_bid.blob_kzg_commitments_root,
+            envelope: envelope.blob_kzg_commitments.tree_hash_root(),
+        });
+    };
+
+    // Verify the withdrawals root
+    envelope_verify!(
+        payload.withdrawals.tree_hash_root() == *state.latest_withdrawals_root()?,
+        EnvelopeProcessingError::WithdrawalsRootMismatch {
+            state: *state.latest_withdrawals_root()?,
+            envelope: payload.withdrawals.tree_hash_root(),
+        }
+    );
+
+    // Verify the gas limit
+    envelope_verify!(
+        payload.gas_limit == committed_bid.gas_limit,
+        EnvelopeProcessingError::GasLimitMismatch {
+            committed_bid: committed_bid.gas_limit,
+            envelope: payload.gas_limit,
+        }
+    );
+
+    // Verify the block hash
+    envelope_verify!(
+        committed_bid.block_hash == payload.block_hash,
+        EnvelopeProcessingError::BlockHashMismatch {
+            committed_bid: committed_bid.block_hash,
+            envelope: payload.block_hash,
+        }
+    );
+
+    // Verify consistency of the parent hash with respect to the previous execution payload
+    envelope_verify!(
+        payload.parent_hash == *state.latest_block_hash()?,
+        EnvelopeProcessingError::ParentHashMismatch {
+            state: *state.latest_block_hash()?,
+            envelope: payload.parent_hash,
+        }
+    );
+
+    // Verify prev_randao
+    envelope_verify!(
+        payload.prev_randao == *state.get_randao_mix(state.current_epoch())?,
+        EnvelopeProcessingError::PrevRandaoMismatch {
+            state: *state.get_randao_mix(state.current_epoch())?,
+            envelope: payload.prev_randao,
+        }
+    );
+
+    // Verify the timestamp
+    let state_timestamp = compute_timestamp_at_slot(state, state.slot(), spec)?;
+    envelope_verify!(
+        payload.timestamp == state_timestamp,
+        EnvelopeProcessingError::TimestampMismatch {
+            state: state_timestamp,
+            envelope: payload.timestamp,
+        }
+    );
+
+    // Verify the commitments are under limit
+    let max_blobs = spec.max_blobs_per_block(state.current_epoch()) as usize;
+    envelope_verify!(
+        envelope.blob_kzg_commitments.len() <= max_blobs,
+        EnvelopeProcessingError::BlobLimitExceeded {
+            max: max_blobs,
+            envelope: envelope.blob_kzg_commitments.len(),
+        }
+    );
+
+    // process electra operations
+    process_deposit_requests(state, &execution_requests.deposits, spec)?;
+    process_withdrawal_requests(state, &execution_requests.withdrawals, spec)?;
+    process_consolidation_requests(state, &execution_requests.consolidations, spec)?;
+
+    // queue the builder payment
+    let payment_index = E::slots_per_epoch()
+        .safe_add(state.slot().as_u64().safe_rem(E::slots_per_epoch())?)?
+        as usize;
+    let mut payment = state
+        .builder_pending_payments()?
+        .get(payment_index)
+        .ok_or(EnvelopeProcessingError::BuilderPaymentIndexOutOfBounds(
+            payment_index,
+        ))?
+        .clone();
+    let amount = payment.withdrawal.amount;
+    if amount > 0 {
+        let exit_queue_epoch = state.compute_exit_epoch_and_update_churn(amount, spec)?;
+        payment.withdrawal.withdrawable_epoch =
+            exit_queue_epoch.safe_add(spec.min_validator_withdrawability_delay)?;
+        state
+            .builder_pending_withdrawals_mut()?
+            .push(payment.withdrawal)
+            .map_err(|e| EnvelopeProcessingError::BeaconStateError(e.into()))?;
+    }
+    *state
+        .builder_pending_payments_mut()?
+        .get_mut(payment_index)
+        .ok_or(EnvelopeProcessingError::BuilderPaymentIndexOutOfBounds(
+            payment_index,
+        ))? = BuilderPendingPayment::default();
+
+    // cache the execution payload hash
+    let availability_index = state
+        .slot()
+        .safe_rem(E::slots_per_historical_root() as u64)?
+        .as_usize();
+    state
+        .execution_payload_availability_mut()?
+        .set(availability_index, true)
+        .map_err(EnvelopeProcessingError::BitFieldError)?;
+    *state.latest_block_hash_mut()? = payload.block_hash;
+
+    // verify the state root
+    envelope_verify!(
+        envelope.state_root == state.canonical_root()?,
+        EnvelopeProcessingError::InvalidStateRoot {
+            state: state.canonical_root()?,
+            envelope: envelope.state_root,
+        }
+    );
+
+    Ok(())
+}

consensus/state_processing/src/lib.rs

@@ -20,6 +20,7 @@ pub mod all_caches;
 pub mod block_replayer;
 pub mod common;
 pub mod consensus_context;
+pub mod envelope_processing;
 pub mod epoch_cache;
 pub mod genesis;
 pub mod per_block_processing;

consensus/state_processing/src/per_block_processing.rs

@@ -1,8 +1,12 @@
+use self::errors::ExecutionPayloadBidInvalid;
 use crate::consensus_context::ConsensusContext;
 use errors::{BlockOperationError, BlockProcessingError, HeaderInvalid};
 use rayon::prelude::*;
 use safe_arith::{ArithError, SafeArith, SafeArithIter};
-use signature_sets::{block_proposal_signature_set, get_pubkey_from_state, randao_signature_set};
+use signature_sets::{
+    block_proposal_signature_set, execution_payload_bid_signature_set, get_pubkey_from_state,
+    randao_signature_set,
+};
 use std::borrow::Cow;
 use tree_hash::TreeHash;
 use typenum::Unsigned;
@@ -15,6 +19,7 @@ pub use self::verify_proposer_slashing::verify_proposer_slashing;
 pub use altair::sync_committee::process_sync_aggregate;
 pub use block_signature_verifier::{BlockSignatureVerifier, ParallelSignatureSets};
 pub use is_valid_indexed_attestation::is_valid_indexed_attestation;
+pub use is_valid_indexed_payload_attestation::is_valid_indexed_payload_attestation;
 pub use process_operations::process_operations;
 pub use verify_attestation::{
     verify_attestation_for_block_inclusion, verify_attestation_for_state,
@@ -30,7 +35,9 @@ pub mod block_signature_verifier;
 pub mod deneb;
 pub mod errors;
 mod is_valid_indexed_attestation;
+mod is_valid_indexed_payload_attestation;
 pub mod process_operations;
+pub mod process_withdrawals;
 pub mod signature_sets;
 pub mod tests;
 mod verify_attestation;
@@ -38,9 +45,9 @@ mod verify_attester_slashing;
 mod verify_bls_to_execution_change;
 mod verify_deposit;
 mod verify_exit;
+mod verify_payload_attestation;
 mod verify_proposer_slashing;
 
-use crate::common::decrease_balance;
 use crate::common::update_progressive_balances_cache::{
     initialize_progressive_balances_cache, update_progressive_balances_metrics,
 };
@@ -172,14 +179,19 @@ pub fn per_block_processing<E: EthSpec, Payload: AbstractExecPayload<E>>(
     // previous block.
     if is_execution_enabled(state, block.body()) {
         let body = block.body();
-        // TODO(EIP-7732): build out process_withdrawals variant for gloas
-        process_withdrawals::<E, Payload>(state, body.execution_payload()?, spec)?;
-        process_execution_payload::<E, Payload>(state, body, spec)?;
+        if state.fork_name_unchecked().gloas_enabled() {
+            process_withdrawals::gloas::process_withdrawals::<E>(state, spec)?;
+            process_execution_payload_bid(state, block, verify_signatures, spec)?;
+        } else {
+            process_withdrawals::capella::process_withdrawals::<E, Payload>(
+                state,
+                body.execution_payload()?,
+                spec,
+            )?;
+            process_execution_payload::<E, Payload>(state, body, spec)?;
+        }
     }
 
-    // TODO(EIP-7732): build out process_execution_bid
-    // process_execution_bid(state, block, verify_signatures, spec)?;
-
     process_randao(state, block, verify_randao, ctxt, spec)?;
     process_eth1_data(state, block.body().eth1_data())?;
     process_operations(state, block.body(), verify_signatures, ctxt, spec)?;
@@ -516,17 +528,70 @@ pub fn compute_timestamp_at_slot<E: EthSpec>(
 
 /// Compute the next batch of withdrawals which should be included in a block.
 ///
-/// https://github.com/ethereum/consensus-specs/blob/dev/specs/electra/beacon-chain.md#new-get_expected_withdrawals
+/// https://ethereum.github.io/consensus-specs/specs/gloas/beacon-chain/#modified-get_expected_withdrawals
+#[allow(clippy::type_complexity)]
 pub fn get_expected_withdrawals<E: EthSpec>(
     state: &BeaconState<E>,
     spec: &ChainSpec,
-) -> Result<(Withdrawals<E>, Option<usize>), BlockProcessingError> {
+) -> Result<(Withdrawals<E>, Option<usize>, Option<usize>), BlockProcessingError> {
     let epoch = state.current_epoch();
     let mut withdrawal_index = state.next_withdrawal_index()?;
     let mut validator_index = state.next_withdrawal_validator_index()?;
     let mut withdrawals = Vec::<Withdrawal>::with_capacity(E::max_withdrawals_per_payload());
     let fork_name = state.fork_name_unchecked();
 
+    // [New in Gloas:EIP7732]
+    // Sweep for builder payments
+    let processed_builder_withdrawals_count =
+        if let Ok(builder_pending_withdrawals) = state.builder_pending_withdrawals() {
+            let mut processed_builder_withdrawals_count = 0;
+            for withdrawal in builder_pending_withdrawals {
+                if withdrawal.withdrawable_epoch > epoch
+                    || withdrawals.len().safe_add(1)? == E::max_withdrawals_per_payload()
+                {
+                    break;
+                }
+
+                if process_withdrawals::is_builder_payment_withdrawable(state, withdrawal)? {
+                    let total_withdrawn = withdrawals
+                        .iter()
+                        .filter_map(|w| {
+                            (w.validator_index == withdrawal.builder_index).then_some(w.amount)
+                        })
+                        .safe_sum()?;
+                    let balance = state
+                        .get_balance(withdrawal.builder_index as usize)?
+                        .safe_sub(total_withdrawn)?;
+                    let builder = state.get_validator(withdrawal.builder_index as usize)?;
+
+                    let withdrawable_balance = if builder.slashed {
+                        std::cmp::min(balance, withdrawal.amount)
+                    } else if balance > spec.min_activation_balance {
+                        std::cmp::min(
+                            balance.safe_sub(spec.min_activation_balance)?,
+                            withdrawal.amount,
+                        )
+                    } else {
+                        0
+                    };
+
+                    if withdrawable_balance > 0 {
+                        withdrawals.push(Withdrawal {
+                            index: withdrawal_index,
+                            validator_index: withdrawal.builder_index,
+                            address: withdrawal.fee_recipient,
+                            amount: withdrawable_balance,
+                        });
+                        withdrawal_index.safe_add_assign(1)?;
+                    }
+                }
+                processed_builder_withdrawals_count.safe_add_assign(1)?;
+            }
+            Some(processed_builder_withdrawals_count)
+        } else {
+            None
+        };
+
     // [New in Electra:EIP7251]
     // Consume pending partial withdrawals
     let processed_partial_withdrawals_count =
@@ -566,7 +631,7 @@ pub fn get_expected_withdrawals<E: EthSpec>(
                         index: withdrawal_index,
                         validator_index: withdrawal.validator_index,
                         address: validator
-                            .get_execution_withdrawal_address(spec)
+                            .get_execution_withdrawal_address(spec, state.fork_name_unchecked())
                             .ok_or(BeaconStateError::NonExecutionAddressWithdrawalCredential)?,
                         amount: withdrawable_balance,
                     });
@@ -603,7 +668,7 @@ pub fn get_expected_withdrawals<E: EthSpec>(
                 index: withdrawal_index,
                 validator_index,
                 address: validator
-                    .get_execution_withdrawal_address(spec)
+                    .get_execution_withdrawal_address(spec, state.fork_name_unchecked())
                     .ok_or(BlockProcessingError::WithdrawalCredentialsInvalid)?,
                 amount: balance,
             });
@@ -613,7 +678,7 @@ pub fn get_expected_withdrawals<E: EthSpec>(
                 index: withdrawal_index,
                 validator_index,
                 address: validator
-                    .get_execution_withdrawal_address(spec)
+                    .get_execution_withdrawal_address(spec, state.fork_name_unchecked())
                     .ok_or(BlockProcessingError::WithdrawalCredentialsInvalid)?,
                 amount: balance.safe_sub(validator.get_max_effective_balance(spec, fork_name))?,
             });
@@ -631,72 +696,163 @@ pub fn get_expected_withdrawals<E: EthSpec>(
         withdrawals
             .try_into()
             .map_err(BlockProcessingError::SszTypesError)?,
+        processed_builder_withdrawals_count,
         processed_partial_withdrawals_count,
     ))
 }
 
-/// Apply withdrawals to the state.
-/// TODO(EIP-7732): abstract this out and create gloas variant
-pub fn process_withdrawals<E: EthSpec, Payload: AbstractExecPayload<E>>(
+pub fn process_execution_payload_bid<E: EthSpec, Payload: AbstractExecPayload<E>>(
     state: &mut BeaconState<E>,
-    payload: Payload::Ref<'_>,
+    block: BeaconBlockRef<'_, E, Payload>,
+    verify_signatures: VerifySignatures,
     spec: &ChainSpec,
 ) -> Result<(), BlockProcessingError> {
-    if state.fork_name_unchecked().capella_enabled() {
-        let (expected_withdrawals, processed_partial_withdrawals_count) =
-            get_expected_withdrawals(state, spec)?;
-        let expected_root = expected_withdrawals.tree_hash_root();
-        let withdrawals_root = payload.withdrawals_root()?;
+    // Verify the bid signature
+    let signed_bid = block.body().signed_execution_payload_bid()?;
 
-        if expected_root != withdrawals_root {
-            return Err(BlockProcessingError::WithdrawalsRootMismatch {
-                expected: expected_root,
-                found: withdrawals_root,
-            });
-        }
+    let bid = &signed_bid.message;
+    let amount = bid.value;
+    let builder_index = bid.builder_index;
+    let builder = state.get_validator(builder_index as usize)?;
 
-        for withdrawal in expected_withdrawals.iter() {
-            decrease_balance(
-                state,
-                withdrawal.validator_index as usize,
-                withdrawal.amount,
-            )?;
-        }
-
-        // Update pending partial withdrawals [New in Electra:EIP7251]
-        if let Some(processed_partial_withdrawals_count) = processed_partial_withdrawals_count {
-            state
-                .pending_partial_withdrawals_mut()?
-                .pop_front(processed_partial_withdrawals_count)?;
-        }
-
-        // Update the next withdrawal index if this block contained withdrawals
-        if let Some(latest_withdrawal) = expected_withdrawals.last() {
-            *state.next_withdrawal_index_mut()? = latest_withdrawal.index.safe_add(1)?;
-
-            // Update the next validator index to start the next withdrawal sweep
-            if expected_withdrawals.len() == E::max_withdrawals_per_payload() {
-                // Next sweep starts after the latest withdrawal's validator index
-                let next_validator_index = latest_withdrawal
-                    .validator_index
-                    .safe_add(1)?
-                    .safe_rem(state.validators().len() as u64)?;
-                *state.next_withdrawal_validator_index_mut()? = next_validator_index;
-            }
-        }
-
-        // Advance sweep by the max length of the sweep if there was not a full set of withdrawals
-        if expected_withdrawals.len() != E::max_withdrawals_per_payload() {
-            let next_validator_index = state
-                .next_withdrawal_validator_index()?
-                .safe_add(spec.max_validators_per_withdrawals_sweep)?
-                .safe_rem(state.validators().len() as u64)?;
-            *state.next_withdrawal_validator_index_mut()? = next_validator_index;
-        }
-
-        Ok(())
+    // For self-builds, amount must be zero regardless of withdrawal credential prefix
+    if builder_index == block.proposer_index() {
+        block_verify!(amount == 0, ExecutionPayloadBidInvalid::BadAmount.into());
+        // TODO(EIP-7732): check with team if we should use ExecutionPayloadBidInvalid::BadSignature or a new error variant for this, like BadSelfBuildSignature
+        block_verify!(
+            signed_bid.signature.is_infinity(),
+            ExecutionPayloadBidInvalid::BadSignature.into()
+        );
     } else {
-        // these shouldn't even be encountered but they're here for completeness
-        Ok(())
+        // Non-self builds require builder withdrawal credential
+        block_verify!(
+            builder.has_builder_withdrawal_credential(spec),
+            ExecutionPayloadBidInvalid::BadWithdrawalCredentials.into()
+        );
+        if verify_signatures.is_true() {
+            block_verify!(
+                execution_payload_bid_signature_set(
+                    state,
+                    |i| get_pubkey_from_state(state, i),
+                    signed_bid,
+                    spec
+                )?
+                .verify(),
+                ExecutionPayloadBidInvalid::BadSignature.into()
+            );
+        }
     }
+
+    // Verify builder is active and not slashed
+    block_verify!(
+        builder.is_active_at(state.current_epoch()),
+        ExecutionPayloadBidInvalid::BuilderNotActive(builder_index).into()
+    );
+    block_verify!(
+        !builder.slashed,
+        ExecutionPayloadBidInvalid::BuilderSlashed(builder_index).into()
+    );
+
+    // Only perform payment related checks if amount > 0
+    if amount > 0 {
+        // Check that the builder has funds to cover the bid
+        let pending_payments = state
+            .builder_pending_payments()?
+            .iter()
+            .filter_map(|payment| {
+                if payment.withdrawal.builder_index == builder_index {
+                    Some(payment.withdrawal.amount)
+                } else {
+                    None
+                }
+            })
+            .safe_sum()?;
+
+        let pending_withdrawals = state
+            .builder_pending_withdrawals()?
+            .iter()
+            .filter_map(|withdrawal| {
+                if withdrawal.builder_index == builder_index {
+                    Some(withdrawal.amount)
+                } else {
+                    None
+                }
+            })
+            .safe_sum()?;
+
+        let builder_balance = state.get_balance(builder_index as usize)?;
+
+        block_verify!(
+            builder_balance
+                >= amount
+                    .safe_add(pending_payments)?
+                    .safe_add(pending_withdrawals)?
+                    .safe_add(spec.min_activation_balance)?,
+            ExecutionPayloadBidInvalid::InsufficientBalance {
+                builder_index,
+                builder_balance,
+                bid_value: amount,
+            }
+            .into()
+        );
+    }
+
+    // Verify that the bid is for the current slot
+    block_verify!(
+        bid.slot == block.slot(),
+        ExecutionPayloadBidInvalid::SlotMismatch {
+            state_slot: block.slot(),
+            bid_slot: bid.slot,
+        }
+        .into()
+    );
+
+    // Verify that the bid is for the right parent block
+    let latest_block_hash = state.latest_block_hash()?;
+    block_verify!(
+        bid.parent_block_hash == *latest_block_hash,
+        ExecutionPayloadBidInvalid::ParentBlockHashMismatch {
+            state_block_hash: *latest_block_hash,
+            bid_parent_hash: bid.parent_block_hash,
+        }
+        .into()
+    );
+
+    block_verify!(
+        bid.parent_block_root == block.parent_root(),
+        ExecutionPayloadBidInvalid::ParentBlockRootMismatch {
+            block_parent_root: block.parent_root(),
+            bid_parent_root: bid.parent_block_root,
+        }
+        .into()
+    );
+
+    // Record the pending payment if there is some payment
+    if amount > 0 {
+        let pending_payment = BuilderPendingPayment {
+            weight: 0,
+            withdrawal: BuilderPendingWithdrawal {
+                fee_recipient: bid.fee_recipient,
+                amount,
+                builder_index,
+                withdrawable_epoch: spec.far_future_epoch,
+            },
+        };
+
+        let payment_index = (E::slots_per_epoch()
+            .safe_add(bid.slot.as_u64().safe_rem(E::slots_per_epoch())?)?)
+            as usize;
+
+        *state
+            .builder_pending_payments_mut()?
+            .get_mut(payment_index)
+            .ok_or(BlockProcessingError::BeaconStateError(
+                BeaconStateError::BuilderPendingPaymentsIndexNotSupported(payment_index),
+            ))? = pending_payment;
+    }
+
+    // Cache the execution bid
+    *state.latest_execution_payload_bid_mut()? = bid.clone();
+
+    Ok(())
 }

consensus/state_processing/src/per_block_processing/errors.rs

@@ -41,6 +41,10 @@ pub enum BlockProcessingError {
         index: usize,
         reason: AttestationInvalid,
     },
+    PayloadAttestationInvalid {
+        index: usize,
+        reason: PayloadAttestationInvalid,
+    },
     DepositInvalid {
         index: usize,
         reason: DepositInvalid,
@@ -91,6 +95,9 @@ pub enum BlockProcessingError {
     },
     WithdrawalCredentialsInvalid,
     PendingAttestationInElectra,
+    ExecutionPayloadBidInvalid {
+        reason: ExecutionPayloadBidInvalid,
+    },
 }
 
 impl From<BeaconStateError> for BlockProcessingError {
@@ -147,6 +154,12 @@ impl From<milhouse::Error> for BlockProcessingError {
     }
 }
 
+impl From<ExecutionPayloadBidInvalid> for BlockProcessingError {
+    fn from(reason: ExecutionPayloadBidInvalid) -> Self {
+        Self::ExecutionPayloadBidInvalid { reason }
+    }
+}
+
 impl From<BlockOperationError<HeaderInvalid>> for BlockProcessingError {
     fn from(e: BlockOperationError<HeaderInvalid>) -> BlockProcessingError {
         match e {
@@ -200,7 +213,8 @@ impl_into_block_processing_error_with_index!(
     AttestationInvalid,
     DepositInvalid,
     ExitInvalid,
-    BlsExecutionChangeInvalid
+    BlsExecutionChangeInvalid,
+    PayloadAttestationInvalid
 );
 
 pub type HeaderValidationError = BlockOperationError<HeaderInvalid>;
@@ -401,6 +415,58 @@ pub enum IndexedAttestationInvalid {
     SignatureSetError(SignatureSetError),
 }
 
+#[derive(Debug, PartialEq, Clone)]
+pub enum PayloadAttestationInvalid {
+    /// Block root does not match the parent beacon block root.
+    BlockRootMismatch {
+        expected: Hash256,
+        found: Hash256,
+    },
+    /// The attestation slot is not the previous slot.
+    SlotMismatch {
+        expected: Slot,
+        found: Slot,
+    },
+    BadIndexedPayloadAttestation(IndexedPayloadAttestationInvalid),
+}
+
+impl From<BlockOperationError<IndexedPayloadAttestationInvalid>>
+    for BlockOperationError<PayloadAttestationInvalid>
+{
+    fn from(e: BlockOperationError<IndexedPayloadAttestationInvalid>) -> Self {
+        match e {
+            BlockOperationError::Invalid(e) => BlockOperationError::invalid(
+                PayloadAttestationInvalid::BadIndexedPayloadAttestation(e),
+            ),
+            BlockOperationError::BeaconStateError(e) => BlockOperationError::BeaconStateError(e),
+            BlockOperationError::SignatureSetError(e) => BlockOperationError::SignatureSetError(e),
+            BlockOperationError::SszTypesError(e) => BlockOperationError::SszTypesError(e),
+            BlockOperationError::BitfieldError(e) => BlockOperationError::BitfieldError(e),
+            BlockOperationError::ConsensusContext(e) => BlockOperationError::ConsensusContext(e),
+            BlockOperationError::ArithError(e) => BlockOperationError::ArithError(e),
+        }
+    }
+}
+
+#[derive(Debug, PartialEq, Clone)]
+pub enum IndexedPayloadAttestationInvalid {
+    /// The number of indices is 0.
+    IndicesEmpty,
+    /// The validator indices were not in increasing order.
+    ///
+    /// The error occurred between the given `index` and `index + 1`
+    BadValidatorIndicesOrdering(usize),
+    /// The validator index is unknown. One cannot slash one who does not exist.
+    UnknownValidator(u64),
+    /// The indexed attestation aggregate signature was not valid.
+    BadSignature,
+    /// There was an error whilst attempting to get a set of signatures. The signatures may have
+    /// been invalid or an internal error occurred.
+    SignatureSetError(SignatureSetError),
+    /// Invalid Payload Status
+    PayloadStatusInvalid,
+}
+
 #[derive(Debug, PartialEq, Clone)]
 pub enum DepositInvalid {
     /// The signature (proof-of-possession) does not match the given pubkey.
@@ -440,6 +506,38 @@ pub enum ExitInvalid {
     PendingWithdrawalInQueue(u64),
 }
 
+#[derive(Debug, PartialEq, Clone)]
+pub enum ExecutionPayloadBidInvalid {
+    /// The builder sent a 0 amount
+    BadAmount,
+    /// The signature is invalid.
+    BadSignature,
+    /// The builder's withdrawal credential is invalid
+    BadWithdrawalCredentials,
+    /// The builder is not an active validator.
+    BuilderNotActive(u64),
+    /// The builder is slashed
+    BuilderSlashed(u64),
+    /// The builder has insufficient balance to cover the bid
+    InsufficientBalance {
+        builder_index: u64,
+        builder_balance: u64,
+        bid_value: u64,
+    },
+    /// Bid slot doesn't match state slot
+    SlotMismatch { state_slot: Slot, bid_slot: Slot },
+    /// The bid's parent block hash doesn't match the state's latest block hash
+    ParentBlockHashMismatch {
+        state_block_hash: ExecutionBlockHash,
+        bid_parent_hash: ExecutionBlockHash,
+    },
+    /// The bid's parent block root doesn't match the block's parent root
+    ParentBlockRootMismatch {
+        block_parent_root: Hash256,
+        bid_parent_root: Hash256,
+    },
+}
+
 #[derive(Debug, PartialEq, Clone)]
 pub enum BlsExecutionChangeInvalid {
     /// The specified validator is not in the state's validator registry.

consensus/state_processing/src/per_block_processing/is_valid_indexed_payload_attestation.rs

@@ -0,0 +1,50 @@
+use super::errors::{BlockOperationError, IndexedPayloadAttestationInvalid as Invalid};
+use super::signature_sets::{get_pubkey_from_state, indexed_payload_attestation_signature_set};
+use crate::VerifySignatures;
+use itertools::Itertools;
+use types::*;
+
+fn error(reason: Invalid) -> BlockOperationError<Invalid> {
+    BlockOperationError::invalid(reason)
+}
+
+pub fn is_valid_indexed_payload_attestation<E: EthSpec>(
+    state: &BeaconState<E>,
+    indexed_payload_attestation: &IndexedPayloadAttestation<E>,
+    verify_signatures: VerifySignatures,
+    spec: &ChainSpec,
+) -> Result<(), BlockOperationError<Invalid>> {
+    // Verify indices are non-empty and sorted (duplicates allowed)
+    let indices = &indexed_payload_attestation.attesting_indices;
+    verify!(!indices.is_empty(), Invalid::IndicesEmpty);
+    let check_sorted = |list: &[u64]| -> Result<(), BlockOperationError<Invalid>> {
+        list.iter()
+            .tuple_windows()
+            .enumerate()
+            .try_for_each(|(i, (x, y))| {
+                if x <= y {
+                    Ok(())
+                } else {
+                    Err(error(Invalid::BadValidatorIndicesOrdering(i)))
+                }
+            })?;
+        Ok(())
+    };
+    check_sorted(indices)?;
+
+    if verify_signatures.is_true() {
+        verify!(
+            indexed_payload_attestation_signature_set(
+                state,
+                |i| get_pubkey_from_state(state, i),
+                &indexed_payload_attestation.signature,
+                indexed_payload_attestation,
+                spec
+            )?
+            .verify(),
+            Invalid::BadSignature
+        );
+    }
+
+    Ok(())
+}

consensus/state_processing/src/per_block_processing/process_operations.rs

@@ -5,6 +5,7 @@ use crate::common::{
     slash_validator,
 };
 use crate::per_block_processing::errors::{BlockProcessingError, IntoWithIndex};
+use crate::per_block_processing::verify_payload_attestation::verify_payload_attestation;
 use ssz_types::FixedVector;
 use typenum::U33;
 use types::consts::altair::{PARTICIPATION_FLAG_WEIGHTS, PROPOSER_WEIGHT, WEIGHT_DENOMINATOR};
@@ -38,7 +39,15 @@ pub fn process_operations<E: EthSpec, Payload: AbstractExecPayload<E>>(
         process_bls_to_execution_changes(state, bls_to_execution_changes, verify_signatures, spec)?;
     }
 
-    if state.fork_name_unchecked().electra_enabled() {
+    if state.fork_name_unchecked().gloas_enabled() {
+        process_payload_attestations(
+            state,
+            block_body.payload_attestations()?.iter(),
+            verify_signatures,
+            ctxt,
+            spec,
+        )?;
+    } else if state.fork_name_unchecked().electra_enabled() {
         state.update_pubkey_cache()?;
         process_deposit_requests(state, &block_body.execution_requests()?.deposits, spec)?;
         process_withdrawal_requests(state, &block_body.execution_requests()?.withdrawals, spec)?;
@@ -514,9 +523,10 @@ pub fn process_withdrawal_requests<E: EthSpec>(
 
         let validator = state.get_validator(validator_index)?;
         // Verify withdrawal credentials
-        let has_correct_credential = validator.has_execution_withdrawal_credential(spec);
+        let has_correct_credential =
+            validator.has_execution_withdrawal_credential(spec, state.fork_name_unchecked());
         let is_correct_source_address = validator
-            .get_execution_withdrawal_address(spec)
+            .get_execution_withdrawal_address(spec, state.fork_name_unchecked())
             .map(|addr| addr == request.source_address)
             .unwrap_or(false);
 
@@ -561,7 +571,7 @@ pub fn process_withdrawal_requests<E: EthSpec>(
                 .safe_add(pending_balance_to_withdraw)?;
 
         // Only allow partial withdrawals with compounding withdrawal credentials
-        if validator.has_compounding_withdrawal_credential(spec)
+        if validator.has_compounding_withdrawal_credential(spec, state.fork_name_unchecked())
             && has_sufficient_effective_balance
             && has_excess_balance
         {
@@ -730,7 +740,9 @@ pub fn process_consolidation_request<E: EthSpec>(
 
     let source_validator = state.get_validator(source_index)?;
     // Verify the source withdrawal credentials
-    if let Some(withdrawal_address) = source_validator.get_execution_withdrawal_address(spec) {
+    if let Some(withdrawal_address) =
+        source_validator.get_execution_withdrawal_address(spec, state.fork_name_unchecked())
+    {
         if withdrawal_address != consolidation_request.source_address {
             return Ok(());
         }
@@ -741,7 +753,7 @@ pub fn process_consolidation_request<E: EthSpec>(
 
     let target_validator = state.get_validator(target_index)?;
     // Verify the target has compounding withdrawal credentials
-    if !target_validator.has_compounding_withdrawal_credential(spec) {
+    if !target_validator.has_compounding_withdrawal_credential(spec, state.fork_name_unchecked()) {
         return Ok(());
     }
 
@@ -787,3 +799,52 @@ pub fn process_consolidation_request<E: EthSpec>(
 
     Ok(())
 }
+
+// TODO(EIP-7732): Add test cases for `process_payload_attestations` to
+// `consensus/state_processing/src/per_block_processing/tests.rs`.
+// The tests will require being able to build Gloas blocks with PayloadAttestations,
+// which currently fails due to incomplete Gloas block structure as mentioned here
+// https://github.com/sigp/lighthouse/pull/8273
+pub fn process_payload_attestation<E: EthSpec>(
+    state: &mut BeaconState<E>,
+    payload_attestation: &PayloadAttestation<E>,
+    att_index: usize,
+    verify_signatures: VerifySignatures,
+    ctxt: &mut ConsensusContext<E>,
+    spec: &ChainSpec,
+) -> Result<(), BlockProcessingError> {
+    verify_payload_attestation(state, payload_attestation, ctxt, verify_signatures, spec)
+        .map_err(|e| e.into_with_index(att_index))
+}
+
+pub fn process_payload_attestations<'a, E: EthSpec, I>(
+    state: &mut BeaconState<E>,
+    payload_attestations: I,
+    verify_signatures: VerifySignatures,
+    ctxt: &mut ConsensusContext<E>,
+    spec: &ChainSpec,
+) -> Result<(), BlockProcessingError>
+where
+    I: Iterator<Item = &'a PayloadAttestation<E>>,
+{
+    // Ensure required caches are all built. These should be no-ops during regular operation.
+    // TODO(EIP-7732): verify necessary caches
+    state.build_committee_cache(RelativeEpoch::Current, spec)?;
+    state.build_committee_cache(RelativeEpoch::Previous, spec)?;
+    initialize_epoch_cache(state, spec)?;
+    initialize_progressive_balances_cache(state, spec)?;
+    state.build_slashings_cache()?;
+
+    payload_attestations
+        .enumerate()
+        .try_for_each(|(i, payload_attestation)| {
+            process_payload_attestation(
+                state,
+                payload_attestation,
+                i,
+                verify_signatures,
+                ctxt,
+                spec,
+            )
+        })
+}

consensus/state_processing/src/per_block_processing/process_withdrawals.rs

@@ -0,0 +1,167 @@
+use super::errors::BlockProcessingError;
+use super::get_expected_withdrawals;
+use crate::common::decrease_balance;
+use milhouse::List;
+use safe_arith::SafeArith;
+use tree_hash::TreeHash;
+use types::{
+    AbstractExecPayload, BeaconState, BuilderPendingWithdrawal, ChainSpec, EthSpec, ExecPayload,
+    Withdrawals,
+};
+
+/// Check if a builder payment is withdrawable.
+/// A builder payment is withdrawable if the builder is not slashed or
+/// the builder's withdrawable epoch has been reached.
+pub fn is_builder_payment_withdrawable<E: EthSpec>(
+    state: &BeaconState<E>,
+    withdrawal: &BuilderPendingWithdrawal,
+) -> Result<bool, BlockProcessingError> {
+    let builder = state.get_validator(withdrawal.builder_index as usize)?;
+    let current_epoch = state.current_epoch();
+
+    Ok(builder.withdrawable_epoch >= current_epoch || !builder.slashed)
+}
+
+fn process_withdrawals_common<E: EthSpec>(
+    state: &mut BeaconState<E>,
+    expected_withdrawals: Withdrawals<E>,
+    partial_withdrawals_count: Option<usize>,
+    spec: &ChainSpec,
+) -> Result<(), BlockProcessingError> {
+    match state {
+        BeaconState::Capella(_)
+        | BeaconState::Deneb(_)
+        | BeaconState::Electra(_)
+        | BeaconState::Fulu(_)
+        | BeaconState::Gloas(_) => {
+            // Update pending partial withdrawals [New in Electra:EIP7251]
+            if let Some(partial_withdrawals_count) = partial_withdrawals_count {
+                state
+                    .pending_partial_withdrawals_mut()?
+                    .pop_front(partial_withdrawals_count)?;
+            }
+
+            // Update the next withdrawal index if this block contained withdrawals
+            if let Some(latest_withdrawal) = expected_withdrawals.last() {
+                *state.next_withdrawal_index_mut()? = latest_withdrawal.index.safe_add(1)?;
+
+                // Update the next validator index to start the next withdrawal sweep
+                if expected_withdrawals.len() == E::max_withdrawals_per_payload() {
+                    // Next sweep starts after the latest withdrawal's validator index
+                    let next_validator_index = latest_withdrawal
+                        .validator_index
+                        .safe_add(1)?
+                        .safe_rem(state.validators().len() as u64)?;
+                    *state.next_withdrawal_validator_index_mut()? = next_validator_index;
+                }
+            }
+
+            // Advance sweep by the max length of the sweep if there was not a full set of withdrawals
+            if expected_withdrawals.len() != E::max_withdrawals_per_payload() {
+                let next_validator_index = state
+                    .next_withdrawal_validator_index()?
+                    .safe_add(spec.max_validators_per_withdrawals_sweep)?
+                    .safe_rem(state.validators().len() as u64)?;
+                *state.next_withdrawal_validator_index_mut()? = next_validator_index;
+            }
+
+            Ok(())
+        }
+        // these shouldn't even be encountered but they're here for completeness
+        BeaconState::Base(_) | BeaconState::Altair(_) | BeaconState::Bellatrix(_) => Ok(()),
+    }
+}
+
+pub mod capella {
+    use super::*;
+    /// Apply withdrawals to the state.
+    pub fn process_withdrawals<E: EthSpec, Payload: AbstractExecPayload<E>>(
+        state: &mut BeaconState<E>,
+        payload: Payload::Ref<'_>,
+        spec: &ChainSpec,
+    ) -> Result<(), BlockProcessingError> {
+        // check if capella enabled because this function will run on the merge block where the fork is technically still Bellatrix
+        if state.fork_name_unchecked().capella_enabled() {
+            let (expected_withdrawals, _, partial_withdrawals_count) =
+                get_expected_withdrawals(state, spec)?;
+
+            let expected_root = expected_withdrawals.tree_hash_root();
+            let withdrawals_root = payload.withdrawals_root()?;
+            if expected_root != withdrawals_root {
+                return Err(BlockProcessingError::WithdrawalsRootMismatch {
+                    expected: expected_root,
+                    found: withdrawals_root,
+                });
+            }
+
+            for withdrawal in expected_withdrawals.iter() {
+                decrease_balance(
+                    state,
+                    withdrawal.validator_index as usize,
+                    withdrawal.amount,
+                )?;
+            }
+
+            process_withdrawals_common(state, expected_withdrawals, partial_withdrawals_count, spec)
+        } else {
+            // these shouldn't even be encountered but they're here for completeness
+            Ok(())
+        }
+    }
+}
+pub mod gloas {
+    use super::*;
+
+    // TODO(EIP-7732): Add comprehensive tests for Gloas `process_withdrawals`:
+    // Similar to Capella version, these will be tested via:
+    // 1. EF consensus-spec tests in `testing/ef_tests/src/cases/operations.rs`
+    // 2. Integration tests via full block processing
+    // These tests would currently fail due to incomplete Gloas block structure as mentioned here, so we will implement them after block and payload processing is in a good state.
+    // https://github.com/sigp/lighthouse/pull/8273
+    /// Apply withdrawals to the state.
+    pub fn process_withdrawals<E: EthSpec>(
+        state: &mut BeaconState<E>,
+        spec: &ChainSpec,
+    ) -> Result<(), BlockProcessingError> {
+        if !state.is_parent_block_full() {
+            return Ok(());
+        }
+
+        let (expected_withdrawals, builder_withdrawals_count, partial_withdrawals_count) =
+            get_expected_withdrawals(state, spec)?;
+
+        *state.latest_withdrawals_root_mut()? = expected_withdrawals.tree_hash_root();
+
+        for withdrawal in expected_withdrawals.iter() {
+            decrease_balance(
+                state,
+                withdrawal.validator_index as usize,
+                withdrawal.amount,
+            )?;
+        }
+
+        if let (Ok(builder_pending_withdrawals), Some(builder_count)) = (
+            state.builder_pending_withdrawals(),
+            builder_withdrawals_count,
+        ) {
+            let mut updated_builder_withdrawals =
+                Vec::with_capacity(E::builder_pending_withdrawals_limit());
+
+            for (i, withdrawal) in builder_pending_withdrawals.iter().enumerate() {
+                if i < builder_count {
+                    if !is_builder_payment_withdrawable(state, withdrawal)? {
+                        updated_builder_withdrawals.push(withdrawal.clone());
+                    }
+                } else {
+                    updated_builder_withdrawals.push(withdrawal.clone());
+                }
+            }
+
+            *state.builder_pending_withdrawals_mut()? = List::new(updated_builder_withdrawals)?;
+        }
+
+        process_withdrawals_common(state, expected_withdrawals, partial_withdrawals_count, spec)?;
+
+        Ok(())
+    }
+}

consensus/state_processing/src/per_block_processing/signature_sets.rs

@@ -10,10 +10,11 @@ use typenum::Unsigned;
 use types::{
     AbstractExecPayload, AttesterSlashingRef, BeaconBlockRef, BeaconState, BeaconStateError,
     ChainSpec, DepositData, Domain, Epoch, EthSpec, Fork, Hash256, InconsistentFork,
-    IndexedAttestation, IndexedAttestationRef, ProposerSlashing, SignedAggregateAndProof,
-    SignedBeaconBlock, SignedBeaconBlockHeader, SignedBlsToExecutionChange,
-    SignedContributionAndProof, SignedRoot, SignedVoluntaryExit, SigningData, Slot, SyncAggregate,
-    SyncAggregatorSelectionData,
+    IndexedAttestation, IndexedAttestationRef, IndexedPayloadAttestation, ProposerSlashing,
+    SignedAggregateAndProof, SignedBeaconBlock, SignedBeaconBlockHeader,
+    SignedBlsToExecutionChange, SignedContributionAndProof, SignedExecutionPayloadBid,
+    SignedExecutionPayloadEnvelope, SignedRoot, SignedVoluntaryExit, SigningData, Slot,
+    SyncAggregate, SyncAggregatorSelectionData,
 };
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -299,6 +300,35 @@ where
     Ok(SignatureSet::multiple_pubkeys(signature, pubkeys, message))
 }
 
+pub fn indexed_payload_attestation_signature_set<'a, 'b, E, F>(
+    state: &'a BeaconState<E>,
+    get_pubkey: F,
+    signature: &'a AggregateSignature,
+    indexed_payload_attestation: &'b IndexedPayloadAttestation<E>,
+    spec: &'a ChainSpec,
+) -> Result<SignatureSet<'a>>
+where
+    E: EthSpec,
+    F: Fn(usize) -> Option<Cow<'a, PublicKey>>,
+{
+    let mut pubkeys = Vec::with_capacity(indexed_payload_attestation.attesting_indices.len());
+    for &validator_idx in indexed_payload_attestation.attesting_indices.iter() {
+        pubkeys.push(
+            get_pubkey(validator_idx as usize).ok_or(Error::ValidatorUnknown(validator_idx))?,
+        );
+    }
+
+    let domain = spec.compute_domain(
+        Domain::PTCAttester,
+        spec.genesis_fork_version,
+        state.genesis_validators_root(),
+    );
+
+    let message = indexed_payload_attestation.data.signing_root(domain);
+
+    Ok(SignatureSet::multiple_pubkeys(signature, pubkeys, message))
+}
+
 /// Returns the signature set for the given `indexed_attestation` but pubkeys are supplied directly
 /// instead of from the state.
 pub fn indexed_attestation_signature_set_from_pubkeys<'a, 'b, E, F>(
@@ -332,6 +362,62 @@ where
     Ok(SignatureSet::multiple_pubkeys(signature, pubkeys, message))
 }
 
+pub fn execution_envelope_signature_set<'a, E, F>(
+    state: &'a BeaconState<E>,
+    get_pubkey: F,
+    signed_envelope: &'a SignedExecutionPayloadEnvelope<E>,
+    spec: &'a ChainSpec,
+) -> Result<SignatureSet<'a>>
+where
+    E: EthSpec,
+    F: Fn(usize) -> Option<Cow<'a, PublicKey>>,
+{
+    let domain = spec.get_domain(
+        state.current_epoch(),
+        Domain::BeaconBuilder,
+        &state.fork(),
+        state.genesis_validators_root(),
+    );
+    let message = signed_envelope.message.signing_root(domain);
+    let pubkey = get_pubkey(signed_envelope.message.builder_index as usize).ok_or(
+        Error::ValidatorUnknown(signed_envelope.message.builder_index),
+    )?;
+
+    Ok(SignatureSet::single_pubkey(
+        &signed_envelope.signature,
+        pubkey,
+        message,
+    ))
+}
+
+pub fn execution_payload_bid_signature_set<'a, E, F>(
+    state: &'a BeaconState<E>,
+    get_pubkey: F,
+    signed_execution_payload_bid: &'a SignedExecutionPayloadBid,
+    spec: &'a ChainSpec,
+) -> Result<SignatureSet<'a>>
+where
+    E: EthSpec,
+    F: Fn(usize) -> Option<Cow<'a, PublicKey>>,
+{
+    let domain = spec.get_domain(
+        state.current_epoch(),
+        Domain::BeaconBuilder,
+        &state.fork(),
+        state.genesis_validators_root(),
+    );
+    let execution_payload_bid = &signed_execution_payload_bid.message;
+    let pubkey = get_pubkey(execution_payload_bid.builder_index as usize)
+        .ok_or(Error::ValidatorUnknown(execution_payload_bid.builder_index))?;
+    let message = execution_payload_bid.signing_root(domain);
+
+    Ok(SignatureSet::single_pubkey(
+        &signed_execution_payload_bid.signature,
+        pubkey,
+        message,
+    ))
+}
+
 /// Returns the signature set for the given `attester_slashing` and corresponding `pubkeys`.
 pub fn attester_slashing_signature_sets<'a, E, F>(
     state: &'a BeaconState<E>,

consensus/state_processing/src/per_block_processing/verify_payload_attestation.rs

@@ -0,0 +1,46 @@
+use super::VerifySignatures;
+use super::errors::{BlockOperationError, PayloadAttestationInvalid as Invalid};
+use crate::ConsensusContext;
+use crate::per_block_processing::is_valid_indexed_payload_attestation;
+use safe_arith::SafeArith;
+use types::*;
+
+pub fn verify_payload_attestation<'ctxt, E: EthSpec>(
+    state: &mut BeaconState<E>,
+    payload_attestation: &'ctxt PayloadAttestation<E>,
+    ctxt: &'ctxt mut ConsensusContext<E>,
+    verify_signatures: VerifySignatures,
+    spec: &ChainSpec,
+) -> Result<(), BlockOperationError<Invalid>> {
+    let data = &payload_attestation.data;
+
+    // Check that the attestation is for the parent beacon block
+    verify!(
+        data.beacon_block_root == state.latest_block_header().parent_root,
+        Invalid::BlockRootMismatch {
+            expected: state.latest_block_header().parent_root,
+            found: data.beacon_block_root,
+        }
+    );
+
+    // Check that the attestation is for the previous slot
+    verify!(
+        data.slot.safe_add(1)? == state.slot(),
+        Invalid::SlotMismatch {
+            expected: state.slot().saturating_sub(Slot::new(1)),
+            found: data.slot,
+        }
+    );
+
+    let indexed_payload_attestation =
+        ctxt.get_indexed_payload_attestation(state, data.slot, payload_attestation, spec)?;
+
+    is_valid_indexed_payload_attestation(
+        state,
+        indexed_payload_attestation,
+        verify_signatures,
+        spec,
+    )?;
+
+    Ok(())
+}

consensus/state_processing/src/per_epoch_processing/single_pass.rs

@@ -15,9 +15,9 @@ use std::collections::{BTreeSet, HashMap};
 use tracing::instrument;
 use typenum::Unsigned;
 use types::{
-    ActivationQueue, BeaconState, BeaconStateError, ChainSpec, Checkpoint, DepositData, Epoch,
-    EthSpec, ExitCache, ForkName, ParticipationFlags, PendingDeposit, ProgressiveBalancesCache,
-    RelativeEpoch, Validator,
+    ActivationQueue, BeaconState, BeaconStateError, BuilderPendingPayment, ChainSpec, Checkpoint,
+    DepositData, Epoch, EthSpec, ExitCache, ForkName, ParticipationFlags, PendingDeposit,
+    ProgressiveBalancesCache, RelativeEpoch, Validator,
     consts::altair::{
         NUM_FLAG_INDICES, PARTICIPATION_FLAG_WEIGHTS, TIMELY_HEAD_FLAG_INDEX,
         TIMELY_TARGET_FLAG_INDEX, WEIGHT_DENOMINATOR,
@@ -33,6 +33,7 @@ pub struct SinglePassConfig {
     pub pending_consolidations: bool,
     pub effective_balance_updates: bool,
     pub proposer_lookahead: bool,
+    pub builder_pending_payments: bool,
 }
 
 impl Default for SinglePassConfig {
@@ -52,6 +53,7 @@ impl SinglePassConfig {
             pending_consolidations: true,
             effective_balance_updates: true,
             proposer_lookahead: true,
+            builder_pending_payments: true,
         }
     }
 
@@ -65,6 +67,7 @@ impl SinglePassConfig {
             pending_consolidations: false,
             effective_balance_updates: false,
             proposer_lookahead: false,
+            builder_pending_payments: false,
         }
     }
 }
@@ -455,6 +458,12 @@ pub fn process_epoch_single_pass<E: EthSpec>(
         )?;
     }
 
+    // Process builder pending payments outside the single-pass loop, as they depend on balances for multiple
+    // validators and cannot be computed accurately inside the loop.
+    if fork_name.gloas_enabled() && conf.builder_pending_payments {
+        process_builder_pending_payments(state, state_ctxt, spec)?;
+    }
+
     // Finally, finish updating effective balance caches. We need this to happen *after* processing
     // of pending consolidations, which recomputes some effective balances.
     if conf.effective_balance_updates {
@@ -473,7 +482,7 @@ pub fn process_epoch_single_pass<E: EthSpec>(
     Ok(summary)
 }
 
-// TOOO(EIP-7917): use balances cache
+// TODO(EIP-7917): use balances cache
 pub fn process_proposer_lookahead<E: EthSpec>(
     state: &mut BeaconState<E>,
     spec: &ChainSpec,
@@ -503,6 +512,68 @@ pub fn process_proposer_lookahead<E: EthSpec>(
     Ok(())
 }
 
+/// Calculate the quorum threshold for builder payments based on total active balance.
+fn get_builder_payment_quorum_threshold<E: EthSpec>(
+    state_ctxt: &StateContext,
+    spec: &ChainSpec,
+) -> Result<u64, Error> {
+    let per_slot_balance = state_ctxt
+        .total_active_balance
+        .safe_div(E::slots_per_epoch())?;
+    let quorum = per_slot_balance.safe_mul(spec.builder_payment_threshold_numerator)?;
+    quorum
+        .safe_div(spec.builder_payment_threshold_denominator)
+        .map_err(Error::from)
+}
+
+/// Process builder pending payments, moving qualifying payments to withdrawals.
+/// TODO(EIP-7732): Add EF consensus-spec tests for `process_builder_pending_payments`
+/// Currently blocked by EF consensus-spec-tests for Gloas not yet integrated.
+fn process_builder_pending_payments<E: EthSpec>(
+    state: &mut BeaconState<E>,
+    state_ctxt: &StateContext,
+    spec: &ChainSpec,
+) -> Result<(), Error> {
+    let quorum = get_builder_payment_quorum_threshold::<E>(state_ctxt, spec)?;
+
+    // Collect qualifying payments
+    let qualifying_payments = state
+        .builder_pending_payments()?
+        .iter()
+        .take(E::slots_per_epoch() as usize)
+        .filter(|payment| payment.weight > quorum)
+        .cloned()
+        .collect::<Vec<_>>();
+
+    // Update `builder_pending_withdrawals` with qualifying `builder_pending_payments`
+    qualifying_payments
+        .into_iter()
+        .try_for_each(|payment| -> Result<(), Error> {
+            let exit_queue_epoch =
+                state.compute_exit_epoch_and_update_churn(payment.withdrawal.amount, spec)?;
+            let withdrawable_epoch =
+                exit_queue_epoch.safe_add(spec.min_validator_withdrawability_delay)?;
+
+            let mut withdrawal = payment.withdrawal.clone();
+            withdrawal.withdrawable_epoch = withdrawable_epoch;
+            state.builder_pending_withdrawals_mut()?.push(withdrawal)?;
+            Ok(())
+        })?;
+
+    // Move remaining `builder_pending_payments` to start of list and set the rest to default
+    let new_payments = state
+        .builder_pending_payments()?
+        .iter()
+        .skip(E::slots_per_epoch() as usize)
+        .cloned()
+        .chain((0..E::slots_per_epoch() as usize).map(|_| BuilderPendingPayment::default()))
+        .collect::<Vec<_>>();
+
+    *state.builder_pending_payments_mut()? = Vector::new(new_payments)?;
+
+    Ok(())
+}
+
 fn process_single_inactivity_update(
     inactivity_score: &mut Cow<u64>,
     validator_info: &ValidatorInfo,

consensus/state_processing/src/per_slot_processing.rs

@@ -14,6 +14,7 @@ pub enum Error {
     EpochProcessingError(EpochProcessingError),
     ArithError(ArithError),
     InconsistentStateFork(InconsistentFork),
+    BitfieldError(ssz::BitfieldError),
 }
 
 impl From<ArithError> for Error {
@@ -22,6 +23,12 @@ impl From<ArithError> for Error {
     }
 }
 
+impl From<ssz::BitfieldError> for Error {
+    fn from(e: ssz::BitfieldError) -> Self {
+        Self::BitfieldError(e)
+    }
+}
+
 /// Advances a state forward by one slot, performing per-epoch processing if required.
 ///
 /// If the root of the supplied `state` is known, then it can be passed as `state_root`. If
@@ -50,6 +57,18 @@ pub fn per_slot_processing<E: EthSpec>(
 
     state.slot_mut().safe_add_assign(1)?;
 
+    // Unset the next payload availability
+    if state.fork_name_unchecked().gloas_enabled() {
+        let next_slot_index = state
+            .slot()
+            .as_usize()
+            .safe_add(1)?
+            .safe_rem(E::slots_per_historical_root())?;
+        state
+            .execution_payload_availability_mut()?
+            .set(next_slot_index, false)?;
+    }
+
     // Process fork upgrades here. Note that multiple upgrades can potentially run
     // in sequence if they are scheduled in the same Epoch (common in testnets)
     if state.slot().safe_rem(E::slots_per_epoch())? == 0 {

consensus/state_processing/src/upgrade/electra.rs

@@ -82,7 +82,7 @@ pub fn upgrade_to_electra<E: EthSpec>(
     // Ensure early adopters of compounding credentials go through the activation churn
     let validators = post.validators().clone();
     for (index, validator) in validators.iter().enumerate() {
-        if validator.has_compounding_withdrawal_credential(spec) {
+        if validator.has_compounding_withdrawal_credential(spec, post.fork_name_unchecked()) {
             post.queue_excess_active_balance(index, spec)?;
         }
     }

consensus/types/src/attestation/mod.rs

@@ -11,6 +11,7 @@ mod payload_attestation;
 mod payload_attestation_data;
 mod payload_attestation_message;
 mod pending_attestation;
+mod ptc;
 mod selection_proof;
 mod shuffling_id;
 mod signed_aggregate_and_proof;
@@ -36,6 +37,7 @@ pub use payload_attestation::PayloadAttestation;
 pub use payload_attestation_data::PayloadAttestationData;
 pub use payload_attestation_message::PayloadAttestationMessage;
 pub use pending_attestation::PendingAttestation;
+pub use ptc::PTC;
 pub use selection_proof::SelectionProof;
 pub use shuffling_id::AttestationShufflingId;
 pub use signed_aggregate_and_proof::{

consensus/types/src/attestation/ptc.rs

@@ -0,0 +1,24 @@
+use crate::EthSpec;
+use ssz_types::FixedVector;
+
+/// TODO(EIP-7732): is it easier to return u64 or usize?
+#[derive(Clone, Debug, PartialEq)]
+pub struct PTC<E: EthSpec>(pub FixedVector<usize, E::PTCSize>);
+
+impl<'a, E: EthSpec> IntoIterator for &'a PTC<E> {
+    type Item = &'a usize;
+    type IntoIter = std::slice::Iter<'a, usize>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        self.0.iter()
+    }
+}
+
+impl<E: EthSpec> IntoIterator for PTC<E> {
+    type Item = usize;
+    type IntoIter = std::vec::IntoIter<usize>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        self.0.into_iter()
+    }
+}

consensus/types/src/execution/signed_execution_payload_envelope.rs

@@ -1,6 +1,9 @@
 use crate::test_utils::TestRandom;
-use crate::{EthSpec, ExecutionPayloadEnvelope};
-use bls::Signature;
+use crate::{
+    BeaconState, BeaconStateError, ChainSpec, Domain, Epoch, EthSpec, ExecutionBlockHash,
+    ExecutionPayloadEnvelope, Fork, Hash256, SignedRoot, Slot,
+};
+use bls::{PublicKey, Signature};
 use educe::Educe;
 use serde::{Deserialize, Serialize};
 use ssz_derive::{Decode, Encode};
@@ -15,6 +18,78 @@ pub struct SignedExecutionPayloadEnvelope<E: EthSpec> {
     pub signature: Signature,
 }
 
+impl<E: EthSpec> SignedExecutionPayloadEnvelope<E> {
+    pub fn slot(&self) -> Slot {
+        self.message.slot
+    }
+
+    pub fn epoch(&self) -> Epoch {
+        self.slot().epoch(E::slots_per_epoch())
+    }
+
+    pub fn beacon_block_root(&self) -> Hash256 {
+        self.message.beacon_block_root
+    }
+
+    pub fn block_hash(&self) -> ExecutionBlockHash {
+        self.message.payload.block_hash
+    }
+
+    /// Verify `self.signature`.
+    ///
+    /// The `parent_state` is the post-state of the beacon block with
+    /// block_root = self.message.beacon_block_root
+    /// TODO(EIP-7732): maybe delete this function later
+    pub fn verify_signature_with_state(
+        &self,
+        parent_state: &BeaconState<E>,
+        spec: &ChainSpec,
+    ) -> Result<bool, BeaconStateError> {
+        let domain = spec.get_domain(
+            parent_state.current_epoch(),
+            Domain::BeaconBuilder,
+            &parent_state.fork(),
+            parent_state.genesis_validators_root(),
+        );
+        let pubkey = parent_state
+            .validators()
+            .get(self.message.builder_index as usize)
+            .and_then(|v| {
+                let pk: Option<PublicKey> = v.pubkey.decompress().ok();
+                pk
+            })
+            .ok_or(BeaconStateError::UnknownValidator(
+                self.message.builder_index as usize,
+            ))?;
+        let message = self.message.signing_root(domain);
+
+        Ok(self.signature.verify(&pubkey, message))
+    }
+
+    /// Verify `self.signature`.
+    ///
+    /// If the root of `block.message` is already known it can be passed in via `object_root_opt`.
+    /// Otherwise, it will be computed locally.
+    pub fn verify_signature(
+        &self,
+        pubkey: &PublicKey,
+        fork: &Fork,
+        genesis_validators_root: Hash256,
+        spec: &ChainSpec,
+    ) -> bool {
+        let domain = spec.get_domain(
+            self.epoch(),
+            Domain::BeaconProposer,
+            fork,
+            genesis_validators_root,
+        );
+
+        let message = self.message.signing_root(domain);
+
+        self.signature.verify(pubkey, message)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

consensus/types/src/state/beacon_state.rs

@@ -25,7 +25,7 @@ use typenum::Unsigned;
 use crate::{
     BuilderPendingPayment, BuilderPendingWithdrawal, ExecutionBlockHash, ExecutionPayloadBid,
     attestation::{
-        AttestationDuty, BeaconCommittee, Checkpoint, CommitteeIndex, ParticipationFlags,
+        AttestationDuty, BeaconCommittee, Checkpoint, CommitteeIndex, PTC, ParticipationFlags,
         PendingAttestation,
     },
     block::{BeaconBlock, BeaconBlockHeader, SignedBeaconBlockHash},
@@ -168,6 +168,7 @@ pub enum BeaconStateError {
     TotalActiveBalanceDiffUninitialized,
     GeneralizedIndexNotSupported(usize),
     IndexNotSupported(usize),
+    BuilderPendingPaymentsIndexNotSupported(usize),
     InvalidFlagIndex(usize),
     MerkleTreeError(merkle_proof::MerkleTreeError),
     PartialWithdrawalCountInvalid(usize),
@@ -195,6 +196,8 @@ pub enum BeaconStateError {
     ProposerLookaheadOutOfBounds {
         i: usize,
     },
+    InvalidIndicesCount,
+    PleaseNotifyTheDevs(String),
 }
 
 /// Control whether an epoch-indexed field can be indexed at the next epoch or not.
@@ -1115,13 +1118,22 @@ impl<E: EthSpec> BeaconState<E> {
             }
         }
 
+        let gloas_enabled = self.fork_name_unchecked().gloas_enabled();
         epoch
             .slot_iter(E::slots_per_epoch())
             .map(|slot| {
                 let mut preimage = seed.to_vec();
                 preimage.append(&mut int_to_bytes8(slot.as_u64()));
                 let seed = hash(&preimage);
-                self.compute_proposer_index(indices, &seed, spec)
+
+                if gloas_enabled {
+                    self.compute_balance_weighted_selection(indices, &seed, 1, true, spec)?
+                        .first()
+                        .copied()
+                        .ok_or(BeaconStateError::InsufficientValidators)
+                } else {
+                    self.compute_proposer_index(indices, &seed, spec)
+                }
             })
             .collect()
     }
@@ -1378,39 +1390,50 @@ impl<E: EthSpec> BeaconState<E> {
         let epoch = self.current_epoch().safe_add(1)?;
 
         let active_validator_indices = self.get_active_validator_indices(epoch, spec)?;
-        let active_validator_count = active_validator_indices.len();
-
         let seed = self.get_seed(epoch, Domain::SyncCommittee, spec)?;
-        let max_effective_balance = spec.max_effective_balance_for_fork(self.fork_name_unchecked());
-        let max_random_value = if self.fork_name_unchecked().electra_enabled() {
-            MAX_RANDOM_VALUE
-        } else {
-            MAX_RANDOM_BYTE
-        };
 
-        let mut i = 0;
-        let mut sync_committee_indices = Vec::with_capacity(E::SyncCommitteeSize::to_usize());
-        while sync_committee_indices.len() < E::SyncCommitteeSize::to_usize() {
-            let shuffled_index = compute_shuffled_index(
-                i.safe_rem(active_validator_count)?,
-                active_validator_count,
+        if self.fork_name_unchecked().gloas_enabled() {
+            self.compute_balance_weighted_selection(
+                &active_validator_indices,
                 seed.as_slice(),
-                spec.shuffle_round_count,
+                E::SyncCommitteeSize::to_usize(),
+                true,
+                spec,
             )
-            .ok_or(BeaconStateError::UnableToShuffle)?;
-            let candidate_index = *active_validator_indices
-                .get(shuffled_index)
-                .ok_or(BeaconStateError::ShuffleIndexOutOfBounds(shuffled_index))?;
-            let random_value = self.shuffling_random_value(i, seed.as_slice())?;
-            let effective_balance = self.get_validator(candidate_index)?.effective_balance;
-            if effective_balance.safe_mul(max_random_value)?
-                >= max_effective_balance.safe_mul(random_value)?
-            {
-                sync_committee_indices.push(candidate_index);
+        } else {
+            let active_validator_count = active_validator_indices.len();
+            let max_effective_balance =
+                spec.max_effective_balance_for_fork(self.fork_name_unchecked());
+            let max_random_value = if self.fork_name_unchecked().electra_enabled() {
+                MAX_RANDOM_VALUE
+            } else {
+                MAX_RANDOM_BYTE
+            };
+
+            let mut i = 0;
+            let mut sync_committee_indices = Vec::with_capacity(E::SyncCommitteeSize::to_usize());
+            while sync_committee_indices.len() < E::SyncCommitteeSize::to_usize() {
+                let shuffled_index = compute_shuffled_index(
+                    i.safe_rem(active_validator_count)?,
+                    active_validator_count,
+                    seed.as_slice(),
+                    spec.shuffle_round_count,
+                )
+                .ok_or(BeaconStateError::UnableToShuffle)?;
+                let candidate_index = *active_validator_indices
+                    .get(shuffled_index)
+                    .ok_or(BeaconStateError::ShuffleIndexOutOfBounds(shuffled_index))?;
+                let random_value = self.shuffling_random_value(i, seed.as_slice())?;
+                let effective_balance = self.get_validator(candidate_index)?.effective_balance;
+                if effective_balance.safe_mul(max_random_value)?
+                    >= max_effective_balance.safe_mul(random_value)?
+                {
+                    sync_committee_indices.push(candidate_index);
+                }
+                i.safe_add_assign(1)?;
             }
-            i.safe_add_assign(1)?;
+            Ok(sync_committee_indices)
         }
-        Ok(sync_committee_indices)
     }
 
     /// Compute the next sync committee.
@@ -2292,6 +2315,7 @@ impl<E: EthSpec> BeaconState<E> {
         }
     }
 
+    /// Return true if the parent block was full (both beacon block and execution payload were present).
     pub fn is_parent_block_full(&self) -> bool {
         match self {
             BeaconState::Base(_) | BeaconState::Altair(_) => false,
@@ -2558,11 +2582,16 @@ impl<E: EthSpec> BeaconState<E> {
             .map_err(Into::into)
     }
 
+    // TODO(EIP-7732): The consensus spec PR for this change mentions that some EF tests will be needed but haven't been created yet.
+    // We should integrate them once they are available.
+    // https://github.com/ethereum/consensus-specs/pull/4513
     pub fn get_pending_balance_to_withdraw(
         &self,
         validator_index: usize,
     ) -> Result<u64, BeaconStateError> {
         let mut pending_balance = 0;
+
+        // Sum pending partial withdrawals
         for withdrawal in self
             .pending_partial_withdrawals()?
             .iter()
@@ -2570,6 +2599,27 @@ impl<E: EthSpec> BeaconState<E> {
         {
             pending_balance.safe_add_assign(withdrawal.amount)?;
         }
+
+        // Sum builder pending withdrawals
+        if let Ok(builder_pending_withdrawals) = self.builder_pending_withdrawals() {
+            for withdrawal in builder_pending_withdrawals
+                .iter()
+                .filter(|withdrawal| withdrawal.builder_index as usize == validator_index)
+            {
+                pending_balance.safe_add_assign(withdrawal.amount)?;
+            }
+        }
+
+        // Sum builder pending payments
+        if let Ok(builder_pending_payments) = self.builder_pending_payments() {
+            for payment in builder_pending_payments
+                .iter()
+                .filter(|payment| payment.withdrawal.builder_index as usize == validator_index)
+            {
+                pending_balance.safe_add_assign(payment.withdrawal.amount)?;
+            }
+        }
+
         Ok(pending_balance)
     }
 
@@ -2842,6 +2892,120 @@ impl<E: EthSpec> BeaconState<E> {
 
         Ok(())
     }
+
+    /// Get the PTC
+    /// Requires the committee cache to be initialized.
+    /// TODO(EIP-7732): definitely gonna have to cache this..
+    pub fn get_ptc(&self, slot: Slot, spec: &ChainSpec) -> Result<PTC<E>, BeaconStateError> {
+        let committee_cache = self.committee_cache_at_slot(slot)?;
+        let committees = committee_cache.get_beacon_committees_at_slot(slot)?;
+
+        let seed = self.get_ptc_attester_seed(slot, spec)?;
+
+        let committee_indices: Vec<usize> = committees
+            .iter()
+            .flat_map(|committee| committee.committee.iter().copied())
+            .collect();
+        let selected_indices = self.compute_balance_weighted_selection(
+            &committee_indices,
+            &seed,
+            E::ptc_size(),
+            false,
+            spec,
+        )?;
+
+        Ok(PTC(FixedVector::new(selected_indices)?))
+    }
+
+    /// Compute the seed to use for the ptc attester selection at the given `slot`.
+    ///
+    /// Spec v0.12.1
+    pub fn get_ptc_attester_seed(
+        &self,
+        slot: Slot,
+        spec: &ChainSpec,
+    ) -> Result<Vec<u8>, BeaconStateError> {
+        let epoch = slot.epoch(E::slots_per_epoch());
+        let mut preimage = self
+            .get_seed(epoch, Domain::PTCAttester, spec)?
+            .as_slice()
+            .to_vec();
+        preimage.append(&mut int_to_bytes8(slot.as_u64()));
+        Ok(hash(&preimage))
+    }
+
+    /// Return size indices sampled by effective balance, using indices as candidates.
+    ///
+    /// If shuffle_indices is True, candidate indices are themselves sampled from indices
+    /// by shuffling it, otherwise indices is traversed in order.
+    fn compute_balance_weighted_selection(
+        &self,
+        indices: &[usize],
+        seed: &[u8],
+        size: usize,
+        shuffle_indices: bool,
+        spec: &ChainSpec,
+    ) -> Result<Vec<usize>, BeaconStateError> {
+        let total = indices.len();
+        if total == 0 {
+            return Err(BeaconStateError::InvalidIndicesCount);
+        }
+
+        let mut selected = Vec::with_capacity(size);
+        let mut count = 0usize;
+
+        while selected.len() < size {
+            let mut next_index = count.safe_rem(total)?;
+
+            if shuffle_indices {
+                next_index =
+                    compute_shuffled_index(next_index, total, seed, spec.shuffle_round_count)
+                        .ok_or(BeaconStateError::UnableToShuffle)?;
+            }
+
+            let candidate_index = indices
+                .get(next_index)
+                .ok_or(BeaconStateError::InvalidIndicesCount)?;
+
+            if self.compute_balance_weighted_acceptance(*candidate_index, seed, count, spec)? {
+                selected.push(*candidate_index);
+            }
+
+            count.safe_add_assign(1)?;
+        }
+
+        Ok(selected)
+    }
+
+    /// Return whether to accept the selection of the validator `index`, with probability
+    /// proportional to its `effective_balance`, and randomness given by `seed` and `iteration`.
+    fn compute_balance_weighted_acceptance(
+        &self,
+        index: usize,
+        seed: &[u8],
+        iteration: usize,
+        spec: &ChainSpec,
+    ) -> Result<bool, BeaconStateError> {
+        // TODO(EIP-7732): Consider grabbing effective balances from the epoch cache here.
+        // Note that this function will be used in a loop, so using cached values could be nice for performance.
+        // However, post-gloas, this function will be used in `compute_proposer_indices`, `get_next_sync_committee_indices`, and `get_ptc`, which has ~15 call sites in total
+        // so we will need to check each one to ensure epoch cache is initialized first, if we deem a good idea.
+        // Currently, we can't test if making the change would work since the test suite is not ready for gloas.
+        let effective_balance = self.get_effective_balance(index)?;
+        let max_effective_balance = spec.max_effective_balance_for_fork(self.fork_name_unchecked());
+
+        let random_value = self.shuffling_random_value(iteration, seed)?;
+
+        // this codepath should technically never be hit pre-gloas, but added this defensively
+        let max_random_value = if self.fork_name_unchecked().electra_enabled() {
+            MAX_RANDOM_VALUE
+        } else {
+            MAX_RANDOM_BYTE
+        };
+
+        Ok(effective_balance.safe_mul(max_random_value)?
+            >= max_effective_balance.safe_mul(random_value)?)
+    }
 }
 
 impl<E: EthSpec> ForkVersionDecode for BeaconState<E> {

consensus/types/src/validator/validator.rs

@@ -165,13 +165,41 @@ impl Validator {
     }
 
     /// Check if ``validator`` has an 0x02 prefixed "compounding" withdrawal credential.
-    pub fn has_compounding_withdrawal_credential(&self, spec: &ChainSpec) -> bool {
+    pub fn has_compounding_withdrawal_credential(
+        &self,
+        spec: &ChainSpec,
+        current_fork: ForkName,
+    ) -> bool {
+        if current_fork.gloas_enabled() {
+            self.has_compounding_withdrawal_credential_gloas(spec)
+        } else {
+            self.has_compounding_withdrawal_credential_electra(spec)
+        }
+    }
+
+    /// Check if ``validator`` has an 0x02 prefixed "compounding" withdrawal credential
+    pub fn has_compounding_withdrawal_credential_electra(&self, spec: &ChainSpec) -> bool {
         is_compounding_withdrawal_credential(self.withdrawal_credentials, spec)
     }
 
+    /// Check if ``validator`` has an 0x02 prefixed "compounding" withdrawal credential or an 0x03 prefixed "builder" withdrawal credential
+    pub fn has_compounding_withdrawal_credential_gloas(&self, spec: &ChainSpec) -> bool {
+        is_compounding_withdrawal_credential(self.withdrawal_credentials, spec)
+            || is_builder_withdrawal_credential(self.withdrawal_credentials, spec)
+    }
+
+    /// Check if ``validator`` has an 0x03 prefixed "builder" withdrawal credential.
+    pub fn has_builder_withdrawal_credential(&self, spec: &ChainSpec) -> bool {
+        is_builder_withdrawal_credential(self.withdrawal_credentials, spec)
+    }
+
     /// Get the execution withdrawal address if this validator has one initialized.
-    pub fn get_execution_withdrawal_address(&self, spec: &ChainSpec) -> Option<Address> {
-        self.has_execution_withdrawal_credential(spec)
+    pub fn get_execution_withdrawal_address(
+        &self,
+        spec: &ChainSpec,
+        current_fork: ForkName,
+    ) -> Option<Address> {
+        self.has_execution_withdrawal_credential(spec, current_fork)
             .then(|| {
                 self.withdrawal_credentials
                     .as_slice()
@@ -202,7 +230,7 @@ impl Validator {
         current_fork: ForkName,
     ) -> bool {
         if current_fork.electra_enabled() {
-            self.is_fully_withdrawable_validator_electra(balance, epoch, spec)
+            self.is_fully_withdrawable_validator_electra(balance, epoch, spec, current_fork)
         } else {
             self.is_fully_withdrawable_validator_capella(balance, epoch, spec)
         }
@@ -226,8 +254,9 @@ impl Validator {
         balance: u64,
         epoch: Epoch,
         spec: &ChainSpec,
+        current_fork: ForkName,
     ) -> bool {
-        self.has_execution_withdrawal_credential(spec)
+        self.has_execution_withdrawal_credential(spec, current_fork)
             && self.withdrawable_epoch <= epoch
             && balance > 0
     }
@@ -267,21 +296,25 @@ impl Validator {
         let max_effective_balance = self.get_max_effective_balance(spec, current_fork);
         let has_max_effective_balance = self.effective_balance == max_effective_balance;
         let has_excess_balance = balance > max_effective_balance;
-        self.has_execution_withdrawal_credential(spec)
+        self.has_execution_withdrawal_credential(spec, current_fork)
             && has_max_effective_balance
             && has_excess_balance
     }
 
     /// Returns `true` if the validator has a 0x01 or 0x02 prefixed withdrawal credential.
-    pub fn has_execution_withdrawal_credential(&self, spec: &ChainSpec) -> bool {
-        self.has_compounding_withdrawal_credential(spec)
+    pub fn has_execution_withdrawal_credential(
+        &self,
+        spec: &ChainSpec,
+        current_fork: ForkName,
+    ) -> bool {
+        self.has_compounding_withdrawal_credential(spec, current_fork)
             || self.has_eth1_withdrawal_credential(spec)
     }
 
     /// Returns the max effective balance for a validator in gwei.
     pub fn get_max_effective_balance(&self, spec: &ChainSpec, current_fork: ForkName) -> u64 {
         if current_fork >= ForkName::Electra {
-            if self.has_compounding_withdrawal_credential(spec) {
+            if self.has_compounding_withdrawal_credential(spec, current_fork) {
                 spec.max_effective_balance_electra
             } else {
                 spec.min_activation_balance
@@ -319,6 +352,15 @@ pub fn is_compounding_withdrawal_credential(
         .unwrap_or(false)
 }
 
+/// Check if the withdrawal credential has the builder withdrawal prefix (0x03).
+pub fn is_builder_withdrawal_credential(withdrawal_credentials: Hash256, spec: &ChainSpec) -> bool {
+    withdrawal_credentials
+        .as_slice()
+        .first()
+        .map(|prefix_byte| *prefix_byte == spec.builder_withdrawal_prefix_byte)
+        .unwrap_or(false)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

consensus/types/tests/state.rs

@@ -55,6 +55,15 @@ async fn build_state<E: EthSpec>(validator_count: usize) -> BeaconState<E> {
         .head_beacon_state_cloned()
 }
 
+// TODO(EIP-7732): Add tests for PTC (Payload Timeliness Committee) functions:
+// - get_ptc: Test committee selection, size, balance-weighted selection
+// - get_ptc_attester_seed: Test seed generation and determinism
+// - compute_balance_weighted_selection: Test selection algorithm with various balances
+// - compute_balance_weighted_acceptance: Test acceptance probability
+// These tests require being able to build Gloas states with initialized committee caches,
+// which currently fails due to incomplete Gloas block structure as mentioned here:
+// https://github.com/sigp/lighthouse/pull/8273
+// Similar to existing committee_consistency_test suite for get_beacon_committee.
 async fn test_beacon_proposer_index<E: EthSpec>() {
     let spec = E::default_spec();
 

testing/ef_tests/src/cases/operations.rs

@@ -419,8 +419,15 @@ impl<E: EthSpec> Operation<E> for WithdrawalsPayload<E> {
         spec: &ChainSpec,
         _: &Operations<E, Self>,
     ) -> Result<(), BlockProcessingError> {
-        // TODO(EIP-7732): implement separate gloas and non-gloas variants of process_withdrawals
-        process_withdrawals::<_, FullPayload<_>>(state, self.payload.to_ref(), spec)
+        if state.fork_name_unchecked().gloas_enabled() {
+            process_withdrawals::gloas::process_withdrawals(state, spec)
+        } else {
+            process_withdrawals::capella::process_withdrawals::<_, FullPayload<_>>(
+                state,
+                self.payload.to_ref(),
+                spec,
+            )
+        }
     }
 }
 

testing/execution_engine_integration/src/test_rig.rs

@@ -135,6 +135,7 @@ impl<Engine: GenericExecutionEngine> TestRig<Engine> {
                 secret_file: None,
                 suggested_fee_recipient: Some(Address::repeat_byte(42)),
                 default_datadir: execution_engine.datadir(),
+                bypass_new_payload_cache: true,
                 ..Default::default()
             };
             let execution_layer = ExecutionLayer::from_config(config, executor.clone()).unwrap();
@@ -153,6 +154,7 @@ impl<Engine: GenericExecutionEngine> TestRig<Engine> {
                 secret_file: None,
                 suggested_fee_recipient: fee_recipient,
                 default_datadir: execution_engine.datadir(),
+                bypass_new_payload_cache: true,
                 ..Default::default()
             };
             let execution_layer = ExecutionLayer::from_config(config, executor).unwrap();