merge upstream

beacon_node/beacon_chain/src/blob_verification.rs

@@ -660,6 +660,15 @@ impl<E: EthSpec> AsBlock<E> for BlockWrapper<E> {
     }
 }
 
+impl<E: EthSpec> BlockWrapper<E> {
+    pub fn n_blobs(&self) -> usize {
+        match self {
+            BlockWrapper::Block(_) => 0,
+            BlockWrapper::BlockAndBlobs(_, blobs) => blobs.len(),
+        }
+    }
+}
+
 impl<E: EthSpec> From<Arc<SignedBeaconBlock<E>>> for BlockWrapper<E> {
     fn from(value: Arc<SignedBeaconBlock<E>>) -> Self {
         Self::Block(value)

beacon_node/beacon_chain/src/block_verification.rs

@@ -912,7 +912,7 @@ impl<T: BeaconChainTypes> GossipVerifiedBlock<T> {
         // We check this *before* we load the parent so that we can return a more detailed error.
         let block = check_block_is_finalized_checkpoint_or_descendant(
             chain,
-            &chain.canonical_head.fork_choice_write_lock(),
+            &chain.canonical_head.fork_choice_read_lock(),
             block,
         )?;
 

beacon_node/beacon_chain/src/builder.rs

@@ -960,9 +960,11 @@ where
         }
 
         // Prune blobs sidecars older than the blob data availability boundary in the background.
-        beacon_chain
-            .store_migrator
-            .process_prune_blobs(beacon_chain.data_availability_boundary());
+        if let Some(data_availability_boundary) = beacon_chain.data_availability_boundary() {
+            beacon_chain
+                .store_migrator
+                .process_prune_blobs(data_availability_boundary);
+        }
 
         Ok(beacon_chain)
     }

beacon_node/beacon_chain/src/canonical_head.rs

@@ -758,8 +758,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         drop(old_cached_head);
 
         // Prune blobs in the background.
-        self.store_migrator
-            .process_prune_blobs(self.data_availability_boundary());
+        if let Some(data_availability_boundary) = self.data_availability_boundary() {
+            self.store_migrator
+                .process_prune_blobs(data_availability_boundary);
+        }
 
         // If the finalized checkpoint changed, perform some updates.
         //

beacon_node/beacon_chain/src/data_availability_checker.rs

@@ -569,6 +569,13 @@ impl<E: EthSpec> AvailableBlock<E> {
             VerifiedBlobs::Available(blobs) => (self.block, Some(blobs)),
         }
     }
