diff --git a/.cargo/config.toml b/.cargo/config.toml
index a408305c4d1..dac01630032 100644
--- a/.cargo/config.toml
+++ b/.cargo/config.toml
@@ -1,3 +1,4 @@
 [env]
 # Set the number of arenas to 16 when using jemalloc.
 JEMALLOC_SYS_WITH_MALLOC_CONF = "abort_conf:true,narenas:16"
+
diff --git a/.github/workflows/test-suite.yml b/.github/workflows/test-suite.yml
index 65663e0cf49..45f3b757e74 100644
--- a/.github/workflows/test-suite.yml
+++ b/.github/workflows/test-suite.yml
@@ -350,7 +350,7 @@ jobs:
     - name: Check formatting with cargo fmt
       run: make cargo-fmt
     - name: Lint code for quality and style with Clippy
-      run: make lint
+      run: make lint-full
     - name: Certify Cargo.lock freshness
       run: git diff --exit-code Cargo.lock
     - name: Typecheck benchmark code without running it
diff --git a/Makefile b/Makefile
index 958abf87058..8faf8a2e54b 100644
--- a/Makefile
+++ b/Makefile
@@ -204,7 +204,7 @@ test-full: cargo-fmt test-release test-debug test-ef test-exec-engine
 # Lints the code for bad style and potentially unsafe arithmetic using Clippy.
 # Clippy lints are opt-in per-crate for now. By default, everything is allowed except for performance and correctness lints.
 lint:
-	RUSTFLAGS="-C debug-assertions=no $(RUSTFLAGS)" cargo clippy --workspace --benches --tests $(EXTRA_CLIPPY_OPTS) --features "$(TEST_FEATURES)" -- \
+	cargo clippy --workspace --benches --tests $(EXTRA_CLIPPY_OPTS) --features "$(TEST_FEATURES)" -- \
 		-D clippy::fn_to_numeric_cast_any \
 		-D clippy::manual_let_else \
 		-D clippy::large_stack_frames \
@@ -220,6 +220,10 @@ lint:
 lint-fix:
 	EXTRA_CLIPPY_OPTS="--fix --allow-staged --allow-dirty" $(MAKE) lint
 
+# Also run the lints on the optimized-only tests
+lint-full:
+	RUSTFLAGS="-C debug-assertions=no $(RUSTFLAGS)" $(MAKE) lint
+
 # Runs the makefile in the `ef_tests` repo.
 #
 # May download and extract an archive of test vectors from the ethereum
diff --git a/beacon_node/beacon_chain/src/beacon_chain.rs b/beacon_node/beacon_chain/src/beacon_chain.rs
index 80766d57b33..7155547d01b 100644
--- a/beacon_node/beacon_chain/src/beacon_chain.rs
+++ b/beacon_node/beacon_chain/src/beacon_chain.rs
@@ -1243,7 +1243,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     pub fn get_blobs(&self, block_root: &Hash256) -> Result<BlobSidecarList<T::EthSpec>, Error> {
         match self.store.get_blobs(block_root)? {
             Some(blobs) => Ok(blobs),
-            None => Ok(BlobSidecarList::default()),
+            None => Ok(BlobSidecarList::empty_uninitialized()),
         }
     }
 
diff --git a/beacon_node/beacon_chain/src/blob_verification.rs b/beacon_node/beacon_chain/src/blob_verification.rs
index 6c87deb8260..786b627bb7e 100644
--- a/beacon_node/beacon_chain/src/blob_verification.rs
+++ b/beacon_node/beacon_chain/src/blob_verification.rs
@@ -400,7 +400,7 @@ pub fn validate_blob_sidecar_for_gossip<T: BeaconChainTypes, O: ObservationStrat
     // since we only subscribe to `MaxBlobsPerBlock` subnets over gossip network.
     // We include this check only for completeness.
     // Getting this error would imply something very wrong with our networking decoding logic.
