diff --git a/beacon_node/beacon_chain/src/block_verification_types.rs b/beacon_node/beacon_chain/src/block_verification_types.rs
index 5978e97c4d9..b20820e0f52 100644
--- a/beacon_node/beacon_chain/src/block_verification_types.rs
+++ b/beacon_node/beacon_chain/src/block_verification_types.rs
@@ -9,8 +9,8 @@ use std::fmt::{Debug, Formatter};
 use std::sync::Arc;
 use types::blob_sidecar::BlobIdentifier;
 use types::{
-    BeaconBlockRef, BeaconState, BlindedPayload, BlobSidecarList, Epoch, EthSpec, Hash256,
-    SignedBeaconBlock, SignedBeaconBlockHeader, Slot,
+    BeaconBlockRef, BeaconState, BlindedPayload, BlobSidecarList, Epoch, EthSpec, ForkName,
+    Hash256, RuntimeVariableList, SignedBeaconBlock, SignedBeaconBlockHeader, Slot,
 };
 
 /// A block that has been received over RPC. It has 2 internal variants:
@@ -28,40 +28,114 @@ use types::{
 /// over rpc do not contain the proposer signature for dos resistance.
 #[derive(Clone, Educe)]
 #[educe(Hash(bound(E: EthSpec)))]
-pub struct RpcBlock<E: EthSpec> {
+pub enum RpcBlock<E: EthSpec> {
+    Available(AvailableRpcBlock<E>),
+    MaybeAvailable(MaybeAvailableRpcBlock<E>),
+}
+
+#[derive(Clone, Educe)]
+#[educe(Hash(bound(E: EthSpec)))]
+pub struct MaybeAvailableRpcBlock<E: EthSpec> {
+    block_root: Hash256,
+    block: RpcBlockInner<E>,
+}
+
+impl<E: EthSpec> MaybeAvailableRpcBlock<E> {
+    #[allow(clippy::type_complexity)]
+    pub fn deconstruct(
+        self,
+    ) -> (
+        Hash256,
+        Arc<SignedBeaconBlock<E>>,
+        Option<BlobSidecarList<E>>,
+        Option<CustodyDataColumnList<E>>,
+    ) {
+        let block_root = self.block_root;
+        match self.block {
+            RpcBlockInner::Block(block) => (block_root, block.clone(), None, None),
+            RpcBlockInner::BlockAndBlobs(block, blobs) => {
+                (block_root, block.clone(), Some(blobs.clone()), None)
+            }
+            RpcBlockInner::BlockAndCustodyColumns(block, data_columns) => {
+                (block_root, block.clone(), None, Some(data_columns.clone()))
+            }
+        }
+    }
+}
+
+#[derive(Clone, Educe)]
+#[educe(Hash(bound(E: EthSpec)))]
+pub struct AvailableRpcBlock<E: EthSpec> {
     block_root: Hash256,
     block: RpcBlockInner<E>,
 }
 
+impl<E: EthSpec> AvailableRpcBlock<E> {
+    #[allow(clippy::type_complexity)]
+    pub fn deconstruct(
+        self,
+    ) -> (
+        Hash256,
+        Arc<SignedBeaconBlock<E>>,
+        Option<BlobSidecarList<E>>,
+        Option<CustodyDataColumnList<E>>,
+    ) {
+        let block_root = self.block_root;
+        match self.block {
+            RpcBlockInner::Block(block) => (block_root, block.clone(), None, None),
+            RpcBlockInner::BlockAndBlobs(block, blobs) => {
+                (block_root, block.clone(), Some(blobs.clone()), None)
+            }
+            RpcBlockInner::BlockAndCustodyColumns(block, data_columns) => {
+                (block_root, block.clone(), None, Some(data_columns.clone()))
+            }
+        }
+    }
+}
+
 impl<E: EthSpec> Debug for RpcBlock<E> {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(f, "RpcBlock({:?})", self.block_root)
+        write!(f, "RpcBlock({:?})", self.block_root())
     }
 }
 
 impl<E: EthSpec> RpcBlock<E> {
     pub fn block_root(&self) -> Hash256 {
-        self.block_root
+        match self {
+            RpcBlock::Available(available_rpc_block) => available_rpc_block.block_root,
+            RpcBlock::MaybeAvailable(maybe_available_rpc_block) => {
+                maybe_available_rpc_block.block_root
+            }
+        }
+    }
+
+    fn rpc_block_inner(&self) -> &RpcBlockInner<E> {
+        match self {
+            RpcBlock::Available(available_rpc_block) => &available_rpc_block.block,
+            RpcBlock::MaybeAvailable(maybe_available_rpc_block) => &maybe_available_rpc_block.block,
+        }
     }
 
     pub fn as_block(&self) -> &SignedBeaconBlock<E> {
-        match &self.block {
-            RpcBlockInner::Block(block) => block,
-            RpcBlockInner::BlockAndBlobs(block, _) => block,
-            RpcBlockInner::BlockAndCustodyColumns(block, _) => block,
+        match self.rpc_block_inner() {
+            RpcBlockInner::Block(signed_beacon_block) => signed_beacon_block,
+            RpcBlockInner::BlockAndBlobs(signed_beacon_block, _) => signed_beacon_block,
+            RpcBlockInner::BlockAndCustodyColumns(signed_beacon_block, _) => signed_beacon_block,
         }
     }
 
     pub fn block_cloned(&self) -> Arc<SignedBeaconBlock<E>> {
-        match &self.block {
-            RpcBlockInner::Block(block) => block.clone(),
-            RpcBlockInner::BlockAndBlobs(block, _) => block.clone(),
-            RpcBlockInner::BlockAndCustodyColumns(block, _) => block.clone(),
+        match self.rpc_block_inner() {
+            RpcBlockInner::Block(signed_beacon_block) => signed_beacon_block.clone(),
+            RpcBlockInner::BlockAndBlobs(signed_beacon_block, _) => signed_beacon_block.clone(),
+            RpcBlockInner::BlockAndCustodyColumns(signed_beacon_block, _) => {
+                signed_beacon_block.clone()
+            }
         }
     }
 
     pub fn blobs(&self) -> Option<&BlobSidecarList<E>> {
-        match &self.block {
+        match self.rpc_block_inner() {
             RpcBlockInner::Block(_) => None,
             RpcBlockInner::BlockAndBlobs(_, blobs) => Some(blobs),
             RpcBlockInner::BlockAndCustodyColumns(_, _) => None,
@@ -69,7 +143,7 @@ impl<E: EthSpec> RpcBlock<E> {
     }
 
     pub fn custody_columns(&self) -> Option<&CustodyDataColumnList<E>> {
-        match &self.block {
+        match self.rpc_block_inner() {
             RpcBlockInner::Block(_) => None,
             RpcBlockInner::BlockAndBlobs(_, _) => None,
             RpcBlockInner::BlockAndCustodyColumns(_, data_columns) => Some(data_columns),
@@ -93,34 +167,18 @@ enum RpcBlockInner<E: EthSpec> {
     BlockAndCustodyColumns(Arc<SignedBeaconBlock<E>>, CustodyDataColumnList<E>),
 }
 
-impl<E: EthSpec> RpcBlock<E> {
-    /// Constructs a `Block` variant.
-    pub fn new_without_blobs(
-        block_root: Option<Hash256>,
-        block: Arc<SignedBeaconBlock<E>>,
-    ) -> Self {
-        let block_root = block_root.unwrap_or_else(|| get_block_root(&block));
-
-        Self {
-            block_root,
-            block: RpcBlockInner::Block(block),
-        }
-    }
-
+impl<E: EthSpec> RpcBlockInner<E> {
     /// Constructs a new `BlockAndBlobs` variant after making consistency
-    /// checks between the provided blocks and blobs. This struct makes no
+    /// checks between the provided block and blobs. This struct makes no
     /// guarantees about whether blobs should be present, only that they are
     /// consistent with the block. An empty list passed in for `blobs` is
     /// viewed the same as `None` passed in.
-    pub fn new(
-        block_root: Option<Hash256>,
+    fn new_block_and_blobs(
         block: Arc<SignedBeaconBlock<E>>,
         blobs: Option<BlobSidecarList<E>>,
     ) -> Result<Self, AvailabilityCheckError> {
-        let block_root = block_root.unwrap_or_else(|| get_block_root(&block));
         // Treat empty blob lists as if they are missing.
         let blobs = blobs.filter(|b| !b.is_empty());
-
         if let (Some(blobs), Ok(block_commitments)) = (
             blobs.as_ref(),
             block.message().body().blob_kzg_commitments(),
@@ -138,37 +196,96 @@ impl<E: EthSpec> RpcBlock<E> {
                 }
             }
         }
-        let inner = match blobs {
-            Some(blobs) => RpcBlockInner::BlockAndBlobs(block, blobs),
-            None => RpcBlockInner::Block(block),
+
+        let blobs = if let Some(blobs) = blobs {
+            blobs
+        } else {
+            RuntimeVariableList::new(vec![], E::max_blob_commitments_per_block())?
         };
-        Ok(Self {
+
+        Ok(Self::BlockAndBlobs(block, blobs))
+    }
+
+    /// Constructs a new `BlockAndCustodyColumns` variant after making consistency
+    /// checks between the provided block and columns. This struct makes no
+    /// guarantees about whether columns should be present, only that they are
+    /// consistent with the block. An empty list passed in for `custody_columns` is
+    /// viewed the same as `None` passed in.
+    fn new_block_and_columns(
+        block: Arc<SignedBeaconBlock<E>>,
+        custody_columns: Option<Vec<CustodyDataColumn<E>>>,
+    ) -> Result<Self, AvailabilityCheckError> {
+        let columns = if let Some(custody_columns) = custody_columns {
+            if block.num_expected_blobs() > 0 && custody_columns.is_empty() {
+                // The number of required custody columns is out of scope here.
+                return Err(AvailabilityCheckError::MissingCustodyColumns);
+            }
+            // Treat empty data column lists as if they are missing.
+            if !custody_columns.is_empty() {
+                VariableList::new(custody_columns)?
+            } else {
+                VariableList::new(vec![])?
+            }
+        } else {
+            VariableList::new(vec![])?
+        };
+
+        Ok(RpcBlockInner::BlockAndCustodyColumns(block, columns))
+    }
+
+    /// Constructs a new `Block` variant.
+    pub fn new(
+        block: Arc<SignedBeaconBlock<E>>,
+        blobs: Option<BlobSidecarList<E>>,
+        columns: Option<Vec<CustodyDataColumn<E>>>,
+    ) -> Result<Self, AvailabilityCheckError> {
+        match block.fork_name_unchecked() {
+            ForkName::Base | ForkName::Altair | ForkName::Bellatrix | ForkName::Capella => {
+                Ok(RpcBlockInner::Block(block))
+            }
+            ForkName::Deneb | ForkName::Electra => Self::new_block_and_blobs(block, blobs),
+            ForkName::Fulu | ForkName::Gloas => Self::new_block_and_columns(block, columns),
+        }
+    }
+}
+
+impl<E: EthSpec> RpcBlock<E> {
+    /// Constructs an `RpcBlock::MaybeAvailable`. Consistency checks for this variant must be made
+    /// before indicating that the block is fully available. An empty list or `None` passed in for
+    /// `blobs` or `columns` could mean that node may have not received blobs for this column yet.
+    pub fn new_maybe_available(
+        block_root: Option<Hash256>,
+        block: Arc<SignedBeaconBlock<E>>,
+        blobs: Option<BlobSidecarList<E>>,
+        columns: Option<Vec<CustodyDataColumn<E>>>,
+    ) -> Result<Self, AvailabilityCheckError> {
+        let block_root = block_root.unwrap_or_else(|| get_block_root(&block));
+        let rpc_block_inner = RpcBlockInner::new(block, blobs, columns)?;
+
+        Ok(RpcBlock::MaybeAvailable(MaybeAvailableRpcBlock {
             block_root,
-            block: inner,
-        })
+            block: rpc_block_inner,
+        }))
     }
 
-    pub fn new_with_custody_columns(
+    /// Constructs an `RpcBlock::Available` variant. All consistency checks
+    /// for the `RpcBlock` should be made before calling this function. Successfully constructing
+    /// an `Available` variant indicates that the block is fully available from the nodes perspective.
+    /// An empty list or `None` passed in for `blobs` or `columns` means that there
+    /// are no blobs for the given block.
+    pub fn new_available(
         block_root: Option<Hash256>,
         block: Arc<SignedBeaconBlock<E>>,
-        custody_columns: Vec<CustodyDataColumn<E>>,
+        blobs: Option<BlobSidecarList<E>>,
+        columns: Option<Vec<CustodyDataColumn<E>>>,
     ) -> Result<Self, AvailabilityCheckError> {
         let block_root = block_root.unwrap_or_else(|| get_block_root(&block));
+        let rpc_block_inner = RpcBlockInner::new(block, blobs, columns)?;
 
-        if block.num_expected_blobs() > 0 && custody_columns.is_empty() {
-            // The number of required custody columns is out of scope here.
-            return Err(AvailabilityCheckError::MissingCustodyColumns);
-        }
-        // Treat empty data column lists as if they are missing.
-        let inner = if !custody_columns.is_empty() {
-            RpcBlockInner::BlockAndCustodyColumns(block, VariableList::new(custody_columns)?)
-        } else {
-            RpcBlockInner::Block(block)
-        };
-        Ok(Self {
+        Ok(RpcBlock::Available(AvailableRpcBlock {
             block_root,
-            block: inner,
-        })
+            block: rpc_block_inner,
+        }))
     }
 
     #[allow(clippy::type_complexity)]
@@ -181,22 +298,24 @@ impl<E: EthSpec> RpcBlock<E> {
         Option<CustodyDataColumnList<E>>,
     ) {
         let block_root = self.block_root();
-        match self.block {
-            RpcBlockInner::Block(block) => (block_root, block, None, None),
-            RpcBlockInner::BlockAndBlobs(block, blobs) => (block_root, block, Some(blobs), None),
+        match self.rpc_block_inner() {
+            RpcBlockInner::Block(block) => (block_root, block.clone(), None, None),
+            RpcBlockInner::BlockAndBlobs(block, blobs) => {
+                (block_root, block.clone(), Some(blobs.clone()), None)
+            }
             RpcBlockInner::BlockAndCustodyColumns(block, data_columns) => {
-                (block_root, block, None, Some(data_columns))
+                (block_root, block.clone(), None, Some(data_columns.clone()))
             }
         }
     }
     pub fn n_blobs(&self) -> usize {
-        match &self.block {
+        match self.rpc_block_inner() {
             RpcBlockInner::Block(_) | RpcBlockInner::BlockAndCustodyColumns(_, _) => 0,
             RpcBlockInner::BlockAndBlobs(_, blobs) => blobs.len(),
         }
     }
     pub fn n_data_columns(&self) -> usize {
-        match &self.block {
+        match self.rpc_block_inner() {
             RpcBlockInner::Block(_) | RpcBlockInner::BlockAndBlobs(_, _) => 0,
             RpcBlockInner::BlockAndCustodyColumns(_, data_columns) => data_columns.len(),
         }
@@ -500,14 +619,14 @@ impl<E: EthSpec> AsBlock<E> for RpcBlock<E> {
         self.as_block().message()
     }
     fn as_block(&self) -> &SignedBeaconBlock<E> {
-        match &self.block {
+        match &self.rpc_block_inner() {
             RpcBlockInner::Block(block) => block,
             RpcBlockInner::BlockAndBlobs(block, _) => block,
             RpcBlockInner::BlockAndCustodyColumns(block, _) => block,
         }
     }
     fn block_cloned(&self) -> Arc<SignedBeaconBlock<E>> {
-        match &self.block {
+        match &self.rpc_block_inner() {
             RpcBlockInner::Block(block) => block.clone(),
             RpcBlockInner::BlockAndBlobs(block, _) => block.clone(),
             RpcBlockInner::BlockAndCustodyColumns(block, _) => block.clone(),
diff --git a/beacon_node/beacon_chain/src/data_availability_checker.rs b/beacon_node/beacon_chain/src/data_availability_checker.rs
index 3e859456b18..45e826a4356 100644
--- a/beacon_node/beacon_chain/src/data_availability_checker.rs
+++ b/beacon_node/beacon_chain/src/data_availability_checker.rs
@@ -2,7 +2,8 @@ use crate::blob_verification::{
     GossipVerifiedBlob, KzgVerifiedBlob, KzgVerifiedBlobList, verify_kzg_for_blob_list,
 };
 use crate::block_verification_types::{
-    AvailabilityPendingExecutedBlock, AvailableExecutedBlock, RpcBlock,
+    AvailabilityPendingExecutedBlock, AvailableExecutedBlock, AvailableRpcBlock,
+    MaybeAvailableRpcBlock, RpcBlock,
 };
 use crate::data_availability_checker::overflow_lru_cache::{
     DataAvailabilityCheckerInner, ReconstructColumnsDecision,
@@ -366,18 +367,15 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
             .remove_pre_execution_block(block_root);
     }
 
-    /// Verifies kzg commitments for an RpcBlock, returns a `MaybeAvailableBlock` that may
-    /// include the fully available block.
-    ///
-    /// WARNING: This function assumes all required blobs are already present, it does NOT
-    ///          check if there are any missing blobs.
-    pub fn verify_kzg_for_rpc_block(
+    pub fn verify_kzg_for_maybe_available_rpc_block(
         &self,
-        block: RpcBlock<T::EthSpec>,
+        block: MaybeAvailableRpcBlock<T::EthSpec>,
     ) -> Result<MaybeAvailableBlock<T::EthSpec>, AvailabilityCheckError> {
         let (block_root, block, blobs, data_columns) = block.deconstruct();
         if self.blobs_required_for_block(&block) {
-            return if let Some(blob_list) = blobs {
+            return if let Some(blob_list) = blobs
+                && blob_list.len() == block.num_expected_blobs()
+            {
                 verify_kzg_for_blob_list(blob_list.iter(), &self.kzg)
                     .map_err(AvailabilityCheckError::InvalidBlobs)?;
                 Ok(MaybeAvailableBlock::Available(AvailableBlock {
@@ -392,7 +390,9 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
             };
         }
         if self.data_columns_required_for_block(&block) {
-            return if let Some(data_column_list) = data_columns.as_ref() {
+            return if let Some(data_column_list) = data_columns.as_ref()
+                && data_column_list.len() == T::EthSpec::number_of_columns()
+            {
                 verify_kzg_for_data_column_list(
                     data_column_list
                         .iter()
@@ -426,6 +426,81 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
         }))
     }
 
+    pub fn verify_kzg_for_available_rpc_block(
+        &self,
+        block: AvailableRpcBlock<T::EthSpec>,
+    ) -> Result<MaybeAvailableBlock<T::EthSpec>, AvailabilityCheckError> {
+        let (block_root, block, blobs, data_columns) = block.deconstruct();
+
+        if self.blobs_required_for_block(&block) {
+            return if let Some(blob_list) = blobs {
+                verify_kzg_for_blob_list(blob_list.iter(), &self.kzg)
+                    .map_err(AvailabilityCheckError::InvalidBlobs)?;
+                Ok(MaybeAvailableBlock::Available(AvailableBlock {
+                    block_root,
+                    block,
+                    blob_data: AvailableBlockData::Blobs(blob_list),
+                    blobs_available_timestamp: None,
+                    spec: self.spec.clone(),
+                }))
+            } else {
+                return Err(AvailabilityCheckError::MissingBlobs);
+            };
+        }
+
+        if self.data_columns_required_for_block(&block) {
+            return if let Some(data_column_list) = data_columns {
+                verify_kzg_for_data_column_list(
+                    data_column_list
+                        .iter()
+                        .map(|custody_column| custody_column.as_data_column()),
+                    &self.kzg,
+                )
+                .map_err(AvailabilityCheckError::InvalidColumn)?;
+                Ok(MaybeAvailableBlock::Available(AvailableBlock {
+                    block_root,
+                    block,
+                    blob_data: AvailableBlockData::DataColumns(
+                        data_column_list
+                            .into_iter()
+                            .map(|d| d.clone_arc())
+                            .collect(),
+                    ),
+                    blobs_available_timestamp: None,
+                    spec: self.spec.clone(),
+                }))
+            } else {
+                return Err(AvailabilityCheckError::MissingCustodyColumns);
+            };
+        }
+        Ok(MaybeAvailableBlock::Available(AvailableBlock {
+            block_root,
+            block,
+            blob_data: AvailableBlockData::NoData,
+            blobs_available_timestamp: None,
+            spec: self.spec.clone(),
+        }))
+    }
+
+    /// Verifies kzg commitments for an RpcBlock, returns a `MaybeAvailableBlock` that may
+    /// include the fully available block.
+    ///
+    /// WARNING: This function assumes all required blobs are already present, it does NOT
+    ///          check if there are any missing blobs.
+    pub fn verify_kzg_for_rpc_block(
+        &self,
+        block: RpcBlock<T::EthSpec>,
+    ) -> Result<MaybeAvailableBlock<T::EthSpec>, AvailabilityCheckError> {
+        match block {
+            RpcBlock::Available(available_rpc_block) => {
+                self.verify_kzg_for_available_rpc_block(available_rpc_block)
+            }
+            RpcBlock::MaybeAvailable(maybe_available_rpc_block) => {
+                self.verify_kzg_for_maybe_available_rpc_block(maybe_available_rpc_block)
+            }
+        }
+    }
+
     /// Checks if a vector of blocks are available. Returns a vector of `MaybeAvailableBlock`
     /// This is more efficient than calling `verify_kzg_for_rpc_block` in a loop as it does
     /// all kzg verification at once
@@ -469,10 +544,17 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
         }
 
         for block in blocks {
+            let is_available = match block {
+                RpcBlock::Available(_) => true,
+                RpcBlock::MaybeAvailable(_) => false,
+            };
+
             let (block_root, block, blobs, data_columns) = block.deconstruct();
 
             let maybe_available_block = if self.blobs_required_for_block(&block) {
-                if let Some(blobs) = blobs {
+                if let Some(blobs) = blobs
+                    && blobs.len() == block.num_expected_blobs()
+                {
                     MaybeAvailableBlock::Available(AvailableBlock {
                         block_root,
                         block,
@@ -481,10 +563,14 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
                         spec: self.spec.clone(),
                     })
                 } else {
+                    if is_available {
+                        return Err(AvailabilityCheckError::MissingBlobs);
+                    }
                     MaybeAvailableBlock::AvailabilityPending { block_root, block }
                 }
             } else if self.data_columns_required_for_block(&block) {
-                if let Some(data_columns) = data_columns {
+                if let Some(data_columns) = data_columns
+                {
                     MaybeAvailableBlock::Available(AvailableBlock {
                         block_root,
                         block,
@@ -495,6 +581,9 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
                         spec: self.spec.clone(),
                     })
                 } else {
+                    if is_available {
+                        return Err(AvailabilityCheckError::MissingCustodyColumns);
+                    }
                     MaybeAvailableBlock::AvailabilityPending { block_root, block }
                 }
             } else {
@@ -1074,7 +1163,7 @@ mod test {
                         .collect::<Vec<_>>()
                 };
 
-                RpcBlock::new_with_custody_columns(None, Arc::new(block), custody_columns)
+                RpcBlock::new_available(None, Arc::new(block), None, Some(custody_columns))
                     .expect("should create RPC block with custody columns")
             })
             .collect::<Vec<_>>();
diff --git a/beacon_node/beacon_chain/src/data_availability_checker/error.rs b/beacon_node/beacon_chain/src/data_availability_checker/error.rs
index c9efb7a4149..1d9b8820d36 100644
--- a/beacon_node/beacon_chain/src/data_availability_checker/error.rs
+++ b/beacon_node/beacon_chain/src/data_availability_checker/error.rs
@@ -22,6 +22,7 @@ pub enum Error {
     BlockReplayError(state_processing::BlockReplayError),
     RebuildingStateCaches(BeaconStateError),
     SlotClockError,
+    InvalidFork,
 }
 
 #[derive(PartialEq, Eq)]
@@ -44,7 +45,8 @@ impl Error {
             | Error::ParentStateMissing(_)
             | Error::BlockReplayError(_)
             | Error::RebuildingStateCaches(_)
-            | Error::SlotClockError => ErrorCategory::Internal,
+            | Error::SlotClockError
+            | Error::InvalidFork => ErrorCategory::Internal,
             Error::InvalidBlobs { .. }
             | Error::InvalidColumn { .. }
             | Error::ReconstructColumnsError { .. }
diff --git a/beacon_node/beacon_chain/src/test_utils.rs b/beacon_node/beacon_chain/src/test_utils.rs
index 05d67e4504a..f915fb6027f 100644
--- a/beacon_node/beacon_chain/src/test_utils.rs
+++ b/beacon_node/beacon_chain/src/test_utils.rs
@@ -2368,7 +2368,7 @@ where
         self.set_current_slot(slot);
         let (block, blob_items) = block_contents;
 
-        let rpc_block = self.build_rpc_block_from_blobs(block_root, block, blob_items)?;
+        let rpc_block = self.build_rpc_block_from_blobs(block_root, block, blob_items, true)?;
         let block_hash: SignedBeaconBlockHash = self
             .chain
             .process_block(
@@ -2392,7 +2392,7 @@ where
         let (block, blob_items) = block_contents;
 
         let block_root = block.canonical_root();
-        let rpc_block = self.build_rpc_block_from_blobs(block_root, block, blob_items)?;
+        let rpc_block = self.build_rpc_block_from_blobs(block_root, block, blob_items, true)?;
         let block_hash: SignedBeaconBlockHash = self
             .chain
             .process_block(
@@ -2423,7 +2423,8 @@ where
             .blob_kzg_commitments()
             .is_ok_and(|c| !c.is_empty());
         if !has_blobs {
-            return RpcBlock::new_without_blobs(Some(block_root), block);
+            // TODO(can refactor so we dont call new_avail a bunch)
+            return RpcBlock::new_available(Some(block_root), block, None, None).unwrap();
         }
 
         // Blobs are stored as data columns from Fulu (PeerDAS)
@@ -2433,10 +2434,10 @@ where
                 .into_iter()
                 .map(CustodyDataColumn::from_asserted_custody)
                 .collect::<Vec<_>>();
-            RpcBlock::new_with_custody_columns(Some(block_root), block, custody_columns).unwrap()
+            RpcBlock::new_available(Some(block_root), block, None, Some(custody_columns)).unwrap()
         } else {
             let blobs = self.chain.get_blobs(&block_root).unwrap().blobs();
-            RpcBlock::new(Some(block_root), block, blobs).unwrap()
+            RpcBlock::new_available(Some(block_root), block, blobs, None).unwrap()
         }
     }
 
@@ -2446,6 +2447,7 @@ where
         block_root: Hash256,
         block: Arc<SignedBeaconBlock<E, FullPayload<E>>>,
         blob_items: Option<(KzgProofs<E>, BlobsList<E>)>,
+        is_available: bool,
     ) -> Result<RpcBlock<E>, BlockError> {
         Ok(if self.spec.is_peer_das_enabled_for_epoch(block.epoch()) {
             let epoch = block.slot().epoch(E::slots_per_epoch());
@@ -2460,9 +2462,16 @@ where
                     .filter(|d| sampling_columns.contains(&d.index))
                     .map(CustodyDataColumn::from_asserted_custody)
                     .collect::<Vec<_>>();
-                RpcBlock::new_with_custody_columns(Some(block_root), block, columns)?
+                // TODO(can combine these and clean it up)
+                if is_available {
+                    RpcBlock::new_available(Some(block_root), block, None, Some(columns))?
+                } else {
+                    RpcBlock::new_maybe_available(Some(block_root), block, None, Some(columns))?
+                }
+            } else if is_available {
+                RpcBlock::new_available(Some(block_root), block, None, None)?
             } else {
-                RpcBlock::new_without_blobs(Some(block_root), block)
+                RpcBlock::new_maybe_available(Some(block_root), block, None, None)?
             }
         } else {
             let blobs = blob_items
@@ -2471,7 +2480,11 @@ where
                 })
                 .transpose()
                 .unwrap();
-            RpcBlock::new(Some(block_root), block, blobs)?
+            if is_available {
+                RpcBlock::new_available(Some(block_root), block, blobs, None)?
+            } else {
+                RpcBlock::new_maybe_available(Some(block_root), block, blobs, None)?
+            }
         })
     }
 
diff --git a/beacon_node/beacon_chain/tests/blob_verification.rs b/beacon_node/beacon_chain/tests/blob_verification.rs
index c42a2828c01..dfb03b3fd2f 100644
--- a/beacon_node/beacon_chain/tests/blob_verification.rs
+++ b/beacon_node/beacon_chain/tests/blob_verification.rs
@@ -76,7 +76,7 @@ async fn rpc_blobs_with_invalid_header_signature() {
     // Process the block without blobs so that it doesn't become available.
     harness.advance_slot();
     let rpc_block = harness
-        .build_rpc_block_from_blobs(block_root, signed_block.clone(), None)
+        .build_rpc_block_from_blobs(block_root, signed_block.clone(), None, false)
         .unwrap();
     let availability = harness
         .chain
@@ -84,11 +84,12 @@ async fn rpc_blobs_with_invalid_header_signature() {
             block_root,
             rpc_block,
             NotifyExecutionLayer::Yes,
-            BlockImportSource::RangeSync,
+            BlockImportSource::Lookup,
             || Ok(()),
         )
         .await
         .unwrap();
+
     assert_eq!(
         availability,
         AvailabilityProcessingStatus::MissingComponents(slot, block_root)
@@ -113,6 +114,8 @@ async fn rpc_blobs_with_invalid_header_signature() {
         .process_rpc_blobs(slot, block_root, blob_sidecars)
         .await
         .unwrap_err();
+
+    println!("{:?}", err);
     assert!(matches!(
         err,
         BlockError::InvalidSignature(InvalidSignature::ProposerSignature)
diff --git a/beacon_node/beacon_chain/tests/block_verification.rs b/beacon_node/beacon_chain/tests/block_verification.rs
index 881885cef23..5076639d843 100644
--- a/beacon_node/beacon_chain/tests/block_verification.rs
+++ b/beacon_node/beacon_chain/tests/block_verification.rs
@@ -146,12 +146,12 @@ fn build_rpc_block(
 ) -> RpcBlock<E> {
     match data_sidecars {
         Some(DataSidecars::Blobs(blobs)) => {
-            RpcBlock::new(None, block, Some(blobs.clone())).unwrap()
+            RpcBlock::new_available(None, block, Some(blobs.clone()), None).unwrap()
         }
         Some(DataSidecars::DataColumns(columns)) => {
-            RpcBlock::new_with_custody_columns(None, block, columns.clone()).unwrap()
+            RpcBlock::new_available(None, block, None, Some(columns.clone())).unwrap()
         }
-        None => RpcBlock::new_without_blobs(None, block),
+        None => RpcBlock::new_available(None, block, None, None).unwrap(),
     }
 }
 
@@ -368,10 +368,13 @@ async fn chain_segment_non_linear_parent_roots() {
 
     let (mut block, signature) = blocks[3].as_block().clone().deconstruct();
     *block.parent_root_mut() = Hash256::zero();
-    blocks[3] = RpcBlock::new_without_blobs(
+    blocks[3] = RpcBlock::new_available(
         None,
         Arc::new(SignedBeaconBlock::from_block(block, signature)),
-    );
+        None,
+        None,
+    )
+    .unwrap();
 
     assert!(
         matches!(
@@ -404,10 +407,13 @@ async fn chain_segment_non_linear_slots() {
         .collect();
     let (mut block, signature) = blocks[3].as_block().clone().deconstruct();
     *block.slot_mut() = Slot::new(0);
-    blocks[3] = RpcBlock::new_without_blobs(
+    blocks[3] = RpcBlock::new_available(
         None,
         Arc::new(SignedBeaconBlock::from_block(block, signature)),
-    );
+        None,
+        None,
+    )
+    .unwrap();
 
     assert!(
         matches!(
@@ -430,10 +436,13 @@ async fn chain_segment_non_linear_slots() {
         .collect();
     let (mut block, signature) = blocks[3].as_block().clone().deconstruct();
     *block.slot_mut() = blocks[2].slot();
-    blocks[3] = RpcBlock::new_without_blobs(
+    blocks[3] = RpcBlock::new_available(
         None,
         Arc::new(SignedBeaconBlock::from_block(block, signature)),
-    );
+        None,
+        None,
+    )
+    .unwrap();
 
     assert!(
         matches!(
@@ -567,7 +576,8 @@ async fn invalid_signature_gossip_block() {
             .into_block_error()
             .expect("should import all blocks prior to the one being tested");
         let signed_block = SignedBeaconBlock::from_block(block, junk_signature());
-        let rpc_block = RpcBlock::new_without_blobs(None, Arc::new(signed_block));
+        let rpc_block =
+            RpcBlock::new_maybe_available(None, Arc::new(signed_block), None, None).unwrap();
         let process_res = harness
             .chain
             .process_block(
@@ -1568,7 +1578,8 @@ async fn add_base_block_to_altair_chain() {
     ));
 
     // Ensure that it would be impossible to import via `BeaconChain::process_block`.
-    let base_rpc_block = RpcBlock::new_without_blobs(None, Arc::new(base_block.clone()));
+    let base_rpc_block =
+        RpcBlock::new_maybe_available(None, Arc::new(base_block.clone()), None, None).unwrap();
     assert!(matches!(
         harness
             .chain
@@ -1592,7 +1603,9 @@ async fn add_base_block_to_altair_chain() {
         harness
             .chain
             .process_chain_segment(
-                vec![RpcBlock::new_without_blobs(None, Arc::new(base_block))],
+                vec![
+                    RpcBlock::new_maybe_available(None, Arc::new(base_block), None, None).unwrap()
+                ],
                 NotifyExecutionLayer::Yes,
             )
             .await,
@@ -1705,7 +1718,8 @@ async fn add_altair_block_to_base_chain() {
     ));
 
     // Ensure that it would be impossible to import via `BeaconChain::process_block`.
-    let altair_rpc_block = RpcBlock::new_without_blobs(None, Arc::new(altair_block.clone()));
+    let altair_rpc_block =
+        RpcBlock::new_maybe_available(None, Arc::new(altair_block.clone()), None, None).unwrap();
     assert!(matches!(
         harness
             .chain
@@ -1729,7 +1743,10 @@ async fn add_altair_block_to_base_chain() {
         harness
             .chain
             .process_chain_segment(
-                vec![RpcBlock::new_without_blobs(None, Arc::new(altair_block))],
+                vec![
+                    RpcBlock::new_maybe_available(None, Arc::new(altair_block), None, None)
+                        .unwrap()
+                ],
                 NotifyExecutionLayer::Yes
             )
             .await,
@@ -1792,7 +1809,8 @@ async fn import_duplicate_block_unrealized_justification() {
     // Create two verified variants of the block, representing the same block being processed in
     // parallel.
     let notify_execution_layer = NotifyExecutionLayer::Yes;
-    let rpc_block = RpcBlock::new_without_blobs(Some(block_root), block.clone());
+    let rpc_block =
+        RpcBlock::new_maybe_available(Some(block_root), block.clone(), None, None).unwrap();
     let verified_block1 = rpc_block
         .clone()
         .into_execution_pending_block(block_root, chain, notify_execution_layer)
diff --git a/beacon_node/beacon_chain/tests/column_verification.rs b/beacon_node/beacon_chain/tests/column_verification.rs
index 229ae1e1998..0dbe939839f 100644
--- a/beacon_node/beacon_chain/tests/column_verification.rs
+++ b/beacon_node/beacon_chain/tests/column_verification.rs
@@ -80,7 +80,7 @@ async fn rpc_columns_with_invalid_header_signature() {
     // Process the block without blobs so that it doesn't become available.
     harness.advance_slot();
     let rpc_block = harness
-        .build_rpc_block_from_blobs(block_root, signed_block.clone(), None)
+        .build_rpc_block_from_blobs(block_root, signed_block.clone(), None, false)
         .unwrap();
     let availability = harness
         .chain
diff --git a/beacon_node/beacon_chain/tests/payload_invalidation.rs b/beacon_node/beacon_chain/tests/payload_invalidation.rs
index 5bd43835e33..b47c5abc164 100644
--- a/beacon_node/beacon_chain/tests/payload_invalidation.rs
+++ b/beacon_node/beacon_chain/tests/payload_invalidation.rs
@@ -685,7 +685,8 @@ async fn invalidates_all_descendants() {
     assert_eq!(fork_parent_state.slot(), fork_parent_slot);
     let ((fork_block, _), _fork_post_state) =
         rig.harness.make_block(fork_parent_state, fork_slot).await;
-    let fork_rpc_block = RpcBlock::new_without_blobs(None, fork_block.clone());
+    let fork_rpc_block =
+        RpcBlock::new_maybe_available(None, fork_block.clone(), None, None).unwrap();
     let fork_block_root = rig
         .harness
         .chain
@@ -787,7 +788,8 @@ async fn switches_heads() {
     let ((fork_block, _), _fork_post_state) =
         rig.harness.make_block(fork_parent_state, fork_slot).await;
     let fork_parent_root = fork_block.parent_root();
-    let fork_rpc_block = RpcBlock::new_without_blobs(None, fork_block.clone());
+    let fork_rpc_block =
+        RpcBlock::new_maybe_available(None, fork_block.clone(), None, None).unwrap();
     let fork_block_root = rig
         .harness
         .chain
@@ -1059,7 +1061,7 @@ async fn invalid_parent() {
     ));
 
     // Ensure the block built atop an invalid payload is invalid for import.
-    let rpc_block = RpcBlock::new_without_blobs(None, block.clone());
+    let rpc_block = RpcBlock::new_maybe_available(None, block.clone(), None, None).unwrap();
     assert!(matches!(
         rig.harness.chain.process_block(rpc_block.block_root(), rpc_block, NotifyExecutionLayer::Yes, BlockImportSource::Lookup,
             || Ok(()),
@@ -1384,7 +1386,8 @@ async fn recover_from_invalid_head_by_importing_blocks() {
     } = InvalidHeadSetup::new().await;
 
     // Import the fork block, it should become the head.
-    let fork_rpc_block = RpcBlock::new_without_blobs(None, fork_block.clone());
+    let fork_rpc_block =
+        RpcBlock::new_maybe_available(None, fork_block.clone(), None, None).unwrap();
     rig.harness
         .chain
         .process_block(
diff --git a/beacon_node/beacon_chain/tests/store_tests.rs b/beacon_node/beacon_chain/tests/store_tests.rs
index 0733d901fc3..1f80b597a52 100644
--- a/beacon_node/beacon_chain/tests/store_tests.rs
+++ b/beacon_node/beacon_chain/tests/store_tests.rs
@@ -3543,7 +3543,8 @@ async fn process_blocks_and_attestations_for_unaligned_checkpoint() {
     assert_eq!(split.block_root, valid_fork_block.parent_root());
     assert_ne!(split.state_root, unadvanced_split_state_root);
 
-    let invalid_fork_rpc_block = RpcBlock::new_without_blobs(None, invalid_fork_block.clone());
+    let invalid_fork_rpc_block =
+        RpcBlock::new_maybe_available(None, invalid_fork_block.clone(), None, None).unwrap();
     // Applying the invalid block should fail.
     let err = harness
         .chain
@@ -3559,7 +3560,8 @@ async fn process_blocks_and_attestations_for_unaligned_checkpoint() {
     assert!(matches!(err, BlockError::WouldRevertFinalizedSlot { .. }));
 
     // Applying the valid block should succeed, but it should not become head.
-    let valid_fork_rpc_block = RpcBlock::new_without_blobs(None, valid_fork_block.clone());
+    let valid_fork_rpc_block =
+        RpcBlock::new_maybe_available(None, valid_fork_block.clone(), None, None).unwrap();
     harness
         .chain
         .process_block(
diff --git a/beacon_node/http_api/src/publish_blocks.rs b/beacon_node/http_api/src/publish_blocks.rs
index 9671a72da26..2dea79ff376 100644
--- a/beacon_node/http_api/src/publish_blocks.rs
+++ b/beacon_node/http_api/src/publish_blocks.rs
@@ -314,9 +314,16 @@ pub async fn publish_block<T: BeaconChainTypes, B: IntoGossipVerifiedBlock<T>>(
                 slot = %block.slot(),
                 "Block previously seen"
             );
+            let Ok(rpc_block) =
+                RpcBlock::new_maybe_available(Some(block_root), block.clone(), None, None)
+            else {
+                return Err(warp_utils::reject::custom_bad_request(
+                    "Unable to construct rpc block".to_string(),
+                ));
+            };
             let import_result = Box::pin(chain.process_block(
                 block_root,
-                RpcBlock::new_without_blobs(Some(block_root), block.clone()),
+                rpc_block,
                 NotifyExecutionLayer::Yes,
                 BlockImportSource::HttpApi,
                 publish_fn,
diff --git a/beacon_node/network/src/network_beacon_processor/tests.rs b/beacon_node/network/src/network_beacon_processor/tests.rs
index d83059ad278..7892f3d9edf 100644
--- a/beacon_node/network/src/network_beacon_processor/tests.rs
+++ b/beacon_node/network/src/network_beacon_processor/tests.rs
@@ -398,7 +398,13 @@ impl TestRig {
         self.network_beacon_processor
             .send_rpc_beacon_block(
                 block_root,
-                RpcBlock::new_without_blobs(Some(block_root), self.next_block.clone()),
+                RpcBlock::new_maybe_available(
+                    Some(block_root),
+                    self.next_block.clone(),
+                    None,
+                    None,
+                )
+                .unwrap(),
                 std::time::Duration::default(),
                 BlockProcessType::SingleBlock { id: 0 },
             )
@@ -410,7 +416,13 @@ impl TestRig {
         self.network_beacon_processor
             .send_rpc_beacon_block(
                 block_root,
-                RpcBlock::new_without_blobs(Some(block_root), self.next_block.clone()),
+                RpcBlock::new_maybe_available(
+                    Some(block_root),
+                    self.next_block.clone(),
+                    None,
+                    None,
+                )
+                .unwrap(),
                 std::time::Duration::default(),
                 BlockProcessType::SingleBlock { id: 1 },
             )
diff --git a/beacon_node/network/src/sync/block_sidecar_coupling.rs b/beacon_node/network/src/sync/block_sidecar_coupling.rs
index 01929cbf906..df7d5bbdca4 100644
--- a/beacon_node/network/src/sync/block_sidecar_coupling.rs
+++ b/beacon_node/network/src/sync/block_sidecar_coupling.rs
@@ -315,7 +315,7 @@ impl<E: EthSpec> RangeBlockComponentsRequest<E> {
                 CouplingError::BlobPeerFailure("Blobs returned exceeds max length".to_string())
             })?;
             responses.push(
-                RpcBlock::new(None, block, Some(blobs))
+                RpcBlock::new_available(None, block, Some(blobs), None)
                     .map_err(|e| CouplingError::BlobPeerFailure(format!("{e:?}")))?,
             )
         }
@@ -414,11 +414,12 @@ impl<E: EthSpec> RangeBlockComponentsRequest<E> {
                     );
                 }
 
-                RpcBlock::new_with_custody_columns(Some(block_root), block, custody_columns)
+                RpcBlock::new_available(Some(block_root), block, None, Some(custody_columns))
                     .map_err(|e| CouplingError::InternalError(format!("{:?}", e)))?
             } else {
                 // Block has no data, expects zero columns
-                RpcBlock::new_without_blobs(Some(block_root), block)
+                RpcBlock::new_available(Some(block_root), block, None, None)
+                    .map_err(|e| CouplingError::InternalError(format!("{:?}", e)))?
             });
         }
 
diff --git a/beacon_node/network/src/sync/network_context.rs b/beacon_node/network/src/sync/network_context.rs
index 2e0c56db23f..1e932588dfe 100644
--- a/beacon_node/network/src/sync/network_context.rs
+++ b/beacon_node/network/src/sync/network_context.rs
@@ -1606,7 +1606,8 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
             .beacon_processor_if_enabled()
             .ok_or(SendErrorProcessor::ProcessorNotAvailable)?;
 
-        let block = RpcBlock::new_without_blobs(Some(block_root), block);
+        let block = RpcBlock::new_maybe_available(Some(block_root), block, None, None)
+            .map_err(|_| SendErrorProcessor::SendError)?;
 
         debug!(block = ?block_root, id, "Sending block for processing");
         // Lookup sync event safety: If `beacon_processor.send_rpc_beacon_block` returns Ok() sync
diff --git a/beacon_node/network/src/sync/tests/lookups.rs b/beacon_node/network/src/sync/tests/lookups.rs
index 63bcd176f52..9a10aadcc22 100644
--- a/beacon_node/network/src/sync/tests/lookups.rs
+++ b/beacon_node/network/src/sync/tests/lookups.rs
@@ -2286,12 +2286,13 @@ mod deneb_only {
             let max_len = self.rig.spec.max_blobs_per_block(block.epoch()) as usize;
             // Now this block is the one we expect requests from
             self.block = block.clone();
-            let block = RpcBlock::new(
+            let block = RpcBlock::new_maybe_available(
                 Some(block.canonical_root()),
                 block,
                 self.unknown_parent_blobs
                     .take()
                     .map(|vec| RuntimeVariableList::new(vec, max_len).unwrap()),
+                None,
             )
             .unwrap();
             self.rig.parent_block_processed(
diff --git a/beacon_node/network/src/sync/tests/range.rs b/beacon_node/network/src/sync/tests/range.rs
index cb728a90c1b..7a30c25f5d1 100644
--- a/beacon_node/network/src/sync/tests/range.rs
+++ b/beacon_node/network/src/sync/tests/range.rs
@@ -446,13 +446,13 @@ fn build_rpc_block(
 ) -> RpcBlock<E> {
     match data_sidecars {
         Some(DataSidecars::Blobs(blobs)) => {
-            RpcBlock::new(None, block, Some(blobs.clone())).unwrap()
+            RpcBlock::new_available(None, block, Some(blobs.clone()), None).unwrap()
         }
         Some(DataSidecars::DataColumns(columns)) => {
-            RpcBlock::new_with_custody_columns(None, block, columns.clone()).unwrap()
+            RpcBlock::new_available(None, block, None, Some(columns.clone())).unwrap()
         }
         // Block has no data, expects zero columns
-        None => RpcBlock::new_without_blobs(None, block),
+        None => RpcBlock::new_available(None, block, None, None).unwrap(),
     }
 }
 
diff --git a/testing/ef_tests/src/cases/fork_choice.rs b/testing/ef_tests/src/cases/fork_choice.rs
index 8e9d438a243..0e4f5e941e2 100644
--- a/testing/ef_tests/src/cases/fork_choice.rs
+++ b/testing/ef_tests/src/cases/fork_choice.rs
@@ -542,13 +542,16 @@ impl<E: EthSpec> Tester<E> {
 
         let block = Arc::new(block);
         let result: Result<Result<Hash256, ()>, _> = self
-            .block_on_dangerous(self.harness.chain.process_block(
-                block_root,
-                RpcBlock::new_without_blobs(Some(block_root), block.clone()),
-                NotifyExecutionLayer::Yes,
-                BlockImportSource::Lookup,
-                || Ok(()),
-            ))?
+            .block_on_dangerous(
+                self.harness.chain.process_block(
+                    block_root,
+                    RpcBlock::new_maybe_available(Some(block_root), block.clone(), None, None)
+                        .map_err(|e| Error::InternalError(format!("{:?}", e)))?,
+                    NotifyExecutionLayer::Yes,
+                    BlockImportSource::Lookup,
+                    || Ok(()),
+                ),
+            )?
             .map(|avail: AvailabilityProcessingStatus| avail.try_into());
         let success = data_column_success && result.as_ref().is_ok_and(|inner| inner.is_ok());
         if success != valid {
@@ -632,13 +635,16 @@ impl<E: EthSpec> Tester<E> {
 
         let block = Arc::new(block);
         let result: Result<Result<Hash256, ()>, _> = self
-            .block_on_dangerous(self.harness.chain.process_block(
-                block_root,
-                RpcBlock::new_without_blobs(Some(block_root), block.clone()),
-                NotifyExecutionLayer::Yes,
-                BlockImportSource::Lookup,
-                || Ok(()),
-            ))?
+            .block_on_dangerous(
+                self.harness.chain.process_block(
+                    block_root,
+                    RpcBlock::new_maybe_available(Some(block_root), block.clone(), None, None)
+                        .map_err(|e| Error::InternalError(format!("{:?}", e)))?,
+                    NotifyExecutionLayer::Yes,
+                    BlockImportSource::Lookup,
+                    || Ok(()),
+                ),
+            )?
             .map(|avail: AvailabilityProcessingStatus| avail.try_into());
         let success = blob_success && result.as_ref().is_ok_and(|inner| inner.is_ok());
         if success != valid {