Remove lookup

Co-authored-by: Jimmy Chen <jchen.tc@gmail.com>

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -132,6 +132,7 @@ use task_executor::{RayonPoolType, ShutdownReason, TaskExecutor};
 use tokio_stream::Stream;
 use tracing::{Span, debug, debug_span, error, info, info_span, instrument, trace, warn};
 use tree_hash::TreeHash;
+use types::consts::gloas::PAYLOAD_STATUS_FULL;
 use types::data::{ColumnIndex, FixedBlobSidecarList};
 use types::execution::BlockProductionVersion;
 use types::*;
@@ -1178,6 +1179,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.
+    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(_)) => {
+                // TODO(gloas) should we return None here?
+                Ok(None)
+            }
+            None => Ok(None),
+        }
+    }
+
     /// Returns the block at the given root, if any.
     ///
     /// ## Errors
@@ -2021,19 +2039,25 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         drop(head_timer);
 
         // Only attest to a block if it is fully verified (i.e. not optimistic or invalid).
-        match self
-            .canonical_head
-            .fork_choice_read_lock()
-            .get_block_execution_status(&beacon_block_root)
-        {
-            Some(execution_status) if execution_status.is_valid_or_irrelevant() => (),
-            Some(execution_status) => {
+        // Also grab the block's slot from fork choice for use later.
+        let payload_status = {
+            let fork_choice = self.canonical_head.fork_choice_read_lock();
+            let proto_block = fork_choice
+                .get_block(&beacon_block_root)
+                .ok_or(Error::HeadMissingFromForkChoice(beacon_block_root))?;
+            if !proto_block.execution_status.is_valid_or_irrelevant() {
                 return Err(Error::HeadBlockNotFullyVerified {
                     beacon_block_root,
-                    execution_status,
+                    execution_status: proto_block.execution_status,
                 });
             }
-            None => return Err(Error::HeadMissingFromForkChoice(beacon_block_root)),
+
+            if proto_block.slot < request_slot && proto_block.payload_status == PAYLOAD_STATUS_FULL
+            {
+                1
+            } else {
+                0
+            }
         };
 
         /*
@@ -2080,7 +2104,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 )
             };
 
-        Ok(Attestation::<T::EthSpec>::empty_for_signing(
+        let mut attestation = Attestation::<T::EthSpec>::empty_for_signing(
             request_index,
             committee_len,
             request_slot,
@@ -2088,7 +2112,17 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             justified_checkpoint,
             target,
             &self.spec,
-        )?)
+        )?;
+
+        if self
+            .spec
+            .fork_name_at_slot::<T::EthSpec>(request_slot)
+            .gloas_enabled()
+        {
+            attestation.data_mut().index = payload_status
+        }
+
+        Ok(attestation)
     }
 
     /// Performs the same validation as `Self::verify_unaggregated_attestation_for_gossip`, but for

beacon_node/beacon_chain/src/block_verification.rs

@@ -95,7 +95,7 @@ use std::sync::Arc;
 use store::{Error as DBError, KeyValueStore};
 use strum::AsRefStr;
 use task_executor::JoinHandle;
-use tracing::{Instrument, Span, debug, debug_span, error, info_span, instrument};
+use tracing::{Instrument, Span, debug, debug_span, error, info_span, instrument, warn};
 use types::{
     BeaconBlockRef, BeaconState, BeaconStateError, BlobsList, ChainSpec, DataColumnSidecarList,
     Epoch, EthSpec, FullPayload, Hash256, InconsistentFork, KzgProofs, RelativeEpoch,
@@ -1031,6 +1031,14 @@ impl<T: BeaconChainTypes> GossipVerifiedBlock<T> {
             })));
         }
 
+        // Beacon API execution_payload_bid events
+        if let Some(event_handler) = chain.event_handler.as_ref()
+            && event_handler.has_execution_payload_bid_subscribers()
+            && let Ok(bid) = block.message().body().signed_execution_payload_bid()
+        {
+            event_handler.register(EventKind::ExecutionPayloadBid(Box::new(bid.clone())));
+        }
+
         // Having checked the proposer index and the block root we can cache them.
         let consensus_context = ConsensusContext::new(block.slot())
             .set_current_block_root(block_root)

beacon_node/beacon_chain/src/errors.rs

@@ -313,6 +313,7 @@ pub enum BlockProductionError {
     MissingSyncAggregate,
     MissingExecutionPayload,
     MissingKzgCommitment(String),
+    MissingStateRoot,
     TokioJoin(JoinError),
     BeaconChain(Box<BeaconChainError>),
     InvalidPayloadFork,
@@ -327,6 +328,15 @@ pub enum BlockProductionError {
     GloasNotImplemented(String),
 }
 
+impl From<task_executor::SpawnBlockingError> for BeaconChainError {
+    fn from(e: task_executor::SpawnBlockingError) -> Self {
+        match e {
+            task_executor::SpawnBlockingError::RuntimeShutdown => BeaconChainError::RuntimeShutdown,
+            task_executor::SpawnBlockingError::JoinError(e) => BeaconChainError::TokioJoin(e),
+        }
+    }
+}
+
 easy_from_to!(BlockProcessingError, BlockProductionError);
 easy_from_to!(BeaconStateError, BlockProductionError);
 easy_from_to!(SlotProcessingError, BlockProductionError);

beacon_node/beacon_chain/src/events.rs

@@ -25,6 +25,8 @@ pub struct ServerSentEventHandler<E: EthSpec> {
     attester_slashing_tx: Sender<EventKind<E>>,
     bls_to_execution_change_tx: Sender<EventKind<E>>,
     block_gossip_tx: Sender<EventKind<E>>,
+    execution_payload_bid_tx: Sender<EventKind<E>>,
+    execution_payload_available_tx: Sender<EventKind<E>>,
 }
 
 impl<E: EthSpec> ServerSentEventHandler<E> {
@@ -51,6 +53,8 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
         let (attester_slashing_tx, _) = broadcast::channel(capacity);
         let (bls_to_execution_change_tx, _) = broadcast::channel(capacity);
         let (block_gossip_tx, _) = broadcast::channel(capacity);
+        let (execution_payload_bid_tx, _) = broadcast::channel(capacity);
+        let (execution_payload_available_tx, _) = broadcast::channel(capacity);
 
         Self {
             attestation_tx,
@@ -71,6 +75,8 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
             attester_slashing_tx,
             bls_to_execution_change_tx,
             block_gossip_tx,
+            execution_payload_bid_tx,
+            execution_payload_available_tx,
         }
     }
 
@@ -155,6 +161,14 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
                 .block_gossip_tx
                 .send(kind)
                 .map(|count| log_count("block gossip", count)),
+            EventKind::ExecutionPayloadBid(_) => self
+                .execution_payload_bid_tx
+                .send(kind)
+                .map(|count| log_count("execution payload bid", count)),
+            EventKind::ExecutionPayloadAvailable(_) => self
+                .execution_payload_available_tx
+                .send(kind)
+                .map(|count| log_count("execution payload available", count)),
         };
         if let Err(SendError(event)) = result {
             trace!(?event, "No receivers registered to listen for event");
@@ -296,4 +310,20 @@ impl<E: EthSpec> ServerSentEventHandler<E> {
     pub fn has_block_gossip_subscribers(&self) -> bool {
         self.block_gossip_tx.receiver_count() > 0
     }
+
+    pub fn subscribe_execution_payload_bid(&self) -> Receiver<EventKind<E>> {
+        self.execution_payload_bid_tx.subscribe()
+    }
+
+    pub fn subscribe_execution_payload_available(&self) -> Receiver<EventKind<E>> {
+        self.execution_payload_available_tx.subscribe()
+    }
+
+    pub fn has_execution_payload_bid_subscribers(&self) -> bool {
+        self.execution_payload_bid_tx.receiver_count() > 0
+    }
+
+    pub fn has_execution_payload_available_subscribers(&self) -> bool {
+        self.execution_payload_available_tx.receiver_count() > 0
+    }
 }

beacon_node/beacon_chain/src/execution_payload.rs

@@ -298,9 +298,18 @@ pub fn get_execution_payload<T: BeaconChainTypes>(
     let timestamp =
         compute_timestamp_at_slot(state, state.slot(), spec).map_err(BeaconStateError::from)?;
     let random = *state.get_randao_mix(current_epoch)?;
-    let latest_execution_payload_header = state.latest_execution_payload_header()?;
-    let latest_execution_payload_header_block_hash = latest_execution_payload_header.block_hash();
-    let latest_execution_payload_header_gas_limit = latest_execution_payload_header.gas_limit();
+    // In GLOAS (ePBS), the execution payload header is replaced by
+    // `latest_block_hash` and `latest_execution_payload_bid`.
+    let (latest_execution_payload_header_block_hash, latest_execution_payload_header_gas_limit) =
+        if state.fork_name_unchecked() == ForkName::Gloas {
+            (
+                *state.latest_block_hash()?,
+                state.latest_execution_payload_bid()?.gas_limit,
+            )
+        } else {
+            let header = state.latest_execution_payload_header()?;
+            (header.block_hash(), header.gas_limit())
+        };
     let withdrawals = if state.fork_name_unchecked().capella_enabled() {
         Some(Withdrawals::<T::EthSpec>::from(get_expected_withdrawals(state, spec)?).into())
     } else {

beacon_node/beacon_chain/src/payload_envelope_verification/execution_pending_envelope.rs

@@ -12,7 +12,7 @@ use crate::{
     PayloadVerificationOutcome,
     block_verification::PayloadVerificationHandle,
     payload_envelope_verification::{
-        EnvelopeError, EnvelopeImportData, MaybeAvailableEnvelope,
+        AvailableEnvelope, EnvelopeError, EnvelopeImportData, MaybeAvailableEnvelope,
         gossip_verified_envelope::GossipVerifiedEnvelope, load_snapshot_from_state_root,
         payload_notifier::PayloadNotifier,
     },
@@ -32,7 +32,6 @@ impl<T: BeaconChainTypes> GossipVerifiedEnvelope<T> {
     ) -> Result<ExecutionPendingEnvelope<T::EthSpec>, EnvelopeError> {
         let signed_envelope = self.signed_envelope;
         let envelope = &signed_envelope.message;
-        let payload = &envelope.payload;
 
         // Define a future that will verify the execution payload with an execution engine.
         //
@@ -91,10 +90,13 @@ impl<T: BeaconChainTypes> GossipVerifiedEnvelope<T> {
         )?;
 
         Ok(ExecutionPendingEnvelope {
-            signed_envelope: MaybeAvailableEnvelope::AvailabilityPending {
-                block_hash: payload.block_hash,
-                envelope: signed_envelope,
-            },
+            signed_envelope: MaybeAvailableEnvelope::Available(AvailableEnvelope {
+                execution_block_hash: signed_envelope.block_hash(),
+                envelope: signed_envelope.clone(),
+                columns: vec![],
+                columns_available_timestamp: None,
+                spec: chain.spec.clone(),
+            }),
             import_data: EnvelopeImportData {
                 block_root,
                 post_state: Box::new(state),

beacon_node/beacon_chain/src/payload_envelope_verification/import.rs

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 use std::time::Duration;
 
+use eth2::types::{EventKind, SseExecutionPayloadAvailable};
 use fork_choice::PayloadVerificationStatus;
 use slot_clock::SlotClock;
 use store::StoreOp;
@@ -13,8 +14,11 @@ use super::{
 };
 use crate::{
     AvailabilityProcessingStatus, BeaconChain, BeaconChainError, BeaconChainTypes,
-    NotifyExecutionLayer, block_verification_types::AvailableBlockData, metrics,
-    payload_envelope_verification::ExecutionPendingEnvelope, validator_monitor::get_slot_delay_ms,
+    NotifyExecutionLayer,
+    block_verification_types::AvailableBlockData,
+    metrics,
+    payload_envelope_verification::ExecutionPendingEnvelope,
+    validator_monitor::{get_slot_delay_ms, timestamp_now},
 };
 
 const ENVELOPE_METRICS_CACHE_SLOT_LIMIT: u32 = 64;
@@ -235,21 +239,20 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             return Err(EnvelopeError::BlockRootUnknown { block_root });
         }
 
-        // TODO(gloas) add defensive check to see if payload envelope is already in fork choice
-        // Note that a duplicate cache/payload status table should prevent this from happening
-        // but it doesnt hurt to be defensive.
-
-        // TODO(gloas) when the code below is implemented we can delete this drop
-        drop(fork_choice_reader);
-
         // TODO(gloas) no fork choice logic yet
         // Take an exclusive write-lock on fork choice. It's very important to prevent deadlocks by
         // avoiding taking other locks whilst holding this lock.
-        // let fork_choice = parking_lot::RwLockUpgradableReadGuard::upgrade(fork_choice_reader);
+        let mut fork_choice = parking_lot::RwLockUpgradableReadGuard::upgrade(fork_choice_reader);
+
+        // Update the node's payload_status from PENDING to FULL in fork choice.
+        fork_choice
+            .on_execution_payload(block_root)
+            .map_err(|e| EnvelopeError::InternalError(format!("{e:?}")))?;
 
         // TODO(gloas) Do we need this check? Do not import a block that doesn't descend from the finalized root.
         // let signed_block = check_block_is_finalized_checkpoint_or_descendant(self, &fork_choice, signed_block)?;
 
+        // TODO(gloas) Do we want to use an early attester cache like mechanism for payload enevelopes?
         // TODO(gloas) emit SSE event if the payload became the new head payload
 
         // It is important NOT to return errors here before the database commit, because the envelope
@@ -257,7 +260,8 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         // state if we returned early without committing. In other words, an error here would
         // corrupt the node's database permanently.
 
-        // Store the envelope, its post-state, and any data columns.
+        // Store the envelope and its state, and execute the confirmation batch for the intermediate
+        // states, which will delete their temporary flags.
         // If the write fails, revert fork choice to the version from disk, else we can
         // end up with envelopes in fork choice that are missing from disk.
         // See https://github.com/sigp/lighthouse/issues/2028
@@ -302,19 +306,19 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
         drop(db_span);
 
-        // TODO(gloas) drop fork choice lock
         // The fork choice write-lock is dropped *after* the on-disk database has been updated.
         // This prevents inconsistency between the two at the expense of concurrency.
-        // drop(fork_choice);
+        drop(fork_choice);
 
         // We're declaring the envelope "imported" at this point, since fork choice and the DB know
         // about it.
-        let envelope_time_imported = self.slot_clock.now_duration().unwrap_or(Duration::MAX);
+        let envelope_time_imported = timestamp_now();
 
         // TODO(gloas) depending on what happens with light clients
         // we might need to do some light client related computations here
 
         metrics::stop_timer(db_write_timer);
+        metrics::inc_counter(&metrics::ENVELOPE_PROCESSING_SUCCESSES);
 
         self.import_envelope_update_metrics_and_events(
             block_root,
@@ -349,6 +353,16 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             );
         }
 
-        // TODO(gloas) emit SSE event for envelope import (similar to SseBlock for blocks).
+        // Beacon API execution_payload_available events
+        if let Some(event_handler) = self.event_handler.as_ref()
+            && event_handler.has_execution_payload_available_subscribers()
+        {
+            event_handler.register(EventKind::ExecutionPayloadAvailable(
+                SseExecutionPayloadAvailable {
+                    slot: envelope_slot,
+                    block_root,
+                },
+            ));
+        }
     }
 }

beacon_node/beacon_chain/src/payload_envelope_verification/mod.rs

@@ -59,6 +59,22 @@ pub struct AvailableEnvelope<E: EthSpec> {
 }
 
 impl<E: EthSpec> AvailableEnvelope<E> {
+    pub fn new(
+        execution_block_hash: ExecutionBlockHash,
+        envelope: Arc<SignedExecutionPayloadEnvelope<E>>,
+        columns: DataColumnSidecarList<E>,
+        columns_available_timestamp: Option<std::time::Duration>,
+        spec: Arc<ChainSpec>,
+    ) -> Self {
+        Self {
+            execution_block_hash,
+            envelope,
+            columns,
+            columns_available_timestamp,
+            spec,
+        }
+    }
+
     pub fn message(&self) -> &ExecutionPayloadEnvelope<E> {
         &self.envelope.message
     }

beacon_node/beacon_processor/src/lib.rs

@@ -416,6 +416,9 @@ pub enum Work<E: EthSpec> {
     RpcBlobs {
         process_fn: AsyncFn,
     },
+    RpcPayloadEnvelope {
+        process_fn: AsyncFn,
+    },
     RpcCustodyColumn(AsyncFn),
     ColumnReconstruction(AsyncFn),
     IgnoredRpcBlock {
@@ -477,6 +480,7 @@ pub enum WorkType {
     GossipLightClientOptimisticUpdate,
     RpcBlock,
     RpcBlobs,
+    RpcPayloadEnvelope,
     RpcCustodyColumn,
     ColumnReconstruction,
     IgnoredRpcBlock,
@@ -538,6 +542,7 @@ impl<E: EthSpec> Work<E> {
             Work::GossipProposerPreferences(_) => WorkType::GossipProposerPreferences,
             Work::RpcBlock { .. } => WorkType::RpcBlock,
             Work::RpcBlobs { .. } => WorkType::RpcBlobs,
+            Work::RpcPayloadEnvelope { .. } => WorkType::RpcPayloadEnvelope,
             Work::RpcCustodyColumn { .. } => WorkType::RpcCustodyColumn,
             Work::ColumnReconstruction(_) => WorkType::ColumnReconstruction,
             Work::IgnoredRpcBlock { .. } => WorkType::IgnoredRpcBlock,
@@ -1169,7 +1174,9 @@ impl<E: EthSpec> BeaconProcessor<E> {
                             Work::GossipLightClientOptimisticUpdate { .. } => work_queues
                                 .lc_gossip_optimistic_update_queue
                                 .push(work, work_id),
-                            Work::RpcBlock { .. } | Work::IgnoredRpcBlock { .. } => {
+                            Work::RpcBlock { .. }
+                            | Work::IgnoredRpcBlock { .. }
+                            | Work::RpcPayloadEnvelope { .. } => {
                                 work_queues.rpc_block_queue.push(work, work_id)
                             }
                             Work::RpcBlobs { .. } => work_queues.rpc_blob_queue.push(work, work_id),
@@ -1301,7 +1308,9 @@ impl<E: EthSpec> BeaconProcessor<E> {
                         WorkType::GossipLightClientOptimisticUpdate => {
                             work_queues.lc_gossip_optimistic_update_queue.len()
                         }
-                        WorkType::RpcBlock => work_queues.rpc_block_queue.len(),
+                        WorkType::RpcBlock | WorkType::RpcPayloadEnvelope => {
+                            work_queues.rpc_block_queue.len()
+                        }
                         WorkType::RpcBlobs | WorkType::IgnoredRpcBlock => {
                             work_queues.rpc_blob_queue.len()
                         }
@@ -1496,6 +1505,7 @@ impl<E: EthSpec> BeaconProcessor<E> {
                 beacon_block_root: _,
             }
             | Work::RpcBlobs { process_fn }
+            | Work::RpcPayloadEnvelope { process_fn }
             | Work::RpcCustodyColumn(process_fn)
             | Work::ColumnReconstruction(process_fn) => task_spawner.spawn_async(process_fn),
             Work::IgnoredRpcBlock { process_fn } => task_spawner.spawn_blocking(process_fn),

beacon_node/execution_layer/src/engine_api.rs

@@ -5,7 +5,7 @@ use crate::http::{
     ENGINE_GET_PAYLOAD_BODIES_BY_HASH_V1, ENGINE_GET_PAYLOAD_BODIES_BY_RANGE_V1,
     ENGINE_GET_PAYLOAD_V1, ENGINE_GET_PAYLOAD_V2, ENGINE_GET_PAYLOAD_V3, ENGINE_GET_PAYLOAD_V4,
     ENGINE_GET_PAYLOAD_V5, ENGINE_NEW_PAYLOAD_V1, ENGINE_NEW_PAYLOAD_V2, ENGINE_NEW_PAYLOAD_V3,
-    ENGINE_NEW_PAYLOAD_V4,
+    ENGINE_NEW_PAYLOAD_V4, ENGINE_NEW_PAYLOAD_V5,
 };
 use eth2::types::{
     BlobsBundle, SsePayloadAttributes, SsePayloadAttributesV1, SsePayloadAttributesV2,
@@ -551,6 +551,7 @@ pub struct EngineCapabilities {
     pub new_payload_v2: bool,
     pub new_payload_v3: bool,
     pub new_payload_v4: bool,
+    pub new_payload_v5: bool,
     pub forkchoice_updated_v1: bool,
     pub forkchoice_updated_v2: bool,
     pub forkchoice_updated_v3: bool,
@@ -581,6 +582,9 @@ impl EngineCapabilities {
         if self.new_payload_v4 {
             response.push(ENGINE_NEW_PAYLOAD_V4);
         }
+        if self.new_payload_v5 {
+            response.push(ENGINE_NEW_PAYLOAD_V5);
+        }
         if self.forkchoice_updated_v1 {
             response.push(ENGINE_FORKCHOICE_UPDATED_V1);
         }

beacon_node/execution_layer/src/engine_api/http.rs

@@ -35,6 +35,7 @@ pub const ENGINE_NEW_PAYLOAD_V1: &str = "engine_newPayloadV1";
 pub const ENGINE_NEW_PAYLOAD_V2: &str = "engine_newPayloadV2";
 pub const ENGINE_NEW_PAYLOAD_V3: &str = "engine_newPayloadV3";
 pub const ENGINE_NEW_PAYLOAD_V4: &str = "engine_newPayloadV4";
+pub const ENGINE_NEW_PAYLOAD_V5: &str = "engine_newPayloadV5";
 pub const ENGINE_NEW_PAYLOAD_TIMEOUT: Duration = Duration::from_secs(8);
 
 pub const ENGINE_GET_PAYLOAD_V1: &str = "engine_getPayloadV1";
@@ -74,6 +75,7 @@ pub static LIGHTHOUSE_CAPABILITIES: &[&str] = &[
     ENGINE_NEW_PAYLOAD_V2,
     ENGINE_NEW_PAYLOAD_V3,
     ENGINE_NEW_PAYLOAD_V4,
+    ENGINE_NEW_PAYLOAD_V5,
     ENGINE_GET_PAYLOAD_V1,
     ENGINE_GET_PAYLOAD_V2,
     ENGINE_GET_PAYLOAD_V3,
@@ -883,7 +885,7 @@ impl HttpJsonRpc {
         Ok(response.into())
     }
 
-    pub async fn new_payload_v4_gloas<E: EthSpec>(
+    pub async fn new_payload_v5_gloas<E: EthSpec>(
         &self,
         new_payload_request_gloas: NewPayloadRequestGloas<'_, E>,
     ) -> Result<PayloadStatusV1, Error> {
@@ -903,7 +905,7 @@ impl HttpJsonRpc {
 
         let response: JsonPayloadStatusV1 = self
             .rpc_request(
-                ENGINE_NEW_PAYLOAD_V4,
+                ENGINE_NEW_PAYLOAD_V5,
                 params,
                 ENGINE_NEW_PAYLOAD_TIMEOUT * self.execution_timeout_multiplier,
             )
@@ -1198,6 +1200,7 @@ impl HttpJsonRpc {
             new_payload_v2: capabilities.contains(ENGINE_NEW_PAYLOAD_V2),
             new_payload_v3: capabilities.contains(ENGINE_NEW_PAYLOAD_V3),
             new_payload_v4: capabilities.contains(ENGINE_NEW_PAYLOAD_V4),
+            new_payload_v5: capabilities.contains(ENGINE_NEW_PAYLOAD_V5),
             forkchoice_updated_v1: capabilities.contains(ENGINE_FORKCHOICE_UPDATED_V1),
             forkchoice_updated_v2: capabilities.contains(ENGINE_FORKCHOICE_UPDATED_V2),
             forkchoice_updated_v3: capabilities.contains(ENGINE_FORKCHOICE_UPDATED_V3),
@@ -1353,10 +1356,10 @@ impl HttpJsonRpc {
                 }
             }
             NewPayloadRequest::Gloas(new_payload_request_gloas) => {
-                if engine_capabilities.new_payload_v4 {
-                    self.new_payload_v4_gloas(new_payload_request_gloas).await
+                if engine_capabilities.new_payload_v5 {
+                    self.new_payload_v5_gloas(new_payload_request_gloas).await
                 } else {
-                    Err(Error::RequiredMethodUnsupported("engine_newPayloadV4"))
+                    Err(Error::RequiredMethodUnsupported("engine_newPayloadV5"))
                 }
             }
         }

beacon_node/execution_layer/src/test_utils/handle_rpc.rs

@@ -102,7 +102,8 @@ pub async fn handle_rpc<E: EthSpec>(
         ENGINE_NEW_PAYLOAD_V1
         | ENGINE_NEW_PAYLOAD_V2
         | ENGINE_NEW_PAYLOAD_V3
-        | ENGINE_NEW_PAYLOAD_V4 => {
+        | ENGINE_NEW_PAYLOAD_V4
+        | ENGINE_NEW_PAYLOAD_V5 => {
             let request = match method {
                 ENGINE_NEW_PAYLOAD_V1 => JsonExecutionPayload::Bellatrix(
                     get_param::<JsonExecutionPayloadBellatrix<E>>(params, 0)
@@ -118,17 +119,16 @@ pub async fn handle_rpc<E: EthSpec>(
                 ENGINE_NEW_PAYLOAD_V3 => get_param::<JsonExecutionPayloadDeneb<E>>(params, 0)
                     .map(|jep| JsonExecutionPayload::Deneb(jep))
                     .map_err(|s| (s, BAD_PARAMS_ERROR_CODE))?,
-                ENGINE_NEW_PAYLOAD_V4 => get_param::<JsonExecutionPayloadGloas<E>>(params, 0)
-                    .map(|jep| JsonExecutionPayload::Gloas(jep))
-                    .or_else(|_| {
-                        get_param::<JsonExecutionPayloadFulu<E>>(params, 0)
-                            .map(|jep| JsonExecutionPayload::Fulu(jep))
-                    })
+                ENGINE_NEW_PAYLOAD_V4 => get_param::<JsonExecutionPayloadFulu<E>>(params, 0)
+                    .map(|jep| JsonExecutionPayload::Fulu(jep))
                     .or_else(|_| {
                         get_param::<JsonExecutionPayloadElectra<E>>(params, 0)
                             .map(|jep| JsonExecutionPayload::Electra(jep))
                     })
                     .map_err(|s| (s, BAD_PARAMS_ERROR_CODE))?,
+                ENGINE_NEW_PAYLOAD_V5 => get_param::<JsonExecutionPayloadGloas<E>>(params, 0)
+                    .map(|jep| JsonExecutionPayload::Gloas(jep))
+                    .map_err(|s| (s, BAD_PARAMS_ERROR_CODE))?,
                 _ => unreachable!(),
             };
 
@@ -192,7 +192,7 @@ pub async fn handle_rpc<E: EthSpec>(
                         ));
                     }
                 }
-                ForkName::Electra | ForkName::Fulu | ForkName::Gloas => {
+                ForkName::Electra | ForkName::Fulu => {
                     if method == ENGINE_NEW_PAYLOAD_V1
                         || method == ENGINE_NEW_PAYLOAD_V2
                         || method == ENGINE_NEW_PAYLOAD_V3
@@ -230,6 +230,14 @@ pub async fn handle_rpc<E: EthSpec>(
                         ));
                     }
                 }
+                ForkName::Gloas => {
+                    if method != ENGINE_NEW_PAYLOAD_V5 {
+                        return Err((
+                            format!("{} called for Gloas fork, use engine_newPayloadV5!", method),
+                            GENERIC_ERROR_CODE,
+                        ));
+                    }
+                }
                 _ => unreachable!(),
             };
 

beacon_node/execution_layer/src/test_utils/mod.rs

@@ -43,6 +43,7 @@ pub const DEFAULT_ENGINE_CAPABILITIES: EngineCapabilities = EngineCapabilities {
     new_payload_v2: true,
     new_payload_v3: true,
     new_payload_v4: true,
+    new_payload_v5: true,
     forkchoice_updated_v1: true,
     forkchoice_updated_v2: true,
     forkchoice_updated_v3: true,

beacon_node/http_api/src/beacon/execution_payload_envelope.rs

@@ -1,15 +1,25 @@
+use crate::block_id::BlockId;
 use crate::task_spawner::{Priority, TaskSpawner};
 use crate::utils::{ChainFilter, EthV1Filter, NetworkTxFilter, ResponseFilter, TaskSpawnerFilter};
-use beacon_chain::{BeaconChain, BeaconChainTypes};
+use crate::version::{
+    ResponseIncludesVersion, add_consensus_version_header, add_ssz_content_type_header,
+    execution_optimistic_finalized_beacon_response,
+};
+use beacon_chain::payload_envelope_verification::gossip_verified_envelope::GossipVerifiedEnvelope;
+use beacon_chain::{
+    BeaconChain, BeaconChainTypes, NotifyExecutionLayer,
+    payload_envelope_verification::EnvelopeError,
+};
 use bytes::Bytes;
+use eth2::types as api_types;
 use eth2::{CONTENT_TYPE_HEADER, SSZ_CONTENT_TYPE_HEADER};
 use lighthouse_network::PubsubMessage;
 use network::NetworkMessage;
-use ssz::Decode;
+use ssz::{Decode, Encode};
 use std::sync::Arc;
 use tokio::sync::mpsc::UnboundedSender;
 use tracing::{info, warn};
-use types::SignedExecutionPayloadEnvelope;
+use types::{BlockImportSource, SignedExecutionPayloadEnvelope};
 use warp::{Filter, Rejection, Reply, reply::Response};
 
 // POST beacon/execution_payload_envelope (SSZ)
@@ -76,7 +86,9 @@ pub(crate) fn post_beacon_execution_payload_envelope<T: BeaconChainTypes>(
         )
         .boxed()
 }
-/// Publishes a signed execution payload envelope to the network.
+/// Locally imports and publishes a signed execution payload envelope to the network.
+///
+/// TODO(gloas): Add gossip verification (BroadcastValidation::Gossip) before import.
 pub async fn publish_execution_payload_envelope<T: BeaconChainTypes>(
     envelope: SignedExecutionPayloadEnvelope<T::EthSpec>,
     chain: Arc<BeaconChain<T>>,
@@ -84,33 +96,131 @@ pub async fn publish_execution_payload_envelope<T: BeaconChainTypes>(
 ) -> Result<Response, Rejection> {
     let slot = envelope.message.slot;
     let beacon_block_root = envelope.message.beacon_block_root;
+    let builder_index = envelope.message.builder_index;
 
-    // TODO(gloas): Replace this check once we have gossip validation.
     if !chain.spec.is_gloas_scheduled() {
         return Err(warp_utils::reject::custom_bad_request(
             "Execution payload envelopes are not supported before the Gloas fork".into(),
         ));
     }
 
-    // TODO(gloas): We should probably add validation here i.e. BroadcastValidation::Gossip
-    info!(
-        %slot,
-        %beacon_block_root,
-        builder_index = envelope.message.builder_index,
-        "Publishing signed execution payload envelope to network"
-    );
+    let signed_envelope = Arc::new(envelope);
 
-    // Publish to the network
-    crate::utils::publish_pubsub_message(
-        network_tx,
-        PubsubMessage::ExecutionPayload(Box::new(envelope)),
-    )
-    .map_err(|_| {
-        warn!(%slot, "Failed to publish execution payload envelope to network");
-        warp_utils::reject::custom_server_error(
-            "Unable to publish execution payload envelope to network".into(),
+    // The publish_fn is called inside process_execution_payload_envelope after consensus
+    // verification but before the EL call.
+    let envelope_for_publish = signed_envelope.clone();
+    let sender = network_tx.clone();
+    let publish_fn = move || {
+        info!(
+            %slot,
+            %beacon_block_root,
+            builder_index,
+            "Publishing signed execution payload envelope to network"
+        );
+        crate::utils::publish_pubsub_message(
+            &sender,
+            PubsubMessage::ExecutionPayload(Box::new((*envelope_for_publish).clone())),
         )
-    })?;
+        .map_err(|_| {
+            warn!(%slot, "Failed to publish execution payload envelope to network");
+            EnvelopeError::InternalError(
+                "Unable to publish execution payload envelope to network".to_owned(),
+            )
+        })
+    };
+
+    let ctx = chain.gossip_verification_context();
+    let Ok(gossip_verifed_envelope) = GossipVerifiedEnvelope::new(signed_envelope, &ctx) else {
+        warn!(%slot, %beacon_block_root, "Execution payload envelope rejected");
+        return Err(warp_utils::reject::custom_bad_request(
+            "execution payload envelope rejected, gossip verification".to_string(),
+        ));
+    };
+
+    // Import the envelope locally (runs state transition and notifies the EL).
+    chain
+        .process_execution_payload_envelope(
+            beacon_block_root,
+            gossip_verifed_envelope,
+            NotifyExecutionLayer::Yes,
+            BlockImportSource::HttpApi,
+            publish_fn,
+        )
+        .await
+        .map_err(|e| {
+            warn!(%slot, %beacon_block_root, reason = ?e, "Execution payload envelope rejected");
+            warp_utils::reject::custom_bad_request(format!(
+                "execution payload envelope rejected: {e:?}"
+            ))
+        })?;
 
     Ok(warp::reply().into_response())
 }
+
+// GET beacon/execution_payload_envelope/{block_id}
+pub(crate) fn get_beacon_execution_payload_envelope<T: BeaconChainTypes>(
+    eth_v1: EthV1Filter,
+    block_id_or_err: impl Filter<Extract = (BlockId,), Error = Rejection>
+    + Clone
+    + Send
+    + Sync
+    + 'static,
+    task_spawner_filter: TaskSpawnerFilter<T>,
+    chain_filter: ChainFilter<T>,
+) -> ResponseFilter {
+    eth_v1
+        .and(warp::path("beacon"))
+        .and(warp::path("execution_payload_envelope"))
+        .and(block_id_or_err)
+        .and(warp::path::end())
+        .and(task_spawner_filter)
+        .and(chain_filter)
+        .and(warp::header::optional::<api_types::Accept>("accept"))
+        .then(
+            |block_id: BlockId,
+             task_spawner: TaskSpawner<T::EthSpec>,
+             chain: Arc<BeaconChain<T>>,
+             accept_header: Option<api_types::Accept>| {
+                task_spawner.blocking_response_task(Priority::P1, move || {
+                    let (root, execution_optimistic, finalized) = block_id.root(&chain)?;
+
+                    let envelope = chain
+                        .get_payload_envelope(&root)
+                        .map_err(warp_utils::reject::unhandled_error)?
+                        .ok_or_else(|| {
+                            warp_utils::reject::custom_not_found(format!(
+                                "execution payload envelope for block root {root}"
+                            ))
+                        })?;
+
+                    let fork_name = chain
+                        .spec
+                        .fork_name_at_slot::<T::EthSpec>(envelope.message.slot);
+
+                    match accept_header {
+                        Some(api_types::Accept::Ssz) => warp::http::Response::builder()
+                            .status(200)
+                            .body(warp::hyper::Body::from(envelope.as_ssz_bytes()))
+                            .map(add_ssz_content_type_header)
+                            .map_err(|e| {
+                                warp_utils::reject::custom_server_error(format!(
+                                    "failed to create response: {}",
+                                    e
+                                ))
+                            }),
+                        _ => {
+                            let res = execution_optimistic_finalized_beacon_response(
+                                ResponseIncludesVersion::Yes(fork_name),
+                                execution_optimistic,
+                                finalized,
+                                &envelope,
+                            )?;
+                            Ok(warp::reply::json(&res).into_response())
+                        }
+                    }
+                    .map(|resp| add_consensus_version_header(resp, fork_name))
+                })
+            },
+        )
+        .boxed()
+}

beacon_node/http_api/src/lib.rs

@@ -35,7 +35,8 @@ mod validators;
 mod version;
 
 use crate::beacon::execution_payload_envelope::{
-    post_beacon_execution_payload_envelope, post_beacon_execution_payload_envelope_ssz,
+    get_beacon_execution_payload_envelope, post_beacon_execution_payload_envelope,
+    post_beacon_execution_payload_envelope_ssz,
 };
 use crate::beacon::pool::*;
 use crate::light_client::{get_light_client_bootstrap, get_light_client_updates};
@@ -1501,6 +1502,14 @@ pub fn serve<T: BeaconChainTypes>(
         network_tx_filter.clone(),
     );
 
+    // GET beacon/execution_payload_envelope/{block_id}
+    let get_beacon_execution_payload_envelope = get_beacon_execution_payload_envelope(
+        eth_v1.clone(),
+        block_id_or_err,
+        task_spawner_filter.clone(),
+        chain_filter.clone(),
+    );
+
     // POST beacon/execution_payload_envelope (SSZ)
     let post_beacon_execution_payload_envelope_ssz = post_beacon_execution_payload_envelope_ssz(
         eth_v1.clone(),
@@ -3158,6 +3167,12 @@ pub fn serve<T: BeaconChainTypes>(
                                 api_types::EventTopic::BlockGossip => {
                                     event_handler.subscribe_block_gossip()
                                 }
+                                api_types::EventTopic::ExecutionPayloadBid => {
+                                    event_handler.subscribe_execution_payload_bid()
+                                }
+                                api_types::EventTopic::ExecutionPayloadAvailable => {
+                                    event_handler.subscribe_execution_payload_available()
+                                }
                             };
 
                             receivers.push(
@@ -3283,6 +3298,7 @@ pub fn serve<T: BeaconChainTypes>(
                 .uor(get_beacon_block_root)
                 .uor(get_blob_sidecars)
                 .uor(get_blobs)
+                .uor(get_beacon_execution_payload_envelope)
                 .uor(get_beacon_pool_attestations)
                 .uor(get_beacon_pool_attester_slashings)
                 .uor(get_beacon_pool_proposer_slashings)

beacon_node/network/src/network_beacon_processor/sync_methods.rs

@@ -73,6 +73,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
         Box::pin(process_fn)
     }
 
+    /// Returns an async closure which processes a payload envelope received via RPC.
     /// Returns the `process_fn` and `ignore_fn` required when requeuing an RPC block.
     pub fn generate_lookup_beacon_block_fns(
         self: Arc<Self>,

beacon_node/network/src/router.rs

@@ -24,7 +24,10 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use tokio::sync::mpsc;
 use tokio_stream::wrappers::UnboundedReceiverStream;
 use tracing::{debug, error, trace, warn};
-use types::{BlobSidecar, DataColumnSidecar, EthSpec, ForkContext, SignedBeaconBlock};
+use types::{
+    BlobSidecar, DataColumnSidecar, EthSpec, ForkContext, SignedBeaconBlock,
+    SignedExecutionPayloadEnvelope,
+};
 
 /// Handles messages from the network and routes them to the appropriate service to be handled.
 pub struct Router<T: BeaconChainTypes> {
@@ -327,10 +330,13 @@ impl<T: BeaconChainTypes> Router<T> {
             Response::DataColumnsByRange(data_column) => {
                 self.on_data_columns_by_range_response(peer_id, app_request_id, data_column);
             }
-            // TODO(EIP-7732): implement outgoing payload envelopes by range and root
-            // responses once sync manager requests them.
-            Response::PayloadEnvelopesByRoot(_) | Response::PayloadEnvelopesByRange(_) => {
-                debug!("Requesting envelopes by root and by range not supported yet");
+            Response::PayloadEnvelopesByRoot(envelope) => {
+                self.on_payload_envelopes_by_root_response(peer_id, app_request_id, envelope);
+            }
+            // TODO(EIP-7732): implement outgoing payload envelopes by range responses once
+            // range sync requests them.
+            Response::PayloadEnvelopesByRange(_) => {
+                unreachable!()
             }
             // Light client responses should not be received
             Response::LightClientBootstrap(_)
@@ -703,6 +709,20 @@ impl<T: BeaconChainTypes> Router<T> {
         });
     }
 
+    /// Handle a `PayloadEnvelopesByRoot` response from the peer.
+    pub fn on_payload_envelopes_by_root_response(
+        &mut self,
+        peer_id: PeerId,
+        _app_request_id: AppRequestId,
+        _envelope: Option<Arc<SignedExecutionPayloadEnvelope<T::EthSpec>>>,
+    ) {
+        // TODO(EIP-7732): Envelope lookup sync not yet implemented on this branch.
+        crit!(
+            %peer_id,
+            "Received unexpected PayloadEnvelopesByRoot response"
+        );
+    }
+
     /// Handle a `BlobsByRoot` response from the peer.
     pub fn on_blobs_by_root_response(
         &mut self,

beacon_node/network/src/sync/block_lookups/single_block_lookup.rs

@@ -287,7 +287,7 @@ impl<T: BeaconChainTypes> SingleBlockLookup<T> {
         expected_blobs: usize,
     ) -> Result<(), LookupRequestError> {
         let id = self.id;
-        let awaiting_parent = self.awaiting_parent.is_some();
+        let awaiting_event = self.awaiting_parent.is_some();
         let request =
             R::request_state_mut(self).map_err(|e| LookupRequestError::BadState(e.to_owned()))?;
 
@@ -331,7 +331,7 @@ impl<T: BeaconChainTypes> SingleBlockLookup<T> {
         // Otherwise, attempt to progress awaiting processing
         // If this request is awaiting a parent lookup to be processed, do not send for processing.
         // The request will be rejected with unknown parent error.
-        } else if !awaiting_parent {
+        } else if !awaiting_event {
             // maybe_start_processing returns Some if state == AwaitingProcess. This pattern is
             // useful to conditionally access the result data.
             if let Some(result) = request.get_state_mut().maybe_start_processing() {

common/eth2/src/types.rs

@@ -1070,6 +1070,12 @@ pub struct BlockGossip {
     pub slot: Slot,
     pub block: Hash256,
 }
+
+#[derive(Debug, PartialEq, Clone, Serialize, Deserialize)]
+pub struct SseExecutionPayloadAvailable {
+    pub slot: Slot,
+    pub block_root: Hash256,
+}
 #[derive(PartialEq, Debug, Serialize, Deserialize, Clone)]
 pub struct SseChainReorg {
     pub slot: Slot,
@@ -1210,6 +1216,8 @@ pub enum EventKind<E: EthSpec> {
     AttesterSlashing(Box<AttesterSlashing<E>>),
     BlsToExecutionChange(Box<SignedBlsToExecutionChange>),
     BlockGossip(Box<BlockGossip>),
+    ExecutionPayloadBid(Box<SignedExecutionPayloadBid<E>>),
+    ExecutionPayloadAvailable(SseExecutionPayloadAvailable),
 }
 
 impl<E: EthSpec> EventKind<E> {
@@ -1233,6 +1241,8 @@ impl<E: EthSpec> EventKind<E> {
             EventKind::AttesterSlashing(_) => "attester_slashing",
             EventKind::BlsToExecutionChange(_) => "bls_to_execution_change",
             EventKind::BlockGossip(_) => "block_gossip",
+            EventKind::ExecutionPayloadBid(_) => "execution_payload_bid",
+            EventKind::ExecutionPayloadAvailable(_) => "execution_payload_available",
         }
     }
 
@@ -1322,6 +1332,19 @@ impl<E: EthSpec> EventKind<E> {
             "block_gossip" => Ok(EventKind::BlockGossip(serde_json::from_str(data).map_err(
                 |e| ServerError::InvalidServerSentEvent(format!("Block Gossip: {:?}", e)),
             )?)),
+            "execution_payload_bid" => Ok(EventKind::ExecutionPayloadBid(
+                serde_json::from_str(data).map_err(|e| {
+                    ServerError::InvalidServerSentEvent(format!("Execution Payload Bid: {:?}", e))
+                })?,
+            )),
+            "execution_payload_available" => Ok(EventKind::ExecutionPayloadAvailable(
+                serde_json::from_str(data).map_err(|e| {
+                    ServerError::InvalidServerSentEvent(format!(
+                        "Execution Payload Available: {:?}",
+                        e
+                    ))
+                })?,
+            )),
             _ => Err(ServerError::InvalidServerSentEvent(
                 "Could not parse event tag".to_string(),
             )),
@@ -1357,6 +1380,8 @@ pub enum EventTopic {
     ProposerSlashing,
     BlsToExecutionChange,
     BlockGossip,
+    ExecutionPayloadBid,
+    ExecutionPayloadAvailable,
 }
 
 impl FromStr for EventTopic {
@@ -1382,6 +1407,8 @@ impl FromStr for EventTopic {
             "proposer_slashing" => Ok(EventTopic::ProposerSlashing),
             "bls_to_execution_change" => Ok(EventTopic::BlsToExecutionChange),
             "block_gossip" => Ok(EventTopic::BlockGossip),
+            "execution_payload_bid" => Ok(EventTopic::ExecutionPayloadBid),
+            "execution_payload_available" => Ok(EventTopic::ExecutionPayloadAvailable),
             _ => Err("event topic cannot be parsed.".to_string()),
         }
     }
@@ -1408,6 +1435,8 @@ impl fmt::Display for EventTopic {
             EventTopic::ProposerSlashing => write!(f, "proposer_slashing"),
             EventTopic::BlsToExecutionChange => write!(f, "bls_to_execution_change"),
             EventTopic::BlockGossip => write!(f, "block_gossip"),
+            EventTopic::ExecutionPayloadBid => write!(f, "execution_payload_bid"),
+            EventTopic::ExecutionPayloadAvailable => write!(f, "execution_payload_available"),
         }
     }
 }

common/task_executor/src/lib.rs

@@ -12,6 +12,26 @@ use crate::rayon_pool_provider::RayonPoolProvider;
 pub use crate::rayon_pool_provider::RayonPoolType;
 pub use tokio::task::JoinHandle;
 
+/// Error type for spawning a blocking task and awaiting its result.
+#[derive(Debug)]
+pub enum SpawnBlockingError {
+    /// The runtime is shutting down.
+    RuntimeShutdown,
+    /// The blocking task failed (e.g. due to a panic).
+    JoinError(tokio::task::JoinError),
+}
+
+impl std::fmt::Display for SpawnBlockingError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            SpawnBlockingError::RuntimeShutdown => write!(f, "runtime shutdown"),
+            SpawnBlockingError::JoinError(e) => write!(f, "join error: {e}"),
+        }
+    }
+}
+
+impl std::error::Error for SpawnBlockingError {}
+
 /// Provides a reason when Lighthouse is shut down.
 #[derive(Copy, Clone, Debug, PartialEq)]
 pub enum ShutdownReason {
@@ -343,6 +363,26 @@ impl TaskExecutor {
         Some(future)
     }
 
+    /// Spawn a blocking task and await its result.
+    ///
+    /// Maps the `Option` (runtime shutdown) and `tokio::JoinError` into a single
+    /// `SpawnBlockingError`.
+    pub async fn spawn_blocking_and_await<F, R>(
+        &self,
+        task: F,
+        name: &'static str,
+    ) -> Result<R, SpawnBlockingError>
+    where
+        F: FnOnce() -> R + Send + 'static,
+        R: Send + 'static,
+    {
+        let handle = self
+            .spawn_blocking_handle(task, name)
+            .ok_or(SpawnBlockingError::RuntimeShutdown)?;
+
+        handle.await.map_err(SpawnBlockingError::JoinError)
+    }
+
     /// Block the current (non-async) thread on the completion of some future.
     ///
     /// ## Warning

consensus/fork_choice/src/fork_choice.rs

@@ -20,6 +20,7 @@ use types::{
     AbstractExecPayload, AttestationShufflingId, AttesterSlashingRef, BeaconBlockRef, BeaconState,
     BeaconStateError, ChainSpec, Checkpoint, Epoch, EthSpec, ExecPayload, ExecutionBlockHash,
     Hash256, IndexedAttestationRef, RelativeEpoch, SignedBeaconBlock, Slot,
+    consts::gloas::{PAYLOAD_STATUS_FULL, PAYLOAD_STATUS_PENDING},
 };
 
 #[derive(Debug)]
@@ -385,6 +386,13 @@ where
         // If the current slot is not provided, use the value that was last provided to the store.
         let current_slot = current_slot.unwrap_or_else(|| fc_store.get_current_slot());
 
+        // TODO(gloas) this probably isnt true, but AI is hallucinating, and we probably dont need
+        // this right now anyways
+        // The anchor is a trusted finalized block. Pre-Gloas blocks always have their payload
+        // inline (FULL). For Gloas anchors, we trust the finalized state was fully processed,
+        // so FULL is appropriate.
+        let payload_status = PAYLOAD_STATUS_FULL;
+
         let proto_array = ProtoArrayForkChoice::new::<E>(
             current_slot,
             finalized_block_slot,
@@ -394,6 +402,7 @@ where
             current_epoch_shuffling_id,
             next_epoch_shuffling_id,
             execution_status,
+            payload_status,
         )?;
 
         let mut fork_choice = Self {
@@ -631,6 +640,14 @@ where
             .map_err(Error::FailedToProcessInvalidExecutionPayload)
     }
 
+    /// Register that a valid execution payload envelope has been received for `block_root`,
+    /// updating the node's `payload_status` from PENDING to FULL.
+    pub fn on_execution_payload(&mut self, block_root: Hash256) -> Result<(), Error<T::Error>> {
+        self.proto_array
+            .on_execution_payload(block_root)
+            .map_err(Error::FailedToProcessValidExecutionPayload)
+    }
+
     /// Add `block` to the fork choice DAG.
     ///
     /// - `block_root` is the root of `block.
@@ -907,6 +924,13 @@ where
                 execution_status,
                 unrealized_justified_checkpoint: Some(unrealized_justified_checkpoint),
                 unrealized_finalized_checkpoint: Some(unrealized_finalized_checkpoint),
+                // Pre-Gloas blocks have their payload inline, so they are always FULL.
+                // Gloas blocks start as PENDING until the payload envelope arrives.
+                payload_status: if block.fork_name_unchecked().gloas_enabled() {
+                    PAYLOAD_STATUS_PENDING
+                } else {
+                    PAYLOAD_STATUS_FULL
+                },
             },
             current_slot,
             self.justified_checkpoint(),

consensus/proto_array/src/fork_choice_test_definition.rs

@@ -89,6 +89,7 @@ impl ForkChoiceTestDefinition {
             junk_shuffling_id.clone(),
             junk_shuffling_id,
             ExecutionStatus::Optimistic(ExecutionBlockHash::zero()),
+            types::consts::gloas::PAYLOAD_STATUS_FULL,
         )
         .expect("should create fork choice struct");
         let equivocating_indices = BTreeSet::new();
@@ -211,6 +212,8 @@ impl ForkChoiceTestDefinition {
                         ),
                         unrealized_justified_checkpoint: None,
                         unrealized_finalized_checkpoint: None,
+                        // Test blocks default to FULL payload status.
+                        payload_status: types::consts::gloas::PAYLOAD_STATUS_FULL,
                     };
                     fork_choice
                         .process_block::<MainnetEthSpec>(

consensus/proto_array/src/proto_array.rs

@@ -10,6 +10,7 @@ use superstruct::superstruct;
 use types::{
     AttestationShufflingId, ChainSpec, Checkpoint, Epoch, EthSpec, ExecutionBlockHash, Hash256,
     Slot,
+    consts::gloas::{PAYLOAD_STATUS_FULL, PAYLOAD_STATUS_PENDING, PayloadStatus},
 };
 
 // Define a "legacy" implementation of `Option<usize>` which uses four bytes for encoding the union
@@ -109,6 +110,13 @@ pub struct ProtoNode {
     pub unrealized_justified_checkpoint: Option<Checkpoint>,
     #[ssz(with = "four_byte_option_checkpoint")]
     pub unrealized_finalized_checkpoint: Option<Checkpoint>,
+    /// The payload status of this block (Gloas).
+    ///
+    /// - `PAYLOAD_STATUS_PENDING` (0): The payload has not yet been received.
+    /// - `PAYLOAD_STATUS_EMPTY` (1): The slot has passed without a payload (pre-Gloas blocks
+    ///   or empty Gloas slots).
+    /// - `PAYLOAD_STATUS_FULL` (2): A valid execution payload has been received.
+    pub payload_status: PayloadStatus,
 }
 
 #[derive(PartialEq, Debug, Encode, Decode, Serialize, Deserialize, Copy, Clone)]
@@ -332,6 +340,7 @@ impl ProtoArray {
             execution_status: block.execution_status,
             unrealized_justified_checkpoint: block.unrealized_justified_checkpoint,
             unrealized_finalized_checkpoint: block.unrealized_finalized_checkpoint,
+            payload_status: block.payload_status,
         };
 
         // If the parent has an invalid execution status, return an error before adding the block to
@@ -369,6 +378,29 @@ impl ProtoArray {
         Ok(())
     }
 
+    /// Register that an execution payload has been received and validated for `block_root`.
+    ///
+    /// Updates the node's `payload_status` from `PENDING` to `FULL`.
+    ///
+    /// Returns an error if the block is unknown to fork choice.
+    pub fn on_execution_payload(&mut self, block_root: Hash256) -> Result<(), Error> {
+        let index = self
+            .indices
+            .get(&block_root)
+            .copied()
+            .ok_or(Error::NodeUnknown(block_root))?;
+        let node = self
+            .nodes
+            .get_mut(index)
+            .ok_or(Error::InvalidNodeIndex(index))?;
+
+        if node.payload_status == PAYLOAD_STATUS_PENDING {
+            node.payload_status = PAYLOAD_STATUS_FULL;
+        }
+
+        Ok(())
+    }
+
     /// Updates the `block_root` and all ancestors to have validated execution payloads.
     ///
     /// Returns an error if:

consensus/proto_array/src/proto_array_fork_choice.rs

@@ -17,7 +17,7 @@ use std::{
 };
 use types::{
     AttestationShufflingId, ChainSpec, Checkpoint, Epoch, EthSpec, ExecutionBlockHash, Hash256,
-    Slot,
+    Slot, consts::gloas::PayloadStatus,
 };
 
 pub const DEFAULT_PRUNE_THRESHOLD: usize = 256;
@@ -159,6 +159,8 @@ pub struct Block {
     pub execution_status: ExecutionStatus,
     pub unrealized_justified_checkpoint: Option<Checkpoint>,
     pub unrealized_finalized_checkpoint: Option<Checkpoint>,
+    /// The payload status for this block (Gloas fork choice).
+    pub payload_status: PayloadStatus,
 }
 
 impl Block {
@@ -422,6 +424,7 @@ impl ProtoArrayForkChoice {
         current_epoch_shuffling_id: AttestationShufflingId,
         next_epoch_shuffling_id: AttestationShufflingId,
         execution_status: ExecutionStatus,
+        payload_status: PayloadStatus,
     ) -> Result<Self, String> {
         let mut proto_array = ProtoArray {
             prune_threshold: DEFAULT_PRUNE_THRESHOLD,
@@ -445,6 +448,7 @@ impl ProtoArrayForkChoice {
             execution_status,
             unrealized_justified_checkpoint: Some(justified_checkpoint),
             unrealized_finalized_checkpoint: Some(finalized_checkpoint),
+            payload_status,
         };
 
         proto_array
@@ -484,6 +488,15 @@ impl ProtoArrayForkChoice {
             .map_err(|e| format!("Failed to process invalid payload: {:?}", e))
     }
 
+    /// Register that a valid execution payload envelope has been received for `block_root`.
+    ///
+    /// See `ProtoArray::on_execution_payload` for documentation.
+    pub fn on_execution_payload(&mut self, block_root: Hash256) -> Result<(), String> {
+        self.proto_array
+            .on_execution_payload(block_root)
+            .map_err(|e| format!("on_execution_payload error: {:?}", e))
+    }
+
     pub fn process_attestation(
         &mut self,
         validator_index: usize,
@@ -873,6 +886,7 @@ impl ProtoArrayForkChoice {
             execution_status: block.execution_status,
             unrealized_justified_checkpoint: block.unrealized_justified_checkpoint,
             unrealized_finalized_checkpoint: block.unrealized_finalized_checkpoint,
+            payload_status: block.payload_status,
         })
     }
 
@@ -882,6 +896,16 @@ impl ProtoArrayForkChoice {
         Some(block.execution_status)
     }
 
+    /// Returns the first *beacon block root* which contains an execution payload with the given
+    /// `block_hash`, if any.
+    pub fn execution_block_hash_to_beacon_block_root(
+        &self,
+        block_hash: &ExecutionBlockHash,
+    ) -> Option<Hash256> {
+        self.proto_array
+            .execution_block_hash_to_beacon_block_root(block_hash)
+    }
+
     /// Returns the weight of a given block.
     pub fn get_weight(&self, block_root: &Hash256) -> Option<u64> {
         let block_index = self.proto_array.indices.get(block_root)?;
@@ -1136,6 +1160,7 @@ mod test_compute_deltas {
             junk_shuffling_id.clone(),
             junk_shuffling_id.clone(),
             execution_status,
+            types::consts::gloas::PAYLOAD_STATUS_FULL,
         )
         .unwrap();
 
@@ -1155,6 +1180,7 @@ mod test_compute_deltas {
                     execution_status,
                     unrealized_justified_checkpoint: Some(genesis_checkpoint),
                     unrealized_finalized_checkpoint: Some(genesis_checkpoint),
+                    payload_status: types::consts::gloas::PAYLOAD_STATUS_FULL,
                 },
                 genesis_slot + 1,
                 genesis_checkpoint,
@@ -1180,6 +1206,7 @@ mod test_compute_deltas {
                     execution_status,
                     unrealized_justified_checkpoint: None,
                     unrealized_finalized_checkpoint: None,
+                    payload_status: types::consts::gloas::PAYLOAD_STATUS_FULL,
                 },
                 genesis_slot + 1,
                 genesis_checkpoint,
@@ -1280,6 +1307,7 @@ mod test_compute_deltas {
             junk_shuffling_id.clone(),
             junk_shuffling_id.clone(),
             execution_status,
+            types::consts::gloas::PAYLOAD_STATUS_FULL,
         )
         .unwrap();
 
@@ -1308,6 +1336,7 @@ mod test_compute_deltas {
                         execution_status,
                         unrealized_justified_checkpoint: Some(genesis_checkpoint),
                         unrealized_finalized_checkpoint: Some(genesis_checkpoint),
+                        payload_status: types::consts::gloas::PAYLOAD_STATUS_FULL,
                     },
                     Slot::from(block.slot),
                     genesis_checkpoint,

consensus/types/src/core/consts.rs

@@ -28,6 +28,8 @@ pub mod deneb {
 pub mod gloas {
     pub const BUILDER_INDEX_SELF_BUILD: u64 = u64::MAX;
     pub const BUILDER_INDEX_FLAG: u64 = 1 << 40;
+    pub const BID_VALUE_SELF_BUILD: u64 = 0;
+    pub const EXECUTION_PAYMENT_TRUSTLESS_BUILD: u64 = 0;
 
     // Fork choice constants
     pub type PayloadStatus = u8;

consensus/types/src/execution/execution_payload_envelope.rs

@@ -10,7 +10,9 @@ use ssz_derive::{Decode, Encode};
 use test_random_derive::TestRandom;
 use tree_hash_derive::TreeHash;
 
-#[derive(Debug, Clone, Serialize, Encode, Decode, Deserialize, TestRandom, TreeHash, Educe)]
+#[derive(
+    Debug, Default, Clone, Serialize, Encode, Decode, Deserialize, TestRandom, TreeHash, Educe,
+)]
 #[educe(PartialEq, Hash(bound(E: EthSpec)))]
 #[context_deserialize(ForkName)]
 #[serde(bound = "E: EthSpec")]

validator_client/validator_services/src/block_service.rs

@@ -145,6 +145,7 @@ impl<S: ValidatorStore, T: SlotClock + 'static> BlockServiceBuilder<S, T> {
 
 // Combines a set of non-block-proposing `beacon_nodes` and only-block-proposing
 // `proposer_nodes`.
+#[derive(Clone)]
 pub struct ProposerFallback<T> {
     beacon_nodes: Arc<BeaconNodeFallback<T>>,
     proposer_nodes: Option<Arc<BeaconNodeFallback<T>>>,