-    if blob_index >= T::EthSpec::max_blobs_per_block() as u64 {
+    if blob_index >= chain.spec.max_blobs_per_block(blob_epoch) {
         return Err(GossipBlobError::InvalidSubnet {
             expected: subnet,
             received: blob_index,
diff --git a/beacon_node/beacon_chain/src/block_verification_types.rs b/beacon_node/beacon_chain/src/block_verification_types.rs
index 420c83081c7..0bf3007e9b0 100644
--- a/beacon_node/beacon_chain/src/block_verification_types.rs
+++ b/beacon_node/beacon_chain/src/block_verification_types.rs
@@ -4,11 +4,10 @@ use crate::data_column_verification::{CustodyDataColumn, CustodyDataColumnList};
 use crate::eth1_finalization_cache::Eth1FinalizationData;
 use crate::{get_block_root, PayloadVerificationOutcome};
 use derivative::Derivative;
-use ssz_types::VariableList;
 use state_processing::ConsensusContext;
 use std::fmt::{Debug, Formatter};
 use std::sync::Arc;
-use types::blob_sidecar::{BlobIdentifier, FixedBlobSidecarList};
+use types::blob_sidecar::BlobIdentifier;
 use types::{
     BeaconBlockRef, BeaconState, BlindedPayload, BlobSidecarList, ChainSpec, Epoch, EthSpec,
     Hash256, RuntimeVariableList, SignedBeaconBlock, SignedBeaconBlockHeader, Slot,
@@ -176,23 +175,6 @@ impl<E: EthSpec> RpcBlock<E> {
         })
     }
 
-    pub fn new_from_fixed(
-        block_root: Hash256,
-        block: Arc<SignedBeaconBlock<E>>,
-        blobs: FixedBlobSidecarList<E>,
-    ) -> Result<Self, AvailabilityCheckError> {
-        let filtered = blobs
-            .into_iter()
-            .filter_map(|b| b.clone())
-            .collect::<Vec<_>>();
-        let blobs = if filtered.is_empty() {
-            None
-        } else {
-            Some(VariableList::from(filtered))
-        };
-        Self::new(Some(block_root), block, blobs)
-    }
-
     #[allow(clippy::type_complexity)]
     pub fn deconstruct(
         self,
diff --git a/beacon_node/beacon_chain/src/data_availability_checker.rs b/beacon_node/beacon_chain/src/data_availability_checker.rs
index 72806a74d27..83de995ddcb 100644
--- a/beacon_node/beacon_chain/src/data_availability_checker.rs
+++ b/beacon_node/beacon_chain/src/data_availability_checker.rs
@@ -215,9 +215,12 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
         // Note: currently not reporting which specific blob is invalid because we fetch all blobs
         // from the same peer for both lookup and range sync.
 
-        let verified_blobs =
-            KzgVerifiedBlobList::new(blobs.iter().flatten().cloned(), &self.kzg, seen_timestamp)
-                .map_err(AvailabilityCheckError::InvalidBlobs)?;
+        let verified_blobs = KzgVerifiedBlobList::new(
+            blobs.into_vec().into_iter().flatten(),
+            &self.kzg,
+            seen_timestamp,
+        )
+        .map_err(AvailabilityCheckError::InvalidBlobs)?;
 
         self.availability_cache
             .put_kzg_verified_blobs(block_root, verified_blobs, &self.log)
@@ -400,14 +403,13 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
         blocks: Vec<RpcBlock<T::EthSpec>>,
     ) -> Result<Vec<MaybeAvailableBlock<T::EthSpec>>, AvailabilityCheckError> {
         let mut results = Vec::with_capacity(blocks.len());
-        let all_blobs: BlobSidecarList<T::EthSpec> = blocks
+        let all_blobs = blocks
             .iter()
             .filter(|block| self.blobs_required_for_block(block.as_block()))
             // this clone is cheap as it's cloning an Arc
             .filter_map(|block| block.blobs().cloned())
             .flatten()
-            .collect::<Vec<_>>()
-            .into();
+            .collect::<Vec<_>>();
 
         // verify kzg for all blobs at once
         if !all_blobs.is_empty() {
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 40361574aff..6245d199350 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
@@ -10,13 +10,12 @@ use crate::BeaconChainTypes;
 use lru::LruCache;
 use parking_lot::RwLock;
 use slog::{debug, Logger};
-use ssz_types::FixedVector;
 use std::num::NonZeroUsize;
 use std::sync::Arc;
 use types::blob_sidecar::BlobIdentifier;
 use types::{
     BlobSidecar, ChainSpec, ColumnIndex, DataColumnIdentifier, DataColumnSidecar, Epoch, EthSpec,
-    Hash256, SignedBeaconBlock,
+    Hash256, RuntimeFixedVector, RuntimeVariableList, SignedBeaconBlock,
 };
 
 /// This represents the components of a partially available block
@@ -28,7 +27,7 @@ use types::{
 #[derive(Clone)]
 pub struct PendingComponents<E: EthSpec> {
     pub block_root: Hash256,
-    pub verified_blobs: FixedVector<Option<KzgVerifiedBlob<E>>, E::MaxBlobsPerBlock>,
+    pub verified_blobs: RuntimeFixedVector<Option<KzgVerifiedBlob<E>>>,
     pub verified_data_columns: Vec<KzgVerifiedCustodyDataColumn<E>>,
     pub executed_block: Option<DietAvailabilityPendingExecutedBlock<E>>,
     pub reconstruction_started: bool,
@@ -41,9 +40,7 @@ impl<E: EthSpec> PendingComponents<E> {
     }
 
     /// Returns an immutable reference to the fixed vector of cached blobs.
-    pub fn get_cached_blobs(
-        &self,
-    ) -> &FixedVector<Option<KzgVerifiedBlob<E>>, E::MaxBlobsPerBlock> {
+    pub fn get_cached_blobs(&self) -> &RuntimeFixedVector<Option<KzgVerifiedBlob<E>>> {
         &self.verified_blobs
     }
 
@@ -64,9 +61,7 @@ impl<E: EthSpec> PendingComponents<E> {
     }
 
     /// Returns a mutable reference to the fixed vector of cached blobs.
-    pub fn get_cached_blobs_mut(
-        &mut self,
-    ) -> &mut FixedVector<Option<KzgVerifiedBlob<E>>, E::MaxBlobsPerBlock> {
+    pub fn get_cached_blobs_mut(&mut self) -> &mut RuntimeFixedVector<Option<KzgVerifiedBlob<E>>> {
         &mut self.verified_blobs
     }
 
@@ -138,10 +133,7 @@ impl<E: EthSpec> PendingComponents<E> {
     /// Blobs are only inserted if:
     /// 1. The blob entry at the index is empty and no block exists.
     /// 2. The block exists and its commitment matches the blob's commitment.
-    pub fn merge_blobs(
-        &mut self,
-        blobs: FixedVector<Option<KzgVerifiedBlob<E>>, E::MaxBlobsPerBlock>,
-    ) {
+    pub fn merge_blobs(&mut self, blobs: RuntimeFixedVector<Option<KzgVerifiedBlob<E>>>) {
         for (index, blob) in blobs.iter().cloned().enumerate() {
             let Some(blob) = blob else { continue };
             self.merge_single_blob(index, blob);
@@ -185,7 +177,7 @@ impl<E: EthSpec> PendingComponents<E> {
     /// Blobs that don't match the new block's commitments are evicted.
     pub fn merge_block(&mut self, block: DietAvailabilityPendingExecutedBlock<E>) {
         self.insert_block(block);
-        let reinsert = std::mem::take(self.get_cached_blobs_mut());
+        let reinsert = self.get_cached_blobs_mut().take();
         self.merge_blobs(reinsert);
     }
 
@@ -240,10 +232,10 @@ impl<E: EthSpec> PendingComponents<E> {
     }
 
     /// Returns an empty `PendingComponents` object with the given block root.
-    pub fn empty(block_root: Hash256) -> Self {
+    pub fn empty(block_root: Hash256, max_len: usize) -> Self {
         Self {
             block_root,
-            verified_blobs: FixedVector::default(),
+            verified_blobs: RuntimeFixedVector::new(vec![None; max_len]),
             verified_data_columns: vec![],
             executed_block: None,
             reconstruction_started: false,
@@ -302,7 +294,11 @@ impl<E: EthSpec> PendingComponents<E> {
             else {
                 return Err(AvailabilityCheckError::Unexpected);
             };
-            (Some(verified_blobs), None)
+            let max_len = spec.max_blobs_per_block(diet_executed_block.as_block().epoch()) as usize;
+            (
+                Some(RuntimeVariableList::new(verified_blobs, max_len)?),
+                None,
+            )
         };
 
         let executed_block = recover(diet_executed_block)?;
@@ -344,10 +340,7 @@ impl<E: EthSpec> PendingComponents<E> {
                 }
 
                 if let Some(kzg_verified_data_column) = self.verified_data_columns.first() {
-                    let epoch = kzg_verified_data_column
-                        .as_data_column()
-                        .slot()
-                        .epoch(E::slots_per_epoch());
+                    let epoch = kzg_verified_data_column.as_data_column().epoch();
                     return Some(epoch);
                 }
 
@@ -460,7 +453,18 @@ impl<T: BeaconChainTypes> DataAvailabilityCheckerInner<T> {
         kzg_verified_blobs: I,
         log: &Logger,
     ) -> Result<Availability<T::EthSpec>, AvailabilityCheckError> {
-        let mut fixed_blobs = FixedVector::default();
+        let mut kzg_verified_blobs = kzg_verified_blobs.into_iter().peekable();
+
+        let Some(epoch) = kzg_verified_blobs
+            .peek()
+            .map(|verified_blob| verified_blob.as_blob().epoch())
+        else {
+            // Verified blobs list should be non-empty.
+            return Err(AvailabilityCheckError::Unexpected);
+        };
+
+        let mut fixed_blobs =
+            RuntimeFixedVector::new(vec![None; self.spec.max_blobs_per_block(epoch) as usize]);
 
         for blob in kzg_verified_blobs {
             if let Some(blob_opt) = fixed_blobs.get_mut(blob.blob_index() as usize) {
@@ -474,7 +478,9 @@ impl<T: BeaconChainTypes> DataAvailabilityCheckerInner<T> {
         let mut pending_components = write_lock
             .pop_entry(&block_root)
             .map(|(_, v)| v)
-            .unwrap_or_else(|| PendingComponents::empty(block_root));
+            .unwrap_or_else(|| {
+                PendingComponents::empty(block_root, self.spec.max_blobs_per_block(epoch) as usize)
+            });
 
         // Merge in the blobs.
         pending_components.merge_blobs(fixed_blobs);
@@ -501,13 +507,24 @@ impl<T: BeaconChainTypes> DataAvailabilityCheckerInner<T> {
         kzg_verified_data_columns: I,
         log: &Logger,
     ) -> Result<Availability<T::EthSpec>, AvailabilityCheckError> {
+        let mut kzg_verified_data_columns = kzg_verified_data_columns.into_iter().peekable();
+        let Some(epoch) = kzg_verified_data_columns
+            .peek()
+            .map(|verified_blob| verified_blob.as_data_column().epoch())
+        else {
+            // Verified data_columns list should be non-empty.
+            return Err(AvailabilityCheckError::Unexpected);
+        };
+
         let mut write_lock = self.critical.write();
 
         // Grab existing entry or create a new entry.
         let mut pending_components = write_lock
             .pop_entry(&block_root)
             .map(|(_, v)| v)
-            .unwrap_or_else(|| PendingComponents::empty(block_root));
+            .unwrap_or_else(|| {
+                PendingComponents::empty(block_root, self.spec.max_blobs_per_block(epoch) as usize)
+            });
 
         // Merge in the data columns.
         pending_components.merge_data_columns(kzg_verified_data_columns)?;
@@ -584,6 +601,7 @@ impl<T: BeaconChainTypes> DataAvailabilityCheckerInner<T> {
         log: &Logger,
     ) -> Result<Availability<T::EthSpec>, AvailabilityCheckError> {
         let mut write_lock = self.critical.write();
+        let epoch = executed_block.as_block().epoch();
         let block_root = executed_block.import_data.block_root;
 
         // register the block to get the diet block
@@ -595,7 +613,9 @@ impl<T: BeaconChainTypes> DataAvailabilityCheckerInner<T> {
         let mut pending_components = write_lock
             .pop_entry(&block_root)
             .map(|(_, v)| v)
-            .unwrap_or_else(|| PendingComponents::empty(block_root));
+            .unwrap_or_else(|| {
+                PendingComponents::empty(block_root, self.spec.max_blobs_per_block(epoch) as usize)
+            });
 
         // Merge in the block.
         pending_components.merge_block(diet_executed_block);
@@ -815,7 +835,8 @@ mod test {
         info!(log, "done printing kzg commitments");
 
         let gossip_verified_blobs = if let Some((kzg_proofs, blobs)) = maybe_blobs {
-            let sidecars = BlobSidecar::build_sidecars(blobs, &block, kzg_proofs).unwrap();
+            let sidecars =
+                BlobSidecar::build_sidecars(blobs, &block, kzg_proofs, &chain.spec).unwrap();
             Vec::from(sidecars)
                 .into_iter()
                 .map(|sidecar| {
@@ -948,6 +969,8 @@ mod test {
                 assert_eq!(cache.critical.read().len(), 1);
             }
         }
+        // remove the blob to simulate successful import
+        cache.remove_pending_components(root);
         assert!(
             cache.critical.read().is_empty(),
             "cache should be empty now that all components available"
@@ -1128,7 +1151,7 @@ mod pending_components_tests {
     use super::*;
     use crate::block_verification_types::BlockImportData;
     use crate::eth1_finalization_cache::Eth1FinalizationData;
-    use crate::test_utils::{generate_rand_block_and_blobs, NumBlobs};
+    use crate::test_utils::{generate_rand_block_and_blobs, test_spec, NumBlobs};
     use crate::PayloadVerificationOutcome;
     use fork_choice::PayloadVerificationStatus;
     use kzg::KzgCommitment;
@@ -1144,15 +1167,19 @@ mod pending_components_tests {
 
     type Setup<E> = (
         SignedBeaconBlock<E>,
-        FixedVector<Option<Arc<BlobSidecar<E>>>, <E as EthSpec>::MaxBlobsPerBlock>,
-        FixedVector<Option<Arc<BlobSidecar<E>>>, <E as EthSpec>::MaxBlobsPerBlock>,
+        RuntimeFixedVector<Option<Arc<BlobSidecar<E>>>>,
+        RuntimeFixedVector<Option<Arc<BlobSidecar<E>>>>,
+        usize,
     );
 
     pub fn pre_setup() -> Setup<E> {
         let mut rng = StdRng::seed_from_u64(0xDEADBEEF0BAD5EEDu64);
+        let spec = test_spec::<E>();
         let (block, blobs_vec) =
-            generate_rand_block_and_blobs::<E>(ForkName::Deneb, NumBlobs::Random, &mut rng);
-        let mut blobs: FixedVector<_, <E as EthSpec>::MaxBlobsPerBlock> = FixedVector::default();
+            generate_rand_block_and_blobs::<E>(ForkName::Deneb, NumBlobs::Random, &mut rng, &spec);
+        let max_len = spec.max_blobs_per_block(block.epoch()) as usize;
+        let mut blobs: RuntimeFixedVector<Option<Arc<BlobSidecar<E>>>> =
+            RuntimeFixedVector::default(max_len);
 
         for blob in blobs_vec {
             if let Some(b) = blobs.get_mut(blob.index as usize) {
@@ -1160,10 +1187,8 @@ mod pending_components_tests {
             }
         }
 
-        let mut invalid_blobs: FixedVector<
-            Option<Arc<BlobSidecar<E>>>,
-            <E as EthSpec>::MaxBlobsPerBlock,
-        > = FixedVector::default();
+        let mut invalid_blobs: RuntimeFixedVector<Option<Arc<BlobSidecar<E>>>> =
+            RuntimeFixedVector::default(max_len);
         for (index, blob) in blobs.iter().enumerate() {
             if let Some(invalid_blob) = blob {
                 let mut blob_copy = invalid_blob.as_ref().clone();
@@ -1172,21 +1197,21 @@ mod pending_components_tests {
             }
         }
 
-        (block, blobs, invalid_blobs)
+        (block, blobs, invalid_blobs, max_len)
     }
 
     type PendingComponentsSetup<E> = (
         DietAvailabilityPendingExecutedBlock<E>,
-        FixedVector<Option<KzgVerifiedBlob<E>>, <E as EthSpec>::MaxBlobsPerBlock>,
-        FixedVector<Option<KzgVerifiedBlob<E>>, <E as EthSpec>::MaxBlobsPerBlock>,
+        RuntimeFixedVector<Option<KzgVerifiedBlob<E>>>,
+        RuntimeFixedVector<Option<KzgVerifiedBlob<E>>>,
     );
 
     pub fn setup_pending_components(
         block: SignedBeaconBlock<E>,
-        valid_blobs: FixedVector<Option<Arc<BlobSidecar<E>>>, <E as EthSpec>::MaxBlobsPerBlock>,
-        invalid_blobs: FixedVector<Option<Arc<BlobSidecar<E>>>, <E as EthSpec>::MaxBlobsPerBlock>,
+        valid_blobs: RuntimeFixedVector<Option<Arc<BlobSidecar<E>>>>,
+        invalid_blobs: RuntimeFixedVector<Option<Arc<BlobSidecar<E>>>>,
     ) -> PendingComponentsSetup<E> {
-        let blobs = FixedVector::from(
+        let blobs = RuntimeFixedVector::new(
             valid_blobs
                 .iter()
                 .map(|blob_opt| {
@@ -1196,7 +1221,7 @@ mod pending_components_tests {
                 })
                 .collect::<Vec<_>>(),
         );
-        let invalid_blobs = FixedVector::from(
+        let invalid_blobs = RuntimeFixedVector::new(
             invalid_blobs
                 .iter()
                 .map(|blob_opt| {
@@ -1228,10 +1253,10 @@ mod pending_components_tests {
         (block.into(), blobs, invalid_blobs)
     }
 
-    pub fn assert_cache_consistent(cache: PendingComponents<E>) {
+    pub fn assert_cache_consistent(cache: PendingComponents<E>, max_len: usize) {
         if let Some(cached_block) = cache.get_cached_block() {
             let cached_block_commitments = cached_block.get_commitments();
-            for index in 0..E::max_blobs_per_block() {
+            for index in 0..max_len {
                 let block_commitment = cached_block_commitments.get(index).copied();
                 let blob_commitment_opt = cache.get_cached_blobs().get(index).unwrap();
                 let blob_commitment = blob_commitment_opt.as_ref().map(|b| *b.get_commitment());
@@ -1250,40 +1275,40 @@ mod pending_components_tests {
 
     #[test]
     fn valid_block_invalid_blobs_valid_blobs() {
-        let (block_commitments, blobs, random_blobs) = pre_setup();
+        let (block_commitments, blobs, random_blobs, max_len) = pre_setup();
         let (block_commitments, blobs, random_blobs) =
             setup_pending_components(block_commitments, blobs, random_blobs);
         let block_root = Hash256::zero();
-        let mut cache = <PendingComponents<E>>::empty(block_root);
+        let mut cache = <PendingComponents<E>>::empty(block_root, max_len);
         cache.merge_block(block_commitments);
         cache.merge_blobs(random_blobs);
         cache.merge_blobs(blobs);
 
-        assert_cache_consistent(cache);
+        assert_cache_consistent(cache, max_len);
     }
 
     #[test]
     fn invalid_blobs_block_valid_blobs() {
-        let (block_commitments, blobs, random_blobs) = pre_setup();
+        let (block_commitments, blobs, random_blobs, max_len) = pre_setup();
         let (block_commitments, blobs, random_blobs) =
             setup_pending_components(block_commitments, blobs, random_blobs);
         let block_root = Hash256::zero();
-        let mut cache = <PendingComponents<E>>::empty(block_root);
+        let mut cache = <PendingComponents<E>>::empty(block_root, max_len);
         cache.merge_blobs(random_blobs);
         cache.merge_block(block_commitments);
         cache.merge_blobs(blobs);
 
-        assert_cache_consistent(cache);
+        assert_cache_consistent(cache, max_len);
     }
 
     #[test]
     fn invalid_blobs_valid_blobs_block() {
-        let (block_commitments, blobs, random_blobs) = pre_setup();
+        let (block_commitments, blobs, random_blobs, max_len) = pre_setup();
         let (block_commitments, blobs, random_blobs) =
             setup_pending_components(block_commitments, blobs, random_blobs);
 
         let block_root = Hash256::zero();
-        let mut cache = <PendingComponents<E>>::empty(block_root);
+        let mut cache = <PendingComponents<E>>::empty(block_root, max_len);
         cache.merge_blobs(random_blobs);
         cache.merge_blobs(blobs);
         cache.merge_block(block_commitments);
@@ -1293,46 +1318,46 @@ mod pending_components_tests {
 
     #[test]
     fn block_valid_blobs_invalid_blobs() {
-        let (block_commitments, blobs, random_blobs) = pre_setup();
+        let (block_commitments, blobs, random_blobs, max_len) = pre_setup();
         let (block_commitments, blobs, random_blobs) =
             setup_pending_components(block_commitments, blobs, random_blobs);
 
         let block_root = Hash256::zero();
-        let mut cache = <PendingComponents<E>>::empty(block_root);
+        let mut cache = <PendingComponents<E>>::empty(block_root, max_len);
         cache.merge_block(block_commitments);
         cache.merge_blobs(blobs);
         cache.merge_blobs(random_blobs);
 
-        assert_cache_consistent(cache);
+        assert_cache_consistent(cache, max_len);
     }
 
     #[test]
     fn valid_blobs_block_invalid_blobs() {
-        let (block_commitments, blobs, random_blobs) = pre_setup();
+        let (block_commitments, blobs, random_blobs, max_len) = pre_setup();
         let (block_commitments, blobs, random_blobs) =
             setup_pending_components(block_commitments, blobs, random_blobs);
 
         let block_root = Hash256::zero();
-        let mut cache = <PendingComponents<E>>::empty(block_root);
+        let mut cache = <PendingComponents<E>>::empty(block_root, max_len);
         cache.merge_blobs(blobs);
         cache.merge_block(block_commitments);
         cache.merge_blobs(random_blobs);
 
-        assert_cache_consistent(cache);
+        assert_cache_consistent(cache, max_len);
     }
 
     #[test]
     fn valid_blobs_invalid_blobs_block() {
-        let (block_commitments, blobs, random_blobs) = pre_setup();
+        let (block_commitments, blobs, random_blobs, max_len) = pre_setup();
         let (block_commitments, blobs, random_blobs) =
             setup_pending_components(block_commitments, blobs, random_blobs);
 
         let block_root = Hash256::zero();
-        let mut cache = <PendingComponents<E>>::empty(block_root);
+        let mut cache = <PendingComponents<E>>::empty(block_root, max_len);
         cache.merge_blobs(blobs);
         cache.merge_blobs(random_blobs);
         cache.merge_block(block_commitments);
 
-        assert_cache_consistent(cache);
+        assert_cache_consistent(cache, max_len);
     }
 }
diff --git a/beacon_node/beacon_chain/src/fetch_blobs.rs b/beacon_node/beacon_chain/src/fetch_blobs.rs
index f740b693fbf..f1646072c96 100644
--- a/beacon_node/beacon_chain/src/fetch_blobs.rs
+++ b/beacon_node/beacon_chain/src/fetch_blobs.rs
@@ -21,8 +21,8 @@ use std::sync::Arc;
 use tokio::sync::mpsc::Receiver;
 use types::blob_sidecar::{BlobSidecarError, FixedBlobSidecarList};
 use types::{
-    BeaconStateError, BlobSidecar, DataColumnSidecar, DataColumnSidecarList, EthSpec, FullPayload,
-    Hash256, SignedBeaconBlock, SignedBeaconBlockHeader,
+    BeaconStateError, BlobSidecar, ChainSpec, DataColumnSidecar, DataColumnSidecarList, EthSpec,
+    FullPayload, Hash256, SignedBeaconBlock, SignedBeaconBlockHeader,
 };
 
 pub enum BlobsOrDataColumns<T: BeaconChainTypes> {
@@ -112,6 +112,7 @@ pub async fn fetch_and_process_engine_blobs<T: BeaconChainTypes>(
         response,
         signed_block_header,
         &kzg_commitments_proof,
+        &chain.spec,
     )?;
 
     let num_fetched_blobs = fixed_blob_sidecar_list
@@ -275,8 +276,11 @@ fn build_blob_sidecars<E: EthSpec>(
     response: Vec<Option<BlobAndProofV1<E>>>,
     signed_block_header: SignedBeaconBlockHeader,
     kzg_commitments_inclusion_proof: &FixedVector<Hash256, E::KzgCommitmentsInclusionProofDepth>,
+    spec: &ChainSpec,
 ) -> Result<FixedBlobSidecarList<E>, FetchEngineBlobError> {
-    let mut fixed_blob_sidecar_list = FixedBlobSidecarList::default();
+    let epoch = block.epoch();
+    let mut fixed_blob_sidecar_list =
+        FixedBlobSidecarList::default(spec.max_blobs_per_block(epoch) as usize);
     for (index, blob_and_proof) in response
         .into_iter()
         .enumerate()
diff --git a/beacon_node/beacon_chain/src/kzg_utils.rs b/beacon_node/beacon_chain/src/kzg_utils.rs
index 1680c0298d1..99db4632281 100644
--- a/beacon_node/beacon_chain/src/kzg_utils.rs
+++ b/beacon_node/beacon_chain/src/kzg_utils.rs
@@ -193,9 +193,11 @@ fn build_data_column_sidecars<E: EthSpec>(
     spec: &ChainSpec,
 ) -> Result<DataColumnSidecarList<E>, String> {
     let number_of_columns = spec.number_of_columns;
-    let mut columns = vec![Vec::with_capacity(E::max_blobs_per_block()); number_of_columns];
-    let mut column_kzg_proofs =
-        vec![Vec::with_capacity(E::max_blobs_per_block()); number_of_columns];
+    let max_blobs_per_block = spec
+        .max_blobs_per_block(signed_block_header.message.slot.epoch(E::slots_per_epoch()))
+        as usize;
+    let mut columns = vec![Vec::with_capacity(max_blobs_per_block); number_of_columns];
+    let mut column_kzg_proofs = vec![Vec::with_capacity(max_blobs_per_block); number_of_columns];
 
     for (blob_cells, blob_cell_proofs) in blob_cells_and_proofs_vec {
         // we iterate over each column, and we construct the column from "top to bottom",
diff --git a/beacon_node/beacon_chain/src/observed_data_sidecars.rs b/beacon_node/beacon_chain/src/observed_data_sidecars.rs
index 53f8c71f54e..a9ab8f4e10b 100644
--- a/beacon_node/beacon_chain/src/observed_data_sidecars.rs
+++ b/beacon_node/beacon_chain/src/observed_data_sidecars.rs
@@ -24,7 +24,7 @@ pub trait ObservableDataSidecar {
     fn slot(&self) -> Slot;
     fn block_proposer_index(&self) -> u64;
     fn index(&self) -> u64;
-    fn max_num_of_items(spec: &ChainSpec) -> usize;
+    fn max_num_of_items(spec: &ChainSpec, slot: Slot) -> usize;
 }
 
 impl<E: EthSpec> ObservableDataSidecar for BlobSidecar<E> {
@@ -40,8 +40,8 @@ impl<E: EthSpec> ObservableDataSidecar for BlobSidecar<E> {
         self.index
     }
 
-    fn max_num_of_items(_spec: &ChainSpec) -> usize {
-        E::max_blobs_per_block()
+    fn max_num_of_items(spec: &ChainSpec, slot: Slot) -> usize {
+        spec.max_blobs_per_block(slot.epoch(E::slots_per_epoch())) as usize
     }
 }
 
@@ -58,7 +58,7 @@ impl<E: EthSpec> ObservableDataSidecar for DataColumnSidecar<E> {
         self.index
     }
 
-    fn max_num_of_items(spec: &ChainSpec) -> usize {
+    fn max_num_of_items(spec: &ChainSpec, _slot: Slot) -> usize {
         spec.number_of_columns
     }
 }
@@ -103,7 +103,9 @@ impl<T: ObservableDataSidecar> ObservedDataSidecars<T> {
                 slot: data_sidecar.slot(),
                 proposer: data_sidecar.block_proposer_index(),
             })
-            .or_insert_with(|| HashSet::with_capacity(T::max_num_of_items(&self.spec)));
+            .or_insert_with(|| {
+                HashSet::with_capacity(T::max_num_of_items(&self.spec, data_sidecar.slot()))
+            });
         let did_not_exist = data_indices.insert(data_sidecar.index());
 
         Ok(!did_not_exist)
@@ -123,7 +125,7 @@ impl<T: ObservableDataSidecar> ObservedDataSidecars<T> {
     }
 
     fn sanitize_data_sidecar(&self, data_sidecar: &T) -> Result<(), Error> {
-        if data_sidecar.index() >= T::max_num_of_items(&self.spec) as u64 {
+        if data_sidecar.index() >= T::max_num_of_items(&self.spec, data_sidecar.slot()) as u64 {
             return Err(Error::InvalidDataIndex(data_sidecar.index()));
         }
         let finalized_slot = self.finalized_slot;
@@ -179,7 +181,7 @@ mod tests {
     use crate::test_utils::test_spec;
     use bls::Hash256;
     use std::sync::Arc;
-    use types::MainnetEthSpec;
+    use types::{Epoch, MainnetEthSpec};
 
     type E = MainnetEthSpec;
 
@@ -333,7 +335,7 @@ mod tests {
     #[test]
     fn simple_observations() {
         let spec = Arc::new(test_spec::<E>());
-        let mut cache = ObservedDataSidecars::<BlobSidecar<E>>::new(spec);
+        let mut cache = ObservedDataSidecars::<BlobSidecar<E>>::new(spec.clone());
 
         // Slot 0, index 0
         let proposer_index_a = 420;
@@ -489,7 +491,7 @@ mod tests {
         );
 
         // Try adding an out of bounds index
-        let invalid_index = E::max_blobs_per_block() as u64;
+        let invalid_index = spec.max_blobs_per_block(Epoch::new(0));
         let sidecar_d = get_blob_sidecar(0, proposer_index_a, invalid_index);
         assert_eq!(
             cache.observe_sidecar(&sidecar_d),
diff --git a/beacon_node/beacon_chain/src/test_utils.rs b/beacon_node/beacon_chain/src/test_utils.rs
index 093ee0c44b4..a722669c2d8 100644
--- a/beacon_node/beacon_chain/src/test_utils.rs
+++ b/beacon_node/beacon_chain/src/test_utils.rs
@@ -511,7 +511,7 @@ where
 
     pub fn mock_execution_layer_with_config(mut self) -> Self {
         let mock = mock_execution_layer_from_parts::<E>(
-            self.spec.as_ref().expect("cannot build without spec"),
+            self.spec.clone().expect("cannot build without spec"),
             self.runtime.task_executor.clone(),
         );
         self.execution_layer = Some(mock.el.clone());
@@ -611,7 +611,7 @@ where
 }
 
 pub fn mock_execution_layer_from_parts<E: EthSpec>(
-    spec: &ChainSpec,
+    spec: Arc<ChainSpec>,
     task_executor: TaskExecutor,
 ) -> MockExecutionLayer<E> {
     let shanghai_time = spec.capella_fork_epoch.map(|epoch| {
@@ -624,7 +624,7 @@ pub fn mock_execution_layer_from_parts<E: EthSpec>(
         HARNESS_GENESIS_TIME + spec.seconds_per_slot * E::slots_per_epoch() * epoch.as_u64()
     });
 
-    let kzg = get_kzg(spec);
+    let kzg = get_kzg(&spec);
 
     MockExecutionLayer::new(
         task_executor,
@@ -633,7 +633,7 @@ pub fn mock_execution_layer_from_parts<E: EthSpec>(
         cancun_time,
         prague_time,
         Some(JwtKey::from_slice(&DEFAULT_JWT_SECRET).unwrap()),
-        spec.clone(),
+        spec,
         Some(kzg),
     )
 }
@@ -2019,7 +2019,7 @@ where
         let (block, blob_items) = block_contents;
 
         let sidecars = blob_items
-            .map(|(proofs, blobs)| BlobSidecar::build_sidecars(blobs, &block, proofs))
+            .map(|(proofs, blobs)| BlobSidecar::build_sidecars(blobs, &block, proofs, &self.spec))
             .transpose()
             .unwrap();
         let block_hash: SignedBeaconBlockHash = self
@@ -2045,7 +2045,7 @@ where
         let (block, blob_items) = block_contents;
 
         let sidecars = blob_items
-            .map(|(proofs, blobs)| BlobSidecar::build_sidecars(blobs, &block, proofs))
+            .map(|(proofs, blobs)| BlobSidecar::build_sidecars(blobs, &block, proofs, &self.spec))
             .transpose()
             .unwrap();
         let block_root = block.canonical_root();
@@ -2816,11 +2816,12 @@ pub fn generate_rand_block_and_blobs<E: EthSpec>(
     fork_name: ForkName,
     num_blobs: NumBlobs,
     rng: &mut impl Rng,
+    spec: &ChainSpec,
 ) -> (SignedBeaconBlock<E, FullPayload<E>>, Vec<BlobSidecar<E>>) {
     let inner = map_fork_name!(fork_name, BeaconBlock, <_>::random_for_test(rng));
 
     let mut block = SignedBeaconBlock::from_block(inner, types::Signature::random_for_test(rng));
-
+    let max_blobs = spec.max_blobs_per_block(block.epoch()) as usize;
     let mut blob_sidecars = vec![];
 
     let bundle = match block {
@@ -2830,7 +2831,7 @@ pub fn generate_rand_block_and_blobs<E: EthSpec>(
             // Get either zero blobs or a random number of blobs between 1 and Max Blobs.
             let payload: &mut FullPayloadDeneb<E> = &mut message.body.execution_payload;
             let num_blobs = match num_blobs {
-                NumBlobs::Random => rng.gen_range(1..=E::max_blobs_per_block()),
+                NumBlobs::Random => rng.gen_range(1..=max_blobs),
                 NumBlobs::Number(n) => n,
                 NumBlobs::None => 0,
             };
@@ -2850,7 +2851,7 @@ pub fn generate_rand_block_and_blobs<E: EthSpec>(
             // Get either zero blobs or a random number of blobs between 1 and Max Blobs.
             let payload: &mut FullPayloadElectra<E> = &mut message.body.execution_payload;
             let num_blobs = match num_blobs {
-                NumBlobs::Random => rng.gen_range(1..=E::max_blobs_per_block()),
+                NumBlobs::Random => rng.gen_range(1..=max_blobs),
                 NumBlobs::Number(n) => n,
                 NumBlobs::None => 0,
             };
@@ -2904,7 +2905,7 @@ pub fn generate_rand_block_and_data_columns<E: EthSpec>(
     DataColumnSidecarList<E>,
 ) {
     let kzg = get_kzg(spec);
-    let (block, blobs) = generate_rand_block_and_blobs(fork_name, num_blobs, rng);
+    let (block, blobs) = generate_rand_block_and_blobs(fork_name, num_blobs, rng, spec);
     let blob_refs = blobs.iter().map(|b| &b.blob).collect::<Vec<_>>();
     let data_columns = blobs_to_data_column_sidecars(&blob_refs, &block, &kzg, spec).unwrap();
 
diff --git a/beacon_node/beacon_chain/tests/block_verification.rs b/beacon_node/beacon_chain/tests/block_verification.rs
index f094a173eec..415aed5831d 100644
--- a/beacon_node/beacon_chain/tests/block_verification.rs
+++ b/beacon_node/beacon_chain/tests/block_verification.rs
@@ -206,7 +206,11 @@ fn update_blob_signed_header<E: EthSpec>(
     signed_block: &SignedBeaconBlock<E>,
     blobs: &mut BlobSidecarList<E>,
 ) {
-    for old_blob_sidecar in blobs.iter_mut() {
+    // Required to prevent as_mut_slice being called on an uninitialized list. Yuck.
+    if blobs.is_empty() {
+        return;
+    }
+    for old_blob_sidecar in blobs.as_mut_slice().unwrap() {
         let new_blob = Arc::new(BlobSidecar::<E> {
             index: old_blob_sidecar.index,
             blob: old_blob_sidecar.blob.clone(),
@@ -1214,7 +1218,7 @@ async fn verify_block_for_gossip_slashing_detection() {
     let slasher = Arc::new(
         Slasher::open(
             SlasherConfig::new(slasher_dir.path().into()),
-            spec,
+            spec.clone(),
             test_logger(),
         )
         .unwrap(),
@@ -1238,7 +1242,7 @@ async fn verify_block_for_gossip_slashing_detection() {
 
     if let Some((kzg_proofs, blobs)) = blobs1 {
         let sidecars =
-            BlobSidecar::build_sidecars(blobs, verified_block.block(), kzg_proofs).unwrap();
+            BlobSidecar::build_sidecars(blobs, verified_block.block(), kzg_proofs, &spec).unwrap();
         for sidecar in sidecars {
             let blob_index = sidecar.index;
             let verified_blob = harness
diff --git a/beacon_node/beacon_chain/tests/events.rs b/beacon_node/beacon_chain/tests/events.rs
index ab784d3be45..c9bd55e0620 100644
--- a/beacon_node/beacon_chain/tests/events.rs
+++ b/beacon_node/beacon_chain/tests/events.rs
@@ -73,7 +73,7 @@ async fn blob_sidecar_event_on_process_rpc_blobs() {
     let blob_1 = Arc::new(blob_1);
     let blob_2 = Arc::new(blob_2);
 
-    let blobs = FixedBlobSidecarList::from(vec![Some(blob_1.clone()), Some(blob_2.clone())]);
+    let blobs = FixedBlobSidecarList::new(vec![Some(blob_1.clone()), Some(blob_2.clone())]);
     let expected_sse_blobs = vec![
         SseBlobSidecar::from_blob_sidecar(blob_1.as_ref()),
         SseBlobSidecar::from_blob_sidecar(blob_2.as_ref()),
diff --git a/beacon_node/beacon_chain/tests/store_tests.rs b/beacon_node/beacon_chain/tests/store_tests.rs
index e1258ccdea7..d32bbdfeaa4 100644
--- a/beacon_node/beacon_chain/tests/store_tests.rs
+++ b/beacon_node/beacon_chain/tests/store_tests.rs
@@ -2341,8 +2341,10 @@ async fn weak_subjectivity_sync_test(slots: Vec<Slot>, checkpoint_slot: Slot) {
 
     let kzg = get_kzg(&spec);
 
-    let mock =
-        mock_execution_layer_from_parts(&harness.spec, harness.runtime.task_executor.clone());
+    let mock = mock_execution_layer_from_parts(
+        harness.spec.clone(),
+        harness.runtime.task_executor.clone(),
+    );
 
     // Initialise a new beacon chain from the finalized checkpoint.
     // The slot clock must be set to a time ahead of the checkpoint state.
diff --git a/beacon_node/client/src/builder.rs b/beacon_node/client/src/builder.rs
index 7c6a253aca4..f47da542885 100644
--- a/beacon_node/client/src/builder.rs
+++ b/beacon_node/client/src/builder.rs
@@ -36,7 +36,6 @@ use network::{NetworkConfig, NetworkSenders, NetworkService};
 use slasher::Slasher;
 use slasher_service::SlasherService;
 use slog::{debug, info, warn, Logger};
-use ssz::Decode;
 use std::net::TcpListener;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
@@ -361,10 +360,11 @@ where
                 let anchor_block = SignedBeaconBlock::from_ssz_bytes(&anchor_block_bytes, &spec)
                     .map_err(|e| format!("Unable to parse weak subj block SSZ: {:?}", e))?;
                 let anchor_blobs = if anchor_block.message().body().has_blobs() {
+                    let max_blobs_len = spec.max_blobs_per_block(anchor_block.epoch()) as usize;
                     let anchor_blobs_bytes = anchor_blobs_bytes
                         .ok_or("Blobs for checkpoint must be provided using --checkpoint-blobs")?;
                     Some(
-                        BlobSidecarList::from_ssz_bytes(&anchor_blobs_bytes)
+                        BlobSidecarList::from_ssz_bytes(&anchor_blobs_bytes, max_blobs_len)
                             .map_err(|e| format!("Unable to parse weak subj blobs SSZ: {e:?}"))?,
                     )
                 } else {
diff --git a/beacon_node/execution_layer/src/engine_api/http.rs b/beacon_node/execution_layer/src/engine_api/http.rs
index 33dc60d0378..830eb9d5d90 100644
--- a/beacon_node/execution_layer/src/engine_api/http.rs
+++ b/beacon_node/execution_layer/src/engine_api/http.rs
@@ -1321,7 +1321,8 @@ mod test {
 
     impl Tester {
         pub fn new(with_auth: bool) -> Self {
-            let server = MockServer::unit_testing();
+            let spec = Arc::new(MainnetEthSpec::default_spec());
+            let server = MockServer::unit_testing(spec);
 
             let rpc_url = SensitiveUrl::parse(&server.url()).unwrap();
             let echo_url = SensitiveUrl::parse(&format!("{}/echo", server.url())).unwrap();
diff --git a/beacon_node/execution_layer/src/test_utils/execution_block_generator.rs b/beacon_node/execution_layer/src/test_utils/execution_block_generator.rs
index 4fab7150ce3..7dd39d58ad8 100644
--- a/beacon_node/execution_layer/src/test_utils/execution_block_generator.rs
+++ b/beacon_node/execution_layer/src/test_utils/execution_block_generator.rs
@@ -153,6 +153,7 @@ pub struct ExecutionBlockGenerator<E: EthSpec> {
     pub blobs_bundles: HashMap<PayloadId, BlobsBundle<E>>,
     pub kzg: Option<Arc<Kzg>>,
     rng: Arc<Mutex<StdRng>>,
+    spec: Arc<ChainSpec>,
 }
 
 fn make_rng() -> Arc<Mutex<StdRng>> {
@@ -162,6 +163,7 @@ fn make_rng() -> Arc<Mutex<StdRng>> {
 }
 
 impl<E: EthSpec> ExecutionBlockGenerator<E> {
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         terminal_total_difficulty: Uint256,
         terminal_block_number: u64,
@@ -169,6 +171,7 @@ impl<E: EthSpec> ExecutionBlockGenerator<E> {
         shanghai_time: Option<u64>,
         cancun_time: Option<u64>,
         prague_time: Option<u64>,
+        spec: Arc<ChainSpec>,
         kzg: Option<Arc<Kzg>>,
     ) -> Self {
         let mut gen = Self {
@@ -188,6 +191,7 @@ impl<E: EthSpec> ExecutionBlockGenerator<E> {
             blobs_bundles: <_>::default(),
             kzg,
             rng: make_rng(),
+            spec,
         };
 
         gen.insert_pow_block(0).unwrap();
@@ -671,7 +675,11 @@ impl<E: EthSpec> ExecutionBlockGenerator<E> {
         if execution_payload.fork_name().deneb_enabled() {
             // get random number between 0 and Max Blobs
             let mut rng = self.rng.lock();
-            let num_blobs = rng.gen::<usize>() % (E::max_blobs_per_block() + 1);
+            let max_blobs = self
+                .spec
+                .max_blobs_per_block_by_fork(execution_payload.fork_name())
+                as usize;
+            let num_blobs = rng.gen::<usize>() % (max_blobs + 1);
             let (bundle, transactions) = generate_blobs(num_blobs)?;
             for tx in Vec::from(transactions) {
                 execution_payload
@@ -874,6 +882,7 @@ mod test {
         const TERMINAL_DIFFICULTY: u64 = 10;
         const TERMINAL_BLOCK: u64 = 10;
         const DIFFICULTY_INCREMENT: u64 = 1;
+        let spec = Arc::new(MainnetEthSpec::default_spec());
 
         let mut generator: ExecutionBlockGenerator<MainnetEthSpec> = ExecutionBlockGenerator::new(
             Uint256::from(TERMINAL_DIFFICULTY),
@@ -882,6 +891,7 @@ mod test {
             None,
             None,
             None,
+            spec,
             None,
         );
 
diff --git a/beacon_node/execution_layer/src/test_utils/mock_execution_layer.rs b/beacon_node/execution_layer/src/test_utils/mock_execution_layer.rs
index 48372a39be1..f472d21482a 100644
--- a/beacon_node/execution_layer/src/test_utils/mock_execution_layer.rs
+++ b/beacon_node/execution_layer/src/test_utils/mock_execution_layer.rs
@@ -13,7 +13,7 @@ pub struct MockExecutionLayer<E: EthSpec> {
     pub server: MockServer<E>,
     pub el: ExecutionLayer<E>,
     pub executor: TaskExecutor,
-    pub spec: ChainSpec,
+    pub spec: Arc<ChainSpec>,
 }
 
 impl<E: EthSpec> MockExecutionLayer<E> {
@@ -29,7 +29,7 @@ impl<E: EthSpec> MockExecutionLayer<E> {
             None,
             None,
             Some(JwtKey::from_slice(&DEFAULT_JWT_SECRET).unwrap()),
-            spec,
+            Arc::new(spec),
             None,
         )
     }
@@ -42,7 +42,7 @@ impl<E: EthSpec> MockExecutionLayer<E> {
         cancun_time: Option<u64>,
         prague_time: Option<u64>,
         jwt_key: Option<JwtKey>,
-        spec: ChainSpec,
+        spec: Arc<ChainSpec>,
         kzg: Option<Arc<Kzg>>,
     ) -> Self {
         let handle = executor.handle().unwrap();
@@ -57,6 +57,7 @@ impl<E: EthSpec> MockExecutionLayer<E> {
             shanghai_time,
             cancun_time,
             prague_time,
+            spec.clone(),
             kzg,
         );
 
@@ -320,7 +321,7 @@ impl<E: EthSpec> MockExecutionLayer<E> {
 
     pub async fn with_terminal_block<'a, U, V>(self, func: U) -> Self
     where
-        U: Fn(ChainSpec, ExecutionLayer<E>, Option<ExecutionBlock>) -> V,
+        U: Fn(Arc<ChainSpec>, ExecutionLayer<E>, Option<ExecutionBlock>) -> V,
         V: Future<Output = ()>,
     {
         let terminal_block_number = self
diff --git a/beacon_node/execution_layer/src/test_utils/mod.rs b/beacon_node/execution_layer/src/test_utils/mod.rs
index faf6d4ef0b6..726874ddb06 100644
--- a/beacon_node/execution_layer/src/test_utils/mod.rs
+++ b/beacon_node/execution_layer/src/test_utils/mod.rs
@@ -21,7 +21,7 @@ use std::marker::PhantomData;
 use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
 use std::sync::{Arc, LazyLock};
 use tokio::{runtime, sync::oneshot};
-use types::{EthSpec, ExecutionBlockHash, Uint256};
+use types::{ChainSpec, EthSpec, ExecutionBlockHash, Uint256};
 use warp::{http::StatusCode, Filter, Rejection};
 
 use crate::EngineCapabilities;
@@ -107,7 +107,7 @@ pub struct MockServer<E: EthSpec> {
 }
 
 impl<E: EthSpec> MockServer<E> {
-    pub fn unit_testing() -> Self {
+    pub fn unit_testing(chain_spec: Arc<ChainSpec>) -> Self {
         Self::new(
             &runtime::Handle::current(),
             JwtKey::from_slice(&DEFAULT_JWT_SECRET).unwrap(),
@@ -117,6 +117,7 @@ impl<E: EthSpec> MockServer<E> {
             None, // FIXME(capella): should this be the default?
             None, // FIXME(deneb): should this be the default?
             None, // FIXME(electra): should this be the default?
+            chain_spec,
             None,
         )
     }
@@ -124,6 +125,7 @@ impl<E: EthSpec> MockServer<E> {
     pub fn new_with_config(
         handle: &runtime::Handle,
         config: MockExecutionConfig,
+        spec: Arc<ChainSpec>,
         kzg: Option<Arc<Kzg>>,
     ) -> Self {
         let MockExecutionConfig {
@@ -145,6 +147,7 @@ impl<E: EthSpec> MockServer<E> {
             shanghai_time,
             cancun_time,
             prague_time,
+            spec,
             kzg,
         );
 
@@ -208,6 +211,7 @@ impl<E: EthSpec> MockServer<E> {
         shanghai_time: Option<u64>,
         cancun_time: Option<u64>,
         prague_time: Option<u64>,
+        spec: Arc<ChainSpec>,
         kzg: Option<Arc<Kzg>>,
     ) -> Self {
         Self::new_with_config(
@@ -222,6 +226,7 @@ impl<E: EthSpec> MockServer<E> {
                 cancun_time,
                 prague_time,
             },
+            spec,
             kzg,
         )
     }
diff --git a/beacon_node/http_api/src/block_id.rs b/beacon_node/http_api/src/block_id.rs
index dba8eb1ef32..6db05c6b80e 100644
--- a/beacon_node/http_api/src/block_id.rs
+++ b/beacon_node/http_api/src/block_id.rs
@@ -296,16 +296,17 @@ impl BlockId {
                     ))
                 })?
         } else {
-            BlobSidecarList::default()
+            BlobSidecarList::empty_uninitialized()
         };
 
         let blob_sidecar_list_filtered = match indices.indices {
             Some(vec) => {
-                let list = blob_sidecar_list
+                let list: Vec<_> = blob_sidecar_list
                     .into_iter()
                     .filter(|blob_sidecar| vec.contains(&blob_sidecar.index))
                     .collect();
-                BlobSidecarList::new(list)
+                let max_len = chain.spec.max_blobs_per_block(block.epoch());
+                BlobSidecarList::new(list, max_len as usize)
                     .map_err(|e| warp_utils::reject::custom_server_error(format!("{:?}", e)))?
             }
             None => blob_sidecar_list,
diff --git a/beacon_node/http_api/tests/broadcast_validation_tests.rs b/beacon_node/http_api/tests/broadcast_validation_tests.rs
index e1ecf2d4fc3..99ed3ef724f 100644
--- a/beacon_node/http_api/tests/broadcast_validation_tests.rs
+++ b/beacon_node/http_api/tests/broadcast_validation_tests.rs
@@ -1460,7 +1460,8 @@ pub async fn block_seen_on_gossip_with_some_blobs() {
     let blobs = blobs.expect("should have some blobs");
     assert!(
         blobs.0.len() >= 2,
-        "need at least 2 blobs for partial reveal"
+        "need at least 2 blobs for partial reveal, got: {}",
+        blobs.0.len()
     );
 
     let partial_kzg_proofs = vec![*blobs.0.first().unwrap()];
diff --git a/beacon_node/lighthouse_network/gossipsub/src/behaviour.rs b/beacon_node/lighthouse_network/gossipsub/src/behaviour.rs
index aafd869beed..c4e20e43972 100644
--- a/beacon_node/lighthouse_network/gossipsub/src/behaviour.rs
+++ b/beacon_node/lighthouse_network/gossipsub/src/behaviour.rs
@@ -679,9 +679,15 @@ where
                 // Gossipsub peers
                 None => {
                     tracing::debug!(topic=%topic_hash, "Topic not in the mesh");
+                    // `fanout_peers` is always non-empty if it's `Some`.
+                    let fanout_peers = self
+                        .fanout
+                        .get(&topic_hash)
+                        .map(|peers| if peers.is_empty() { None } else { Some(peers) })
+                        .unwrap_or(None);
                     // If we have fanout peers add them to the map.
-                    if self.fanout.contains_key(&topic_hash) {
-                        for peer in self.fanout.get(&topic_hash).expect("Topic must exist") {
+                    if let Some(peers) = fanout_peers {
+                        for peer in peers {
                             recipient_peers.insert(*peer);
                         }
                     } else {
diff --git a/beacon_node/lighthouse_network/src/rpc/methods.rs b/beacon_node/lighthouse_network/src/rpc/methods.rs
index bb8bfb0e206..e35af6fb40a 100644
--- a/beacon_node/lighthouse_network/src/rpc/methods.rs
+++ b/beacon_node/lighthouse_network/src/rpc/methods.rs
@@ -25,6 +25,13 @@ use types::{
 pub type MaxErrorLen = U256;
 pub const MAX_ERROR_LEN: u64 = 256;
 
+/// The max number of blobs we expect in the configs to set for compile time params.
+/// Note: This value is an estimate that we should use only for rate limiting,
+/// bounds checking and other non-consensus critical operations.
+///
+/// For exact value, we should always check the chainspec.
+pub const MAX_BLOBS_PER_BLOCK_CEILING: u64 = 16;
+
 /// Wrapper over SSZ List to represent error message in rpc responses.
 #[derive(Debug, Clone)]
 pub struct ErrorType(pub VariableList<u8, MaxErrorLen>);
@@ -327,8 +334,13 @@ pub struct BlobsByRangeRequest {
 }
 
 impl BlobsByRangeRequest {
+    /// This function provides an upper bound on number of blobs expected in
+    /// a certain slot range.
+    ///
+    /// Note: **must not** use for anything consensus critical, only for
+    /// bounds checking and rate limiting.
     pub fn max_blobs_requested<E: EthSpec>(&self) -> u64 {
-        self.count.saturating_mul(E::max_blobs_per_block() as u64)
+        self.count.saturating_mul(MAX_BLOBS_PER_BLOCK_CEILING)
     }
 }
 
@@ -851,3 +863,16 @@ impl slog::KV for StatusMessage {
         slog::Result::Ok(())
     }
 }
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use types::{ForkName, MainnetEthSpec};
+
+    #[test]
+    fn max_blobs_per_block_ceiling() {
+        let spec = MainnetEthSpec::default_spec();
+        let latest_fork = ForkName::latest();
+        assert!(spec.max_blobs_per_block_by_fork(latest_fork) <= MAX_BLOBS_PER_BLOCK_CEILING);
+    }
+}
diff --git a/beacon_node/lighthouse_network/src/rpc/protocol.rs b/beacon_node/lighthouse_network/src/rpc/protocol.rs
index 57c2795b04f..e4e8232e18f 100644
--- a/beacon_node/lighthouse_network/src/rpc/protocol.rs
+++ b/beacon_node/lighthouse_network/src/rpc/protocol.rs
@@ -93,7 +93,7 @@ pub static SIGNED_BEACON_BLOCK_DENEB_MAX: LazyLock<usize> = LazyLock::new(|| {
     *SIGNED_BEACON_BLOCK_CAPELLA_MAX_WITHOUT_PAYLOAD
     + types::ExecutionPayload::<MainnetEthSpec>::max_execution_payload_deneb_size() // adding max size of execution payload (~16gb)
     + ssz::BYTES_PER_LENGTH_OFFSET // Adding the additional offsets for the `ExecutionPayload`
-    + (<types::KzgCommitment as Encode>::ssz_fixed_len() * <MainnetEthSpec>::max_blobs_per_block())
+    + (<types::KzgCommitment as Encode>::ssz_fixed_len() * MAX_BLOBS_PER_BLOCK_CEILING as usize)
     + ssz::BYTES_PER_LENGTH_OFFSET
 }); // Length offset for the blob commitments field.
     //
@@ -101,7 +101,7 @@ pub static SIGNED_BEACON_BLOCK_ELECTRA_MAX: LazyLock<usize> = LazyLock::new(|| {
     *SIGNED_BEACON_BLOCK_ELECTRA_MAX_WITHOUT_PAYLOAD
     + types::ExecutionPayload::<MainnetEthSpec>::max_execution_payload_electra_size() // adding max size of execution payload (~16gb)
     + ssz::BYTES_PER_LENGTH_OFFSET // Adding the additional ssz offset for the `ExecutionPayload` field
-    + (<types::KzgCommitment as Encode>::ssz_fixed_len() * <MainnetEthSpec>::max_blobs_per_block())
+    + (<types::KzgCommitment as Encode>::ssz_fixed_len() * MAX_BLOBS_PER_BLOCK_CEILING as usize)
     + ssz::BYTES_PER_LENGTH_OFFSET
 }); // Length offset for the blob commitments field.
 
@@ -111,14 +111,6 @@ pub static BLOB_SIDECAR_SIZE: LazyLock<usize> =
 pub static BLOB_SIDECAR_SIZE_MINIMAL: LazyLock<usize> =
     LazyLock::new(BlobSidecar::<MinimalEthSpec>::max_size);
 
-pub static DATA_COLUMNS_SIDECAR_MIN: LazyLock<usize> = LazyLock::new(|| {
-    DataColumnSidecar::<MainnetEthSpec>::empty()
-        .as_ssz_bytes()
-        .len()
-});
-pub static DATA_COLUMNS_SIDECAR_MAX: LazyLock<usize> =
-    LazyLock::new(DataColumnSidecar::<MainnetEthSpec>::max_size);
-
 pub static ERROR_TYPE_MIN: LazyLock<usize> = LazyLock::new(|| {
     VariableList::<u8, MaxErrorLen>::from(Vec::<u8>::new())
         .as_ssz_bytes()
@@ -611,8 +603,8 @@ impl ProtocolId {
             Protocol::BlocksByRoot => rpc_block_limits_by_fork(fork_context.current_fork()),
             Protocol::BlobsByRange => rpc_blob_limits::<E>(),
             Protocol::BlobsByRoot => rpc_blob_limits::<E>(),
-            Protocol::DataColumnsByRoot => rpc_data_column_limits(),
-            Protocol::DataColumnsByRange => rpc_data_column_limits(),
+            Protocol::DataColumnsByRoot => rpc_data_column_limits::<E>(),
+            Protocol::DataColumnsByRange => rpc_data_column_limits::<E>(),
             Protocol::Ping => RpcLimits::new(
                 <Ping as Encode>::ssz_fixed_len(),
                 <Ping as Encode>::ssz_fixed_len(),
@@ -692,8 +684,11 @@ pub fn rpc_blob_limits<E: EthSpec>() -> RpcLimits {
     }
 }
 
-pub fn rpc_data_column_limits() -> RpcLimits {
-    RpcLimits::new(*DATA_COLUMNS_SIDECAR_MIN, *DATA_COLUMNS_SIDECAR_MAX)
+pub fn rpc_data_column_limits<E: EthSpec>() -> RpcLimits {
+    RpcLimits::new(
+        DataColumnSidecar::<E>::empty().as_ssz_bytes().len(),
+        DataColumnSidecar::<E>::max_size(MAX_BLOBS_PER_BLOCK_CEILING as usize),
+    )
 }
 
 /* Inbound upgrade */
diff --git a/beacon_node/network/src/network_beacon_processor/tests.rs b/beacon_node/network/src/network_beacon_processor/tests.rs
index 7e27a91bd6b..8238fa146dd 100644
--- a/beacon_node/network/src/network_beacon_processor/tests.rs
+++ b/beacon_node/network/src/network_beacon_processor/tests.rs
@@ -259,7 +259,7 @@ impl TestRig {
         assert!(beacon_processor.is_ok());
         let block = next_block_tuple.0;
         let blob_sidecars = if let Some((kzg_proofs, blobs)) = next_block_tuple.1 {
-            Some(BlobSidecar::build_sidecars(blobs, &block, kzg_proofs).unwrap())
+            Some(BlobSidecar::build_sidecars(blobs, &block, kzg_proofs, &chain.spec).unwrap())
         } else {
             None
         };
@@ -344,7 +344,7 @@ impl TestRig {
     }
     pub fn enqueue_single_lookup_rpc_blobs(&self) {
         if let Some(blobs) = self.next_blobs.clone() {
-            let blobs = FixedBlobSidecarList::from(blobs.into_iter().map(Some).collect::<Vec<_>>());
+            let blobs = FixedBlobSidecarList::new(blobs.into_iter().map(Some).collect::<Vec<_>>());
             self.network_beacon_processor
                 .send_rpc_blobs(
                     self.next_block.canonical_root(),
@@ -1130,7 +1130,12 @@ async fn test_blobs_by_range() {
             .block_root_at_slot(Slot::new(slot), WhenSlotSkipped::None)
             .unwrap();
         blob_count += root
-            .map(|root| rig.chain.get_blobs(&root).unwrap_or_default().len())
+            .map(|root| {
+                rig.chain
+                    .get_blobs(&root)
+                    .map(|list| list.len())
+                    .unwrap_or(0)
+            })
             .unwrap_or(0);
     }
     let mut actual_count = 0;
diff --git a/beacon_node/network/src/sync/block_sidecar_coupling.rs b/beacon_node/network/src/sync/block_sidecar_coupling.rs
index 966ce55fabe..7a234eaef04 100644
--- a/beacon_node/network/src/sync/block_sidecar_coupling.rs
+++ b/beacon_node/network/src/sync/block_sidecar_coupling.rs
@@ -2,13 +2,13 @@ use beacon_chain::{
     block_verification_types::RpcBlock, data_column_verification::CustodyDataColumn, get_block_root,
 };
 use lighthouse_network::PeerId;
-use ssz_types::VariableList;
 use std::{
     collections::{HashMap, VecDeque},
     sync::Arc,
 };
 use types::{
-    BlobSidecar, ChainSpec, ColumnIndex, DataColumnSidecar, EthSpec, Hash256, SignedBeaconBlock,
+    BlobSidecar, ChainSpec, ColumnIndex, DataColumnSidecar, EthSpec, Hash256, RuntimeVariableList,
+    SignedBeaconBlock,
 };
 
 #[derive(Debug)]
@@ -31,6 +31,7 @@ pub struct RangeBlockComponentsRequest<E: EthSpec> {
     num_custody_column_requests: Option<usize>,
     /// The peers the request was made to.
     pub(crate) peer_ids: Vec<PeerId>,
+    max_blobs_per_block: usize,
 }
 
 impl<E: EthSpec> RangeBlockComponentsRequest<E> {
@@ -39,6 +40,7 @@ impl<E: EthSpec> RangeBlockComponentsRequest<E> {
         expects_custody_columns: Option<Vec<ColumnIndex>>,
         num_custody_column_requests: Option<usize>,
         peer_ids: Vec<PeerId>,
+        max_blobs_per_block: usize,
     ) -> Self {
         Self {
             blocks: <_>::default(),
@@ -51,6 +53,7 @@ impl<E: EthSpec> RangeBlockComponentsRequest<E> {
             expects_custody_columns,
             num_custody_column_requests,
             peer_ids,
+            max_blobs_per_block,
         }
     }
 
@@ -100,7 +103,7 @@ impl<E: EthSpec> RangeBlockComponentsRequest<E> {
         let mut responses = Vec::with_capacity(blocks.len());
         let mut blob_iter = blobs.into_iter().peekable();
         for block in blocks.into_iter() {
-            let mut blob_list = Vec::with_capacity(E::max_blobs_per_block());
+            let mut blob_list = Vec::with_capacity(self.max_blobs_per_block);
             while {
                 let pair_next_blob = blob_iter
                     .peek()
@@ -111,7 +114,7 @@ impl<E: EthSpec> RangeBlockComponentsRequest<E> {
                 blob_list.push(blob_iter.next().ok_or("Missing next blob".to_string())?);
             }
 
-            let mut blobs_buffer = vec![None; E::max_blobs_per_block()];
+            let mut blobs_buffer = vec![None; self.max_blobs_per_block];
             for blob in blob_list {
                 let blob_index = blob.index as usize;
                 let Some(blob_opt) = blobs_buffer.get_mut(blob_index) else {
@@ -123,7 +126,11 @@ impl<E: EthSpec> RangeBlockComponentsRequest<E> {
                     *blob_opt = Some(blob);
                 }
             }
-            let blobs = VariableList::from(blobs_buffer.into_iter().flatten().collect::<Vec<_>>());
+            let blobs = RuntimeVariableList::new(
+                blobs_buffer.into_iter().flatten().collect::<Vec<_>>(),
+                self.max_blobs_per_block,
+            )
+            .map_err(|_| "Blobs returned exceeds max length".to_string())?;
             responses.push(RpcBlock::new(None, block, Some(blobs)).map_err(|e| format!("{e:?}"))?)
         }
 
@@ -245,12 +252,18 @@ mod tests {
 
     #[test]
     fn no_blobs_into_responses() {
+        let spec = test_spec::<E>();
         let peer_id = PeerId::random();
-        let mut info = RangeBlockComponentsRequest::<E>::new(false, None, None, vec![peer_id]);
         let mut rng = XorShiftRng::from_seed([42; 16]);
         let blocks = (0..4)
-            .map(|_| generate_rand_block_and_blobs::<E>(ForkName::Base, NumBlobs::None, &mut rng).0)
+            .map(|_| {
+                generate_rand_block_and_blobs::<E>(ForkName::Base, NumBlobs::None, &mut rng, &spec)
+                    .0
+            })
             .collect::<Vec<_>>();
+        let max_len = spec.max_blobs_per_block(blocks.first().unwrap().epoch()) as usize;
+        let mut info =
+            RangeBlockComponentsRequest::<E>::new(false, None, None, vec![peer_id], max_len);
 
         // Send blocks and complete terminate response
         for block in blocks {
@@ -265,15 +278,24 @@ mod tests {
 
     #[test]
     fn empty_blobs_into_responses() {
+        let spec = test_spec::<E>();
         let peer_id = PeerId::random();
-        let mut info = RangeBlockComponentsRequest::<E>::new(true, None, None, vec![peer_id]);
         let mut rng = XorShiftRng::from_seed([42; 16]);
         let blocks = (0..4)
             .map(|_| {
                 // Always generate some blobs.
-                generate_rand_block_and_blobs::<E>(ForkName::Deneb, NumBlobs::Number(3), &mut rng).0
+                generate_rand_block_and_blobs::<E>(
+                    ForkName::Deneb,
+                    NumBlobs::Number(3),
+                    &mut rng,
+                    &spec,
+                )
+                .0
             })
             .collect::<Vec<_>>();
+        let max_len = spec.max_blobs_per_block(blocks.first().unwrap().epoch()) as usize;
+        let mut info =
+            RangeBlockComponentsRequest::<E>::new(true, None, None, vec![peer_id], max_len);
 
         // Send blocks and complete terminate response
         for block in blocks {
@@ -294,12 +316,7 @@ mod tests {
     fn rpc_block_with_custody_columns() {
         let spec = test_spec::<E>();
         let expects_custody_columns = vec![1, 2, 3, 4];
-        let mut info = RangeBlockComponentsRequest::<E>::new(
-            false,
-            Some(expects_custody_columns.clone()),
-            Some(expects_custody_columns.len()),
-            vec![PeerId::random()],
-        );
+
         let mut rng = XorShiftRng::from_seed([42; 16]);
         let blocks = (0..4)
             .map(|_| {
@@ -311,7 +328,14 @@ mod tests {
                 )
             })
             .collect::<Vec<_>>();
-
+        let max_len = spec.max_blobs_per_block(blocks.first().unwrap().0.epoch()) as usize;
+        let mut info = RangeBlockComponentsRequest::<E>::new(
+            false,
+            Some(expects_custody_columns.clone()),
+            Some(expects_custody_columns.len()),
+            vec![PeerId::random()],
+            max_len,
+        );
         // Send blocks and complete terminate response
         for block in &blocks {
             info.add_block_response(Some(block.0.clone().into()));
@@ -355,12 +379,7 @@ mod tests {
         let spec = test_spec::<E>();
         let expects_custody_columns = vec![1, 2, 3, 4];
         let num_of_data_column_requests = 2;
-        let mut info = RangeBlockComponentsRequest::<E>::new(
-            false,
-            Some(expects_custody_columns.clone()),
-            Some(num_of_data_column_requests),
-            vec![PeerId::random()],
-        );
+
         let mut rng = XorShiftRng::from_seed([42; 16]);
         let blocks = (0..4)
             .map(|_| {
@@ -372,7 +391,14 @@ mod tests {
                 )
             })
             .collect::<Vec<_>>();
-
+        let max_len = spec.max_blobs_per_block(blocks.first().unwrap().0.epoch()) as usize;
+        let mut info = RangeBlockComponentsRequest::<E>::new(
+            false,
+            Some(expects_custody_columns.clone()),
+            Some(num_of_data_column_requests),
+            vec![PeerId::random()],
+            max_len,
+        );
         // Send blocks and complete terminate response
         for block in &blocks {
             info.add_block_response(Some(block.0.clone().into()));
diff --git a/beacon_node/network/src/sync/manager.rs b/beacon_node/network/src/sync/manager.rs
index 5d02be2b4c1..2df8b5f94c0 100644
--- a/beacon_node/network/src/sync/manager.rs
+++ b/beacon_node/network/src/sync/manager.rs
@@ -1234,6 +1234,7 @@ impl<T: BeaconChainTypes> SyncManager<T> {
             .network
             .range_block_and_blob_response(id, block_or_blob)
         {
+            let epoch = resp.sender_id.batch_id();
             match resp.responses {
                 Ok(blocks) => {
                     match resp.sender_id {
@@ -1277,6 +1278,7 @@ impl<T: BeaconChainTypes> SyncManager<T> {
                             resp.expects_custody_columns,
                             None,
                             vec![],
+                            self.chain.spec.max_blobs_per_block(epoch) as usize,
                         ),
                     );
                     // inform range that the request needs to be treated as failed
diff --git a/beacon_node/network/src/sync/network_context.rs b/beacon_node/network/src/sync/network_context.rs
index b6b7b315f3f..ed413cb4597 100644
--- a/beacon_node/network/src/sync/network_context.rs
+++ b/beacon_node/network/src/sync/network_context.rs
@@ -67,6 +67,15 @@ pub enum RangeRequestId {
     },
 }
 
+impl RangeRequestId {
+    pub fn batch_id(&self) -> BatchId {
+        match self {
+            RangeRequestId::RangeSync { batch_id, .. } => *batch_id,
+            RangeRequestId::BackfillSync { batch_id, .. } => *batch_id,
+        }
+    }
+}
+
 #[derive(Debug)]
 pub enum RpcEvent<T> {
     StreamTermination,
@@ -445,11 +454,14 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
                 (None, None)
             };
 
+        // TODO(pawan): this would break if a batch contains multiple epochs
+        let max_blobs_len = self.chain.spec.max_blobs_per_block(epoch);
         let info = RangeBlockComponentsRequest::new(
             expected_blobs,
             expects_columns,
             num_of_column_req,
             requested_peers,
+            max_blobs_len as usize,
         );
         self.range_block_components_requests
             .insert(id, (sender_id, info));
@@ -950,12 +962,17 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
     ) -> Option<RpcResponseResult<FixedBlobSidecarList<T::EthSpec>>> {
         let response = self.blobs_by_root_requests.on_response(id, rpc_event);
         let response = response.map(|res| {
-            res.and_then(
-                |(blobs, seen_timestamp)| match to_fixed_blob_sidecar_list(blobs) {
+            res.and_then(|(blobs, seen_timestamp)| {
+                let max_len = if let Some(blob) = blobs.first() {
+                    self.chain.spec.max_blobs_per_block(blob.epoch()) as usize
+                } else {
+                    6
+                };
+                match to_fixed_blob_sidecar_list(blobs, max_len) {
                     Ok(blobs) => Ok((blobs, seen_timestamp)),
                     Err(e) => Err(e.into()),
-                },
-            )
+                }
+            })
         });
         if let Some(Err(RpcResponseError::VerifyError(e))) = &response {
             self.report_peer(peer_id, PeerAction::LowToleranceError, e.into());
@@ -1150,8 +1167,9 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
 
 fn to_fixed_blob_sidecar_list<E: EthSpec>(
     blobs: Vec<Arc<BlobSidecar<E>>>,
+    max_len: usize,
 ) -> Result<FixedBlobSidecarList<E>, LookupVerifyError> {
-    let mut fixed_list = FixedBlobSidecarList::default();
+    let mut fixed_list = FixedBlobSidecarList::new(vec![None; max_len]);
     for blob in blobs.into_iter() {
         let index = blob.index as usize;
         *fixed_list
diff --git a/beacon_node/network/src/sync/tests/lookups.rs b/beacon_node/network/src/sync/tests/lookups.rs
index 94aacad3e81..ee96db0f636 100644
--- a/beacon_node/network/src/sync/tests/lookups.rs
+++ b/beacon_node/network/src/sync/tests/lookups.rs
@@ -119,6 +119,8 @@ impl TestRig {
             .network_globals
             .set_sync_state(SyncState::Synced);
 
+        let spec = chain.spec.clone();
+
         let rng = XorShiftRng::from_seed([42; 16]);
         TestRig {
             beacon_processor_rx,
@@ -142,6 +144,7 @@ impl TestRig {
             harness,
             fork_name,
             log,
+            spec,
         }
     }
 
@@ -213,7 +216,7 @@ impl TestRig {
     ) -> (SignedBeaconBlock<E>, Vec<BlobSidecar<E>>) {
         let fork_name = self.fork_name;
         let rng = &mut self.rng;
-        generate_rand_block_and_blobs::<E>(fork_name, num_blobs, rng)
+        generate_rand_block_and_blobs::<E>(fork_name, num_blobs, rng, &self.spec)
     }
 
     fn rand_block_and_data_columns(
@@ -1328,8 +1331,10 @@ impl TestRig {
 
 #[test]
 fn stable_rng() {
+    let spec = types::MainnetEthSpec::default_spec();
     let mut rng = XorShiftRng::from_seed([42; 16]);
-    let (block, _) = generate_rand_block_and_blobs::<E>(ForkName::Base, NumBlobs::None, &mut rng);
+    let (block, _) =
+        generate_rand_block_and_blobs::<E>(ForkName::Base, NumBlobs::None, &mut rng, &spec);
     assert_eq!(
         block.canonical_root(),
         Hash256::from_slice(
@@ -2187,8 +2192,8 @@ mod deneb_only {
         block_verification_types::{AsBlock, RpcBlock},
         data_availability_checker::AvailabilityCheckError,
     };
-    use ssz_types::VariableList;
     use std::collections::VecDeque;
+    use types::RuntimeVariableList;
 
     struct DenebTester {
         rig: TestRig,
@@ -2546,12 +2551,15 @@ mod deneb_only {
         fn parent_block_unknown_parent(mut self) -> Self {
             self.rig.log("parent_block_unknown_parent");
             let block = self.unknown_parent_block.take().unwrap();
+            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(
                 Some(block.canonical_root()),
                 block,
-                self.unknown_parent_blobs.take().map(VariableList::from),
+                self.unknown_parent_blobs
+                    .take()
+                    .map(|vec| RuntimeVariableList::from_vec(vec, max_len)),
             )
             .unwrap();
             self.rig.parent_block_processed(
diff --git a/beacon_node/network/src/sync/tests/mod.rs b/beacon_node/network/src/sync/tests/mod.rs
index 47666b413c5..6ed5c7f8fab 100644
--- a/beacon_node/network/src/sync/tests/mod.rs
+++ b/beacon_node/network/src/sync/tests/mod.rs
@@ -12,7 +12,7 @@ use slot_clock::ManualSlotClock;
 use std::sync::Arc;
 use store::MemoryStore;
 use tokio::sync::mpsc;
-use types::{test_utils::XorShiftRng, ForkName, MinimalEthSpec as E};
+use types::{test_utils::XorShiftRng, ChainSpec, ForkName, MinimalEthSpec as E};
 
 mod lookups;
 mod range;
@@ -64,4 +64,5 @@ struct TestRig {
     rng: XorShiftRng,
     fork_name: ForkName,
     log: Logger,
+    spec: Arc<ChainSpec>,
 }
diff --git a/beacon_node/store/src/hot_cold_store.rs b/beacon_node/store/src/hot_cold_store.rs
index da3e6d4ebcb..b313051b410 100644
--- a/beacon_node/store/src/hot_cold_store.rs
+++ b/beacon_node/store/src/hot_cold_store.rs
@@ -2057,7 +2057,20 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
             .get_bytes(DBColumn::BeaconBlob.into(), block_root.as_slice())?
         {
             Some(ref blobs_bytes) => {
-                let blobs = BlobSidecarList::from_ssz_bytes(blobs_bytes)?;
+                // We insert a VariableList of BlobSidecars into the db, but retrieve
+                // a plain vec since we don't know the length limit of the list without
+                // knowing the slot.
+                // The encoding of a VariableList is the same as a regular vec.
+                let blobs: Vec<Arc<BlobSidecar<E>>> = Vec::<_>::from_ssz_bytes(blobs_bytes)?;
+                let blobs = if let Some(max_blobs_per_block) = blobs
+                    .first()
+                    .map(|blob| self.spec.max_blobs_per_block(blob.epoch()))
+                {
+                    BlobSidecarList::from_vec(blobs, max_blobs_per_block as usize)
+                } else {
+                    // This always implies that there were no blobs for this block_root
+                    BlobSidecarList::empty_uninitialized()
+                };
                 self.block_cache
                     .lock()
                     .put_blobs(*block_root, blobs.clone());
diff --git a/beacon_node/store/src/impls/execution_payload.rs b/beacon_node/store/src/impls/execution_payload.rs
index 14fc10ad6de..3f7c4172bc9 100644
--- a/beacon_node/store/src/impls/execution_payload.rs
+++ b/beacon_node/store/src/impls/execution_payload.rs
@@ -1,7 +1,7 @@
 use crate::{DBColumn, Error, StoreItem};
 use ssz::{Decode, Encode};
 use types::{
-    BlobSidecarList, EthSpec, ExecutionPayload, ExecutionPayloadBellatrix, ExecutionPayloadCapella,
+    EthSpec, ExecutionPayload, ExecutionPayloadBellatrix, ExecutionPayloadCapella,
     ExecutionPayloadDeneb, ExecutionPayloadElectra,
 };
 
@@ -26,7 +26,6 @@ impl_store_item!(ExecutionPayloadBellatrix);
 impl_store_item!(ExecutionPayloadCapella);
 impl_store_item!(ExecutionPayloadDeneb);
 impl_store_item!(ExecutionPayloadElectra);
-impl_store_item!(BlobSidecarList);
 
 /// This fork-agnostic implementation should be only used for writing.
 ///
diff --git a/common/eth2_network_config/built_in_network_configs/chiado/config.yaml b/common/eth2_network_config/built_in_network_configs/chiado/config.yaml
index 1eca01bbeef..2fc3c7e945a 100644
--- a/common/eth2_network_config/built_in_network_configs/chiado/config.yaml
+++ b/common/eth2_network_config/built_in_network_configs/chiado/config.yaml
@@ -136,9 +136,11 @@ MAX_REQUEST_BLOB_SIDECARS: 768
 MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUESTS: 16384
 # `6`
 BLOB_SIDECAR_SUBNET_COUNT: 6
+# `uint64(6)`
+MAX_BLOBS_PER_BLOCK: 6
 
 # DAS
 CUSTODY_REQUIREMENT: 4
 DATA_COLUMN_SIDECAR_SUBNET_COUNT: 128
 NUMBER_OF_COLUMNS: 128
-SAMPLES_PER_SLOT: 8
\ No newline at end of file
+SAMPLES_PER_SLOT: 8
diff --git a/common/eth2_network_config/built_in_network_configs/gnosis/config.yaml b/common/eth2_network_config/built_in_network_configs/gnosis/config.yaml
index 500555a2694..b8fe4d51857 100644
--- a/common/eth2_network_config/built_in_network_configs/gnosis/config.yaml
+++ b/common/eth2_network_config/built_in_network_configs/gnosis/config.yaml
@@ -119,9 +119,11 @@ MAX_REQUEST_BLOB_SIDECARS: 768
 MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUESTS: 16384
 # `6`
 BLOB_SIDECAR_SUBNET_COUNT: 6
+# `uint64(6)`
+MAX_BLOBS_PER_BLOCK: 6
 
 # DAS
 CUSTODY_REQUIREMENT: 4
 DATA_COLUMN_SIDECAR_SUBNET_COUNT: 128
 NUMBER_OF_COLUMNS: 128
-SAMPLES_PER_SLOT: 8
\ No newline at end of file
+SAMPLES_PER_SLOT: 8
diff --git a/common/eth2_network_config/built_in_network_configs/holesky/config.yaml b/common/eth2_network_config/built_in_network_configs/holesky/config.yaml
index d67d77d3bea..1ba0f03641a 100644
--- a/common/eth2_network_config/built_in_network_configs/holesky/config.yaml
+++ b/common/eth2_network_config/built_in_network_configs/holesky/config.yaml
@@ -123,9 +123,11 @@ MAX_REQUEST_BLOB_SIDECARS: 768
 MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUESTS: 4096
 # `6`
 BLOB_SIDECAR_SUBNET_COUNT: 6
+# `uint64(6)`
+MAX_BLOBS_PER_BLOCK: 6
 
 # DAS
 CUSTODY_REQUIREMENT: 4
 DATA_COLUMN_SIDECAR_SUBNET_COUNT: 128
 NUMBER_OF_COLUMNS: 128
-SAMPLES_PER_SLOT: 8
\ No newline at end of file
+SAMPLES_PER_SLOT: 8
diff --git a/common/eth2_network_config/built_in_network_configs/mainnet/config.yaml b/common/eth2_network_config/built_in_network_configs/mainnet/config.yaml
index 18591fecdcd..ea3469e3f1d 100644
--- a/common/eth2_network_config/built_in_network_configs/mainnet/config.yaml
+++ b/common/eth2_network_config/built_in_network_configs/mainnet/config.yaml
@@ -145,9 +145,11 @@ MAX_REQUEST_BLOB_SIDECARS: 768
 MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUESTS: 4096
 # `6`
 BLOB_SIDECAR_SUBNET_COUNT: 6
+# `uint64(6)`
+MAX_BLOBS_PER_BLOCK: 6
 
 # DAS
 CUSTODY_REQUIREMENT: 4
 DATA_COLUMN_SIDECAR_SUBNET_COUNT: 128
 NUMBER_OF_COLUMNS: 128
-SAMPLES_PER_SLOT: 8
\ No newline at end of file
+SAMPLES_PER_SLOT: 8
diff --git a/common/eth2_network_config/built_in_network_configs/sepolia/config.yaml b/common/eth2_network_config/built_in_network_configs/sepolia/config.yaml
index b08a6180bf0..38185188976 100644
--- a/common/eth2_network_config/built_in_network_configs/sepolia/config.yaml
+++ b/common/eth2_network_config/built_in_network_configs/sepolia/config.yaml
@@ -119,9 +119,11 @@ MAX_REQUEST_BLOB_SIDECARS: 768
 MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUESTS: 4096
 # `6`
 BLOB_SIDECAR_SUBNET_COUNT: 6
+# `uint64(6)`
+MAX_BLOBS_PER_BLOCK: 6
 
 # DAS
 CUSTODY_REQUIREMENT: 4
 DATA_COLUMN_SIDECAR_SUBNET_COUNT: 128
 NUMBER_OF_COLUMNS: 128
-SAMPLES_PER_SLOT: 8
\ No newline at end of file
+SAMPLES_PER_SLOT: 8
diff --git a/consensus/state_processing/src/per_block_processing.rs b/consensus/state_processing/src/per_block_processing.rs
index 1f3d973603f..6aee217efd6 100644
--- a/consensus/state_processing/src/per_block_processing.rs
+++ b/consensus/state_processing/src/per_block_processing.rs
@@ -391,10 +391,12 @@ pub fn partially_verify_execution_payload<E: EthSpec, Payload: AbstractExecPaylo
 
     if let Ok(blob_commitments) = body.blob_kzg_commitments() {
         // Verify commitments are under the limit.
+        let max_blobs_per_block =
+            spec.max_blobs_per_block(block_slot.epoch(E::slots_per_epoch())) as usize;
         block_verify!(
-            blob_commitments.len() <= E::max_blobs_per_block(),
+            blob_commitments.len() <= max_blobs_per_block,
             BlockProcessingError::ExecutionInvalidBlobsLen {
-                max: E::max_blobs_per_block(),
+                max: max_blobs_per_block,
                 actual: blob_commitments.len(),
             }
         );
diff --git a/consensus/types/presets/gnosis/deneb.yaml b/consensus/types/presets/gnosis/deneb.yaml
index d2d7d0abed3..9a46a6dafe5 100644
--- a/consensus/types/presets/gnosis/deneb.yaml
+++ b/consensus/types/presets/gnosis/deneb.yaml
@@ -8,7 +8,5 @@
 FIELD_ELEMENTS_PER_BLOB: 4096
 # `uint64(2**12)` (= 4096)
 MAX_BLOB_COMMITMENTS_PER_BLOCK: 4096
-# `uint64(6)`
-MAX_BLOBS_PER_BLOCK: 6
 # `floorlog2(BLOB_KZG_COMMITMENTS_GINDEX) + 1 + ceillog2(MAX_BLOB_COMMITMENTS_PER_BLOCK)` = 4 + 1 + 12 = 17
 KZG_COMMITMENT_INCLUSION_PROOF_DEPTH: 17
diff --git a/consensus/types/src/beacon_block_body.rs b/consensus/types/src/beacon_block_body.rs
index b896dc46932..07cf6f41deb 100644
--- a/consensus/types/src/beacon_block_body.rs
+++ b/consensus/types/src/beacon_block_body.rs
@@ -13,8 +13,6 @@ use tree_hash_derive::TreeHash;
 
 pub type KzgCommitments<E> =
     VariableList<KzgCommitment, <E as EthSpec>::MaxBlobCommitmentsPerBlock>;
-pub type KzgCommitmentOpts<E> =
-    FixedVector<Option<KzgCommitment>, <E as EthSpec>::MaxBlobsPerBlock>;
 
 /// The number of leaves (including padding) on the `BeaconBlockBody` Merkle tree.
 ///
diff --git a/consensus/types/src/blob_sidecar.rs b/consensus/types/src/blob_sidecar.rs
index 5a330388cce..e241fd4b84e 100644
--- a/consensus/types/src/blob_sidecar.rs
+++ b/consensus/types/src/blob_sidecar.rs
@@ -1,10 +1,10 @@
 use crate::test_utils::TestRandom;
-use crate::ForkName;
 use crate::{
     beacon_block_body::BLOB_KZG_COMMITMENTS_INDEX, BeaconBlockHeader, BeaconStateError, Blob,
-    Epoch, EthSpec, FixedVector, Hash256, SignedBeaconBlockHeader, Slot, VariableList,
+    Epoch, EthSpec, FixedVector, ForkVersionDeserialize, Hash256, KzgProofs, RuntimeFixedVector,
+    RuntimeVariableList, SignedBeaconBlock, SignedBeaconBlockHeader, Slot, VariableList,
 };
-use crate::{ForkVersionDeserialize, KzgProofs, SignedBeaconBlock};
+use crate::{ChainSpec, ForkName};
 use bls::Signature;
 use derivative::Derivative;
 use kzg::{Blob as KzgBlob, Kzg, KzgCommitment, KzgProof, BYTES_PER_BLOB, BYTES_PER_FIELD_ELEMENT};
@@ -30,19 +30,6 @@ pub struct BlobIdentifier {
     pub index: u64,
 }
 
-impl BlobIdentifier {
-    pub fn get_all_blob_ids<E: EthSpec>(block_root: Hash256) -> Vec<BlobIdentifier> {
-        let mut blob_ids = Vec::with_capacity(E::max_blobs_per_block());
-        for i in 0..E::max_blobs_per_block() {
-            blob_ids.push(BlobIdentifier {
-                block_root,
-                index: i as u64,
-            });
-        }
-        blob_ids
-    }
-}
-
 impl PartialOrd for BlobIdentifier {
     fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
         Some(self.cmp(other))
@@ -291,19 +278,23 @@ impl<E: EthSpec> BlobSidecar<E> {
         blobs: BlobsList<E>,
         block: &SignedBeaconBlock<E>,
         kzg_proofs: KzgProofs<E>,
+        spec: &ChainSpec,
     ) -> Result<BlobSidecarList<E>, BlobSidecarError> {
         let mut blob_sidecars = vec![];
         for (i, (kzg_proof, blob)) in kzg_proofs.iter().zip(blobs).enumerate() {
             let blob_sidecar = BlobSidecar::new(i, blob, block, *kzg_proof)?;
             blob_sidecars.push(Arc::new(blob_sidecar));
         }
-        Ok(VariableList::from(blob_sidecars))
+        Ok(RuntimeVariableList::from_vec(
+            blob_sidecars,
+            spec.max_blobs_per_block(block.epoch()) as usize,
+        ))
     }
 }
 
-pub type BlobSidecarList<E> = VariableList<Arc<BlobSidecar<E>>, <E as EthSpec>::MaxBlobsPerBlock>;
-pub type FixedBlobSidecarList<E> =
-    FixedVector<Option<Arc<BlobSidecar<E>>>, <E as EthSpec>::MaxBlobsPerBlock>;
+pub type BlobSidecarList<E> = RuntimeVariableList<Arc<BlobSidecar<E>>>;
+/// Alias for a non length-constrained list of `BlobSidecar`s.
+pub type FixedBlobSidecarList<E> = RuntimeFixedVector<Option<Arc<BlobSidecar<E>>>>;
 pub type BlobsList<E> = VariableList<Blob<E>, <E as EthSpec>::MaxBlobCommitmentsPerBlock>;
 
 impl<E: EthSpec> ForkVersionDeserialize for BlobSidecarList<E> {
diff --git a/consensus/types/src/chain_spec.rs b/consensus/types/src/chain_spec.rs
index 0b33a76ff19..05d5142853a 100644
--- a/consensus/types/src/chain_spec.rs
+++ b/consensus/types/src/chain_spec.rs
@@ -229,6 +229,7 @@ pub struct ChainSpec {
     pub max_request_data_column_sidecars: u64,
     pub min_epochs_for_blob_sidecars_requests: u64,
     pub blob_sidecar_subnet_count: u64,
+    max_blobs_per_block: u64,
 
     /*
      * Networking Derived
@@ -605,6 +606,17 @@ impl ChainSpec {
         }
     }
 
+    /// Return the value of `MAX_BLOBS_PER_BLOCK` appropriate for the fork at `epoch`.
+    pub fn max_blobs_per_block(&self, epoch: Epoch) -> u64 {
+        self.max_blobs_per_block_by_fork(self.fork_name_at_epoch(epoch))
+    }
+
+    /// Return the value of `MAX_BLOBS_PER_BLOCK` appropriate for `fork`.
+    pub fn max_blobs_per_block_by_fork(&self, _fork_name: ForkName) -> u64 {
+        // TODO(electra): add Electra blobs per block change here
+        self.max_blobs_per_block
+    }
+
     pub fn data_columns_per_subnet(&self) -> usize {
         self.number_of_columns
             .safe_div(self.data_column_sidecar_subnet_count as usize)
@@ -841,6 +853,7 @@ impl ChainSpec {
             max_request_data_column_sidecars: default_max_request_data_column_sidecars(),
             min_epochs_for_blob_sidecars_requests: default_min_epochs_for_blob_sidecars_requests(),
             blob_sidecar_subnet_count: default_blob_sidecar_subnet_count(),
+            max_blobs_per_block: default_max_blobs_per_block(),
 
             /*
              * Derived Deneb Specific
@@ -1159,6 +1172,7 @@ impl ChainSpec {
             max_request_data_column_sidecars: default_max_request_data_column_sidecars(),
             min_epochs_for_blob_sidecars_requests: 16384,
             blob_sidecar_subnet_count: default_blob_sidecar_subnet_count(),
+            max_blobs_per_block: default_max_blobs_per_block(),
 
             /*
              * Derived Deneb Specific
@@ -1352,6 +1366,9 @@ pub struct Config {
     #[serde(default = "default_blob_sidecar_subnet_count")]
     #[serde(with = "serde_utils::quoted_u64")]
     blob_sidecar_subnet_count: u64,
+    #[serde(default = "default_max_blobs_per_block")]
+    #[serde(with = "serde_utils::quoted_u64")]
+    max_blobs_per_block: u64,
 
     #[serde(default = "default_min_per_epoch_churn_limit_electra")]
     #[serde(with = "serde_utils::quoted_u64")]
@@ -1482,6 +1499,12 @@ const fn default_blob_sidecar_subnet_count() -> u64 {
     6
 }
 
+/// Its important to keep this consistent with the deneb preset value for
+/// `MAX_BLOBS_PER_BLOCK` else we might run into consensus issues.
+const fn default_max_blobs_per_block() -> u64 {
+    6
+}
+
 const fn default_min_per_epoch_churn_limit_electra() -> u64 {
     128_000_000_000
 }
@@ -1699,6 +1722,7 @@ impl Config {
             max_request_data_column_sidecars: spec.max_request_data_column_sidecars,
             min_epochs_for_blob_sidecars_requests: spec.min_epochs_for_blob_sidecars_requests,
             blob_sidecar_subnet_count: spec.blob_sidecar_subnet_count,
+            max_blobs_per_block: spec.max_blobs_per_block,
 
             min_per_epoch_churn_limit_electra: spec.min_per_epoch_churn_limit_electra,
             max_per_epoch_activation_exit_churn_limit: spec
@@ -1774,6 +1798,7 @@ impl Config {
             max_request_data_column_sidecars,
             min_epochs_for_blob_sidecars_requests,
             blob_sidecar_subnet_count,
+            max_blobs_per_block,
 
             min_per_epoch_churn_limit_electra,
             max_per_epoch_activation_exit_churn_limit,
@@ -1840,6 +1865,7 @@ impl Config {
             max_request_data_column_sidecars,
             min_epochs_for_blob_sidecars_requests,
             blob_sidecar_subnet_count,
+            max_blobs_per_block,
 
             min_per_epoch_churn_limit_electra,
             max_per_epoch_activation_exit_churn_limit,
diff --git a/consensus/types/src/data_column_sidecar.rs b/consensus/types/src/data_column_sidecar.rs
index 57251e319a4..b2a050e9d5e 100644
--- a/consensus/types/src/data_column_sidecar.rs
+++ b/consensus/types/src/data_column_sidecar.rs
@@ -1,7 +1,7 @@
 use crate::beacon_block_body::{KzgCommitments, BLOB_KZG_COMMITMENTS_INDEX};
 use crate::test_utils::TestRandom;
 use crate::BeaconStateError;
-use crate::{BeaconBlockHeader, EthSpec, Hash256, KzgProofs, SignedBeaconBlockHeader, Slot};
+use crate::{BeaconBlockHeader, Epoch, EthSpec, Hash256, KzgProofs, SignedBeaconBlockHeader, Slot};
 use bls::Signature;
 use derivative::Derivative;
 use kzg::Error as KzgError;
@@ -11,7 +11,6 @@ use safe_arith::ArithError;
 use serde::{Deserialize, Serialize};
 use ssz::Encode;
 use ssz_derive::{Decode, Encode};
-use ssz_types::typenum::Unsigned;
 use ssz_types::Error as SszError;
 use ssz_types::{FixedVector, VariableList};
 use std::hash::Hash;
@@ -68,6 +67,10 @@ impl<E: EthSpec> DataColumnSidecar<E> {
         self.signed_block_header.message.slot
     }
 
+    pub fn epoch(&self) -> Epoch {
+        self.slot().epoch(E::slots_per_epoch())
+    }
+
     pub fn block_root(&self) -> Hash256 {
         self.signed_block_header.message.tree_hash_root()
     }
@@ -110,18 +113,16 @@ impl<E: EthSpec> DataColumnSidecar<E> {
         .len()
     }
 
-    pub fn max_size() -> usize {
+    pub fn max_size(max_blobs_per_block: usize) -> usize {
         Self {
             index: 0,
-            column: VariableList::new(vec![Cell::<E>::default(); E::MaxBlobsPerBlock::to_usize()])
-                .unwrap(),
+            column: VariableList::new(vec![Cell::<E>::default(); max_blobs_per_block]).unwrap(),
             kzg_commitments: VariableList::new(vec![
                 KzgCommitment::empty_for_testing();
-                E::MaxBlobsPerBlock::to_usize()
+                max_blobs_per_block
             ])
             .unwrap(),
-            kzg_proofs: VariableList::new(vec![KzgProof::empty(); E::MaxBlobsPerBlock::to_usize()])
-                .unwrap(),
+            kzg_proofs: VariableList::new(vec![KzgProof::empty(); max_blobs_per_block]).unwrap(),
             signed_block_header: SignedBeaconBlockHeader {
                 message: BeaconBlockHeader::empty(),
                 signature: Signature::empty(),
diff --git a/consensus/types/src/eth_spec.rs b/consensus/types/src/eth_spec.rs
index b14934cf4c0..0bc074072f6 100644
--- a/consensus/types/src/eth_spec.rs
+++ b/consensus/types/src/eth_spec.rs
@@ -4,7 +4,7 @@ use safe_arith::SafeArith;
 use serde::{Deserialize, Serialize};
 use ssz_types::typenum::{
     bit::B0, UInt, U0, U1, U10, U1024, U1048576, U1073741824, U1099511627776, U128, U131072,
-    U134217728, U16, U16777216, U17, U2, U2048, U256, U262144, U32, U4, U4096, U512, U6, U625, U64,
+    U134217728, U16, U16777216, U17, U2, U2048, U256, U262144, U32, U4, U4096, U512, U625, U64,
     U65536, U8, U8192,
 };
 use std::fmt::{self, Debug};
@@ -108,7 +108,6 @@ pub trait EthSpec:
     /*
      * New in Deneb
      */
-    type MaxBlobsPerBlock: Unsigned + Clone + Sync + Send + Debug + PartialEq + Unpin;
     type MaxBlobCommitmentsPerBlock: Unsigned + Clone + Sync + Send + Debug + PartialEq + Unpin;
     type FieldElementsPerBlob: Unsigned + Clone + Sync + Send + Debug + PartialEq;
     type BytesPerFieldElement: Unsigned + Clone + Sync + Send + Debug + PartialEq;
@@ -280,11 +279,6 @@ pub trait EthSpec:
         Self::MaxWithdrawalsPerPayload::to_usize()
     }
 
-    /// Returns the `MAX_BLOBS_PER_BLOCK` constant for this specification.
-    fn max_blobs_per_block() -> usize {
-        Self::MaxBlobsPerBlock::to_usize()
-    }
-
     /// Returns the `MAX_BLOB_COMMITMENTS_PER_BLOCK` constant for this specification.
     fn max_blob_commitments_per_block() -> usize {
         Self::MaxBlobCommitmentsPerBlock::to_usize()
@@ -420,7 +414,6 @@ impl EthSpec for MainnetEthSpec {
     type GasLimitDenominator = U1024;
     type MinGasLimit = U5000;
     type MaxExtraDataBytes = U32;
-    type MaxBlobsPerBlock = U6;
     type MaxBlobCommitmentsPerBlock = U4096;
     type BytesPerFieldElement = U32;
     type FieldElementsPerBlob = U4096;
@@ -504,7 +497,6 @@ impl EthSpec for MinimalEthSpec {
         MinGasLimit,
         MaxExtraDataBytes,
         MaxBlsToExecutionChanges,
-        MaxBlobsPerBlock,
         BytesPerFieldElement,
         PendingDepositsLimit,
         MaxPendingDepositsPerEpoch,
@@ -558,7 +550,6 @@ impl EthSpec for GnosisEthSpec {
     type SlotsPerEth1VotingPeriod = U1024; // 64 epochs * 16 slots per epoch
     type MaxBlsToExecutionChanges = U16;
     type MaxWithdrawalsPerPayload = U8;
-    type MaxBlobsPerBlock = U6;
     type MaxBlobCommitmentsPerBlock = U4096;
     type FieldElementsPerBlob = U4096;
     type BytesPerFieldElement = U32;
diff --git a/consensus/types/src/lib.rs b/consensus/types/src/lib.rs
index dd304c6296c..728266f19c3 100644
--- a/consensus/types/src/lib.rs
+++ b/consensus/types/src/lib.rs
@@ -108,6 +108,7 @@ pub mod data_column_sidecar;
 pub mod data_column_subnet_id;
 pub mod light_client_header;
 pub mod non_zero_usize;
+pub mod runtime_fixed_vector;
 pub mod runtime_var_list;
 
 pub use crate::activation_queue::ActivationQueue;
@@ -219,6 +220,7 @@ pub use crate::preset::{
 pub use crate::proposer_preparation_data::ProposerPreparationData;
 pub use crate::proposer_slashing::ProposerSlashing;
 pub use crate::relative_epoch::{Error as RelativeEpochError, RelativeEpoch};
+pub use crate::runtime_fixed_vector::RuntimeFixedVector;
 pub use crate::runtime_var_list::RuntimeVariableList;
 pub use crate::selection_proof::SelectionProof;
 pub use crate::shuffling_id::AttestationShufflingId;
diff --git a/consensus/types/src/preset.rs b/consensus/types/src/preset.rs
index b469b7b777a..949b6b2abef 100644
--- a/consensus/types/src/preset.rs
+++ b/consensus/types/src/preset.rs
@@ -205,8 +205,6 @@ impl CapellaPreset {
 #[derive(Debug, PartialEq, Clone, Serialize, Deserialize)]
 #[serde(rename_all = "UPPERCASE")]
 pub struct DenebPreset {
-    #[serde(with = "serde_utils::quoted_u64")]
-    pub max_blobs_per_block: u64,
     #[serde(with = "serde_utils::quoted_u64")]
     pub max_blob_commitments_per_block: u64,
     #[serde(with = "serde_utils::quoted_u64")]
@@ -216,7 +214,6 @@ pub struct DenebPreset {
 impl DenebPreset {
     pub fn from_chain_spec<E: EthSpec>(_spec: &ChainSpec) -> Self {
         Self {
-            max_blobs_per_block: E::max_blobs_per_block() as u64,
             max_blob_commitments_per_block: E::max_blob_commitments_per_block() as u64,
             field_elements_per_blob: E::field_elements_per_blob() as u64,
         }
diff --git a/consensus/types/src/runtime_fixed_vector.rs b/consensus/types/src/runtime_fixed_vector.rs
new file mode 100644
index 00000000000..f169df83bf5
--- /dev/null
+++ b/consensus/types/src/runtime_fixed_vector.rs
@@ -0,0 +1,78 @@
+/// Emulates a SSZ `Vector`.
+#[derive(Clone, Debug)]
+pub struct RuntimeFixedVector<T> {
+    vec: Vec<T>,
+    len: usize,
+}
+
+impl<T: Clone + Default> RuntimeFixedVector<T> {
+    pub fn new(vec: Vec<T>) -> Self {
+        let len = vec.len();
+        Self { vec, len }
+    }
+
+    pub fn to_vec(&self) -> Vec<T> {
+        self.vec.clone()
+    }
+
+    pub fn as_slice(&self) -> &[T] {
+        self.vec.as_slice()
+    }
+
+    #[allow(clippy::len_without_is_empty)]
+    pub fn len(&self) -> usize {
+        self.len
+    }
+
+    pub fn into_vec(self) -> Vec<T> {
+        self.vec
+    }
+
+    pub fn default(max_len: usize) -> Self {
+        Self {
+            vec: vec![T::default(); max_len],
+            len: max_len,
+        }
+    }
+
+    pub fn take(&mut self) -> Self {
+        let new = std::mem::take(&mut self.vec);
+        *self = Self::new(vec![T::default(); self.len]);
+        Self {
+            vec: new,
+            len: self.len,
+        }
+    }
+}
+
+impl<T> std::ops::Deref for RuntimeFixedVector<T> {
+    type Target = [T];
+
+    fn deref(&self) -> &[T] {
+        &self.vec[..]
+    }
+}
+
+impl<T> std::ops::DerefMut for RuntimeFixedVector<T> {
+    fn deref_mut(&mut self) -> &mut [T] {
+        &mut self.vec[..]
+    }
+}
+
+impl<T> IntoIterator for RuntimeFixedVector<T> {
+    type Item = T;
+    type IntoIter = std::vec::IntoIter<T>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        self.vec.into_iter()
+    }
+}
+
+impl<'a, T> IntoIterator for &'a RuntimeFixedVector<T> {
+    type Item = &'a T;
+    type IntoIter = std::slice::Iter<'a, T>;
+
+    fn into_iter(self) -> Self::IntoIter {
+        self.vec.iter()
+    }
+}
diff --git a/consensus/types/src/runtime_var_list.rs b/consensus/types/src/runtime_var_list.rs
index 8290876fa1f..db13da01f65 100644
--- a/consensus/types/src/runtime_var_list.rs
+++ b/consensus/types/src/runtime_var_list.rs
@@ -2,7 +2,7 @@ use derivative::Derivative;
 use serde::{Deserialize, Serialize};
 use ssz::Decode;
 use ssz_types::Error;
-use std::ops::{Deref, DerefMut, Index, IndexMut};
+use std::ops::{Deref, Index, IndexMut};
 use std::slice::SliceIndex;
 
 /// Emulates a SSZ `List`.
@@ -10,6 +10,11 @@ use std::slice::SliceIndex;
 /// An ordered, heap-allocated, variable-length, homogeneous collection of `T`, with no more than
 /// `max_len` values.
 ///
+/// In cases where the `max_length` of the container is unknown at time of initialization, we provide
+/// a `Self::empty_uninitialized` constructor that initializes a runtime list without setting the max_len.
+///
+/// To ensure there are no inconsistent states, we do not allow any mutating operation if `max_len` is not set.
+///
 /// ## Example
 ///
 /// ```
@@ -35,14 +40,26 @@ use std::slice::SliceIndex;
 ///
 /// // Push a value to if it _does_ exceed the maximum.
 /// assert!(long.push(6).is_err());
+///
+/// let mut uninit = RuntimeVariableList::empty_uninitialized();
+/// assert!(uninit.push(5).is_err());
+///
+/// // Set max_len to allow mutation.
+/// uninit.set_max_len(5usize);
+///
+/// uninit.push(5).unwrap();
+/// assert_eq!(&uninit[..], &[5]);
+///
 /// ```
 #[derive(Debug, Clone, Serialize, Deserialize, Derivative)]
 #[derivative(PartialEq, Eq, Hash(bound = "T: std::hash::Hash"))]
 #[serde(transparent)]
 pub struct RuntimeVariableList<T> {
     vec: Vec<T>,
+    /// A `None` here indicates an uninitialized `Self`.
+    /// No mutating operation will be allowed until `max_len` is Some
     #[serde(skip)]
-    max_len: usize,
+    max_len: Option<usize>,
 }
 
 impl<T> RuntimeVariableList<T> {
@@ -50,7 +67,10 @@ impl<T> RuntimeVariableList<T> {
     /// `Err(OutOfBounds { .. })`.
     pub fn new(vec: Vec<T>, max_len: usize) -> Result<Self, Error> {
         if vec.len() <= max_len {
-            Ok(Self { vec, max_len })
+            Ok(Self {
+                vec,
+                max_len: Some(max_len),
+            })
         } else {
             Err(Error::OutOfBounds {
                 i: vec.len(),
@@ -62,14 +82,17 @@ impl<T> RuntimeVariableList<T> {
     pub fn from_vec(mut vec: Vec<T>, max_len: usize) -> Self {
         vec.truncate(max_len);
 
-        Self { vec, max_len }
+        Self {
+            vec,
+            max_len: Some(max_len),
+        }
     }
 
-    /// Create an empty list.
+    /// Create an empty list with the given `max_len`.
     pub fn empty(max_len: usize) -> Self {
         Self {
             vec: vec![],
-            max_len,
+            max_len: Some(max_len),
         }
     }
 
@@ -77,6 +100,28 @@ impl<T> RuntimeVariableList<T> {
         self.vec.as_slice()
     }
 
+    pub fn as_mut_slice(&mut self) -> Option<&mut [T]> {
+        if self.max_len.is_none() {
+            return None;
+        };
+        Some(self.vec.as_mut_slice())
+    }
+
+    /// Returns an instance of `Self` with max_len = None.
+    ///
+    /// No mutating operation can be performed on an uninitialized instance
+    /// without first setting `max_len`.
+    pub fn empty_uninitialized() -> Self {
+        Self {
+            vec: vec![],
+            max_len: None,
+        }
+    }
+
+    pub fn set_max_len(&mut self, max_len: usize) {
+        self.max_len = Some(max_len);
+    }
+
     /// Returns the number of values presently in `self`.
     pub fn len(&self) -> usize {
         self.vec.len()
@@ -88,7 +133,9 @@ impl<T> RuntimeVariableList<T> {
     }
 
     /// Returns the type-level maximum length.
-    pub fn max_len(&self) -> usize {
+    ///
+    /// Returns `None` if self is uninitialized with a max_len.
+    pub fn max_len(&self) -> Option<usize> {
         self.max_len
     }
 
@@ -96,13 +143,17 @@ impl<T> RuntimeVariableList<T> {
     ///
     /// Returns `Err(())` when appending `value` would exceed the maximum length.
     pub fn push(&mut self, value: T) -> Result<(), Error> {
-        if self.vec.len() < self.max_len {
+        let Some(max_len) = self.max_len else {
+            // TODO(pawan): set a better error?
+            return Err(Error::MissingLengthInformation);
+        };
+        if self.vec.len() < max_len {
             self.vec.push(value);
             Ok(())
         } else {
             Err(Error::OutOfBounds {
                 i: self.vec.len().saturating_add(1),
-                len: self.max_len,
+                len: max_len,
             })
         }
     }
@@ -135,7 +186,10 @@ impl<T: Decode> RuntimeVariableList<T> {
         } else {
             ssz::decode_list_of_variable_length_items(bytes, Some(max_len))?
         };
-        Ok(Self { vec, max_len })
+        Ok(Self {
+            vec,
+            max_len: Some(max_len),
+        })
     }
 }
 
@@ -169,12 +223,6 @@ impl<T> Deref for RuntimeVariableList<T> {
     }
 }
 
-impl<T> DerefMut for RuntimeVariableList<T> {
-    fn deref_mut(&mut self) -> &mut [T] {
-        &mut self.vec[..]
-    }
-}
-
 impl<'a, T> IntoIterator for &'a RuntimeVariableList<T> {
     type Item = &'a T;
     type IntoIter = std::slice::Iter<'a, T>;
@@ -283,7 +331,7 @@ mod test {
     }
 
     fn round_trip<T: Encode + Decode + PartialEq + Debug>(item: RuntimeVariableList<T>) {
-        let max_len = item.max_len();
+        let max_len = item.max_len().unwrap();
         let encoded = &item.as_ssz_bytes();
         assert_eq!(item.ssz_bytes_len(), encoded.len());
         assert_eq!(
diff --git a/lcli/src/mock_el.rs b/lcli/src/mock_el.rs
index 8d3220b1df8..aad7c89ddad 100644
--- a/lcli/src/mock_el.rs
+++ b/lcli/src/mock_el.rs
@@ -9,6 +9,7 @@ use execution_layer::{
 };
 use std::net::Ipv4Addr;
 use std::path::PathBuf;
+use std::sync::Arc;
 use types::*;
 
 pub fn run<E: EthSpec>(mut env: Environment<E>, matches: &ArgMatches) -> Result<(), String> {
@@ -21,7 +22,7 @@ pub fn run<E: EthSpec>(mut env: Environment<E>, matches: &ArgMatches) -> Result<
     let prague_time = parse_optional(matches, "prague-time")?;
 
     let handle = env.core_context().executor.handle().unwrap();
-    let spec = &E::default_spec();
+    let spec = Arc::new(E::default_spec());
     let jwt_key = JwtKey::from_slice(&DEFAULT_JWT_SECRET).unwrap();
     std::fs::write(jwt_path, hex::encode(DEFAULT_JWT_SECRET)).unwrap();
 
@@ -39,7 +40,7 @@ pub fn run<E: EthSpec>(mut env: Environment<E>, matches: &ArgMatches) -> Result<
         prague_time,
     };
     let kzg = None;
-    let server: MockServer<E> = MockServer::new_with_config(&handle, config, kzg);
+    let server: MockServer<E> = MockServer::new_with_config(&handle, config, spec, kzg);
 
     if all_payloads_valid {
         eprintln!(
diff --git a/testing/node_test_rig/src/lib.rs b/testing/node_test_rig/src/lib.rs
index ac01c84b9d0..6e632ccf549 100644
--- a/testing/node_test_rig/src/lib.rs
+++ b/testing/node_test_rig/src/lib.rs
@@ -7,6 +7,7 @@ use environment::RuntimeContext;
 use eth2::{reqwest::ClientBuilder, BeaconNodeHttpClient, Timeouts};
 use sensitive_url::SensitiveUrl;
 use std::path::PathBuf;
+use std::sync::Arc;
 use std::time::Duration;
 use std::time::{SystemTime, UNIX_EPOCH};
 use tempfile::{Builder as TempBuilder, TempDir};
@@ -248,8 +249,14 @@ impl<E: EthSpec> LocalExecutionNode<E> {
         if let Err(e) = std::fs::write(jwt_file_path, config.jwt_key.hex_string()) {
             panic!("Failed to write jwt file {}", e);
         }
+        let spec = Arc::new(E::default_spec());
         Self {
-            server: MockServer::new_with_config(&context.executor.handle().unwrap(), config, None),
+            server: MockServer::new_with_config(
+                &context.executor.handle().unwrap(),
+                config,
+                spec,
+                None,
+            ),
             datadir,
         }
     }