+
+    pub fn blobs(&self) -> Option<&BlobSidecarList<E>> {
+        match &self.blobs {
+            VerifiedBlobs::Available(blobs) => Some(blobs),
+            _ => None,
+        }
+    }
 }
 
 impl<E: EthSpec> AsBlock<E> for AvailableBlock<E> {

beacon_node/beacon_chain/src/historical_blocks.rs

@@ -1,3 +1,4 @@
+use crate::data_availability_checker::AvailableBlock;
 use crate::{errors::BeaconChainError as Error, metrics, BeaconChain, BeaconChainTypes};
 use itertools::Itertools;
 use slog::debug;
@@ -7,10 +8,9 @@ use state_processing::{
 };
 use std::borrow::Cow;
 use std::iter;
-use std::sync::Arc;
 use std::time::Duration;
 use store::{chunked_vector::BlockRoots, AnchorInfo, ChunkWriter, KeyValueStore};
-use types::{Hash256, SignedBlindedBeaconBlock, Slot};
+use types::{Hash256, Slot};
 
 /// Use a longer timeout on the pubkey cache.
 ///
@@ -59,7 +59,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     /// Return the number of blocks successfully imported.
     pub fn import_historical_block_batch(
         &self,
-        blocks: Vec<Arc<SignedBlindedBeaconBlock<T::EthSpec>>>,
+        mut blocks: Vec<AvailableBlock<T::EthSpec>>,
     ) -> Result<usize, Error> {
         let anchor_info = self
             .store
@@ -67,19 +67,21 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             .ok_or(HistoricalBlockError::NoAnchorInfo)?;
 
         // Take all blocks with slots less than the oldest block slot.
-        let num_relevant =
-            blocks.partition_point(|block| block.slot() < anchor_info.oldest_block_slot);
-        let blocks_to_import = &blocks
-            .get(..num_relevant)
-            .ok_or(HistoricalBlockError::IndexOutOfBounds)?;
+        let num_relevant = blocks.partition_point(|available_block| {
+            available_block.block().slot() < anchor_info.oldest_block_slot
+        });
 
-        if blocks_to_import.len() != blocks.len() {
+        let total_blocks = blocks.len();
+        blocks.truncate(num_relevant);
+        let blocks_to_import = blocks;
+
+        if blocks_to_import.len() != total_blocks {
             debug!(
                 self.log,
                 "Ignoring some historic blocks";
                 "oldest_block_slot" => anchor_info.oldest_block_slot,
-                "total_blocks" => blocks.len(),
-                "ignored" => blocks.len().saturating_sub(blocks_to_import.len()),
+                "total_blocks" => total_blocks,
+                "ignored" => total_blocks.saturating_sub(blocks_to_import.len()),
             );
         }
 
@@ -87,15 +89,23 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             return Ok(0);
         }
 
+        let n_blobs_to_import = blocks_to_import
+            .iter()
+            .map(|available_block| available_block.blobs().map_or(0, |blobs| blobs.len()))
+            .sum::<usize>();
+
         let mut expected_block_root = anchor_info.oldest_block_parent;
         let mut prev_block_slot = anchor_info.oldest_block_slot;
         let mut chunk_writer =
             ChunkWriter::<BlockRoots, _, _>::new(&self.store.cold_db, prev_block_slot.as_usize())?;
 
-        let mut cold_batch = Vec::with_capacity(blocks.len());
-        let mut hot_batch = Vec::with_capacity(blocks.len());
+        let mut cold_batch = Vec::with_capacity(blocks_to_import.len());
+        let mut hot_batch = Vec::with_capacity(blocks_to_import.len() + n_blobs_to_import);
+        let mut signed_blocks = Vec::with_capacity(blocks_to_import.len());
+
+        for available_block in blocks_to_import.into_iter().rev() {
+            let (block, maybe_blobs) = available_block.deconstruct();
 
-        for block in blocks_to_import.iter().rev() {
             // Check chain integrity.
             let block_root = block.canonical_root();
 
@@ -107,9 +117,15 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 .into());
             }
 
+            let blinded_block = block.clone_as_blinded();
             // Store block in the hot database without payload.
             self.store
-                .blinded_block_as_kv_store_ops(&block_root, block, &mut hot_batch);
+                .blinded_block_as_kv_store_ops(&block_root, &blinded_block, &mut hot_batch);
+            // Store the blobs too
+            if let Some(blobs) = maybe_blobs {
+                self.store
+                    .blobs_as_kv_store_ops(&block_root, blobs, &mut hot_batch);
+            }
 
             // Store block roots, including at all skip slots in the freezer DB.
             for slot in (block.slot().as_usize()..prev_block_slot.as_usize()).rev() {
@@ -132,8 +148,11 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 expected_block_root = Hash256::zero();
                 break;
             }
+            signed_blocks.push(block);
         }
         chunk_writer.write(&mut cold_batch)?;
+        // these were pushed in reverse order so we reverse again
+        signed_blocks.reverse();
 
         // Verify signatures in one batch, holding the pubkey cache lock for the shortest duration
         // possible. For each block fetch the parent root from its successor. Slicing from index 1
@@ -144,13 +163,13 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             .validator_pubkey_cache
             .try_read_for(PUBKEY_CACHE_LOCK_TIMEOUT)
             .ok_or(HistoricalBlockError::ValidatorPubkeyCacheTimeout)?;
-        let block_roots = blocks_to_import
+        let block_roots = signed_blocks
             .get(1..)
             .ok_or(HistoricalBlockError::IndexOutOfBounds)?
             .iter()
             .map(|block| block.parent_root())
             .chain(iter::once(anchor_info.oldest_block_parent));
-        let signature_set = blocks_to_import
+        let signature_set = signed_blocks
             .iter()
             .zip_eq(block_roots)
             .filter_map(|(block, block_root)| {
@@ -207,6 +226,6 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             self.store_migrator.process_reconstruction();
         }
 
-        Ok(blocks_to_import.len())
+        Ok(num_relevant)
     }
 }

beacon_node/beacon_chain/src/migrate.rs

@@ -86,7 +86,7 @@ pub enum PruningError {
 pub enum Notification {
     Finalization(FinalizationNotification),
     Reconstruction,
-    PruneBlobs(Option<Epoch>),
+    PruneBlobs(Epoch),
 }
 
 pub struct FinalizationNotification {
@@ -153,7 +153,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
         }
     }
 
-    pub fn process_prune_blobs(&self, data_availability_boundary: Option<Epoch>) {
+    pub fn process_prune_blobs(&self, data_availability_boundary: Epoch) {
         if let Some(Notification::PruneBlobs(data_availability_boundary)) =
             self.send_background_notification(Notification::PruneBlobs(data_availability_boundary))
         {
@@ -173,7 +173,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
 
     pub fn run_prune_blobs(
         db: Arc<HotColdDB<E, Hot, Cold>>,
-        data_availability_boundary: Option<Epoch>,
+        data_availability_boundary: Epoch,
         log: &Logger,
     ) {
         if let Err(e) = db.try_prune_blobs(false, data_availability_boundary) {
@@ -606,7 +606,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> BackgroundMigrator<E, Ho
                     StoreOp::DeleteBlock(block_root),
                     StoreOp::DeleteExecutionPayload(block_root),
                 ];
-                if let Ok(true) = store.blobs_sidecar_exists(&block_root) {
+                if store.blobs_sidecar_exists(&block_root).unwrap_or(false) {
                     // Keep track of non-empty orphaned blobs sidecars.
                     store_ops.extend([
                         StoreOp::DeleteBlobs(block_root),