diff --git a/beacon_node/beacon_chain/src/beacon_chain.rs b/beacon_node/beacon_chain/src/beacon_chain.rs
index 76e7b217ce..7a8a747d8c 100644
--- a/beacon_node/beacon_chain/src/beacon_chain.rs
+++ b/beacon_node/beacon_chain/src/beacon_chain.rs
@@ -70,7 +70,8 @@ use crate::payload_envelope_streamer::{EnvelopeRequestSource, launch_payload_env
 use crate::pending_payload_cache::PendingPayloadCache;
 use crate::pending_payload_cache::{
     Availability as PayloadAvailability,
-    DataColumnReconstructionResult as DataColumnReconstructionResultGloas, PendingPayloadBid,
+    DataColumnReconstructionResult as DataColumnReconstructionResultGloas,
+    signed_payload_bid_from_block,
 };
 use crate::pending_payload_envelopes::PendingPayloadEnvelopes;
 use crate::persisted_beacon_chain::PersistedBeaconChain;
@@ -3431,7 +3432,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                     .put_kzg_verified_custody_data_columns(
                         block_root,
                         bid,
-                        merge_result.full_columns.clone(),
+                        &merge_result.full_columns,
                     )
                     .map_err(BlockError::from)?;
                 self.process_payload_availability(slot, availability, || Ok(()))
@@ -3818,7 +3819,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
             let block = execution_pending.block.block_cloned();
             if block.fork_name_unchecked().gloas_enabled() {
-                let bid = PendingPayloadBid::from_block(block.as_ref())?;
+                let bid = signed_payload_bid_from_block(block.as_ref())?;
                 chain
                     .pending_payload_cache
                     .init_pending_bid(block_root, bid);
@@ -4107,7 +4108,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                         .ok_or(BlockError::EnvelopeBlockRootUnknown { block_root })?;
                     let availability = self
                         .pending_payload_cache
-                        .put_kzg_verified_custody_data_columns(block_root, bid, data_columns)
+                        .put_kzg_verified_custody_data_columns(block_root, bid, &data_columns)
                         .map_err(BlockError::from)?;
                     Ok(self
                         .process_payload_availability(slot, availability, || Ok(()))
diff --git a/beacon_node/beacon_chain/src/data_column_verification.rs b/beacon_node/beacon_chain/src/data_column_verification.rs
index 7036ecdc70..9a48477ce8 100644
--- a/beacon_node/beacon_chain/src/data_column_verification.rs
+++ b/beacon_node/beacon_chain/src/data_column_verification.rs
@@ -9,7 +9,7 @@ use crate::kzg_utils::{
 use crate::observed_data_sidecars::{
     Error as ObservedDataSidecarsError, ObservationKey, ObservationStrategy, Observe,
 };
-use crate::pending_payload_cache::PendingPayloadBid;
+use crate::pending_payload_cache::signed_payload_bid_from_block;
 use crate::{BeaconChain, BeaconChainError, BeaconChainTypes, metrics};
 use educe::Educe;
 use fork_choice::ProtoBlock;
@@ -32,7 +32,7 @@ use types::data::{
 use types::{
     BeaconStateError, ChainSpec, DataColumnSidecar, DataColumnSidecarFulu, DataColumnSidecarGloas,
     DataColumnSubnetId, EthSpec, Hash256, KzgCommitment, PartialDataColumnSidecarRef,
-    SignedBeaconBlockHeader, Slot,
+    SignedBeaconBlockHeader, SignedExecutionPayloadBid, Slot,
 };
 
 /// An error occurred while validating a gossip data column.
@@ -373,15 +373,15 @@ impl<T: BeaconChainTypes, O: ObservationStrategy> GossipVerifiedDataColumn<T, O>
                 )?;
             }
             DataColumnSidecar::Gloas(_) => {
-                let kzg_commitments = load_gloas_payload_bid(column_sidecar.block_root(), chain)?
-                    .ok_or(GossipDataColumnError::BlockRootUnknown {
+                let bid = load_gloas_payload_bid(column_sidecar.block_root(), chain)?.ok_or(
+                    GossipDataColumnError::BlockRootUnknown {
                         block_root: column_sidecar.block_root(),
                         slot: column_sidecar.slot(),
-                    })?
-                    .blob_kzg_commitments;
+                    },
+                )?;
                 verify_data_column_sidecar_with_commitments_len(
                     &column_sidecar,
-                    kzg_commitments.len(),
+                    bid.message.blob_kzg_commitments.len(),
                     &chain.spec,
                 )?;
             }
@@ -1091,12 +1091,13 @@ pub fn validate_data_column_sidecar_for_gossip_gloas<
     verify_slot_greater_than_latest_finalized_slot(chain, column_slot)?;
     verify_is_unknown_sidecar(chain, &data_column)?;
 
-    let kzg_commitments = load_gloas_payload_bid(data_column.block_root(), chain)?
-        .ok_or(GossipDataColumnError::BlockRootUnknown {
+    let bid = load_gloas_payload_bid(data_column.block_root(), chain)?.ok_or(
+        GossipDataColumnError::BlockRootUnknown {
             block_root: data_column.block_root(),
             slot: column_slot,
-        })?
-        .blob_kzg_commitments;
+        },
+    )?;
+    let kzg_commitments = &bid.message.blob_kzg_commitments;
     verify_data_column_sidecar_with_commitments_len(
         &data_column,
         kzg_commitments.len(),
@@ -1306,13 +1307,13 @@ fn verify_data_column_sidecar_with_commitments_len<E: EthSpec>(
 pub(crate) fn load_gloas_payload_bid<T: BeaconChainTypes>(
     block_root: Hash256,
     chain: &BeaconChain<T>,
-) -> Result<Option<PendingPayloadBid<T::EthSpec>>, BeaconChainError> {
+) -> Result<Option<Arc<SignedExecutionPayloadBid<T::EthSpec>>>, BeaconChainError> {
     if let Some(bid) = chain.pending_payload_cache.get_bid(&block_root) {
         return Ok(Some(bid));
     }
 
     let bid = if let Some(block) = chain.early_attester_cache.get_block(block_root) {
-        PendingPayloadBid::from_block(block.as_ref()).map_err(BeaconChainError::BeaconStateError)?
+        signed_payload_bid_from_block(block.as_ref()).map_err(BeaconChainError::BeaconStateError)?
     } else {
         match chain
             .store
@@ -1320,10 +1321,10 @@ pub(crate) fn load_gloas_payload_bid<T: BeaconChainTypes>(
             .map_err(BeaconChainError::DBError)?
         {
             Some(DatabaseBlock::Full(block)) => {
-                PendingPayloadBid::from_block(&block).map_err(BeaconChainError::BeaconStateError)?
+                signed_payload_bid_from_block(&block).map_err(BeaconChainError::BeaconStateError)?
             }
             Some(DatabaseBlock::Blinded(block)) => {
-                PendingPayloadBid::from_block(&block).map_err(BeaconChainError::BeaconStateError)?
+                signed_payload_bid_from_block(&block).map_err(BeaconChainError::BeaconStateError)?
             }
             None => {
                 return Ok(None);
diff --git a/beacon_node/beacon_chain/src/pending_payload_cache/mod.rs b/beacon_node/beacon_chain/src/pending_payload_cache/mod.rs
index b490b11522..0d5900c374 100644
--- a/beacon_node/beacon_chain/src/pending_payload_cache/mod.rs
+++ b/beacon_node/beacon_chain/src/pending_payload_cache/mod.rs
@@ -68,8 +68,9 @@ use crate::metrics::{
     KZG_DATA_COLUMN_RECONSTRUCTION_ATTEMPTS, KZG_DATA_COLUMN_RECONSTRUCTION_FAILURES,
 };
 use crate::observed_data_sidecars::ObservationStrategy;
-pub use pending_components::PendingPayloadBid;
+pub use pending_components::signed_payload_bid_from_block;
 use pending_components::{PendingComponents, ReconstructColumnsDecision};
+use types::SignedExecutionPayloadBid;
 use types::new_non_zero_usize;
 
 /// The LRU Cache stores `PendingComponents`, which store the block root, the execution payload bid, and its associated column data.
@@ -153,7 +154,7 @@ impl<T: BeaconChainTypes> PendingPayloadCache<T> {
         block_root: Hash256,
     ) -> Option<DataColumnSidecarList<T::EthSpec>> {
         self.peek_pending_components(&block_root, |components| {
-            components.map(|c| c.get_cached_data_columns(block_root))
+            components.map(|c| c.get_cached_data_columns())
         })
     }
 
@@ -165,8 +166,11 @@ impl<T: BeaconChainTypes> PendingPayloadCache<T> {
         })
     }
 
-    /// Return the cached Gloas payload bid metadata for `block_root`, if present.
-    pub fn get_bid(&self, block_root: &Hash256) -> Option<PendingPayloadBid<T::EthSpec>> {
+    /// Return the cached Gloas payload bid for `block_root`, if present.
+    pub fn get_bid(
+        &self,
+        block_root: &Hash256,
+    ) -> Option<Arc<SignedExecutionPayloadBid<T::EthSpec>>> {
         self.peek_pending_components(block_root, |components| {
             components.map(|components| components.bid.clone())
         })
@@ -201,7 +205,7 @@ impl<T: BeaconChainTypes> PendingPayloadCache<T> {
     /// Insert an executed payload envelope into the cache and performs an availability check
     pub fn put_executed_payload_envelope(
         &self,
-        bid: PendingPayloadBid<T::EthSpec>,
+        bid: Arc<SignedExecutionPayloadBid<T::EthSpec>>,
         executed_envelope: AvailabilityPendingExecutedEnvelope<T::EthSpec>,
     ) -> Result<Availability<T::EthSpec>, AvailabilityCheckError> {
         let epoch = executed_envelope.envelope.epoch();
@@ -226,9 +230,13 @@ impl<T: BeaconChainTypes> PendingPayloadCache<T> {
     }
 
     /// Initialize pending components for a block's Gloas bid.
-    pub fn init_pending_bid(&self, block_root: Hash256, bid: PendingPayloadBid<T::EthSpec>) {
+    pub fn init_pending_bid(
+        &self,
+        block_root: Hash256,
+        bid: Arc<SignedExecutionPayloadBid<T::EthSpec>>,
+    ) {
         let mut write_lock = self.availability_cache.write();
-        write_lock.get_or_insert_mut(block_root, || PendingComponents::empty(block_root, bid));
+        write_lock.get_or_insert_mut(block_root, || PendingComponents::new(block_root, bid));
     }
 
     /// Perform KZG verification on RPC custody columns and insert them into the cache.
@@ -237,17 +245,17 @@ impl<T: BeaconChainTypes> PendingPayloadCache<T> {
     pub fn put_rpc_custody_columns(
         &self,
         block_root: Hash256,
-        bid: PendingPayloadBid<T::EthSpec>,
+        bid: Arc<SignedExecutionPayloadBid<T::EthSpec>>,
         custody_columns: DataColumnSidecarList<T::EthSpec>,
     ) -> Result<Availability<T::EthSpec>, AvailabilityCheckError> {
         let kzg_verified_columns = KzgVerifiedDataColumn::from_batch_with_scoring_and_commitments(
             custody_columns,
-            bid.blob_kzg_commitments.as_ref(),
+            bid.message.blob_kzg_commitments.as_ref(),
             &self.kzg,
         )
         .map_err(AvailabilityCheckError::InvalidColumn)?;
 
-        let epoch = bid.slot.epoch(T::EthSpec::slots_per_epoch());
+        let epoch = bid.message.slot.epoch(T::EthSpec::slots_per_epoch());
         let sampling_columns = self
             .custody_context
             .sampling_columns_for_epoch(epoch, &self.spec);
@@ -257,7 +265,7 @@ impl<T: BeaconChainTypes> PendingPayloadCache<T> {
             .map(KzgVerifiedCustodyDataColumn::from_asserted_custody)
             .collect::<Vec<_>>();
 
-        self.put_kzg_verified_custody_data_columns(block_root, bid, verified_custody_columns)
+        self.put_kzg_verified_custody_data_columns(block_root, bid, &verified_custody_columns)
     }
 
     /// Perform KZG verification on gossip verified custody columns and insert them into the cache.
@@ -266,10 +274,10 @@ impl<T: BeaconChainTypes> PendingPayloadCache<T> {
     pub fn put_gossip_verified_data_columns<O: ObservationStrategy>(
         &self,
         block_root: Hash256,
-        bid: PendingPayloadBid<T::EthSpec>,
+        bid: Arc<SignedExecutionPayloadBid<T::EthSpec>>,
         data_columns: Vec<GossipVerifiedDataColumn<T, O>>,
     ) -> Result<Availability<T::EthSpec>, AvailabilityCheckError> {
-        let epoch = bid.slot.epoch(T::EthSpec::slots_per_epoch());
+        let epoch = bid.message.slot.epoch(T::EthSpec::slots_per_epoch());
         let sampling_columns = self
             .custody_context
             .sampling_columns_for_epoch(epoch, &self.spec);
@@ -279,7 +287,7 @@ impl<T: BeaconChainTypes> PendingPayloadCache<T> {
             .map(|c| KzgVerifiedCustodyDataColumn::from_asserted_custody(c.into_inner()))
             .collect::<Vec<_>>();
 
-        self.put_kzg_verified_custody_data_columns(block_root, bid, custody_columns)
+        self.put_kzg_verified_custody_data_columns(block_root, bid, &custody_columns)
     }
 
     /// Insert KZG verified columns into the cache.
@@ -287,8 +295,8 @@ impl<T: BeaconChainTypes> PendingPayloadCache<T> {
     pub fn put_kzg_verified_custody_data_columns(
         &self,
         block_root: Hash256,
-        bid: PendingPayloadBid<T::EthSpec>,
-        kzg_verified_data_columns: Vec<KzgVerifiedCustodyDataColumn<T::EthSpec>>,
+        bid: Arc<SignedExecutionPayloadBid<T::EthSpec>>,
+        kzg_verified_data_columns: &[KzgVerifiedCustodyDataColumn<T::EthSpec>],
     ) -> Result<Availability<T::EthSpec>, AvailabilityCheckError> {
         let pending_components =
             self.get_pending_components(block_root, bid, |pending_components| {
@@ -312,7 +320,7 @@ impl<T: BeaconChainTypes> PendingPayloadCache<T> {
     pub fn reconstruct_data_columns(
         &self,
         block_root: &Hash256,
-        bid: PendingPayloadBid<T::EthSpec>,
+        bid: Arc<SignedExecutionPayloadBid<T::EthSpec>>,
     ) -> Result<DataColumnReconstructionResult<T::EthSpec>, AvailabilityCheckError> {
         let verified_data_columns = match self.check_and_set_reconstruction_started(block_root) {
             ReconstructColumnsDecision::Yes(verified_data_columns) => verified_data_columns,
@@ -378,7 +386,7 @@ impl<T: BeaconChainTypes> PendingPayloadCache<T> {
         self.put_kzg_verified_custody_data_columns(
             *block_root,
             bid,
-            data_columns_to_import_and_publish.clone(),
+            &data_columns_to_import_and_publish,
         )
         .map(|availability| {
             DataColumnReconstructionResult::Success((
@@ -413,9 +421,7 @@ impl<T: BeaconChainTypes> PendingPayloadCache<T> {
         pending_components: MappedRwLockReadGuard<'_, PendingComponents<T::EthSpec>>,
         num_expected_columns: usize,
     ) -> Result<Availability<T::EthSpec>, AvailabilityCheckError> {
-        if let Some(available_envelope) =
-            pending_components.make_available(block_root, num_expected_columns)?
-        {
+        if let Some(available_envelope) = pending_components.make_available(num_expected_columns)? {
             // Explicitly drop read lock before acquiring write lock
             drop(pending_components);
             if let Some(components) = self.availability_cache.write().get_mut(&block_root) {
@@ -439,7 +445,7 @@ impl<T: BeaconChainTypes> PendingPayloadCache<T> {
     fn get_pending_components<F>(
         &self,
         block_root: Hash256,
-        bid: PendingPayloadBid<T::EthSpec>,
+        bid: Arc<SignedExecutionPayloadBid<T::EthSpec>>,
         update_fn: F,
     ) -> Result<MappedRwLockReadGuard<'_, PendingComponents<T::EthSpec>>, AvailabilityCheckError>
     where
@@ -449,7 +455,7 @@ impl<T: BeaconChainTypes> PendingPayloadCache<T> {
 
         {
             let pending_components = write_lock
-                .get_or_insert_mut(block_root, || PendingComponents::empty(block_root, bid));
+                .get_or_insert_mut(block_root, || PendingComponents::new(block_root, bid));
             update_fn(pending_components)?
         }
 
@@ -499,7 +505,7 @@ impl<T: BeaconChainTypes> PendingPayloadCache<T> {
         }
 
         pending_components.reconstruction_started = true;
-        ReconstructColumnsDecision::Yes(pending_components.get_cached_data_columns(*block_root))
+        ReconstructColumnsDecision::Yes(pending_components.get_cached_data_columns())
     }
 
     /// This could mean some invalid data columns made it through to the `DataAvailabilityChecker`.
@@ -741,12 +747,16 @@ mod data_availability_checker_tests {
         spec: &ChainSpec,
         num_blobs: NumBlobs,
         seed: u64,
-    ) -> (PendingPayloadBid<E>, Hash256, DataColumnSidecarList<E>) {
+    ) -> (
+        Arc<SignedExecutionPayloadBid<E>>,
+        Hash256,
+        DataColumnSidecarList<E>,
+    ) {
         let mut rng = StdRng::seed_from_u64(seed);
         let (block, data_columns) =
             generate_rand_block_and_data_columns::<E>(ForkName::Gloas, num_blobs, &mut rng, spec);
         let block_root = block.canonical_root();
-        let bid = PendingPayloadBid::from_block(&block).expect("should get payload bid");
+        let bid = signed_payload_bid_from_block(&block).expect("should get payload bid");
         cache.init_pending_bid(block_root, bid.clone());
         (bid, block_root, data_columns)
     }
@@ -756,7 +766,7 @@ mod data_availability_checker_tests {
         let (harness, cache, _path) = setup().await;
         let (bid, block_root, data_columns) =
             init_block(&cache, &harness.spec, NumBlobs::Number(1), RNG_SEED);
-        let epoch = bid.slot.epoch(E::slots_per_epoch());
+        let epoch = bid.message.slot.epoch(E::slots_per_epoch());
         let sampling_cols = cache
             .custody_context()
             .sampling_columns_for_epoch(epoch, &harness.spec);
@@ -790,7 +800,7 @@ mod data_availability_checker_tests {
         let (bid, block_root, data_columns) =
             init_block(&cache, &harness.spec, NumBlobs::Number(1), RNG_SEED);
 
-        let epoch = bid.slot.epoch(E::slots_per_epoch());
+        let epoch = bid.message.slot.epoch(E::slots_per_epoch());
         let num_sampling_columns = cache
             .custody_context()
             .sampling_columns_for_epoch(epoch, &harness.spec)
@@ -849,7 +859,7 @@ mod data_availability_checker_tests {
         let (bid, block_root, data_columns) =
             init_block(&cache, &harness.spec, NumBlobs::Number(1), RNG_SEED);
 
-        let epoch = bid.slot.epoch(E::slots_per_epoch());
+        let epoch = bid.message.slot.epoch(E::slots_per_epoch());
         let sampling_cols = cache
             .custody_context()
             .sampling_columns_for_epoch(epoch, &harness.spec);
@@ -899,7 +909,7 @@ mod data_availability_checker_tests {
         let (harness, cache, _path) = setup().await;
         let (bid, block_root, data_columns) =
             init_block(&cache, &harness.spec, NumBlobs::Number(1), RNG_SEED);
-        let block_epoch = bid.slot.epoch(E::slots_per_epoch());
+        let block_epoch = bid.message.slot.epoch(E::slots_per_epoch());
         let column = data_columns.first().cloned().expect("should have column");
 
         cache
diff --git a/beacon_node/beacon_chain/src/pending_payload_cache/pending_column.rs b/beacon_node/beacon_chain/src/pending_payload_cache/pending_column.rs
index 021983f8a5..5ace1e25b2 100644
--- a/beacon_node/beacon_chain/src/pending_payload_cache/pending_column.rs
+++ b/beacon_node/beacon_chain/src/pending_payload_cache/pending_column.rs
@@ -1,37 +1,28 @@
 use kzg::KzgProof;
 use ssz_types::VariableList;
+use std::collections::HashMap;
 use std::sync::Arc;
 use types::{Cell, ColumnIndex, DataColumnSidecar, DataColumnSidecarGloas, EthSpec, Hash256, Slot};
 
-#[derive(Clone)]
+#[derive(Clone, Default)]
 pub struct PendingColumn<E: EthSpec> {
-    cells: Vec<Option<(Cell<E>, KzgProof)>>,
+    cells: HashMap<usize, (Cell<E>, KzgProof)>,
 }
 
 impl<E: EthSpec> PendingColumn<E> {
-    pub fn new_with_capacity(blobs: usize) -> Self {
-        Self {
-            cells: vec![None; blobs],
-        }
-    }
-
     pub fn insert(&mut self, index: usize, cell: &Cell<E>, proof: &KzgProof) {
-        if let Some(existing_cell) = self.cells.get_mut(index)
-            && existing_cell.is_none()
-        {
-            *existing_cell = Some((cell.clone(), *proof));
-        }
+        self.cells
+            .entry(index)
+            .or_insert_with(|| (cell.clone(), *proof));
     }
 
     pub fn cell_matches(&self, index: usize, cell: &Cell<E>, proof: &KzgProof) -> Option<bool> {
-        self.cells
-            .get(index)?
-            .as_ref()
-            .map(|(c, p)| c == cell && p == proof)
+        let (c, p) = self.cells.get(&index)?;
+        Some(c == cell && p == proof)
     }
 
     pub fn is_complete(&self, blob_count: usize) -> bool {
-        self.cells.len() == blob_count && self.cells.iter().all(|cell| cell.is_some())
+        (0..blob_count).all(|i| self.cells.contains_key(&i))
     }
 
     pub fn try_to_sidecar(
@@ -41,17 +32,11 @@ impl<E: EthSpec> PendingColumn<E> {
         beacon_block_root: Hash256,
         blob_count: usize,
     ) -> Option<Arc<DataColumnSidecar<E>>> {
-        if self.cells.len() != blob_count {
-            return None;
-        }
+        let mut column = Vec::with_capacity(blob_count);
+        let mut kzg_proofs = Vec::with_capacity(blob_count);
 
-        let mut column = Vec::with_capacity(self.cells.len());
-        let mut kzg_proofs = Vec::with_capacity(self.cells.len());
-
-        for cell in self.cells.iter() {
-            let Some((cell, proof)) = cell else {
-                return None;
-            };
+        for i in 0..blob_count {
+            let (cell, proof) = self.cells.get(&i)?;
             // TODO(gloas): we likely want to go and arc all cells
             column.push(cell.clone());
             kzg_proofs.push(*proof);
diff --git a/beacon_node/beacon_chain/src/pending_payload_cache/pending_components.rs b/beacon_node/beacon_chain/src/pending_payload_cache/pending_components.rs
index 779470e3ce..de351cd527 100644
--- a/beacon_node/beacon_chain/src/pending_payload_cache/pending_components.rs
+++ b/beacon_node/beacon_chain/src/pending_payload_cache/pending_components.rs
@@ -10,26 +10,23 @@ use std::sync::Arc;
 use tracing::{Span, debug, debug_span};
 use types::DataColumnSidecar;
 use types::{
-    AbstractExecPayload, BeaconStateError, ColumnIndex, Epoch, EthSpec, Hash256, KzgCommitments,
-    SignedBeaconBlock, Slot,
+    AbstractExecPayload, BeaconStateError, ColumnIndex, Epoch, EthSpec, Hash256, SignedBeaconBlock,
+    SignedExecutionPayloadBid,
 };
 
-#[derive(Clone)]
-pub struct PendingPayloadBid<E: EthSpec> {
-    pub slot: Slot,
-    pub blob_kzg_commitments: KzgCommitments<E>,
-}
-
-impl<E: EthSpec> PendingPayloadBid<E> {
-    pub fn from_block<Payload: AbstractExecPayload<E>>(
-        block: &SignedBeaconBlock<E, Payload>,
-    ) -> Result<Self, BeaconStateError> {
-        let signed_bid = block.message().body().signed_execution_payload_bid()?;
-        Ok(Self {
-            slot: block.slot(),
-            blob_kzg_commitments: signed_bid.message.blob_kzg_commitments.clone(),
-        })
-    }
+/// Extract the signed execution payload bid from a Gloas block as a shareable `Arc`.
+///
+/// Returns `Err` if the block is not a Gloas block.
+pub fn signed_payload_bid_from_block<E: EthSpec, P: AbstractExecPayload<E>>(
+    block: &SignedBeaconBlock<E, P>,
+) -> Result<Arc<SignedExecutionPayloadBid<E>>, BeaconStateError> {
+    Ok(Arc::new(
+        block
+            .message()
+            .body()
+            .signed_execution_payload_bid()?
+            .clone(),
+    ))
 }
 
 /// This represents the components of a payload pending data availability.
@@ -37,7 +34,8 @@ impl<E: EthSpec> PendingPayloadBid<E> {
 /// The columns are all gossip and kzg verified.
 /// The payload is considered "available" when all required columns are received.
 pub struct PendingComponents<E: EthSpec> {
-    pub bid: PendingPayloadBid<E>,
+    pub block_root: Hash256,
+    pub bid: Arc<SignedExecutionPayloadBid<E>>,
     /// a cached post executed payload envelope
     pub envelope: Option<AvailabilityPendingExecutedEnvelope<E>>,
     pub verified_data_columns: HashMap<ColumnIndex, PendingColumn<E>>,
@@ -47,18 +45,18 @@ pub struct PendingComponents<E: EthSpec> {
 
 impl<E: EthSpec> PendingComponents<E> {
     pub fn num_blobs_expected(&self) -> usize {
-        self.bid.blob_kzg_commitments.len()
+        self.bid.message.blob_kzg_commitments.len()
     }
 
     /// Returns the completed custody columns
-    pub fn get_cached_data_columns(&self, block_root: Hash256) -> Vec<Arc<DataColumnSidecar<E>>> {
+    pub fn get_cached_data_columns(&self) -> Vec<Arc<DataColumnSidecar<E>>> {
         self.verified_data_columns
             .iter()
             .filter_map(|(col_idx, col)| {
                 col.try_to_sidecar(
                     *col_idx,
-                    self.bid.slot,
-                    block_root,
+                    self.bid.message.slot,
+                    self.block_root,
                     self.num_blobs_expected(),
                 )
             })
@@ -77,17 +75,16 @@ impl<E: EthSpec> PendingComponents<E> {
     }
 
     /// Merges a given set of data columns into the cache.
-    pub(crate) fn merge_data_columns<I: IntoIterator<Item = KzgVerifiedCustodyDataColumn<E>>>(
+    pub(crate) fn merge_data_columns(
         &mut self,
-        kzg_verified_data_columns: I,
+        kzg_verified_data_columns: &[KzgVerifiedCustodyDataColumn<E>],
     ) -> Result<(), AvailabilityCheckError> {
-        let num_blobs_expected = self.num_blobs_expected();
         for data_column in kzg_verified_data_columns {
             let data_column = data_column.as_data_column();
             let col = self
                 .verified_data_columns
                 .entry(*data_column.index())
-                .or_insert_with(|| PendingColumn::new_with_capacity(num_blobs_expected));
+                .or_default();
             for (cell_idx, (cell, proof)) in data_column
                 .column()
                 .iter()
@@ -121,7 +118,6 @@ impl<E: EthSpec> PendingComponents<E> {
     /// Returns `Some` if the envelope and all required data columns have been received.
     pub fn make_available(
         &self,
-        block_hash: Hash256,
         num_expected_columns: usize,
     ) -> Result<Option<AvailableExecutedEnvelope<E>>, AvailabilityCheckError> {
         // Check if the payload has been received and executed
@@ -154,17 +150,7 @@ impl<E: EthSpec> PendingComponents<E> {
                         debug!("All data columns received, data is available");
                     });
 
-                    self.verified_data_columns
-                        .iter()
-                        .filter_map(|(col_idx, col)| {
-                            col.try_to_sidecar(
-                                *col_idx,
-                                self.bid.slot,
-                                block_hash,
-                                self.num_blobs_expected(),
-                            )
-                        })
-                        .collect()
+                    self.get_cached_data_columns()
                 }
                 Ordering::Less => {
                     // Not enough data columns received yet
@@ -187,11 +173,12 @@ impl<E: EthSpec> PendingComponents<E> {
         }))
     }
 
-    /// Returns an empty `PendingComponents` object with the given block root.
-    pub fn empty(block_root: Hash256, bid: PendingPayloadBid<E>) -> Self {
+    /// Constructs a fresh `PendingComponents` with no envelope and no columns yet.
+    pub fn new(block_root: Hash256, bid: Arc<SignedExecutionPayloadBid<E>>) -> Self {
         let span = debug_span!(parent: None, "lh_pending_components", %block_root);
         let _guard = span.clone().entered();
         Self {
+            block_root,
             bid,
             envelope: None,
             verified_data_columns: HashMap::new(),
@@ -202,7 +189,7 @@ impl<E: EthSpec> PendingComponents<E> {
 
     /// Returns the epoch of the bid or first data column, if available.
     pub fn epoch(&self) -> Epoch {
-        self.bid.slot.epoch(E::slots_per_epoch())
+        self.bid.message.slot.epoch(E::slots_per_epoch())
     }
 
     pub fn status_str(&self, num_expected_columns: usize) -> String {
diff --git a/beacon_node/beacon_chain/tests/events.rs b/beacon_node/beacon_chain/tests/events.rs
index 26d004d069..b45083bfad 100644
--- a/beacon_node/beacon_chain/tests/events.rs
+++ b/beacon_node/beacon_chain/tests/events.rs
@@ -1,6 +1,5 @@
 use beacon_chain::blob_verification::GossipVerifiedBlob;
 use beacon_chain::data_column_verification::GossipVerifiedDataColumn;
-use beacon_chain::pending_payload_cache::PendingPayloadBid;
 use beacon_chain::test_utils::{
     BeaconChainHarness, fork_name_from_env, generate_data_column_sidecars_from_block, test_spec,
 };
@@ -12,8 +11,8 @@ use types::data::FixedBlobSidecarList;
 use types::test_utils::TestRandom;
 use types::{
     BlobSidecar, DataColumnSidecar, DataColumnSidecarFulu, DataColumnSidecarGloas, Domain, EthSpec,
-    KzgCommitments, MinimalEthSpec, PayloadAttestationData, PayloadAttestationMessage, SignedRoot,
-    Slot,
+    MinimalEthSpec, PayloadAttestationData, PayloadAttestationMessage, SignedExecutionPayloadBid,
+    SignedRoot, Slot,
 };
 
 type E = MinimalEthSpec;
@@ -88,13 +87,12 @@ async fn data_column_sidecar_event_on_process_gossip_data_column() {
             random_sidecar.index = harness.chain.sampling_columns_for_epoch(epoch)[0];
 
             // For gloas, the bid must be known, e.g. in the pending payload cache
-            harness.chain.pending_payload_cache.init_pending_bid(
-                random_sidecar.beacon_block_root,
-                PendingPayloadBid {
-                    slot: Slot::new(10),
-                    blob_kzg_commitments: KzgCommitments::<E>::empty(),
-                },
-            );
+            let mut bid = SignedExecutionPayloadBid::<E>::empty();
+            bid.message.slot = Slot::new(10);
+            harness
+                .chain
+                .pending_payload_cache
+                .init_pending_bid(random_sidecar.beacon_block_root, Arc::new(bid));
 
             DataColumnSidecar::Gloas(random_sidecar)
         } else {