diff --git a/Cargo.lock b/Cargo.lock
index 4fc8ef8fe97..1023d743bd7 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -7840,6 +7840,7 @@ dependencies = [
  "strum",
  "tempfile",
  "types",
+ "zstd",
 ]
 
 [[package]]
diff --git a/Cargo.toml b/Cargo.toml
index ce3b47012c8..91b6e6ec3f7 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -174,6 +174,7 @@ uuid = { version = "0.8", features = ["serde", "v4"] }
 warp = { git = "https://github.com/seanmonstar/warp.git", default-features = false, features = ["tls"] }
 zeroize = { version = "1", features = ["zeroize_derive"] }
 zip = "0.6"
+zstd = "0.11.2"
 
 # Local crates.
 account_utils = { path = "common/account_utils" }
diff --git a/beacon_node/beacon_chain/src/beacon_block_streamer.rs b/beacon_node/beacon_chain/src/beacon_block_streamer.rs
index 9312d4511d5..a9c4167cd19 100644
--- a/beacon_node/beacon_chain/src/beacon_block_streamer.rs
+++ b/beacon_node/beacon_chain/src/beacon_block_streamer.rs
@@ -430,7 +430,7 @@ impl<T: BeaconChainTypes> BeaconBlockStreamer<T> {
                 continue;
             }
 
