Address remaining review comments

- pending_column.rs: split try_to_sidecar into is_complete-checked to_sidecar
  with typed PendingColumnError so 'incomplete column' is no longer conflated
  with VariableList size-bound failures (jimmygchen, dapplion).
- pending_components.rs: get_cached_data_columns filters by is_complete first,
  then logs an error if a complete column fails to assemble (dknopik's sanity
  check on filter_map silent drops).
- data_column_verification.rs: add the missing column.slot == bid.slot
  consistency check in validate_data_column_sidecar_for_gossip_gloas, using
  the previously-defined-but-unused BlockSlotMismatch error variant
  (jimmygchen).

beacon_node/beacon_chain/src/data_column_verification.rs

@@ -1097,6 +1097,12 @@ pub fn validate_data_column_sidecar_for_gossip_gloas<
             slot: column_slot,
         },
     )?;
+    if bid.message.slot != column_slot {
+        return Err(GossipDataColumnError::BlockSlotMismatch {
+            block_slot: bid.message.slot,
+            data_column_slot: column_slot,
+        });
+    }
     let kzg_commitments = &bid.message.blob_kzg_commitments;
     verify_data_column_sidecar_with_commitments_len(
         &data_column,

beacon_node/beacon_chain/src/pending_payload_cache/pending_column.rs

@@ -36,24 +36,22 @@ impl<E: EthSpec> PendingColumn<E> {
         self.cells.len() == blob_count && self.cells.iter().all(|cell| cell.is_some())
     }
 
-    pub fn try_to_sidecar(
+    /// Build a `DataColumnSidecar` from the cached cells.
+    ///
+    /// Caller MUST have checked `is_complete(blob_count)` first; this returns `Err` only on the
+    /// (currently theoretically impossible) `VariableList` size-bound failures, which we surface
+    /// as a typed error so the caller can log/metric it instead of silently producing nothing.
+    pub fn to_sidecar(
         &self,
         index: ColumnIndex,
         slot: Slot,
         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);
+    ) -> Result<Arc<DataColumnSidecar<E>>, PendingColumnError> {
+        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;
-            };
+            let (cell, proof) = cell.as_ref().ok_or(PendingColumnError::IncompleteColumn)?;
             // TODO(gloas): we likely want to go and arc all cells
             column.push(cell.clone());
             kzg_proofs.push(*proof);
@@ -61,13 +59,24 @@ impl<E: EthSpec> PendingColumn<E> {
 
         // TODO(gloas): this hard-codes the Gloas sidecar variant. Pass the fork in once
         // post-Gloas variants are introduced (or move construction to a fork-aware helper).
-        Some(Arc::new(DataColumnSidecar::Gloas(DataColumnSidecarGloas {
+        Ok(Arc::new(DataColumnSidecar::Gloas(DataColumnSidecarGloas {
             index,
-            // TODO(gloas): this should not error, but we need to catch it
-            column: VariableList::try_from(column).ok()?,
-            kzg_proofs: VariableList::try_from(kzg_proofs).ok()?,
+            column: VariableList::try_from(column)
+                .map_err(|_| PendingColumnError::ColumnSizeExceedsBound)?,
+            kzg_proofs: VariableList::try_from(kzg_proofs)
+                .map_err(|_| PendingColumnError::ProofsSizeExceedsBound)?,
             slot,
             beacon_block_root,
         })))
     }
 }
+
+/// Errors returned by [`PendingColumn::to_sidecar`]. `IncompleteColumn` should never fire if the
+/// caller checks [`PendingColumn::is_complete`] first; the size-bound variants reflect spec-bound
+/// invariants and should never fire in practice.
+#[derive(Debug, Clone)]
+pub enum PendingColumnError {
+    IncompleteColumn,
+    ColumnSizeExceedsBound,
+    ProofsSizeExceedsBound,
+}

beacon_node/beacon_chain/src/pending_payload_cache/pending_components.rs

@@ -7,7 +7,7 @@ use crate::pending_payload_cache::pending_column::PendingColumn;
 use std::cmp::Ordering;
 use std::collections::HashMap;
 use std::sync::Arc;
-use tracing::{Span, debug, debug_span};
+use tracing::{Span, debug, debug_span, error};
 use types::DataColumnSidecar;
 use types::{
     AbstractExecPayload, BeaconStateError, ColumnIndex, EthSpec, Hash256, SignedBeaconBlock,
@@ -48,18 +48,31 @@ impl<E: EthSpec> PendingComponents<E> {
         self.bid.message.blob_kzg_commitments.len()
     }
 
-    /// Returns the completed custody columns
+    /// Returns the completed custody columns.
+    ///
+    /// Skips columns that are not yet complete and logs an error if a complete column fails to
+    /// build (a spec-bound invariant would have to be violated; should never happen in practice).
     pub fn get_cached_data_columns(&self) -> Vec<Arc<DataColumnSidecar<E>>> {
+        let blob_count = self.num_blobs_expected();
+        let slot = self.bid.message.slot;
+        let block_root = self.block_root;
         self.verified_data_columns
             .iter()
-            .filter_map(|(col_idx, col)| {
-                col.try_to_sidecar(
-                    *col_idx,
-                    self.bid.message.slot,
-                    self.block_root,
-                    self.num_blobs_expected(),
-                )
-            })
+            .filter(|(_, col)| col.is_complete(blob_count))
+            .filter_map(
+                |(col_idx, col)| match col.to_sidecar(*col_idx, slot, block_root) {
+                    Ok(sidecar) => Some(sidecar),
+                    Err(e) => {
+                        error!(
+                            ?e,
+                            column_index = %col_idx,
+                            ?block_root,
+                            "Failed to build sidecar for complete column"
+                        );
+                        None
+                    }
+                },
+            )
             .collect()
     }