diff --git a/beacon_node/beacon_chain/src/beacon_chain.rs b/beacon_node/beacon_chain/src/beacon_chain.rs
index f05b972679..e226c707a4 100644
--- a/beacon_node/beacon_chain/src/beacon_chain.rs
+++ b/beacon_node/beacon_chain/src/beacon_chain.rs
@@ -130,7 +130,7 @@ use store::{
 };
 use task_executor::{RayonPoolType, ShutdownReason, TaskExecutor};
 use tokio_stream::Stream;
-use tracing::{Span, debug, debug_span, error, info, info_span, instrument, trace, warn};
+use tracing::{debug, debug_span, error, info, info_span, instrument, trace, warn};
 use tree_hash::TreeHash;
 use types::data::{ColumnIndex, FixedBlobSidecarList};
 use types::execution::BlockProductionVersion;
@@ -2762,6 +2762,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     /// or already-known).
     ///
     /// This method is potentially long-running and should not run on the core executor.
+    #[instrument(skip_all, level = "debug")]
     pub fn filter_chain_segment(
         self: &Arc<Self>,
         chain_segment: Vec<RangeSyncBlock<T::EthSpec>>,
@@ -2889,12 +2890,8 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
         // Filter uninteresting blocks from the chain segment in a blocking task.
         let chain = self.clone();
-        let filter_chain_segment = debug_span!("filter_chain_segment");
         let filtered_chain_segment_future = self.spawn_blocking_handle(
-            move || {
-                let _guard = filter_chain_segment.enter();
-                chain.filter_chain_segment(chain_segment)
-            },
+            move || chain.filter_chain_segment(chain_segment),
             "filter_chain_segment",
         );
         let mut filtered_chain_segment = match filtered_chain_segment_future.await {
@@ -2925,12 +2922,8 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             std::mem::swap(&mut blocks, &mut filtered_chain_segment);
 
             let chain = self.clone();
-            let current_span = Span::current();
             let signature_verification_future = self.spawn_blocking_handle(
-                move || {
-                    let _guard = current_span.enter();
-                    signature_verify_chain_segment(blocks, &chain)
-                },
+                move || signature_verify_chain_segment(blocks, &chain),
                 "signature_verify_chain_segment",
             );
 
@@ -3020,12 +3013,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         block: Arc<SignedBeaconBlock<T::EthSpec>>,
     ) -> Result<GossipVerifiedBlock<T>, BlockError> {
         let chain = self.clone();
-        let span = Span::current();
         self.task_executor
             .clone()
             .spawn_blocking_handle(
                 move || {
-                    let _guard = span.enter();
                     let slot = block.slot();
                     let graffiti_string = block.message().body().graffiti().as_utf8_lossy();
 
@@ -3333,11 +3324,9 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
         let data_availability_checker = self.data_availability_checker.clone();
 
-        let current_span = Span::current();
         let result = self
             .task_executor
             .spawn_blocking_with_rayon_async(RayonPoolType::HighPriority, move || {
-                let _guard = current_span.enter();
                 data_availability_checker.reconstruct_data_columns(&block_root)
             })
             .await
@@ -3802,7 +3791,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             consensus_context,
         } = import_data;
 
-        // Record the time at which this block's blobs became available.
+        // Record the time at which this block's blobs/data columns became available.
         if let Some(blobs_available) = block.blobs_available_timestamp() {
             self.block_times_cache.write().set_time_blob_observed(
                 block_root,
@@ -3811,16 +3800,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             );
         }
 
-        // TODO(das) record custody column available timestamp
-
         let block_root = {
-            // Capture the current span before moving into the blocking task
-            let current_span = tracing::Span::current();
             let chain = self.clone();
             self.spawn_blocking_handle(
                 move || {
-                    // Enter the captured span in the blocking thread
-                    let _guard = current_span.enter();
                     chain.import_block(
                         block,
                         block_root,
@@ -4531,15 +4514,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         //
         // Load the parent state from disk.
         let chain = self.clone();
-        let span = Span::current();
         let (state, state_root_opt) = self
             .task_executor
             .spawn_blocking_handle(
-                move || {
-                    let _guard =
-                        debug_span!(parent: span, "load_state_for_block_production").entered();
-                    chain.load_state_for_block_production(slot)
-                },
+                move || chain.load_state_for_block_production(slot),
                 "load_state_for_block_production",
             )
             .ok_or(BlockProductionError::ShuttingDown)?
@@ -4973,13 +4951,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             .graffiti_calculator
             .get_graffiti(graffiti_settings)
             .await;
-        let span = Span::current();
         let mut partial_beacon_block = self
             .task_executor
             .spawn_blocking_handle(
                 move || {
-                    let _guard =
-                        debug_span!(parent: span, "produce_partial_beacon_block").entered();
                     chain.produce_partial_beacon_block(
                         state,
                         state_root_opt,
@@ -5015,14 +4990,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             match block_contents_type {
                 BlockProposalContentsType::Full(block_contents) => {
                     let chain = self.clone();
-                    let span = Span::current();
                     let beacon_block_response = self
                         .task_executor
                         .spawn_blocking_handle(
                             move || {
-                                let _guard =
-                                    debug_span!(parent: span, "complete_partial_beacon_block")
-                                        .entered();
                                 chain.complete_partial_beacon_block(
                                     partial_beacon_block,
                                     Some(block_contents),
@@ -5039,14 +5010,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 }
                 BlockProposalContentsType::Blinded(block_contents) => {
                     let chain = self.clone();
-                    let span = Span::current();
                     let beacon_block_response = self
                         .task_executor
                         .spawn_blocking_handle(
                             move || {
-                                let _guard =
-                                    debug_span!(parent: span, "complete_partial_beacon_block")
-                                        .entered();
                                 chain.complete_partial_beacon_block(
                                     partial_beacon_block,
                                     Some(block_contents),
@@ -5064,13 +5031,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             }
         } else {
             let chain = self.clone();
-            let span = Span::current();
             let beacon_block_response = self
                 .task_executor
                 .spawn_blocking_handle(
                     move || {
-                        let _guard =
-                            debug_span!(parent: span, "complete_partial_beacon_block").entered();
                         chain.complete_partial_beacon_block(
                             partial_beacon_block,
                             None,
@@ -5088,6 +5052,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     }
 
     #[allow(clippy::too_many_arguments)]
+    #[instrument(skip_all, level = "debug")]
     fn produce_partial_beacon_block(
         self: &Arc<Self>,
         mut state: BeaconState<T::EthSpec>,
@@ -5332,6 +5297,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         })
     }
 
+    #[instrument(skip_all, level = "debug")]
     fn complete_partial_beacon_block<Payload: AbstractExecPayload<T::EthSpec>>(
         &self,
         partial_beacon_block: PartialBeaconBlock<T::EthSpec>,
diff --git a/beacon_node/beacon_chain/src/block_production/gloas.rs b/beacon_node/beacon_chain/src/block_production/gloas.rs
index 2fc4fb51f7..51caf63b7a 100644
--- a/beacon_node/beacon_chain/src/block_production/gloas.rs
+++ b/beacon_node/beacon_chain/src/block_production/gloas.rs
@@ -19,7 +19,7 @@ use state_processing::{
 };
 use state_processing::{VerifyOperation, state_advance::complete_state_advance};
 use task_executor::JoinHandle;
-use tracing::{Instrument, Span, debug, debug_span, error, instrument, trace, warn};
+use tracing::{Instrument, debug, debug_span, error, instrument, trace, warn};
 use tree_hash::TreeHash;
 use types::consts::gloas::BUILDER_INDEX_SELF_BUILD;
 use types::{
@@ -87,15 +87,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         //
         // Load the parent state from disk.
         let chain = self.clone();
-        let span = Span::current();
         let (state, state_root_opt) = self
             .task_executor
             .spawn_blocking_handle(
-                move || {
-                    let _guard =
-                        debug_span!(parent: span, "load_state_for_block_production").entered();
-                    chain.load_state_for_block_production(slot)
-                },
+                move || chain.load_state_for_block_production(slot),
                 "load_state_for_block_production",
             )
             .ok_or(BlockProductionError::ShuttingDown)?
@@ -135,13 +130,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             .graffiti_calculator
             .get_graffiti(graffiti_settings)
             .await;
-        let span = Span::current();
         let (partial_beacon_block, state) = self
             .task_executor
             .spawn_blocking_handle(
                 move || {
-                    let _guard =
-                        debug_span!(parent: span, "produce_partial_beacon_block_gloas").entered();
                     chain.produce_partial_beacon_block_gloas(
                         state,
                         state_root_opt,
@@ -175,12 +167,9 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         //
         // Complete the block with the execution payload bid.
         let chain = self.clone();
-        let span = Span::current();
         self.task_executor
             .spawn_blocking_handle(
                 move || {
-                    let _guard =
-                        debug_span!(parent: span, "complete_partial_beacon_block_gloas").entered();
                     chain.complete_partial_beacon_block_gloas(
                         partial_beacon_block,
                         execution_payload_bid,
@@ -198,6 +187,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
     #[allow(clippy::too_many_arguments)]
     #[allow(clippy::type_complexity)]
+    #[instrument(skip_all, level = "debug")]
     fn produce_partial_beacon_block_gloas(
         self: &Arc<Self>,
         mut state: BeaconState<T::EthSpec>,
@@ -432,6 +422,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     /// - `pending_state` is the state post block application (prior to payload application)
     /// - `block_value` is the consensus-layer rewards for `block`
     #[allow(clippy::type_complexity)]
+    #[instrument(skip_all, level = "debug")]
     fn complete_partial_beacon_block_gloas(
         &self,
         partial_beacon_block: PartialBeaconBlock<T::EthSpec>,
diff --git a/beacon_node/beacon_chain/src/block_production/mod.rs b/beacon_node/beacon_chain/src/block_production/mod.rs
index 60cc3e919a..bf42923cbe 100644
--- a/beacon_node/beacon_chain/src/block_production/mod.rs
+++ b/beacon_node/beacon_chain/src/block_production/mod.rs
@@ -15,6 +15,7 @@ mod gloas;
 impl<T: BeaconChainTypes> BeaconChain<T> {
     /// Load a beacon state from the database for block production. This is a long-running process
     /// that should not be performed in an `async` context.
+    #[instrument(skip_all, level = "debug")]
     pub(crate) fn load_state_for_block_production(
         self: &Arc<Self>,
         slot: Slot,
diff --git a/beacon_node/beacon_chain/src/block_verification.rs b/beacon_node/beacon_chain/src/block_verification.rs
index ae9acdefd5..1fe013f754 100644
--- a/beacon_node/beacon_chain/src/block_verification.rs
+++ b/beacon_node/beacon_chain/src/block_verification.rs
@@ -1705,7 +1705,7 @@ impl<T: BeaconChainTypes> ExecutionPendingBlock<T> {
                     indexed_payload_attestation,
                     AttestationFromBlock::True,
                     &ptc.0,
-                ) && !matches!(e, ForkChoiceError::InvalidAttestation(_))
+                ) && !matches!(e, ForkChoiceError::InvalidPayloadAttestation(_))
                 {
                     return Err(BlockError::BeaconChainError(Box::new(e.into())));
                 }
diff --git a/beacon_node/beacon_chain/src/canonical_head.rs b/beacon_node/beacon_chain/src/canonical_head.rs
index bfd3d79512..cd53d0ef7c 100644
--- a/beacon_node/beacon_chain/src/canonical_head.rs
+++ b/beacon_node/beacon_chain/src/canonical_head.rs
@@ -58,7 +58,6 @@ use store::{
     Error as StoreError, KeyValueStore, KeyValueStoreOp, StoreConfig, iter::StateRootsIterator,
 };
 use task_executor::{JoinHandle, ShutdownReason};
-use tracing::info_span;
 use tracing::{debug, error, info, instrument, warn};
 use types::*;
 
@@ -539,22 +538,15 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     /// such a case it's critical that the `BeaconChain` keeps importing blocks so that the
     /// situation can be rectified. We avoid returning an error here so that calling functions
     /// can't abort block import because an error is returned here.
+    #[instrument(name = "lh_recompute_head_at_slot", skip(self), level = "info", fields(slot = %current_slot))]
     pub async fn recompute_head_at_slot(self: &Arc<Self>, current_slot: Slot) {
-        let span = info_span!(
-            "lh_recompute_head_at_slot",
-            slot = %current_slot
-        );
-
         metrics::inc_counter(&metrics::FORK_CHOICE_REQUESTS);
         let _timer = metrics::start_timer(&metrics::FORK_CHOICE_TIMES);
 
         let chain = self.clone();
         match self
             .spawn_blocking_handle(
-                move || {
-                    let _guard = span.enter();
-                    chain.recompute_head_at_slot_internal(current_slot)
-                },
+                move || chain.recompute_head_at_slot_internal(current_slot),
                 "recompute_head_internal",
             )
             .await
@@ -1392,8 +1384,8 @@ fn observe_head_block_delays<E: EthSpec, S: SlotClock>(
                 .as_millis() as i64,
         );
 
-        // The time from the start of the slot when all blobs have been observed. Technically this
-        // is the time we last saw a blob related to this block/slot.
+        // The time from the start of the slot when all blobs/data columns have been observed. Technically this
+        // is the time we last saw a blob/data column related to this block/slot.
         metrics::set_gauge(
             &metrics::BEACON_BLOB_DELAY_ALL_OBSERVED_SLOT_START,
             block_delays
diff --git a/beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs b/beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs
index c0403595ee..8f1d4464e1 100644
--- a/beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs
+++ b/beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs
@@ -282,8 +282,11 @@ impl<E: EthSpec> PendingComponents<E> {
                 .flatten()
                 .map(|blob| blob.seen_timestamp())
                 .max(),
-            // TODO(das): To be fixed with https://github.com/sigp/lighthouse/pull/6850
-            AvailableBlockData::DataColumns(_) => None,
+            AvailableBlockData::DataColumns(_) => self
+                .verified_data_columns
+                .iter()
+                .map(|data_column| data_column.seen_timestamp())
+                .max(),
         };
 
         let AvailabilityPendingExecutedBlock {
diff --git a/beacon_node/beacon_chain/src/data_column_verification.rs b/beacon_node/beacon_chain/src/data_column_verification.rs
index dde9fad342..f47de01ddc 100644
--- a/beacon_node/beacon_chain/src/data_column_verification.rs
+++ b/beacon_node/beacon_chain/src/data_column_verification.rs
@@ -5,6 +5,7 @@ use crate::kzg_utils::{reconstruct_data_columns, validate_data_columns};
 use crate::observed_data_sidecars::{
     Error as ObservedDataSidecarsError, ObservationKey, ObservationStrategy, Observe,
 };
+use crate::validator_monitor::timestamp_now;
 use crate::{BeaconChain, BeaconChainError, BeaconChainTypes, metrics};
 use educe::Educe;
 use fork_choice::ProtoBlock;
@@ -16,6 +17,7 @@ use ssz_types::VariableList;
 use std::iter;
 use std::marker::PhantomData;
 use std::sync::Arc;
+use std::time::Duration;
 use tracing::{debug, instrument};
 use types::data::ColumnIndex;
 use types::{
@@ -320,25 +322,34 @@ impl<T: BeaconChainTypes, O: ObservationStrategy> GossipVerifiedDataColumn<T, O>
 #[ssz(struct_behaviour = "transparent")]
 pub struct KzgVerifiedDataColumn<E: EthSpec> {
     data: Arc<DataColumnSidecar<E>>,
+    #[ssz(skip_serializing, skip_deserializing)]
+    seen_timestamp: Duration,
 }
 
 impl<E: EthSpec> KzgVerifiedDataColumn<E> {
     pub fn new(
         data_column: Arc<DataColumnSidecar<E>>,
         kzg: &Kzg,
+        seen_timestamp: Duration,
     ) -> Result<Self, (Option<ColumnIndex>, KzgError)> {
-        verify_kzg_for_data_column(data_column, kzg)
+        verify_kzg_for_data_column(data_column, kzg, seen_timestamp)
     }
 
     /// Mark a data column as KZG verified. Caller must ONLY use this on columns constructed
     /// from EL blobs.
     pub fn from_execution_verified(data_column: Arc<DataColumnSidecar<E>>) -> Self {
-        Self { data: data_column }
+        Self {
+            data: data_column,
+            seen_timestamp: timestamp_now(),
+        }
     }
 
     /// Create a `KzgVerifiedDataColumn` from `DataColumnSidecar` for testing ONLY.
     pub(crate) fn __new_for_testing(data_column: Arc<DataColumnSidecar<E>>) -> Self {
-        Self { data: data_column }
+        Self {
+            data: data_column,
+            seen_timestamp: timestamp_now(),
+        }
     }
 
     pub fn from_batch_with_scoring(
@@ -348,7 +359,10 @@ impl<E: EthSpec> KzgVerifiedDataColumn<E> {
         verify_kzg_for_data_column_list(data_columns.iter(), kzg)?;
         Ok(data_columns
             .into_iter()
-            .map(|column| Self { data: column })
+            .map(|column| Self {
+                data: column,
+                seen_timestamp: timestamp_now(),
+            })
             .collect())
     }
 
@@ -407,6 +421,8 @@ impl<E: EthSpec> CustodyDataColumn<E> {
 #[ssz(struct_behaviour = "transparent")]
 pub struct KzgVerifiedCustodyDataColumn<E: EthSpec> {
     data: Arc<DataColumnSidecar<E>>,
+    #[ssz(skip_serializing, skip_deserializing)]
+    seen_timestamp: Duration,
 }
 
 impl<E: EthSpec> KzgVerifiedCustodyDataColumn<E> {
@@ -414,6 +430,7 @@ impl<E: EthSpec> KzgVerifiedCustodyDataColumn<E> {
     /// include this column
     pub fn from_asserted_custody(kzg_verified: KzgVerifiedDataColumn<E>) -> Self {
         Self {
+            seen_timestamp: kzg_verified.seen_timestamp,
             data: kzg_verified.to_data_column(),
         }
     }
@@ -422,10 +439,12 @@ impl<E: EthSpec> KzgVerifiedCustodyDataColumn<E> {
     pub fn new(
         data_column: CustodyDataColumn<E>,
         kzg: &Kzg,
+        seen_timestamp: Duration,
     ) -> Result<Self, (Option<ColumnIndex>, KzgError)> {
-        verify_kzg_for_data_column(data_column.clone_arc(), kzg)?;
+        verify_kzg_for_data_column(data_column.clone_arc(), kzg, seen_timestamp)?;
         Ok(Self {
             data: data_column.data,
+            seen_timestamp,
         })
     }
 
@@ -443,10 +462,15 @@ impl<E: EthSpec> KzgVerifiedCustodyDataColumn<E> {
             spec,
         )?;
 
+        let seen_timestamp = timestamp_now();
+
         Ok(all_data_columns
             .into_iter()
             .map(|data| {
-                KzgVerifiedCustodyDataColumn::from_asserted_custody(KzgVerifiedDataColumn { data })
+                KzgVerifiedCustodyDataColumn::from_asserted_custody(KzgVerifiedDataColumn {
+                    data,
+                    seen_timestamp,
+                })
             })
             .collect::<Vec<_>>())
     }
@@ -464,6 +488,10 @@ impl<E: EthSpec> KzgVerifiedCustodyDataColumn<E> {
     pub fn index(&self) -> ColumnIndex {
         *self.data.index()
     }
+
+    pub fn seen_timestamp(&self) -> Duration {
+        self.seen_timestamp
+    }
 }
 
 /// Complete kzg verification for a `DataColumnSidecar`.
@@ -473,10 +501,14 @@ impl<E: EthSpec> KzgVerifiedCustodyDataColumn<E> {
 pub fn verify_kzg_for_data_column<E: EthSpec>(
     data_column: Arc<DataColumnSidecar<E>>,
     kzg: &Kzg,
+    seen_timestamp: Duration,
 ) -> Result<KzgVerifiedDataColumn<E>, (Option<ColumnIndex>, KzgError)> {
     let _timer = metrics::start_timer(&metrics::KZG_VERIFICATION_DATA_COLUMN_SINGLE_TIMES);
     validate_data_columns(kzg, iter::once(&data_column))?;
-    Ok(KzgVerifiedDataColumn { data: data_column })
+    Ok(KzgVerifiedDataColumn {
+        data: data_column,
+        seen_timestamp,
+    })
 }
 
 /// Complete kzg verification for a list of `DataColumnSidecar`s.
@@ -538,8 +570,9 @@ pub fn validate_data_column_sidecar_for_gossip_fulu<T: BeaconChainTypes, O: Obse
     verify_slot_higher_than_parent(&parent_block, column_slot)?;
     verify_proposer_and_signature(data_column_fulu, &parent_block, chain)?;
     let kzg = &chain.kzg;
-    let kzg_verified_data_column = verify_kzg_for_data_column(data_column.clone(), kzg)
-        .map_err(|(_, e)| GossipDataColumnError::InvalidKzgProof(e))?;
+    let kzg_verified_data_column =
+        verify_kzg_for_data_column(data_column.clone(), kzg, timestamp_now())
+            .map_err(|(_, e)| GossipDataColumnError::InvalidKzgProof(e))?;
 
     chain
         .observed_slashable
diff --git a/beacon_node/beacon_chain/src/fetch_blobs/mod.rs b/beacon_node/beacon_chain/src/fetch_blobs/mod.rs
index bae61767cc..db76ff887d 100644
--- a/beacon_node/beacon_chain/src/fetch_blobs/mod.rs
+++ b/beacon_node/beacon_chain/src/fetch_blobs/mod.rs
@@ -32,7 +32,7 @@ use mockall_double::double;
 use ssz_types::FixedVector;
 use state_processing::per_block_processing::deneb::kzg_commitment_to_versioned_hash;
 use std::sync::Arc;
-use tracing::{Span, debug, instrument, warn};
+use tracing::{debug, instrument, warn};
 use types::data::{BlobSidecarError, DataColumnSidecarError};
 use types::{
     BeaconStateError, Blob, BlobSidecar, ColumnIndex, EthSpec, FullPayload, Hash256, KzgProofs,
@@ -356,12 +356,10 @@ async fn compute_custody_columns_to_import<T: BeaconChainTypes>(
     let spec = chain_adapter.spec().clone();
     let chain_adapter_cloned = chain_adapter.clone();
     let custody_columns_indices = custody_columns_indices.to_vec();
-    let current_span = Span::current();
     chain_adapter
         .executor()
         .spawn_blocking_handle(
             move || {
-                let _guard = current_span.enter();
                 let mut timer = metrics::start_timer_vec(
                     &metrics::DATA_COLUMN_SIDECAR_COMPUTATION,
                     &[&blobs.len().to_string()],
diff --git a/beacon_node/beacon_chain/src/payload_envelope_verification/gossip_verified_envelope.rs b/beacon_node/beacon_chain/src/payload_envelope_verification/gossip_verified_envelope.rs
index 9a4ed2d044..4d40a29332 100644
--- a/beacon_node/beacon_chain/src/payload_envelope_verification/gossip_verified_envelope.rs
+++ b/beacon_node/beacon_chain/src/payload_envelope_verification/gossip_verified_envelope.rs
@@ -4,7 +4,7 @@ use educe::Educe;
 use eth2::types::{EventKind, SseExecutionPayloadGossip};
 use parking_lot::{Mutex, RwLock};
 use store::DatabaseBlock;
-use tracing::{Span, debug};
+use tracing::debug;
 use types::{
     ChainSpec, EthSpec, ExecutionPayloadBid, ExecutionPayloadEnvelope, Hash256, SignedBeaconBlock,
     SignedExecutionPayloadEnvelope, Slot, consts::gloas::BUILDER_INDEX_SELF_BUILD,
@@ -270,12 +270,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         envelope: Arc<SignedExecutionPayloadEnvelope<T::EthSpec>>,
     ) -> Result<GossipVerifiedEnvelope<T>, EnvelopeError> {
         let chain = self.clone();
-        let span = Span::current();
         self.task_executor
             .clone()
             .spawn_blocking_handle(
                 move || {
-                    let _guard = span.enter();
                     let slot = envelope.slot();
                     let beacon_block_root = envelope.message.beacon_block_root;
 
diff --git a/beacon_node/beacon_chain/src/payload_envelope_verification/import.rs b/beacon_node/beacon_chain/src/payload_envelope_verification/import.rs
index 85332a12ea..7e79799310 100644
--- a/beacon_node/beacon_chain/src/payload_envelope_verification/import.rs
+++ b/beacon_node/beacon_chain/src/payload_envelope_verification/import.rs
@@ -1,12 +1,12 @@
 use std::sync::Arc;
 use std::time::Duration;
 
-use eth2::types::{EventKind, SseExecutionPayloadAvailable};
+use eth2::types::{EventKind, SseExecutionPayload};
 use fork_choice::PayloadVerificationStatus;
 use slot_clock::SlotClock;
 use store::StoreOp;
 use tracing::{debug, error, info, info_span, instrument, warn};
-use types::{BeaconState, BlockImportSource, Hash256, Slot};
+use types::{BeaconState, BlockImportSource, Hash256, SignedExecutionPayloadEnvelope};
 
 use super::{
     AvailableEnvelope, AvailableExecutedEnvelope, EnvelopeError, EnvelopeImportData,
@@ -168,6 +168,16 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             .map_err(BeaconChainError::TokioJoin)?
             .ok_or(BeaconChainError::RuntimeShutdown)??;
 
+        // TODO(gloas): optimistic sync is not supported for Gloas, maybe we could re-add it
+        if payload_verification_outcome
+            .payload_verification_status
+            .is_optimistic()
+        {
+            return Err(EnvelopeError::OptimisticSyncNotSupported {
+                block_root: import_data.block_root,
+            });
+        }
+
         Ok(ExecutedEnvelope::new(
             signed_envelope,
             import_data,
@@ -192,13 +202,9 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         } = import_data;
 
         let block_root = {
-            // Capture the current span before moving into the blocking task
-            let current_span = tracing::Span::current();
             let chain = self.clone();
             self.spawn_blocking_handle(
                 move || {
-                    // Enter the captured span in the blocking thread
-                    let _guard = current_span.enter();
                     chain.import_execution_payload_envelope(
                         envelope,
                         block_root,
@@ -226,7 +232,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         signed_envelope: AvailableEnvelope<T::EthSpec>,
         block_root: Hash256,
         state: BeaconState<T::EthSpec>,
-        _payload_verification_status: PayloadVerificationStatus,
+        payload_verification_status: PayloadVerificationStatus,
     ) -> Result<Hash256, EnvelopeError> {
         // Everything in this initial section is on the hot path for processing the envelope.
         // Take an upgradable read lock on fork choice so we can check if this block has already
@@ -244,9 +250,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         // avoiding taking other locks whilst holding this lock.
         let mut fork_choice = parking_lot::RwLockUpgradableReadGuard::upgrade(fork_choice_reader);
 
-        // Update the node's payload_status from PENDING to FULL in fork choice.
+        // Update the block's payload to received in fork choice, which creates the `Full` virtual
+        // node which can be eligible for head.
         fork_choice
-            .on_execution_payload(block_root)
+            .on_valid_payload_envelope_received(block_root)
             .map_err(|e| EnvelopeError::InternalError(format!("{e:?}")))?;
 
         // TODO(gloas) emit SSE event if the payload became the new head payload
@@ -315,8 +322,9 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         metrics::stop_timer(db_write_timer);
 
         self.import_envelope_update_metrics_and_events(
+            signed_envelope,
             block_root,
-            signed_envelope.slot(),
+            payload_verification_status,
             envelope_time_imported,
         );
 
@@ -325,10 +333,12 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
     fn import_envelope_update_metrics_and_events(
         &self,
+        signed_envelope: Arc<SignedExecutionPayloadEnvelope<T::EthSpec>>,
         block_root: Hash256,
-        envelope_slot: Slot,
+        payload_verification_status: PayloadVerificationStatus,
         envelope_time_imported: Duration,
     ) {
+        let envelope_slot = signed_envelope.slot();
         let envelope_delay_total =
             get_slot_delay_ms(envelope_time_imported, envelope_slot, &self.slot_clock);
 
@@ -348,14 +358,16 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         }
 
         if let Some(event_handler) = self.event_handler.as_ref()
-            && event_handler.has_execution_payload_available_subscribers()
+            && event_handler.has_execution_payload_subscribers()
         {
-            event_handler.register(EventKind::ExecutionPayloadAvailable(
-                SseExecutionPayloadAvailable {
-                    slot: envelope_slot,
-                    block_root,
-                },
-            ));
+            event_handler.register(EventKind::ExecutionPayload(SseExecutionPayload {
+                slot: envelope_slot,
+                builder_index: signed_envelope.message.builder_index,
+                block_hash: signed_envelope.block_hash(),
+                block_root,
+                state_root: signed_envelope.message.state_root,
+                execution_optimistic: payload_verification_status.is_optimistic(),
+            }));
         }
     }
 }
diff --git a/beacon_node/beacon_chain/src/payload_envelope_verification/mod.rs b/beacon_node/beacon_chain/src/payload_envelope_verification/mod.rs
index 8ca6871dda..6580146512 100644
--- a/beacon_node/beacon_chain/src/payload_envelope_verification/mod.rs
+++ b/beacon_node/beacon_chain/src/payload_envelope_verification/mod.rs
@@ -198,6 +198,8 @@ pub enum EnvelopeError {
         payload_slot: Slot,
         latest_finalized_slot: Slot,
     },
+    /// Optimistic sync is not supported for Gloas payload envelopes.
+    OptimisticSyncNotSupported { block_root: Hash256 },
     /// Some Beacon Chain Error
     BeaconChainError(Arc<BeaconChainError>),
     /// Some Beacon State error
diff --git a/beacon_node/beacon_chain/src/schema_change/migration_schema_v29.rs b/beacon_node/beacon_chain/src/schema_change/migration_schema_v29.rs
index 3069200fce..77d4be3443 100644
--- a/beacon_node/beacon_chain/src/schema_change/migration_schema_v29.rs
+++ b/beacon_node/beacon_chain/src/schema_change/migration_schema_v29.rs
@@ -1,7 +1,9 @@
 use crate::beacon_chain::{BeaconChainTypes, FORK_CHOICE_DB_KEY};
 use crate::persisted_fork_choice::{PersistedForkChoiceV28, PersistedForkChoiceV29};
+use std::collections::HashMap;
 use store::hot_cold_store::HotColdDB;
 use store::{DBColumn, Error as StoreError, KeyValueStore, KeyValueStoreOp};
+use tracing::warn;
 use types::EthSpec;
 
 /// Upgrade from schema v28 to v29.
@@ -49,8 +51,49 @@ pub fn upgrade_to_v29<T: BeaconChainTypes>(
         }
     }
 
-    // Convert to v29 and encode.
-    let persisted_v29 = PersistedForkChoiceV29::from(persisted_v28);
+    // Read the previous proposer boost before converting to V29 (V29 no longer stores it).
+    let previous_proposer_boost = persisted_v28
+        .fork_choice_v28
+        .proto_array_v28
+        .previous_proposer_boost;
+
+    // Convert to v29.
+    let mut persisted_v29 = PersistedForkChoiceV29::from(persisted_v28);
+
+    // Subtract the proposer boost from the boosted node and all its ancestors.
+    //
+    // In the V28 schema, `apply_score_changes` baked the proposer boost directly into node
+    // weights and back-propagated it up the parent chain. In V29, the boost is computed
+    // on-the-fly during the virtual tree walk. If we don't subtract the baked-in boost here,
+    // it will be double-counted after the upgrade.
+    if !previous_proposer_boost.root.is_zero() && previous_proposer_boost.score > 0 {
+        let score = previous_proposer_boost.score;
+        let indices: HashMap<_, _> = persisted_v29
+            .fork_choice
+            .proto_array
+            .indices
+            .iter()
+            .cloned()
+            .collect();
+
+        if let Some(node_index) = indices.get(&previous_proposer_boost.root).copied() {
+            let nodes = &mut persisted_v29.fork_choice.proto_array.nodes;
+            let mut current = Some(node_index);
+            while let Some(idx) = current {
+                if let Some(node) = nodes.get_mut(idx) {
+                    *node.weight_mut() = node.weight().saturating_sub(score);
+                    current = node.parent();
+                } else {
+                    break;
+                }
+            }
+        } else {
+            warn!(
+                root = ?previous_proposer_boost.root,
+                "Proposer boost node missing from fork choice"
+            );
+        }
+    }
 
     Ok(vec![
         persisted_v29.as_kv_store_op(FORK_CHOICE_DB_KEY, db.get_config())?,
diff --git a/beacon_node/beacon_chain/tests/payload_invalidation.rs b/beacon_node/beacon_chain/tests/payload_invalidation.rs
index 13672bbb63..947024e8c2 100644
--- a/beacon_node/beacon_chain/tests/payload_invalidation.rs
+++ b/beacon_node/beacon_chain/tests/payload_invalidation.rs
@@ -1438,7 +1438,7 @@ async fn weights_after_resetting_optimistic_status() {
         .canonical_head
         .fork_choice_write_lock()
         .proto_array_mut()
-        .set_all_blocks_to_optimistic::<E>(&rig.harness.chain.spec)
+        .set_all_blocks_to_optimistic::<E>()
         .unwrap();
 
     let new_weights = rig
diff --git a/beacon_node/http_api/src/publish_blocks.rs b/beacon_node/http_api/src/publish_blocks.rs
index 43dfbeb836..eb7e56e9cc 100644
--- a/beacon_node/http_api/src/publish_blocks.rs
+++ b/beacon_node/http_api/src/publish_blocks.rs
@@ -146,12 +146,8 @@ pub async fn publish_block<T: BeaconChainTypes, B: IntoGossipVerifiedBlock<T>>(
     let slot = block.message().slot();
     let sender_clone = network_tx.clone();
 
-    let build_sidecar_task_handle = spawn_build_data_sidecar_task(
-        chain.clone(),
-        block.clone(),
-        unverified_blobs,
-        current_span.clone(),
-    )?;
+    let build_sidecar_task_handle =
+        spawn_build_data_sidecar_task(chain.clone(), block.clone(), unverified_blobs)?;
 
     // Gossip verify the block and blobs/data columns separately.
     let gossip_verified_block_result = unverified_block.into_gossip_verified_block(&chain);
@@ -358,7 +354,6 @@ fn spawn_build_data_sidecar_task<T: BeaconChainTypes>(
     chain: Arc<BeaconChain<T>>,
     block: Arc<SignedBeaconBlock<T::EthSpec, FullPayload<T::EthSpec>>>,
     proofs_and_blobs: UnverifiedBlobs<T>,
-    current_span: Span,
 ) -> Result<impl Future<Output = BuildDataSidecarTaskResult<T>>, Rejection> {
     chain
         .clone()
@@ -368,7 +363,7 @@ fn spawn_build_data_sidecar_task<T: BeaconChainTypes>(
                 let Some((kzg_proofs, blobs)) = proofs_and_blobs else {
                     return Ok((vec![], vec![]));
                 };
-                let _guard = debug_span!(parent: current_span, "build_data_sidecars").entered();
+                let _span = debug_span!("build_data_sidecars").entered();
 
                 let peer_das_enabled = chain.spec.is_peer_das_enabled_for_epoch(block.epoch());
                 if !peer_das_enabled {
diff --git a/beacon_node/lighthouse_network/src/rpc/protocol.rs b/beacon_node/lighthouse_network/src/rpc/protocol.rs
index 2c92e17c44..c949dfe17d 100644
--- a/beacon_node/lighthouse_network/src/rpc/protocol.rs
+++ b/beacon_node/lighthouse_network/src/rpc/protocol.rs
@@ -11,7 +11,7 @@ use std::io;
 use std::marker::PhantomData;
 use std::sync::{Arc, LazyLock};
 use std::time::Duration;
-use strum::{AsRefStr, Display, EnumString, IntoStaticStr};
+use strum::{AsRefStr, Display, EnumIter, EnumString, IntoStaticStr};
 use tokio_util::{
     codec::Framed,
     compat::{Compat, FuturesAsyncReadCompatExt},
@@ -329,7 +329,7 @@ pub enum Encoding {
 }
 
 /// All valid protocol name and version combinations.
-#[derive(Debug, Clone, Copy, PartialEq)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, EnumIter)]
 pub enum SupportedProtocol {
     StatusV1,
     StatusV2,
@@ -499,6 +499,10 @@ impl<E: EthSpec> UpgradeInfo for RPCProtocol<E> {
                 SupportedProtocol::LightClientFinalityUpdateV1,
                 Encoding::SSZSnappy,
             ));
+            supported_protocols.push(ProtocolId::new(
+                SupportedProtocol::LightClientUpdatesByRangeV1,
+                Encoding::SSZSnappy,
+            ));
         }
         supported_protocols
     }
@@ -1133,3 +1137,101 @@ impl RPCError {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use libp2p::core::UpgradeInfo;
+    use std::collections::HashSet;
+    use strum::IntoEnumIterator;
+    use types::{Hash256, Slot};
+
+    type E = MainnetEthSpec;
+
+    /// Whether this protocol should appear in `currently_supported()` for the given context.
+    ///
+    /// Uses an exhaustive match so that adding a new `SupportedProtocol` variant
+    /// causes a compile error until this function is updated.
+    fn expected_in_currently_supported(
+        protocol: SupportedProtocol,
+        fork_context: &ForkContext,
+    ) -> bool {
+        use SupportedProtocol::*;
+        match protocol {
+            StatusV1 | StatusV2 | GoodbyeV1 | PingV1 | BlocksByRangeV1 | BlocksByRangeV2
+            | BlocksByRootV1 | BlocksByRootV2 | MetaDataV1 | MetaDataV2 => true,
+
+            BlobsByRangeV1 | BlobsByRootV1 => fork_context.fork_exists(ForkName::Deneb),
+
+            DataColumnsByRootV1 | DataColumnsByRangeV1 | MetaDataV3 => {
+                fork_context.spec.is_peer_das_scheduled()
+            }
+
+            PayloadEnvelopesByRangeV1 | PayloadEnvelopesByRootV1 => {
+                fork_context.fork_exists(ForkName::Gloas)
+            }
+
+            // Light client protocols are not in currently_supported()
+            LightClientBootstrapV1
+            | LightClientOptimisticUpdateV1
+            | LightClientFinalityUpdateV1
+            | LightClientUpdatesByRangeV1 => false,
+        }
+    }
+
+    /// Whether this protocol should appear in `protocol_info()` when light client server is
+    /// enabled.
+    ///
+    /// Uses an exhaustive match so that adding a new `SupportedProtocol` variant
+    /// causes a compile error until this function is updated.
+    fn expected_in_protocol_info(protocol: SupportedProtocol, fork_context: &ForkContext) -> bool {
+        use SupportedProtocol::*;
+        match protocol {
+            LightClientBootstrapV1
+            | LightClientOptimisticUpdateV1
+            | LightClientFinalityUpdateV1
+            | LightClientUpdatesByRangeV1 => true,
+
+            _ => expected_in_currently_supported(protocol, fork_context),
+        }
+    }
+
+    #[test]
+    fn all_protocols_registered() {
+        for fork in ForkName::list_all() {
+            let spec = fork.make_genesis_spec(E::default_spec());
+            let fork_context = Arc::new(ForkContext::new::<E>(Slot::new(0), Hash256::ZERO, &spec));
+
+            let currently_supported: HashSet<SupportedProtocol> =
+                SupportedProtocol::currently_supported(&fork_context)
+                    .into_iter()
+                    .map(|pid| pid.versioned_protocol)
+                    .collect();
+
+            let rpc_protocol = RPCProtocol::<E> {
+                fork_context: fork_context.clone(),
+                max_rpc_size: spec.max_payload_size as usize,
+                enable_light_client_server: true,
+                phantom: PhantomData,
+            };
+            let protocol_info: HashSet<SupportedProtocol> = rpc_protocol
+                .protocol_info()
+                .into_iter()
+                .map(|pid| pid.versioned_protocol)
+                .collect();
+
+            for protocol in SupportedProtocol::iter() {
+                assert_eq!(
+                    currently_supported.contains(&protocol),
+                    expected_in_currently_supported(protocol, &fork_context),
+                    "{protocol:?} registration mismatch in currently_supported() at {fork:?}"
+                );
+                assert_eq!(
+                    protocol_info.contains(&protocol),
+                    expected_in_protocol_info(protocol, &fork_context),
+                    "{protocol:?} registration mismatch in protocol_info() at {fork:?}"
+                );
+            }
+        }
+    }
+}
diff --git a/common/task_executor/src/lib.rs b/common/task_executor/src/lib.rs
index d3d862f96c..07716fa2e7 100644
--- a/common/task_executor/src/lib.rs
+++ b/common/task_executor/src/lib.rs
@@ -6,7 +6,7 @@ use futures::channel::mpsc::Sender;
 use futures::prelude::*;
 use std::sync::{Arc, Weak};
 use tokio::runtime::{Handle, Runtime};
-use tracing::debug;
+use tracing::{Span, debug};
 
 use crate::rayon_pool_provider::RayonPoolProvider;
 pub use crate::rayon_pool_provider::RayonPoolType;
@@ -225,9 +225,11 @@ impl TaskExecutor {
         F: FnOnce() + Send + 'static,
     {
         let thread_pool = self.rayon_pool_provider.get_thread_pool(rayon_pool_type);
+        let span = Span::current();
         self.spawn_blocking(
             move || {
                 thread_pool.install(|| {
+                    let _guard = span.enter();
                     task();
                 });
             },
@@ -247,8 +249,10 @@ impl TaskExecutor {
     {
         let thread_pool = self.rayon_pool_provider.get_thread_pool(rayon_pool_type);
         let (tx, rx) = tokio::sync::oneshot::channel();
+        let span = Span::current();
 
         thread_pool.spawn(move || {
+            let _guard = span.enter();
             let result = task();
             let _ = tx.send(result);
         });
@@ -320,8 +324,12 @@ impl TaskExecutor {
         let timer = metrics::start_timer_vec(&metrics::BLOCKING_TASKS_HISTOGRAM, &[name]);
         metrics::inc_gauge_vec(&metrics::BLOCKING_TASKS_COUNT, &[name]);
 
+        let span = Span::current();
         let join_handle = if let Some(handle) = self.handle() {
-            handle.spawn_blocking(task)
+            handle.spawn_blocking(move || {
+                let _guard = span.enter();
+                task()
+            })
         } else {
             debug!("Couldn't spawn task. Runtime shutting down");
             return None;
diff --git a/consensus/fork_choice/src/fork_choice.rs b/consensus/fork_choice/src/fork_choice.rs
index cedd42cf01..0993ae95a3 100644
--- a/consensus/fork_choice/src/fork_choice.rs
+++ b/consensus/fork_choice/src/fork_choice.rs
@@ -26,6 +26,7 @@ use types::{
 #[derive(Debug)]
 pub enum Error<T> {
     InvalidAttestation(InvalidAttestation),
+    InvalidPayloadAttestation(InvalidPayloadAttestation),
     InvalidAttesterSlashing(AttesterSlashingValidationError),
     InvalidBlock(InvalidBlock),
     ProtoArrayStringError(String),
@@ -85,6 +86,12 @@ impl<T> From<InvalidAttestation> for Error<T> {
     }
 }
 
+impl<T> From<InvalidPayloadAttestation> for Error<T> {
+    fn from(e: InvalidPayloadAttestation) -> Self {
+        Error::InvalidPayloadAttestation(e)
+    }
+}
+
 impl<T> From<AttesterSlashingValidationError> for Error<T> {
     fn from(e: AttesterSlashingValidationError) -> Self {
         Error::InvalidAttesterSlashing(e)
@@ -177,14 +184,26 @@ pub enum InvalidAttestation {
     /// Post-Gloas: attestation with index == 1 (payload_present) requires the block's
     /// payload to have been received (`root in store.payload_states`).
     PayloadNotReceived { beacon_block_root: Hash256 },
-    /// A payload attestation votes payload_present for a block in the current slot, which is
-    /// invalid because the payload cannot be known yet.
-    PayloadPresentDuringSameSlot { slot: Slot },
+}
+
+#[derive(Debug, Clone, PartialEq)]
+pub enum InvalidPayloadAttestation {
+    /// The payload attestation's attesting indices were empty.
+    EmptyAggregationBitfield,
+    /// The `payload_attestation.data.beacon_block_root` block is unknown.
+    UnknownHeadBlock { beacon_block_root: Hash256 },
+    /// The payload attestation is attesting to a block that is later than itself.
+    AttestsToFutureBlock { block: Slot, attestation: Slot },
     /// A gossip payload attestation must be for the current slot.
     PayloadAttestationNotCurrentSlot {
         attestation_slot: Slot,
         current_slot: Slot,
     },
+    /// One or more payload attesters are not part of the PTC.
+    PayloadAttestationAttestersNotInPtc {
+        attesting_indices_len: usize,
+        attesting_indices_in_ptc: usize,
+    },
 }
 
 impl<T> From<String> for Error<T> {
@@ -654,6 +673,20 @@ where
         }
     }
 
+    /// Mark a Gloas payload envelope as valid and received.
+    ///
+    /// This must only be called for valid Gloas payloads.
+    pub fn on_valid_payload_envelope_received(
+        &mut self,
+        block_root: Hash256,
+    ) -> Result<(), Error<T::Error>> {
+        self.proto_array
+            .on_valid_payload_envelope_received(block_root)
+            .map_err(Error::FailedToProcessValidExecutionPayload)
+    }
+
+    /// Pre-Gloas only.
+    ///
     /// See `ProtoArrayForkChoice::process_execution_payload_validation` for documentation.
     pub fn on_valid_execution_payload(
         &mut self,
@@ -664,6 +697,8 @@ where
             .map_err(Error::FailedToProcessValidExecutionPayload)
     }
 
+    /// Pre-Gloas only.
+    ///
     /// See `ProtoArrayForkChoice::process_execution_payload_invalidation` for documentation.
     pub fn on_invalid_execution_payload(
         &mut self,
@@ -977,12 +1012,6 @@ where
         Ok(())
     }
 
-    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)
-    }
-
     /// Update checkpoints in store if necessary
     fn update_checkpoints(
         &mut self,
@@ -1158,50 +1187,48 @@ where
         &self,
         indexed_payload_attestation: &IndexedPayloadAttestation<E>,
         is_from_block: AttestationFromBlock,
-    ) -> Result<(), InvalidAttestation> {
+    ) -> Result<(), InvalidPayloadAttestation> {
+        // This check is from `is_valid_indexed_payload_attestation`, but we do it immediately to
+        // avoid wasting time on junk attestations.
         if indexed_payload_attestation.attesting_indices.is_empty() {
-            return Err(InvalidAttestation::EmptyAggregationBitfield);
+            return Err(InvalidPayloadAttestation::EmptyAggregationBitfield);
         }
 
+        // PTC attestation must be for a known block. If block is unknown, delay consideration until
+        // the block is found (responsibility of caller).
         let block = self
             .proto_array
             .get_block(&indexed_payload_attestation.data.beacon_block_root)
-            .ok_or(InvalidAttestation::UnknownHeadBlock {
+            .ok_or(InvalidPayloadAttestation::UnknownHeadBlock {
                 beacon_block_root: indexed_payload_attestation.data.beacon_block_root,
             })?;
 
+        // Not strictly part of the spec, but payload attestations to future slots are MORE INVALID
+        // than payload attestations to blocks at previous slots.
         if block.slot > indexed_payload_attestation.data.slot {
-            return Err(InvalidAttestation::AttestsToFutureBlock {
+            return Err(InvalidPayloadAttestation::AttestsToFutureBlock {
                 block: block.slot,
                 attestation: indexed_payload_attestation.data.slot,
             });
         }
 
-        // Spec: `if data.slot != state.slot: return` — PTC votes can only
-        // change the vote for their assigned beacon block.
+        // PTC votes can only change the vote for their assigned beacon block, return early otherwise
         if block.slot != indexed_payload_attestation.data.slot {
             return Ok(());
         }
 
         // Gossip payload attestations must be for the current slot.
+        // NOTE: signature is assumed to have been verified by caller.
         // https://github.com/ethereum/consensus-specs/blob/master/specs/gloas/fork-choice.md
         if matches!(is_from_block, AttestationFromBlock::False)
             && indexed_payload_attestation.data.slot != self.fc_store.get_current_slot()
         {
-            return Err(InvalidAttestation::PayloadAttestationNotCurrentSlot {
-                attestation_slot: indexed_payload_attestation.data.slot,
-                current_slot: self.fc_store.get_current_slot(),
-            });
-        }
-
-        // A payload attestation voting payload_present for a block in the current slot is
-        // invalid: the payload cannot be known yet. This only applies to gossip attestations;
-        // payload attestations from blocks have already been validated by the block producer.
-        if matches!(is_from_block, AttestationFromBlock::False)
-            && self.fc_store.get_current_slot() == block.slot
-            && indexed_payload_attestation.data.payload_present
-        {
-            return Err(InvalidAttestation::PayloadPresentDuringSameSlot { slot: block.slot });
+            return Err(
+                InvalidPayloadAttestation::PayloadAttestationNotCurrentSlot {
+                    attestation_slot: indexed_payload_attestation.data.slot,
+                    current_slot: self.fc_store.get_current_slot(),
+                },
+            );
         }
 
         Ok(())
@@ -1308,10 +1335,22 @@ where
 
         // Resolve validator indices to PTC committee positions.
         let ptc_indices: Vec<usize> = attestation
-            .attesting_indices_iter()
+            .attesting_indices
+            .iter()
             .filter_map(|vi| ptc.iter().position(|&p| p == *vi as usize))
             .collect();
 
+        // Check that all the attesters are in the PTC
+        if ptc_indices.len() != attestation.attesting_indices.len() {
+            return Err(
+                InvalidPayloadAttestation::PayloadAttestationAttestersNotInPtc {
+                    attesting_indices_len: attestation.attesting_indices.len(),
+                    attesting_indices_in_ptc: ptc_indices.len(),
+                }
+                .into(),
+            );
+        }
+
         for &ptc_index in &ptc_indices {
             self.proto_array.process_payload_attestation(
                 attestation.data.beacon_block_root,
@@ -1614,7 +1653,6 @@ where
         persisted_proto_array: proto_array::core::SszContainer,
         justified_balances: JustifiedBalances,
         reset_payload_statuses: ResetPayloadStatuses,
-        spec: &ChainSpec,
     ) -> Result<ProtoArrayForkChoice, Error<T::Error>> {
         let mut proto_array = ProtoArrayForkChoice::from_container(
             persisted_proto_array.clone(),
@@ -1639,7 +1677,7 @@ where
 
         // Reset all blocks back to being "optimistic". This helps recover from an EL consensus
         // fault where an invalid payload becomes valid.
-        if let Err(e) = proto_array.set_all_blocks_to_optimistic::<E>(spec) {
+        if let Err(e) = proto_array.set_all_blocks_to_optimistic::<E>() {
             // If there is an error resetting the optimistic status then log loudly and revert
             // back to a proto-array which does not have the reset applied. This indicates a
             // significant error in Lighthouse and warrants detailed investigation.
@@ -1669,7 +1707,6 @@ where
             persisted.proto_array,
             justified_balances,
             reset_payload_statuses,
-            spec,
         )?;
 
         let current_slot = fc_store.get_current_slot();
@@ -1703,7 +1740,7 @@ where
             // get a different result.
             fork_choice
                 .proto_array
-                .set_all_blocks_to_optimistic::<E>(spec)?;
+                .set_all_blocks_to_optimistic::<E>()?;
             // If the second attempt at finding a head fails, return an error since we do not
             // expect this scenario.
             let _ = fork_choice.get_head(current_slot, spec)?;
diff --git a/consensus/fork_choice/src/lib.rs b/consensus/fork_choice/src/lib.rs
index 8f479125b7..159eab0ec0 100644
--- a/consensus/fork_choice/src/lib.rs
+++ b/consensus/fork_choice/src/lib.rs
@@ -4,8 +4,9 @@ mod metrics;
 
 pub use crate::fork_choice::{
     AttestationFromBlock, Error, ForkChoice, ForkChoiceView, ForkchoiceUpdateParameters,
-    InvalidAttestation, InvalidBlock, PayloadVerificationStatus, PersistedForkChoice,
-    PersistedForkChoiceV28, PersistedForkChoiceV29, QueuedAttestation, ResetPayloadStatuses,
+    InvalidAttestation, InvalidBlock, InvalidPayloadAttestation, PayloadVerificationStatus,
+    PersistedForkChoice, PersistedForkChoiceV28, PersistedForkChoiceV29, QueuedAttestation,
+    ResetPayloadStatuses,
 };
 pub use fork_choice_store::ForkChoiceStore;
 pub use proto_array::{
diff --git a/consensus/fork_choice/tests/tests.rs b/consensus/fork_choice/tests/tests.rs
index 241e25d3e2..d6f937c0ca 100644
--- a/consensus/fork_choice/tests/tests.rs
+++ b/consensus/fork_choice/tests/tests.rs
@@ -11,7 +11,7 @@ use bls::AggregateSignature;
 use fixed_bytes::FixedBytesExtended;
 use fork_choice::{
     AttestationFromBlock, ForkChoiceStore, InvalidAttestation, InvalidBlock,
-    PayloadVerificationStatus, QueuedAttestation,
+    InvalidPayloadAttestation, PayloadVerificationStatus, QueuedAttestation,
 };
 use state_processing::state_advance::complete_state_advance;
 use std::fmt;
@@ -969,8 +969,8 @@ async fn non_block_payload_attestation_for_previous_slot_is_rejected() {
     assert!(
         matches!(
             result,
-            Err(ForkChoiceError::InvalidAttestation(
-                InvalidAttestation::PayloadAttestationNotCurrentSlot { .. }
+            Err(ForkChoiceError::InvalidPayloadAttestation(
+                InvalidPayloadAttestation::PayloadAttestationNotCurrentSlot { .. }
             ))
         ),
         "gossip payload attestation for previous slot should be rejected, got: {:?}",
diff --git a/consensus/proto_array/src/fork_choice_test_definition.rs b/consensus/proto_array/src/fork_choice_test_definition.rs
index ff9d70bad5..c9764d3e44 100644
--- a/consensus/proto_array/src/fork_choice_test_definition.rs
+++ b/consensus/proto_array/src/fork_choice_test_definition.rs
@@ -98,7 +98,7 @@ pub enum Operation {
     },
     /// Simulate receiving and validating an execution payload for `block_root`.
     /// Sets `payload_received = true` on the V29 node via the live validation path.
-    ProcessExecutionPayload {
+    ProcessExecutionPayloadEnvelope {
         block_root: Hash256,
     },
     AssertPayloadReceived {
@@ -500,9 +500,9 @@ impl ForkChoiceTestDefinition {
                     // the payload to be in payload_states (payload_received).
                     node_v29.payload_received = is_timely || is_data_available;
                 }
-                Operation::ProcessExecutionPayload { block_root } => {
+                Operation::ProcessExecutionPayloadEnvelope { block_root } => {
                     fork_choice
-                        .on_execution_payload(block_root)
+                        .on_valid_payload_envelope_received(block_root)
                         .unwrap_or_else(|e| {
                             panic!(
                                 "on_execution_payload op at index {} returned error: {}",
diff --git a/consensus/proto_array/src/fork_choice_test_definition/gloas_payload.rs b/consensus/proto_array/src/fork_choice_test_definition/gloas_payload.rs
index 18d7a40b82..ea37780795 100644
--- a/consensus/proto_array/src/fork_choice_test_definition/gloas_payload.rs
+++ b/consensus/proto_array/src/fork_choice_test_definition/gloas_payload.rs
@@ -53,7 +53,7 @@ pub fn get_gloas_chain_following_test_definition() -> ForkChoiceTestDefinition {
 
     // Mark root_1 as having received its execution payload so that
     // its FULL virtual node exists in the Gloas fork choice tree.
-    ops.push(Operation::ProcessExecutionPayload {
+    ops.push(Operation::ProcessExecutionPayloadEnvelope {
         block_root: get_root(1),
     });
 
@@ -263,7 +263,7 @@ pub fn get_gloas_find_head_vote_transition_test_definition() -> ForkChoiceTestDe
 
     // Mark root_1 as having received its execution payload so that
     // its FULL virtual node exists in the Gloas fork choice tree.
-    ops.push(Operation::ProcessExecutionPayload {
+    ops.push(Operation::ProcessExecutionPayloadEnvelope {
         block_root: get_root(1),
     });
 
@@ -368,7 +368,7 @@ pub fn get_gloas_weight_priority_over_payload_preference_test_definition()
 
     // Mark root_1 as having received its execution payload so that
     // its FULL virtual node exists in the Gloas fork choice tree.
-    ops.push(Operation::ProcessExecutionPayload {
+    ops.push(Operation::ProcessExecutionPayloadEnvelope {
         block_root: get_root(1),
     });
 
@@ -538,7 +538,7 @@ pub fn get_gloas_interleaved_attestations_test_definition() -> ForkChoiceTestDef
 
     // Mark root_1 as having received its execution payload so that
     // its FULL virtual node exists in the Gloas fork choice tree.
-    ops.push(Operation::ProcessExecutionPayload {
+    ops.push(Operation::ProcessExecutionPayloadEnvelope {
         block_root: get_root(1),
     });
 
@@ -674,8 +674,8 @@ pub fn get_gloas_payload_received_interleaving_test_definition() -> ForkChoiceTe
         expected_payload_status: None,
     });
 
-    // ProcessExecutionPayload on genesis is a no-op (already received at init).
-    ops.push(Operation::ProcessExecutionPayload {
+    // ProcessExecutionPayloadEnvelope on genesis is a no-op (already received at init).
+    ops.push(Operation::ProcessExecutionPayloadEnvelope {
         block_root: get_root(0),
     });
 
@@ -778,7 +778,7 @@ mod tests {
 
         // Mark root 2's execution payload as received so the Full virtual child exists.
         if first_gloas_block_full {
-            ops.push(Operation::ProcessExecutionPayload {
+            ops.push(Operation::ProcessExecutionPayloadEnvelope {
                 block_root: get_root(2),
             });
         }
diff --git a/consensus/proto_array/src/proto_array.rs b/consensus/proto_array/src/proto_array.rs
index 361e4a86e2..1f7291b260 100644
--- a/consensus/proto_array/src/proto_array.rs
+++ b/consensus/proto_array/src/proto_array.rs
@@ -165,8 +165,6 @@ pub struct ProtoNode {
     pub payload_data_availability_votes: BitVector<U512>,
     /// Whether the execution payload for this block has been received and validated locally.
     /// Maps to `root in store.payload_states` in the spec.
-    /// When true, `is_payload_timely` and `is_payload_data_available` return true
-    /// regardless of PTC vote counts.
     #[superstruct(only(V29), partial_getter(copy))]
     pub payload_received: bool,
     /// The proposer index for this block, used by `should_apply_proposer_boost`
@@ -213,7 +211,7 @@ impl ProtoNode {
             return false;
         }
 
-        node.payload_timeliness_votes.num_set_bits() > E::ptc_size() / 2
+        node.payload_timeliness_votes.num_set_bits() > E::payload_timely_threshold()
     }
 
     pub fn is_payload_data_available<E: EthSpec>(&self) -> bool {
@@ -226,8 +224,8 @@ impl ProtoNode {
             return false;
         }
 
-        // TODO(gloas): add function on EthSpec for DATA_AVAILABILITY_TIMELY_THRESHOLD
-        node.payload_data_availability_votes.num_set_bits() > E::ptc_size() / 2
+        node.payload_data_availability_votes.num_set_bits()
+            > E::data_availability_timely_threshold()
     }
 }
 
@@ -369,7 +367,6 @@ pub struct ProtoArray {
     pub prune_threshold: usize,
     pub nodes: Vec<ProtoNode>,
     pub indices: HashMap<Hash256, usize>,
-    pub previous_proposer_boost: ProposerBoost,
 }
 
 impl ProtoArray {
@@ -492,20 +489,14 @@ impl ProtoArray {
                             .ok_or(Error::DeltaOverflow(parent_index))?;
                     }
                 } else {
-                    // V17 child of a V29 parent (fork transition): treat as FULL
-                    // since V17 nodes always have execution payloads inline.
-                    parent_delta.full_delta = parent_delta
-                        .full_delta
-                        .checked_add(delta)
-                        .ok_or(Error::DeltaOverflow(parent_index))?;
+                    // This is a v17 node with a v17 parent.
+                    // There is no empty or full weight for v17 nodes, so nothing to propagate.
+                    // In the tree walk, the v17 nodes have an empty child with 0 weight, which
+                    // wins by default (it is the only child).
                 }
             }
         }
 
-        // Proposer boost is now applied on-the-fly in `get_weight` during the
-        // walk, so clear any stale boost from a prior call.
-        self.previous_proposer_boost = ProposerBoost::default();
-
         Ok(())
     }
 
@@ -641,11 +632,9 @@ impl ProtoArray {
                 // Anchor gets [True, True]. Others computed from time_into_slot.
                 block_timeliness_attestation_threshold: is_genesis
                     || (is_current_slot
-                        && time_into_slot < spec.get_unaggregated_attestation_due()),
-                // TODO(gloas): use Gloas-specific PTC due threshold once
-                // `get_payload_attestation_due_ms` is on ChainSpec.
+                        && time_into_slot < spec.get_attestation_due::<E>(current_slot)),
                 block_timeliness_ptc_threshold: is_genesis
-                    || (is_current_slot && time_into_slot < spec.get_slot_duration() / 2),
+                    || (is_current_slot && time_into_slot < spec.get_payload_attestation_due()),
                 equivocating_attestation_score: 0,
             })
         };
@@ -682,11 +671,17 @@ impl ProtoArray {
     }
 
     /// Spec: `is_head_weak`.
-    ///
-    /// The spec adds weight from equivocating validators in the head slot's
-    /// committees. We approximate this with `equivocating_attestation_score`
-    /// which tracks equivocating validators that voted for this block (close
-    /// but not identical to committee membership).
+    // TODO(gloas): the spec adds weight from equivocating validators in the
+    // head slot's *committees*, regardless of who they voted for. We approximate
+    // with `equivocating_attestation_score` which only tracks equivocating
+    // validators whose vote pointed at this block. This under-counts when an
+    // equivocating validator is in the committee but voted for a different fork,
+    // which could allow a re-org the spec wouldn't. In practice the deviation
+    // is small — it requires equivocating validators voting for competing forks
+    // AND the head weight to be exactly at the reorg threshold boundary.
+    // Fixing this properly requires committee computation from BeaconState,
+    // which is not available in proto_array. The fix would be to pass
+    // pre-computed equivocating committee weight from the beacon_chain caller.
     fn is_head_weak<E: EthSpec>(
         &self,
         head_node: &ProtoNode,
@@ -729,7 +724,6 @@ impl ProtoArray {
             .nodes
             .get(block_index)
             .ok_or(Error::InvalidNodeIndex(block_index))?;
-        // TODO(gloas): handle parent unknown case?
         let parent_index = block
             .parent()
             .ok_or(Error::NodeUnknown(proposer_boost_root))?;
@@ -753,7 +747,6 @@ impl ProtoArray {
         // the parent's slot from the same proposer.
         let parent_slot = parent.slot();
         let parent_root = parent.root();
-        // TODO(gloas): handle proposer index for pre-Gloas blocks?
         let parent_proposer = parent.proposer_index();
 
         let has_equivocation = self.nodes.iter().any(|node| {
@@ -773,12 +766,10 @@ impl ProtoArray {
         Ok(!has_equivocation)
     }
 
-    /// Process an execution payload for a Gloas block.
+    /// Process a valid execution payload envelope for a Gloas block.
     ///
-    /// Sets `payload_received` to true, which makes `is_payload_timely` and
-    /// `is_payload_data_available` return true regardless of PTC votes.
-    /// This maps to `store.payload_states[root] = state` in the spec.
-    pub fn on_valid_execution_payload(&mut self, block_root: Hash256) -> Result<(), Error> {
+    /// Sets `payload_received` to true.
+    pub fn on_valid_payload_envelope_received(&mut self, block_root: Hash256) -> Result<(), Error> {
         let index = *self
             .indices
             .get(&block_root)
@@ -814,6 +805,8 @@ impl ProtoArray {
 
     /// Updates the `verified_node_index` and all ancestors to have validated execution payloads.
     ///
+    /// This function is a no-op if called for a Gloas block.
+    ///
     /// Returns an error if:
     ///
     /// - The `verified_node_index` is unknown.
@@ -857,18 +850,10 @@ impl ProtoArray {
                         });
                     }
                 },
-                // Gloas nodes don't carry `ExecutionStatus`. Mark the validated
-                // block as payload-received so that `is_payload_timely` /
-                // `is_payload_data_available` and `index == 1` attestations work.
-                ProtoNode::V29(node) => {
-                    if index == verified_node_index {
-                        node.payload_received = true;
-                    }
-                    if let Some(parent_index) = node.parent {
-                        parent_index
-                    } else {
-                        return Ok(());
-                    }
+                // Gloas nodes should not be marked valid by this function, which exists only
+                // for pre-Gloas fork choice.
+                ProtoNode::V29(_) => {
+                    return Ok(());
                 }
             };
 
@@ -879,6 +864,7 @@ impl ProtoArray {
     /// Invalidate zero or more blocks, as specified by the `InvalidationOperation`.
     ///
     /// See the documentation of `InvalidationOperation` for usage.
+    // TODO(gloas): this needs some tests for the mixed Gloas/pre-Gloas case.
     pub fn propagate_execution_payload_invalidation<E: EthSpec>(
         &mut self,
         op: &InvalidationOperation,
@@ -978,7 +964,7 @@ impl ProtoArray {
                     // This block is pre-merge, therefore it has no execution status. Nor do its
                     // ancestors.
                     Ok(ExecutionStatus::Irrelevant(_)) => break,
-                    Err(_) => (),
+                    Err(_) => break,
                 }
             }
 
@@ -1087,9 +1073,6 @@ impl ProtoArray {
             });
         }
 
-        // In the post-Gloas world, always use a virtual tree walk.
-        //
-        // Best child/best descendant is dead.
         let best_fc_node = self.find_head_walk::<E>(
             justified_index,
             current_slot,
@@ -1125,26 +1108,6 @@ impl ProtoArray {
         Ok((best_fc_node.root, best_fc_node.payload_status))
     }
 
-    /// Build a parent->children index. Invalid nodes are excluded
-    /// (they aren't in store.blocks in the spec).
-    fn build_children_index(&self) -> Vec<Vec<usize>> {
-        let mut children = vec![vec![]; self.nodes.len()];
-        for (i, node) in self.nodes.iter().enumerate() {
-            if node
-                .execution_status()
-                .is_ok_and(|status| status.is_invalid())
-            {
-                continue;
-            }
-            if let Some(parent) = node.parent()
-                && parent < children.len()
-            {
-                children[parent].push(i);
-            }
-        }
-        children
-    }
-
     /// Spec: `get_filtered_block_tree`.
     ///
     /// Returns the set of node indices on viable branches — those with at least
@@ -1155,7 +1118,6 @@ impl ProtoArray {
         current_slot: Slot,
         best_justified_checkpoint: Checkpoint,
         best_finalized_checkpoint: Checkpoint,
-        children_index: &[Vec<usize>],
     ) -> HashSet<usize> {
         let mut viable = HashSet::new();
         self.filter_block_tree::<E>(
@@ -1163,7 +1125,6 @@ impl ProtoArray {
             current_slot,
             best_justified_checkpoint,
             best_finalized_checkpoint,
-            children_index,
             &mut viable,
         );
         viable
@@ -1176,17 +1137,25 @@ impl ProtoArray {
         current_slot: Slot,
         best_justified_checkpoint: Checkpoint,
         best_finalized_checkpoint: Checkpoint,
-        children_index: &[Vec<usize>],
         viable: &mut HashSet<usize>,
     ) -> bool {
         let Some(node) = self.nodes.get(node_index) else {
             return false;
         };
 
-        let children = children_index
-            .get(node_index)
-            .map(|c| c.as_slice())
-            .unwrap_or(&[]);
+        // Skip invalid children — they aren't in store.blocks in the spec.
+        let children: Vec<usize> = self
+            .nodes
+            .iter()
+            .enumerate()
+            .filter(|(_, child)| {
+                child.parent() == Some(node_index)
+                    && !child
+                        .execution_status()
+                        .is_ok_and(|status| status.is_invalid())
+            })
+            .map(|(i, _)| i)
+            .collect();
 
         if !children.is_empty() {
             // Evaluate ALL children (no short-circuit) to mark all viable branches.
@@ -1198,7 +1167,6 @@ impl ProtoArray {
                         current_slot,
                         best_justified_checkpoint,
                         best_finalized_checkpoint,
-                        children_index,
                         viable,
                     )
                 })
@@ -1243,16 +1211,12 @@ impl ProtoArray {
             payload_status: PayloadStatus::Pending,
         };
 
-        // Build parent->children index once for O(1) lookups.
-        let children_index = self.build_children_index();
-
         // Spec: `get_filtered_block_tree`.
         let viable_nodes = self.get_filtered_block_tree::<E>(
             start_index,
             current_slot,
             best_justified_checkpoint,
             best_finalized_checkpoint,
-            &children_index,
         );
 
         // Compute once rather than per-child per-level.
@@ -1261,7 +1225,7 @@ impl ProtoArray {
 
         loop {
             let children: Vec<_> = self
-                .get_node_children(&head, &children_index)?
+                .get_node_children(&head)?
                 .into_iter()
                 .filter(|(fc_node, _)| viable_nodes.contains(&fc_node.proto_node_index))
                 .collect();
@@ -1272,11 +1236,7 @@ impl ProtoArray {
 
             head = children
                 .into_iter()
-                .map(|(child, _)| -> Result<_, Error> {
-                    let proto_node = self
-                        .nodes
-                        .get(child.proto_node_index)
-                        .ok_or(Error::InvalidNodeIndex(child.proto_node_index))?;
+                .map(|(child, ref proto_node)| -> Result<_, Error> {
                     let weight = self.get_weight::<E>(
                         &child,
                         proto_node,
@@ -1424,7 +1384,6 @@ impl ProtoArray {
     fn get_node_children(
         &self,
         node: &IndexedForkChoiceNode,
-        children_index: &[Vec<usize>],
     ) -> Result<Vec<(IndexedForkChoiceNode, ProtoNode)>, Error> {
         if node.payload_status == PayloadStatus::Pending {
             let proto_node = self
@@ -1438,25 +1397,23 @@ impl ProtoArray {
             }
             Ok(children)
         } else {
-            let child_indices = children_index
-                .get(node.proto_node_index)
-                .map(|c| c.as_slice())
-                .unwrap_or(&[]);
-            Ok(child_indices
+            Ok(self
+                .nodes
                 .iter()
-                .filter_map(|&child_index| {
-                    let child_node = self.nodes.get(child_index)?;
-                    if child_node.get_parent_payload_status() != node.payload_status {
-                        return None;
-                    }
-                    Some((
+                .enumerate()
+                .filter(|(_, child_node)| {
+                    child_node.parent() == Some(node.proto_node_index)
+                        && child_node.get_parent_payload_status() == node.payload_status
+                })
+                .map(|(child_index, child_node)| {
+                    (
                         IndexedForkChoiceNode {
                             root: child_node.root(),
                             proto_node_index: child_index,
                             payload_status: PayloadStatus::Pending,
                         },
                         child_node.clone(),
-                    ))
+                    )
                 })
                 .collect())
         }
diff --git a/consensus/proto_array/src/proto_array_fork_choice.rs b/consensus/proto_array/src/proto_array_fork_choice.rs
index 72440b83b8..0ecaea3971 100644
--- a/consensus/proto_array/src/proto_array_fork_choice.rs
+++ b/consensus/proto_array/src/proto_array_fork_choice.rs
@@ -2,8 +2,7 @@ use crate::{
     JustifiedBalances,
     error::Error,
     proto_array::{
-        InvalidationOperation, Iter, NodeDelta, ProposerBoost, ProtoArray, ProtoNode,
-        calculate_committee_fraction,
+        InvalidationOperation, Iter, NodeDelta, ProtoArray, ProtoNode, calculate_committee_fraction,
     },
     ssz_container::SszContainer,
 };
@@ -74,6 +73,7 @@ impl From<VoteTracker> for VoteTrackerV28 {
     }
 }
 
+/// Spec's `LatestMessage` type. Only used in tests.
 pub struct LatestMessage {
     pub slot: Slot,
     pub root: Hash256,
@@ -527,7 +527,6 @@ impl ProtoArrayForkChoice {
             prune_threshold: DEFAULT_PRUNE_THRESHOLD,
             nodes: Vec::with_capacity(1),
             indices: HashMap::with_capacity(1),
-            previous_proposer_boost: ProposerBoost::default(),
         };
 
         let block = Block {
@@ -569,11 +568,18 @@ impl ProtoArrayForkChoice {
         })
     }
 
-    pub fn on_execution_payload(&mut self, block_root: Hash256) -> Result<(), String> {
+    /// Mark a Gloas payload envelope as valid and received.
+    ///
+    /// This must only be called for valid Gloas payloads.
+    pub fn on_valid_payload_envelope_received(
+        &mut self,
+        block_root: Hash256,
+    ) -> Result<(), String> {
         self.proto_array
-            .on_valid_execution_payload(block_root)
+            .on_valid_payload_envelope_received(block_root)
             .map_err(|e| format!("Failed to process execution payload: {:?}", e))
     }
+
     /// See `ProtoArray::propagate_execution_payload_validation` for documentation.
     pub fn process_execution_payload_validation(
         &mut self,
@@ -880,10 +886,7 @@ impl ProtoArrayForkChoice {
     /// status to be optimistic.
     ///
     /// In practice this means forgetting any `VALID` or `INVALID` statuses.
-    pub fn set_all_blocks_to_optimistic<E: EthSpec>(
-        &mut self,
-        spec: &ChainSpec,
-    ) -> Result<(), String> {
+    pub fn set_all_blocks_to_optimistic<E: EthSpec>(&mut self) -> Result<(), String> {
         // Iterate backwards through all nodes in the `proto_array`. Whilst it's not strictly
         // required to do this process in reverse, it seems natural when we consider how LMD votes
         // are counted.
@@ -906,7 +909,7 @@ impl ProtoArrayForkChoice {
 
                     // Restore the weight of the node, it would have been set to `0` in
                     // `apply_score_changes` when it was invalidated.
-                    let mut restored_weight: u64 = self
+                    let restored_weight: u64 = self
                         .votes
                         .0
                         .iter()
@@ -922,26 +925,6 @@ impl ProtoArrayForkChoice {
                         })
                         .sum();
 
-                    // If the invalid root was boosted, apply the weight to it and
-                    // ancestors.
-                    if let Some(proposer_score_boost) = spec.proposer_score_boost
-                        && self.proto_array.previous_proposer_boost.root == node.root()
-                    {
-                        // Compute the score based upon the current balances. We can't rely on
-                        // the `previous_proposr_boost.score` since it is set to zero with an
-                        // invalid node.
-                        let proposer_score =
-                            calculate_committee_fraction::<E>(&self.balances, proposer_score_boost)
-                                .ok_or("Failed to compute proposer boost")?;
-                        // Store the score we've applied here so it can be removed in
-                        // a later call to `apply_score_changes`.
-                        self.proto_array.previous_proposer_boost.score = proposer_score;
-                        // Apply this boost to this node.
-                        restored_weight = restored_weight
-                            .checked_add(proposer_score)
-                            .ok_or("Overflow when adding boost to weight")?;
-                    }
-
                     // Add the restored weight to the node and all ancestors.
                     if restored_weight > 0 {
                         let mut node_or_ancestor = node;
@@ -1082,10 +1065,9 @@ impl ProtoArrayForkChoice {
             .is_finalized_checkpoint_or_descendant::<E>(descendant_root, best_finalized_checkpoint)
     }
 
+    /// NOTE: only used in tests.
     pub fn latest_message(&self, validator_index: usize) -> Option<LatestMessage> {
-        if validator_index < self.votes.0.len() {
-            let vote = &self.votes.0[validator_index];
-
+        if let Some(vote) = self.votes.0.get(validator_index) {
             if *vote == VoteTracker::default() {
                 None
             } else {
diff --git a/consensus/proto_array/src/ssz_container.rs b/consensus/proto_array/src/ssz_container.rs
index 80a6702210..69efb35027 100644
--- a/consensus/proto_array/src/ssz_container.rs
+++ b/consensus/proto_array/src/ssz_container.rs
@@ -38,6 +38,7 @@ pub struct SszContainer {
     #[superstruct(only(V29))]
     pub nodes: Vec<ProtoNode>,
     pub indices: Vec<(Hash256, usize)>,
+    #[superstruct(only(V28))]
     pub previous_proposer_boost: ProposerBoost,
 }
 
@@ -50,7 +51,6 @@ impl SszContainerV29 {
             prune_threshold: proto_array.prune_threshold,
             nodes: proto_array.nodes.clone(),
             indices: proto_array.indices.iter().map(|(k, v)| (*k, *v)).collect(),
-            previous_proposer_boost: proto_array.previous_proposer_boost,
         }
     }
 }
@@ -63,7 +63,6 @@ impl TryFrom<(SszContainerV29, JustifiedBalances)> for ProtoArrayForkChoice {
             prune_threshold: from.prune_threshold,
             nodes: from.nodes,
             indices: from.indices.into_iter().collect::<HashMap<_, _>>(),
-            previous_proposer_boost: from.previous_proposer_boost,
         };
 
         Ok(Self {
@@ -92,7 +91,6 @@ impl From<SszContainerV28> for SszContainerV29 {
                 })
                 .collect(),
             indices: v28.indices,
-            previous_proposer_boost: v28.previous_proposer_boost,
         }
     }
 }
@@ -116,7 +114,8 @@ impl From<SszContainerV29> for SszContainerV28 {
                 })
                 .collect(),
             indices: v29.indices,
-            previous_proposer_boost: v29.previous_proposer_boost,
+            // Proposer boost is not tracked in V29 (computed on-the-fly), so reset it.
+            previous_proposer_boost: ProposerBoost::default(),
         }
     }
 }
diff --git a/consensus/types/src/attestation/indexed_payload_attestation.rs b/consensus/types/src/attestation/indexed_payload_attestation.rs
index 4de805570c..bb2087e330 100644
--- a/consensus/types/src/attestation/indexed_payload_attestation.rs
+++ b/consensus/types/src/attestation/indexed_payload_attestation.rs
@@ -2,7 +2,6 @@ use crate::test_utils::TestRandom;
 use crate::{EthSpec, ForkName, PayloadAttestationData};
 use bls::AggregateSignature;
 use context_deserialize::context_deserialize;
-use core::slice::Iter;
 use serde::{Deserialize, Serialize};
 use ssz_derive::{Decode, Encode};
 use ssz_types::VariableList;
@@ -21,12 +20,6 @@ pub struct IndexedPayloadAttestation<E: EthSpec> {
     pub signature: AggregateSignature,
 }
 
-impl<E: EthSpec> IndexedPayloadAttestation<E> {
-    pub fn attesting_indices_iter(&self) -> Iter<'_, u64> {
-        self.attesting_indices.iter()
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/consensus/types/src/core/chain_spec.rs b/consensus/types/src/core/chain_spec.rs
index cc79d3fc29..e612c8b6db 100644
--- a/consensus/types/src/core/chain_spec.rs
+++ b/consensus/types/src/core/chain_spec.rs
@@ -107,6 +107,8 @@ pub struct ChainSpec {
     pub shard_committee_period: u64,
     pub proposer_reorg_cutoff_bps: u64,
     pub attestation_due_bps: u64,
+    pub attestation_due_bps_gloas: u64,
+    pub payload_attestation_due_bps: u64,
     pub aggregate_due_bps: u64,
     pub sync_message_due_bps: u64,
     pub contribution_due_bps: u64,
@@ -115,6 +117,8 @@ pub struct ChainSpec {
      * Derived time values (computed at startup via `compute_derived_values()`)
      */
     pub unaggregated_attestation_due: Duration,
+    pub unaggregated_attestation_due_gloas: Duration,
+    pub payload_attestation_due: Duration,
     pub aggregate_attestation_due: Duration,
     pub sync_message_due: Duration,
     pub contribution_and_proof_due: Duration,
@@ -877,6 +881,20 @@ impl ChainSpec {
         self.unaggregated_attestation_due
     }
 
+    /// Spec: `get_attestation_due_ms`. Returns the epoch-appropriate threshold.
+    pub fn get_attestation_due<E: EthSpec>(&self, slot: Slot) -> Duration {
+        if self.fork_name_at_slot::<E>(slot).gloas_enabled() {
+            self.unaggregated_attestation_due_gloas
+        } else {
+            self.unaggregated_attestation_due
+        }
+    }
+
+    /// Spec: `get_payload_attestation_due_ms`.
+    pub fn get_payload_attestation_due(&self) -> Duration {
+        self.payload_attestation_due
+    }
+
     /// Get the duration into a slot in which an aggregated attestation is due.
     /// Returns the pre-computed value from `compute_derived_values()`.
     pub fn get_aggregate_attestation_due(&self) -> Duration {
@@ -949,6 +967,12 @@ impl ChainSpec {
         self.unaggregated_attestation_due = self
             .compute_slot_component_duration(self.attestation_due_bps)
             .expect("invalid chain spec: cannot compute unaggregated_attestation_due");
+        self.unaggregated_attestation_due_gloas = self
+            .compute_slot_component_duration(self.attestation_due_bps_gloas)
+            .expect("invalid chain spec: cannot compute unaggregated_attestation_due_gloas");
+        self.payload_attestation_due = self
+            .compute_slot_component_duration(self.payload_attestation_due_bps)
+            .expect("invalid chain spec: cannot compute payload_attestation_due");
         self.aggregate_attestation_due = self
             .compute_slot_component_duration(self.aggregate_due_bps)
             .expect("invalid chain spec: cannot compute aggregate_attestation_due");
@@ -1079,6 +1103,8 @@ impl ChainSpec {
             shard_committee_period: 256,
             proposer_reorg_cutoff_bps: 1667,
             attestation_due_bps: 3333,
+            attestation_due_bps_gloas: 2500,
+            payload_attestation_due_bps: 7500,
             aggregate_due_bps: 6667,
             sync_message_due_bps: 3333,
             contribution_due_bps: 6667,
@@ -1087,6 +1113,8 @@ impl ChainSpec {
              * Derived time values (set by `compute_derived_values()`)
              */
             unaggregated_attestation_due: Duration::from_millis(3999),
+            unaggregated_attestation_due_gloas: Duration::from_millis(3000),
+            payload_attestation_due: Duration::from_millis(9000),
             aggregate_attestation_due: Duration::from_millis(8000),
             sync_message_due: Duration::from_millis(3999),
             contribution_and_proof_due: Duration::from_millis(8000),
@@ -1390,6 +1418,8 @@ impl ChainSpec {
              * Precomputed for 6000ms slot: 3333 bps = 1999ms, 6667 bps = 4000ms
              */
             unaggregated_attestation_due: Duration::from_millis(1999),
+            unaggregated_attestation_due_gloas: Duration::from_millis(1500),
+            payload_attestation_due: Duration::from_millis(4500),
             aggregate_attestation_due: Duration::from_millis(4000),
             sync_message_due: Duration::from_millis(1999),
             contribution_and_proof_due: Duration::from_millis(4000),
@@ -1479,6 +1509,8 @@ impl ChainSpec {
             shard_committee_period: 256,
             proposer_reorg_cutoff_bps: 1667,
             attestation_due_bps: 3333,
+            attestation_due_bps_gloas: 2500,
+            payload_attestation_due_bps: 7500,
             aggregate_due_bps: 6667,
 
             /*
@@ -1486,6 +1518,8 @@ impl ChainSpec {
              * Precomputed for 5000ms slot: 3333 bps = 1666ms, 6667 bps = 3333ms
              */
             unaggregated_attestation_due: Duration::from_millis(1666),
+            unaggregated_attestation_due_gloas: Duration::from_millis(1250),
+            payload_attestation_due: Duration::from_millis(3750),
             aggregate_attestation_due: Duration::from_millis(3333),
             sync_message_due: Duration::from_millis(1666),
             contribution_and_proof_due: Duration::from_millis(3333),
@@ -2062,6 +2096,12 @@ pub struct Config {
     #[serde(default = "default_attestation_due_bps")]
     #[serde(with = "serde_utils::quoted_u64")]
     attestation_due_bps: u64,
+    #[serde(default = "default_attestation_due_bps_gloas")]
+    #[serde(with = "serde_utils::quoted_u64")]
+    attestation_due_bps_gloas: u64,
+    #[serde(default = "default_payload_attestation_due_bps")]
+    #[serde(with = "serde_utils::quoted_u64")]
+    payload_attestation_due_bps: u64,
     #[serde(default = "default_aggregate_due_bps")]
     #[serde(with = "serde_utils::quoted_u64")]
     aggregate_due_bps: u64,
@@ -2288,6 +2328,14 @@ const fn default_attestation_due_bps() -> u64 {
     3333
 }
 
+const fn default_attestation_due_bps_gloas() -> u64 {
+    2500
+}
+
+const fn default_payload_attestation_due_bps() -> u64 {
+    7500
+}
+
 const fn default_aggregate_due_bps() -> u64 {
     6667
 }
@@ -2539,6 +2587,8 @@ impl Config {
 
             proposer_reorg_cutoff_bps: spec.proposer_reorg_cutoff_bps,
             attestation_due_bps: spec.attestation_due_bps,
+            attestation_due_bps_gloas: spec.attestation_due_bps_gloas,
+            payload_attestation_due_bps: spec.payload_attestation_due_bps,
             aggregate_due_bps: spec.aggregate_due_bps,
             sync_message_due_bps: spec.sync_message_due_bps,
             contribution_due_bps: spec.contribution_due_bps,
@@ -2632,6 +2682,8 @@ impl Config {
             min_epochs_for_data_column_sidecars_requests,
             proposer_reorg_cutoff_bps,
             attestation_due_bps,
+            attestation_due_bps_gloas,
+            payload_attestation_due_bps,
             aggregate_due_bps,
             sync_message_due_bps,
             contribution_due_bps,
@@ -2731,6 +2783,8 @@ impl Config {
 
             proposer_reorg_cutoff_bps,
             attestation_due_bps,
+            attestation_due_bps_gloas,
+            payload_attestation_due_bps,
             aggregate_due_bps,
             sync_message_due_bps,
             contribution_due_bps,
@@ -3634,11 +3688,9 @@ mod yaml_tests {
         "EIP7928_FORK_VERSION",
         "EIP7928_FORK_EPOCH",
         // Gloas params not yet in Config
-        "ATTESTATION_DUE_BPS_GLOAS",
         "AGGREGATE_DUE_BPS_GLOAS",
         "SYNC_MESSAGE_DUE_BPS_GLOAS",
         "CONTRIBUTION_DUE_BPS_GLOAS",
-        "PAYLOAD_ATTESTATION_DUE_BPS",
         "MAX_REQUEST_PAYLOADS",
         // Gloas fork choice params not yet in Config
         "REORG_HEAD_WEIGHT_THRESHOLD",
diff --git a/consensus/types/src/core/eth_spec.rs b/consensus/types/src/core/eth_spec.rs
index 36d61fbbf9..4159091f5d 100644
--- a/consensus/types/src/core/eth_spec.rs
+++ b/consensus/types/src/core/eth_spec.rs
@@ -448,6 +448,11 @@ pub trait EthSpec: 'static + Default + Sync + Send + Clone + Debug + PartialEq +
     fn payload_timely_threshold() -> usize {
         Self::PTCSize::to_usize() / 2
     }
+
+    /// Returns the `DATA_AVAILABILITY_TIMELY_THRESHOLD` constant (PTC_SIZE / 2).
+    fn data_availability_timely_threshold() -> usize {
+        Self::PTCSize::to_usize() / 2
+    }
 }
 
 /// Macro to inherit some type values from another EthSpec.
diff --git a/testing/ef_tests/src/cases/fork_choice.rs b/testing/ef_tests/src/cases/fork_choice.rs
index 22e8453e14..06f204ab01 100644
--- a/testing/ef_tests/src/cases/fork_choice.rs
+++ b/testing/ef_tests/src/cases/fork_choice.rs
@@ -1018,7 +1018,7 @@ impl<E: EthSpec> Tester<E> {
             .chain
             .canonical_head
             .fork_choice_write_lock()
-            .on_execution_payload(block_root);
+            .on_valid_payload_envelope_received(block_root);
 
         if valid {
             result.map_err(|e| {