-            match self.beacon_chain.store.try_get_full_block(&root) {
+            match self.beacon_chain.store.try_get_full_block(&root, None) {
                 Err(e) => db_blocks.push((root, Err(e.into()))),
                 Ok(opt_block) => db_blocks.push((
                     root,
diff --git a/beacon_node/beacon_chain/src/beacon_chain.rs b/beacon_node/beacon_chain/src/beacon_chain.rs
index b05ba01ee7e..d8b4e266c5e 100644
--- a/beacon_node/beacon_chain/src/beacon_chain.rs
+++ b/beacon_node/beacon_chain/src/beacon_chain.rs
@@ -176,6 +176,7 @@ pub const INVALID_FINALIZED_MERGE_TRANSITION_BLOCK_SHUTDOWN_REASON: &str =
     "Finalized merge transition block is invalid.";
 
 /// Defines the behaviour when a block/block-root for a skipped slot is requested.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum WhenSlotSkipped {
     /// If the slot is a skip slot, return `None`.
     ///
@@ -925,8 +926,14 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     ) -> Result<Option<SignedBlindedBeaconBlock<T::EthSpec>>, Error> {
         let root = self.block_root_at_slot(request_slot, skips)?;
 
+        // Only hint the slot if expect a block at this exact slot.
+        let slot_hint = match skips {
+            WhenSlotSkipped::Prev => None,
+            WhenSlotSkipped::None => Some(request_slot),
+        };
+
         if let Some(block_root) = root {
-            Ok(self.store.get_blinded_block(&block_root)?)
+            Ok(self.store.get_blinded_block(&block_root, slot_hint)?)
         } else {
             Ok(None)
         }
@@ -1189,7 +1196,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     ) -> Result<Option<SignedBeaconBlock<T::EthSpec>>, Error> {
         // Load block from database, returning immediately if we have the full block w payload
         // stored.
-        let blinded_block = match self.store.try_get_full_block(block_root)? {
+        let blinded_block = match self.store.try_get_full_block(block_root, None)? {
             Some(DatabaseBlock::Full(block)) => return Ok(Some(block)),
             Some(DatabaseBlock::Blinded(block)) => block,
             None => return Ok(None),
@@ -1257,7 +1264,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         &self,
         block_root: &Hash256,
     ) -> Result<Option<SignedBlindedBeaconBlock<T::EthSpec>>, Error> {
-        Ok(self.store.get_blinded_block(block_root)?)
+        Ok(self.store.get_blinded_block(block_root, None)?)
     }
 
     /// Returns the state at the given root, if any.
@@ -6365,7 +6372,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
             let beacon_block = self
                 .store
-                .get_blinded_block(&beacon_block_root)?
+                .get_blinded_block(&beacon_block_root, None)?
                 .ok_or_else(|| {
                     Error::DBInconsistent(format!("Missing block {}", beacon_block_root))
                 })?;
diff --git a/beacon_node/beacon_chain/src/beacon_fork_choice_store.rs b/beacon_node/beacon_chain/src/beacon_fork_choice_store.rs
index 2a42b49b422..900c6b1d8c3 100644
--- a/beacon_node/beacon_chain/src/beacon_fork_choice_store.rs
+++ b/beacon_node/beacon_chain/src/beacon_fork_choice_store.rs
@@ -315,7 +315,7 @@ where
             metrics::inc_counter(&metrics::BALANCES_CACHE_MISSES);
             let justified_block = self
                 .store
-                .get_blinded_block(&self.justified_checkpoint.root)
+                .get_blinded_block(&self.justified_checkpoint.root, None)
                 .map_err(Error::FailedToReadBlock)?
                 .ok_or(Error::MissingBlock(self.justified_checkpoint.root))?
                 .deconstruct()
diff --git a/beacon_node/beacon_chain/src/builder.rs b/beacon_node/beacon_chain/src/builder.rs
index abd0e6b6a82..d721987b32b 100644
--- a/beacon_node/beacon_chain/src/builder.rs
+++ b/beacon_node/beacon_chain/src/builder.rs
@@ -292,7 +292,7 @@ where
             .ok_or("Fork choice not found in store")?;
 
         let genesis_block = store
-            .get_blinded_block(&chain.genesis_block_root)
+            .get_blinded_block(&chain.genesis_block_root, Some(Slot::new(0)))
             .map_err(|e| descriptive_db_error("genesis block", &e))?
             .ok_or("Genesis block not found in store")?;
         let genesis_state = store
@@ -710,7 +710,7 @@ where
         // Try to decode the head block according to the current fork, if that fails, try
         // to backtrack to before the most recent fork.
         let (head_block_root, head_block, head_reverted) =
-            match store.get_full_block(&initial_head_block_root) {
+            match store.get_full_block(&initial_head_block_root, None) {
                 Ok(Some(block)) => (initial_head_block_root, block, false),
                 Ok(None) => return Err("Head block not found in store".into()),
                 Err(StoreError::SszDecodeError(_)) => {
diff --git a/beacon_node/beacon_chain/src/canonical_head.rs b/beacon_node/beacon_chain/src/canonical_head.rs
index ced4eda05cf..3113d15fcfc 100644
--- a/beacon_node/beacon_chain/src/canonical_head.rs
+++ b/beacon_node/beacon_chain/src/canonical_head.rs
@@ -297,7 +297,7 @@ impl<T: BeaconChainTypes> CanonicalHead<T> {
         let fork_choice_view = fork_choice.cached_fork_choice_view();
         let beacon_block_root = fork_choice_view.head_block_root;
         let beacon_block = store
-            .get_full_block(&beacon_block_root)?
+            .get_full_block(&beacon_block_root, None)?
             .ok_or(Error::MissingBeaconBlock(beacon_block_root))?;
         let current_slot = fork_choice.fc_store().get_current_slot();
         let (_, beacon_state) = store
@@ -667,7 +667,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 .unwrap_or_else(|| {
                     let beacon_block = self
                         .store
-                        .get_full_block(&new_view.head_block_root)?
+                        .get_full_block(&new_view.head_block_root, None)?
                         .ok_or(Error::MissingBeaconBlock(new_view.head_block_root))?;
 
                     let (_, beacon_state) = self
diff --git a/beacon_node/beacon_chain/src/fork_revert.rs b/beacon_node/beacon_chain/src/fork_revert.rs
index dc0e34277c9..52cfc474b49 100644
--- a/beacon_node/beacon_chain/src/fork_revert.rs
+++ b/beacon_node/beacon_chain/src/fork_revert.rs
@@ -110,7 +110,7 @@ pub fn reset_fork_choice_to_finalization<E: EthSpec, Hot: ItemStore<E>, Cold: It
     let finalized_checkpoint = head_state.finalized_checkpoint();
     let finalized_block_root = finalized_checkpoint.root;
     let finalized_block = store
-        .get_full_block(&finalized_block_root)
+        .get_full_block(&finalized_block_root, None)
         .map_err(|e| format!("Error loading finalized block: {:?}", e))?
         .ok_or_else(|| {
             format!(
diff --git a/beacon_node/beacon_chain/src/light_client_server_cache.rs b/beacon_node/beacon_chain/src/light_client_server_cache.rs
index 1397e3fc9df..10df90c3188 100644
--- a/beacon_node/beacon_chain/src/light_client_server_cache.rs
+++ b/beacon_node/beacon_chain/src/light_client_server_cache.rs
@@ -83,7 +83,7 @@ impl<T: BeaconChainTypes> LightClientServerCache<T> {
         let signature_slot = block_slot;
         let attested_block_root = block_parent_root;
 
-        let attested_block = store.get_blinded_block(attested_block_root)?.ok_or(
+        let attested_block = store.get_blinded_block(attested_block_root, None)?.ok_or(
             BeaconChainError::DBInconsistent(format!(
                 "Block not available {:?}",
                 attested_block_root
@@ -127,7 +127,7 @@ impl<T: BeaconChainTypes> LightClientServerCache<T> {
         if is_latest_finality & !cached_parts.finalized_block_root.is_zero() {
             // Immediately after checkpoint sync the finalized block may not be available yet.
             if let Some(finalized_block) =
-                store.get_blinded_block(&cached_parts.finalized_block_root)?
+                store.get_blinded_block(&cached_parts.finalized_block_root, None)?
             {
                 *self.latest_finality_update.write() = Some(LightClientFinalityUpdate {
                     // TODO: may want to cache this result from latest_optimistic_update if producing a
diff --git a/beacon_node/beacon_chain/src/migrate.rs b/beacon_node/beacon_chain/src/migrate.rs
index ad597bf92aa..31900256840 100644
--- a/beacon_node/beacon_chain/src/migrate.rs
+++ b/beacon_node/beacon_chain/src/migrate.rs
@@ -527,7 +527,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
             // so delete it from the head tracker but leave it and its states in the database
             // This is suboptimal as it wastes disk space, but it's difficult to fix. A re-sync
             // can be used to reclaim the space.
-            let head_state_root = match store.get_blinded_block(&head_hash) {
+            let head_state_root = match store.get_blinded_block(&head_hash, Some(head_slot)) {
                 Ok(Some(block)) => block.state_root(),
                 Ok(None) => {
                     return Err(BeaconStateError::MissingBeaconBlock(head_hash.into()).into())
diff --git a/beacon_node/beacon_chain/src/pre_finalization_cache.rs b/beacon_node/beacon_chain/src/pre_finalization_cache.rs
index 22b76e026cb..3b337d4228b 100644
--- a/beacon_node/beacon_chain/src/pre_finalization_cache.rs
+++ b/beacon_node/beacon_chain/src/pre_finalization_cache.rs
@@ -73,7 +73,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         }
 
         // 2. Check on disk.
-        if self.store.get_blinded_block(&block_root)?.is_some() {
+        if self.store.get_blinded_block(&block_root, None)?.is_some() {
             cache.block_roots.put(block_root, ());
             return Ok(true);
         }
diff --git a/beacon_node/beacon_chain/src/schema_change/migration_schema_v18.rs b/beacon_node/beacon_chain/src/schema_change/migration_schema_v18.rs
index 04a9da84128..07294f6951f 100644
--- a/beacon_node/beacon_chain/src/schema_change/migration_schema_v18.rs
+++ b/beacon_node/beacon_chain/src/schema_change/migration_schema_v18.rs
@@ -17,7 +17,7 @@ fn get_slot_clock<T: BeaconChainTypes>(
     log: &Logger,
 ) -> Result<Option<T::SlotClock>, Error> {
     let spec = db.get_chain_spec();
-    let Some(genesis_block) = db.get_blinded_block(&Hash256::zero())? else {
+    let Some(genesis_block) = db.get_blinded_block(&Hash256::zero(), Some(Slot::new(0)))? else {
         error!(log, "Missing genesis block");
         return Ok(None);
     };
diff --git a/beacon_node/src/cli.rs b/beacon_node/src/cli.rs
index 4ed60d27b0f..2885db2c3ed 100644
--- a/beacon_node/src/cli.rs
+++ b/beacon_node/src/cli.rs
@@ -736,6 +736,14 @@ pub fn cli_app<'a, 'b>() -> App<'a, 'b> {
                 .takes_value(true)
                 .default_value("true")
         )
+        .arg(
+            Arg::with_name("compression-level")
+                .long("compression-level")
+                .value_name("LEVEL")
+                .help("Compression level (-99 to 22) for zstd compression applied to states on disk \
+                       [default: 1]. You may change the compression level freely without re-syncing.")
+                .takes_value(true)
+        )
         .arg(
             Arg::with_name("prune-payloads")
                 .long("prune-payloads")
diff --git a/beacon_node/store/Cargo.toml b/beacon_node/store/Cargo.toml
index 7bf1ef76bef..5a11df817f1 100644
--- a/beacon_node/store/Cargo.toml
+++ b/beacon_node/store/Cargo.toml
@@ -25,3 +25,4 @@ lru = { workspace = true }
 sloggers = { workspace = true }
 directory = { workspace = true }
 strum = { workspace = true }
+zstd = { workspace = true }
diff --git a/beacon_node/store/src/config.rs b/beacon_node/store/src/config.rs
index 681d424e282..e74fa753487 100644
--- a/beacon_node/store/src/config.rs
+++ b/beacon_node/store/src/config.rs
@@ -9,6 +9,8 @@ use types::{EthSpec, MinimalEthSpec};
 pub const PREV_DEFAULT_SLOTS_PER_RESTORE_POINT: u64 = 2048;
 pub const DEFAULT_SLOTS_PER_RESTORE_POINT: u64 = 8192;
 pub const DEFAULT_BLOCK_CACHE_SIZE: NonZeroUsize = new_non_zero_usize(5);
+pub const DEFAULT_COMPRESSION_LEVEL: i32 = 1;
+const EST_COMPRESSION_FACTOR: usize = 2;
 pub const DEFAULT_HISTORIC_STATE_CACHE_SIZE: NonZeroUsize = new_non_zero_usize(1);
 pub const DEFAULT_EPOCHS_PER_BLOB_PRUNE: u64 = 1;
 pub const DEFAULT_BLOB_PUNE_MARGIN_EPOCHS: u64 = 0;
@@ -22,6 +24,8 @@ pub struct StoreConfig {
     pub slots_per_restore_point_set_explicitly: bool,
     /// Maximum number of blocks to store in the in-memory block cache.
     pub block_cache_size: NonZeroUsize,
+    /// Compression level for blocks, state diffs and other compressed values.
+    pub compression_level: i32,
     /// Maximum number of states from freezer database to store in the in-memory state cache.
     pub historic_state_cache_size: NonZeroUsize,
     /// Whether to compact the database on initialization.
@@ -30,6 +34,8 @@ pub struct StoreConfig {
     pub compact_on_prune: bool,
     /// Whether to prune payloads on initialization and finalization.
     pub prune_payloads: bool,
+    /// Whether to store finalized blocks compressed and linearised in the freezer database.
+    pub linear_blocks: bool,
     /// Whether to prune blobs older than the blob data availability boundary.
     pub prune_blobs: bool,
     /// Frequency of blob pruning in epochs. Default: 1 (every epoch).
@@ -57,10 +63,12 @@ impl Default for StoreConfig {
             slots_per_restore_point: MinimalEthSpec::slots_per_historical_root() as u64,
             slots_per_restore_point_set_explicitly: false,
             block_cache_size: DEFAULT_BLOCK_CACHE_SIZE,
+            compression_level: DEFAULT_COMPRESSION_LEVEL,
             historic_state_cache_size: DEFAULT_HISTORIC_STATE_CACHE_SIZE,
             compact_on_init: false,
             compact_on_prune: true,
             prune_payloads: true,
+            linear_blocks: true,
             prune_blobs: true,
             epochs_per_blob_prune: DEFAULT_EPOCHS_PER_BLOB_PRUNE,
             blob_prune_margin_epochs: DEFAULT_BLOB_PUNE_MARGIN_EPOCHS,
@@ -87,6 +95,25 @@ impl StoreConfig {
         }
         Ok(())
     }
+
+    /// Estimate the size of `len` bytes after compression at the current compression level.
+    pub fn estimate_compressed_size(&self, len: usize) -> usize {
+        if self.compression_level == 0 {
+            len
+        } else {
+            len / EST_COMPRESSION_FACTOR
+        }
+    }
+
+    /// Estimate the size of `len` compressed bytes after decompression at the current compression
+    /// level.
+    pub fn estimate_decompressed_size(&self, len: usize) -> usize {
+        if self.compression_level == 0 {
+            len
+        } else {
+            len * EST_COMPRESSION_FACTOR
+        }
+    }
 }
 
 impl StoreItem for OnDiskStoreConfig {
diff --git a/beacon_node/store/src/errors.rs b/beacon_node/store/src/errors.rs
index 96e02b80ff8..9a6c6d9f886 100644
--- a/beacon_node/store/src/errors.rs
+++ b/beacon_node/store/src/errors.rs
@@ -43,6 +43,7 @@ pub enum Error {
         computed: Hash256,
     },
     BlockReplayError(BlockReplayError),
+    Compression(std::io::Error),
     AddPayloadLogicError,
     SlotClockUnavailableForMigration,
     InvalidKey,
diff --git a/beacon_node/store/src/hot_cold_store.rs b/beacon_node/store/src/hot_cold_store.rs
index 49bc75393c1..71fa244c37b 100644
--- a/beacon_node/store/src/hot_cold_store.rs
+++ b/beacon_node/store/src/hot_cold_store.rs
@@ -6,7 +6,10 @@ use crate::config::{
     PREV_DEFAULT_SLOTS_PER_RESTORE_POINT,
 };
 use crate::forwards_iter::{HybridForwardsBlockRootsIterator, HybridForwardsStateRootsIterator};
-use crate::impls::beacon_state::{get_full_state, store_full_state};
+use crate::impls::{
+    beacon_state::{get_full_state, store_full_state},
+    frozen_block_slot::FrozenBlockSlot,
+};
 use crate::iter::{BlockRootsIterator, ParentRootBlockIterator, RootsIterator};
 use crate::leveldb_store::BytesKey;
 use crate::leveldb_store::LevelDB;
@@ -34,6 +37,7 @@ use state_processing::{
 };
 use std::cmp::min;
 use std::convert::TryInto;
+use std::io::{Read, Write};
 use std::marker::PhantomData;
 use std::num::NonZeroUsize;
 use std::path::Path;
@@ -41,6 +45,7 @@ use std::sync::Arc;
 use std::time::Duration;
 use types::blob_sidecar::BlobSidecarList;
 use types::*;
+use zstd::{Decoder, Encoder};
 
 /// On-disk database that stores finalized states efficiently.
 ///
@@ -413,6 +418,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
     pub fn try_get_full_block(
         &self,
         block_root: &Hash256,
+        slot: Option<Slot>,
     ) -> Result<Option<DatabaseBlock<E>>, Error> {
         metrics::inc_counter(&metrics::BEACON_BLOCK_GET_COUNT);
 
@@ -423,7 +429,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
         }
 
         // Load the blinded block.
-        let Some(blinded_block) = self.get_blinded_block(block_root)? else {
+        let Some(blinded_block) = self.get_blinded_block(block_root, slot)? else {
             return Ok(None);
         };
 
@@ -471,8 +477,9 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
     pub fn get_full_block(
         &self,
         block_root: &Hash256,
+        slot: Option<Slot>,
     ) -> Result<Option<SignedBeaconBlock<E>>, Error> {
-        match self.try_get_full_block(block_root)? {
+        match self.try_get_full_block(block_root, slot)? {
             Some(DatabaseBlock::Full(block)) => Ok(Some(block)),
             Some(DatabaseBlock::Blinded(block)) => Err(
                 HotColdDBError::MissingFullBlockExecutionPayloadPruned(*block_root, block.slot())
@@ -503,12 +510,115 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
     pub fn get_blinded_block(
         &self,
         block_root: &Hash256,
-    ) -> Result<Option<SignedBeaconBlock<E, BlindedPayload<E>>>, Error> {
+        slot: Option<Slot>,
+    ) -> Result<Option<SignedBlindedBeaconBlock<E>>, Error> {
+        let split = self.get_split_info();
+        if let Some(slot) = slot {
+            if (slot < split.slot || slot == 0) && *block_root != split.block_root {
+                // To the freezer DB.
+                self.get_cold_blinded_block_by_slot(slot)
+            } else {
+                self.get_hot_blinded_block(block_root)
+            }
+        } else {
+            match self.get_hot_blinded_block(block_root)? {
+                Some(block) => Ok(Some(block)),
+                None => self.get_cold_blinded_block_by_root(block_root),
+            }
+        }
+    }
+
+    pub fn get_hot_blinded_block(
+        &self,
+        block_root: &Hash256,
+    ) -> Result<Option<SignedBlindedBeaconBlock<E>>, Error> {
         self.get_block_with(block_root, |bytes| {
             SignedBeaconBlock::from_ssz_bytes(bytes, &self.spec)
         })
     }
 
+    pub fn get_cold_blinded_block_by_root(
+        &self,
+        block_root: &Hash256,
+    ) -> Result<Option<SignedBlindedBeaconBlock<E>>, Error> {
+        // Load slot.
+        if let Some(FrozenBlockSlot(block_slot)) = self.cold_db.get(block_root)? {
+            self.get_cold_blinded_block_by_slot(block_slot)
+        } else {
+            Ok(None)
+        }
+    }
+
+    pub fn get_cold_blinded_block_by_slot(
+        &self,
+        slot: Slot,
+    ) -> Result<Option<SignedBlindedBeaconBlock<E>>, Error> {
+        let Some(bytes) = self.cold_db.get_bytes(
+            DBColumn::BeaconBlockFrozen.into(),
+            &slot.as_u64().to_be_bytes(),
+        )?
+        else {
+            return Ok(None);
+        };
+
+        let mut ssz_bytes = Vec::with_capacity(self.config.estimate_decompressed_size(bytes.len()));
+        let mut decoder = Decoder::new(&*bytes).map_err(Error::Compression)?;
+        decoder
+            .read_to_end(&mut ssz_bytes)
+            .map_err(Error::Compression)?;
+        Ok(Some(SignedBeaconBlock::from_ssz_bytes(
+            &ssz_bytes, &self.spec,
+        )?))
+    }
+
+    pub fn put_cold_blinded_block(
+        &self,
+        block_root: &Hash256,
+        block: &SignedBlindedBeaconBlock<E>,
+    ) -> Result<(), Error> {
+        let mut ops = Vec::with_capacity(2);
+        self.blinded_block_as_cold_kv_store_ops(block_root, block, &mut ops)?;
+        self.cold_db.do_atomically(ops)
+    }
+
+    pub fn blinded_block_as_cold_kv_store_ops(
+        &self,
+        block_root: &Hash256,
+        block: &SignedBlindedBeaconBlock<E>,
+        kv_store_ops: &mut Vec<KeyValueStoreOp>,
+    ) -> Result<(), Error> {
+        // Write the block root to slot mapping.
+        let slot = block.slot();
+        kv_store_ops.push(FrozenBlockSlot(slot).as_kv_store_op(*block_root));
+
+        // Write the slot to block root mapping.
+        kv_store_ops.push(KeyValueStoreOp::PutKeyValue(
+            get_key_for_col(
+                DBColumn::BeaconBlockRoots.into(),
+                &slot.as_u64().to_be_bytes(),
+            ),
+            block_root.as_bytes().to_vec(),
+        ));
+
+        // Write the block keyed by slot.
+        let db_key = get_key_for_col(
+            DBColumn::BeaconBlockFrozen.into(),
+            &slot.as_u64().to_be_bytes(),
+        );
+
+        let ssz_bytes = block.as_ssz_bytes();
+        let mut compressed_value =
+            Vec::with_capacity(self.config.estimate_compressed_size(ssz_bytes.len()));
+        let mut encoder = Encoder::new(&mut compressed_value, self.config.compression_level)
+            .map_err(Error::Compression)?;
+        encoder.write_all(&ssz_bytes).map_err(Error::Compression)?;
+        encoder.finish().map_err(Error::Compression)?;
+
+        kv_store_ops.push(KeyValueStoreOp::PutKeyValue(db_key, compressed_value));
+
+        Ok(())
+    }
+
     /// Fetch a block from the store, ignoring which fork variant it *should* be for.
     pub fn get_block_any_variant<Payload: AbstractExecPayload<E>>(
         &self,
@@ -570,10 +680,14 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
             .key_exists(DBColumn::BeaconBlob.into(), block_root.as_bytes())
     }
 
-    /// Determine whether a block exists in the database.
+    /// Determine whether a block exists in the database (hot *or* cold).
     pub fn block_exists(&self, block_root: &Hash256) -> Result<bool, Error> {
-        self.hot_db
-            .key_exists(DBColumn::BeaconBlock.into(), block_root.as_bytes())
+        Ok(self
+            .hot_db
+            .key_exists(DBColumn::BeaconBlock.into(), block_root.as_bytes())?
+            || self
+                .cold_db
+                .key_exists(DBColumn::BeaconBlock.into(), block_root.as_bytes())?)
     }
 
     /// Delete a block from the store and the block cache.
@@ -2215,7 +2329,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
     }
 
     /// This function fills in missing block roots between last restore point slot and split
-    /// slot, if any.  
+    /// slot, if any.
     pub fn heal_freezer_block_roots_at_split(&self) -> Result<(), Error> {
         let split = self.get_split_info();
         let last_restore_point_slot = (split.slot - 1) / self.config.slots_per_restore_point
diff --git a/beacon_node/store/src/impls.rs b/beacon_node/store/src/impls.rs
index 736585a72aa..b2af9a408ef 100644
--- a/beacon_node/store/src/impls.rs
+++ b/beacon_node/store/src/impls.rs
@@ -1,2 +1,3 @@
 pub mod beacon_state;
 pub mod execution_payload;
+pub mod frozen_block_slot;
diff --git a/beacon_node/store/src/impls/frozen_block_slot.rs b/beacon_node/store/src/impls/frozen_block_slot.rs
new file mode 100644
index 00000000000..67d11b4f081
--- /dev/null
+++ b/beacon_node/store/src/impls/frozen_block_slot.rs
@@ -0,0 +1,19 @@
+use crate::{DBColumn, Error, StoreItem};
+use ssz::{Decode, Encode};
+use types::Slot;
+
+pub struct FrozenBlockSlot(pub Slot);
+
+impl StoreItem for FrozenBlockSlot {
+    fn db_column() -> DBColumn {
+        DBColumn::BeaconBlock
+    }
+
+    fn as_store_bytes(&self) -> Vec<u8> {
+        self.0.as_ssz_bytes()
+    }
+
+    fn from_store_bytes(bytes: &[u8]) -> Result<Self, Error> {
+        Ok(FrozenBlockSlot(Slot::from_ssz_bytes(bytes)?))
+    }
+}
diff --git a/beacon_node/store/src/iter.rs b/beacon_node/store/src/iter.rs
index 07c99e5a4ef..b073a76e17d 100644
--- a/beacon_node/store/src/iter.rs
+++ b/beacon_node/store/src/iter.rs
@@ -189,7 +189,7 @@ impl<'a, T: EthSpec, Hot: ItemStore<T>, Cold: ItemStore<T>> RootsIterator<'a, T,
         block_hash: Hash256,
     ) -> Result<Self, Error> {
         let block = store
-            .get_blinded_block(&block_hash)?
+            .get_blinded_block(&block_hash, None)?
             .ok_or_else(|| BeaconStateError::MissingBeaconBlock(block_hash.into()))?;
         let state = store
             .get_state(&block.state_root(), Some(block.slot()))?
@@ -286,7 +286,7 @@ impl<'a, E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>>
             let block = if self.decode_any_variant {
                 self.store.get_block_any_variant(&block_root)
             } else {
-                self.store.get_blinded_block(&block_root)
+                self.store.get_blinded_block(&block_root, None)
             }?
             .ok_or(Error::BlockNotFound(block_root))?;
             self.next_block_root = block.message().parent_root();
@@ -329,7 +329,8 @@ impl<'a, T: EthSpec, Hot: ItemStore<T>, Cold: ItemStore<T>> BlockIterator<'a, T,
     fn do_next(&mut self) -> Result<Option<SignedBeaconBlock<T, BlindedPayload<T>>>, Error> {
         if let Some(result) = self.roots.next() {
             let (root, _slot) = result?;
-            self.roots.inner.store.get_blinded_block(&root)
+            // Don't use slot hint here as it could be a skipped slot.
+            self.roots.inner.store.get_blinded_block(&root, None)
         } else {
             Ok(None)
         }
diff --git a/beacon_node/store/src/lib.rs b/beacon_node/store/src/lib.rs
index eacd28d2db6..3e5af0787c5 100644
--- a/beacon_node/store/src/lib.rs
+++ b/beacon_node/store/src/lib.rs
@@ -205,8 +205,19 @@ pub enum DBColumn {
     /// For data related to the database itself.
     #[strum(serialize = "bma")]
     BeaconMeta,
+    /// Data related to blocks.
+    ///
+    /// - Key: `Hash256` block root.
+    /// - Value in hot DB: SSZ-encoded blinded block.
+    /// - Value in cold DB: 8-byte slot of block.
     #[strum(serialize = "blk")]
     BeaconBlock,
+    /// Frozen beacon blocks.
+    ///
+    /// - Key: 8-byte slot.
+    /// - Value: ZSTD-compressed SSZ-encoded blinded block.
+    #[strum(serialize = "bbf")]
+    BeaconBlockFrozen,
     #[strum(serialize = "blb")]
     BeaconBlob,
     /// For full `BeaconState`s in the hot database (finalized or fork-boundary states).
@@ -295,7 +306,8 @@ impl DBColumn {
             | Self::BeaconStateRoots
             | Self::BeaconHistoricalRoots
             | Self::BeaconHistoricalSummaries
-            | Self::BeaconRandaoMixes => 8,
+            | Self::BeaconRandaoMixes
+            | Self::BeaconBlockFrozen => 8,
         }
     }
 }
diff --git a/beacon_node/store/src/reconstruct.rs b/beacon_node/store/src/reconstruct.rs
index 8fe13777ac4..29db82e443d 100644
--- a/beacon_node/store/src/reconstruct.rs
+++ b/beacon_node/store/src/reconstruct.rs
@@ -75,7 +75,7 @@ where
                     None
                 } else {
                     Some(
-                        self.get_blinded_block(&block_root)?
+                        self.get_blinded_block(&block_root, Some(slot))?
                             .ok_or(Error::BlockNotFound(block_root))?,
                     )
                 };