add processing and processed caching to the DA checker

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -118,6 +118,7 @@ use store::{
 use task_executor::{ShutdownReason, TaskExecutor};
 use tokio_stream::Stream;
 use tree_hash::TreeHash;
+use types::beacon_block_body::KzgCommitmentOpts;
 use types::beacon_state::CloneConfig;
 use types::blob_sidecar::{BlobSidecarList, FixedBlobSidecarList};
 use types::sidecar::BlobItems;
@@ -2791,11 +2792,76 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             .map_err(BeaconChainError::TokioJoin)?
     }
 
-    pub async fn process_blob(
+    pub async fn process_gossip_blob(
         self: &Arc<Self>,
         blob: GossipVerifiedBlob<T>,
     ) -> Result<AvailabilityProcessingStatus, BlockError<T::EthSpec>> {
-        self.check_gossip_blob_availability_and_import(blob).await
+        // TODO: fewer places where we initialize this fixed vec
+        let block_root = blob.block_root();
+        let mut commitments = KzgCommitmentOpts::<T::EthSpec>::default();
+        if let Some(commitment_opt) = commitments.get_mut(blob.as_blob().index as usize) {
+            *commitment_opt = Some(blob.as_blob().kzg_commitment);
+        }
+        self.data_availability_checker
+            .notify_blob_commitments(block_root, commitments);
+        let r = self.check_gossip_blob_availability_and_import(blob).await;
+        self.remove_notified(&block_root, r)
+    }
+
+    pub async fn process_rpc_blobs(
+        self: &Arc<Self>,
+        slot: Slot,
+        block_root: Hash256,
+        blobs: FixedBlobSidecarList<T::EthSpec>,
+    ) -> Result<AvailabilityProcessingStatus, BlockError<T::EthSpec>> {
+        let mut commitments = KzgCommitmentOpts::<T::EthSpec>::default();
+        for blob in blobs.iter().flatten() {
+            if let Some(commitment) = commitments.get_mut(blob.index as usize) {
+                *commitment = Some(blob.kzg_commitment);
+            }
+        }
+        self.data_availability_checker
+            .notify_blob_commitments(block_root, commitments);
+        let r = self
+            .check_rpc_blob_availability_and_import(slot, block_root, blobs)
+            .await;
+        self.remove_notified(&block_root, r)
+    }
+
+    fn remove_notified(
+        &self,
+        block_root: &Hash256,
+        r: Result<AvailabilityProcessingStatus, BlockError<T::EthSpec>>,
+    ) -> Result<AvailabilityProcessingStatus, BlockError<T::EthSpec>> {
+        let has_missing_components =
+            matches!(r, Ok(AvailabilityProcessingStatus::MissingComponents(_, _)));
+        if !has_missing_components {
+            self.data_availability_checker.remove_notified(block_root);
+        }
+        r
+    }
+
+    pub async fn process_block_with_early_caching<B: IntoExecutionPendingBlock<T>>(
+        self: &Arc<Self>,
+        block_root: Hash256,
+        unverified_block: B,
+        notify_execution_layer: NotifyExecutionLayer,
+    ) -> Result<AvailabilityProcessingStatus, BlockError<T::EthSpec>> {
+        if let Ok(commitments) = unverified_block
+            .block()
+            .message()
+            .body()
+            .blob_kzg_commitments()
+        {
+            self.data_availability_checker
+                .notify_block_commitments(block_root, commitments.clone());
+        };
+        let r = self
+            .process_block(block_root, unverified_block, notify_execution_layer, || {
+                Ok(())
+            })
+            .await;
+        self.remove_notified(&block_root, r)
     }
 
     /// Returns `Ok(block_root)` if the given `unverified_block` was successfully verified and
@@ -2961,7 +3027,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
     /// Checks if the block is available, and imports immediately if so, otherwise caches the block
     /// in the data availability checker.
-    pub async fn check_block_availability_and_import(
+    async fn check_block_availability_and_import(
         self: &Arc<Self>,
         block: AvailabilityPendingExecutedBlock<T::EthSpec>,
     ) -> Result<AvailabilityProcessingStatus, BlockError<T::EthSpec>> {
@@ -2974,7 +3040,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
     /// Checks if the provided blob can make any cached blocks available, and imports immediately
     /// if so, otherwise caches the blob in the data availability checker.
-    pub async fn check_gossip_blob_availability_and_import(
+    async fn check_gossip_blob_availability_and_import(
         self: &Arc<Self>,
         blob: GossipVerifiedBlob<T>,
     ) -> Result<AvailabilityProcessingStatus, BlockError<T::EthSpec>> {
@@ -2986,7 +3052,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
     /// Checks if the provided blobs can make any cached blocks available, and imports immediately
     /// if so, otherwise caches the blob in the data availability checker.
-    pub async fn check_rpc_blob_availability_and_import(
+    async fn check_rpc_blob_availability_and_import(
         self: &Arc<Self>,
         slot: Slot,
         block_root: Hash256,
@@ -5250,7 +5316,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             return Ok(());
         }
 
-        // Fetch payoad attributes from the execution layer's cache, or compute them from scratch
+        // Fetch payload attributes from the execution layer's cache, or compute them from scratch
         // if no matching entry is found. This saves recomputing the withdrawals which can take
         // considerable time to compute if a state load is required.
         let head_root = forkchoice_update_params.head_root;

beacon_node/beacon_chain/src/blob_verification.rs

@@ -9,11 +9,12 @@ use crate::beacon_chain::{
 use crate::block_verification::cheap_state_advance_to_obtain_committees;
 use crate::data_availability_checker::AvailabilityCheckError;
 use crate::kzg_utils::{validate_blob, validate_blobs};
-use crate::BeaconChainError;
+use crate::{metrics, BeaconChainError};
 use kzg::Kzg;
 use slog::{debug, warn};
 use ssz_derive::{Decode, Encode};
 use ssz_types::VariableList;
+use tree_hash::TreeHash;
 use types::blob_sidecar::BlobIdentifier;
 use types::{
     BeaconStateError, BlobSidecar, BlobSidecarList, CloneConfig, EthSpec, Hash256,
@@ -166,6 +167,9 @@ impl<T: BeaconChainTypes> GossipVerifiedBlob<T> {
     pub fn to_blob(self) -> Arc<BlobSidecar<T::EthSpec>> {
         self.blob.message
     }
+    pub fn as_blob(&self) -> &BlobSidecar<T::EthSpec> {
+        &self.blob.message
+    }
     pub fn signed_blob(&self) -> SignedBlobSidecar<T::EthSpec> {
         self.blob.clone()
     }
@@ -197,6 +201,8 @@ pub fn validate_blob_sidecar_for_gossip<T: BeaconChainTypes>(
         });
     }
 
+    let blob_root = get_blob_root(&signed_blob_sidecar);
+
     // Verify that the sidecar is not from a future slot.
     let latest_permissible_slot = chain
         .slot_clock
@@ -387,7 +393,7 @@ pub fn validate_blob_sidecar_for_gossip<T: BeaconChainTypes>(
             .ok_or_else(|| GossipBlobError::UnknownValidator(proposer_index as u64))?;
 
         signed_blob_sidecar.verify_signature(
-            None,
+            Some(blob_root),
             pubkey,
             &fork,
             chain.genesis_validators_root,
@@ -467,6 +473,15 @@ impl<T: EthSpec> KzgVerifiedBlob<T> {
     }
 }
 
+#[cfg(test)]
+impl<T: EthSpec> KzgVerifiedBlob<T> {
+    pub fn new(blob: BlobSidecar<T>) -> Self {
+        Self {
+            blob: Arc::new(blob),
+        }
+    }
+}
+
 /// Complete kzg verification for a `GossipVerifiedBlob`.
 ///
 /// Returns an error if the kzg verification check fails.
@@ -512,3 +527,16 @@ pub fn verify_kzg_for_blob_list<T: EthSpec>(
         Err(AvailabilityCheckError::KzgVerificationFailed)
     }
 }
+
+/// Returns the canonical root of the given `blob`.
+///
+/// Use this function to ensure that we report the blob hashing time Prometheus metric.
+pub fn get_blob_root<E: EthSpec>(blob: &SignedBlobSidecar<E>) -> Hash256 {
+    let blob_root_timer = metrics::start_timer(&metrics::BLOCK_PROCESSING_BLOB_ROOT);
+
+    let blob_root = blob.message.tree_hash_root();
+
+    metrics::stop_timer(blob_root_timer);
+
+    blob_root
+}

beacon_node/beacon_chain/src/block_verification_types.rs

@@ -3,7 +3,7 @@ use crate::block_verification::BlockError;
 use crate::data_availability_checker::AvailabilityCheckError;
 pub use crate::data_availability_checker::{AvailableBlock, MaybeAvailableBlock};
 use crate::eth1_finalization_cache::Eth1FinalizationData;
-use crate::{data_availability_checker, GossipVerifiedBlock, PayloadVerificationOutcome};
+use crate::{GossipVerifiedBlock, PayloadVerificationOutcome};
 use derivative::Derivative;
 use ssz_derive::{Decode, Encode};
 use ssz_types::VariableList;
@@ -65,8 +65,22 @@ impl<E: EthSpec> RpcBlock<E> {
         block: Arc<SignedBeaconBlock<E>>,
         blobs: Option<BlobSidecarList<E>>,
     ) -> Result<Self, AvailabilityCheckError> {
-        if let Some(blobs) = blobs.as_ref() {
-            data_availability_checker::consistency_checks(&block, blobs)?;
+        if let (Some(blobs), Ok(block_commitments)) = (
+            blobs.as_ref(),
+            block.message().body().blob_kzg_commitments(),
+        ) {
+            if blobs.len() != block_commitments.len() {
+                return Err(AvailabilityCheckError::MissingBlobs);
+            }
+            for (blob, &block_commitment) in blobs.iter().zip(block_commitments.iter()) {
+                let blob_commitment = blob.kzg_commitment;
+                if blob_commitment != block_commitment {
+                    return Err(AvailabilityCheckError::KzgCommitmentMismatch {
+                        block_commitment,
+                        blob_commitment,
+                    });
+                }
+            }
         }
         let inner = match blobs {
             Some(blobs) => RpcBlockInner::BlockAndBlobs(block, blobs),
@@ -235,27 +249,17 @@ impl<E: EthSpec> AvailabilityPendingExecutedBlock<E> {
         }
     }
 
+    pub fn as_block(&self) -> &SignedBeaconBlock<E> {
+        &self.block
+    }
+
     pub fn num_blobs_expected(&self) -> usize {
         self.block
             .message()
             .body()
             .blob_kzg_commitments()
             .map_or(0, |commitments| commitments.len())
     }
-
-    pub fn get_all_blob_ids(&self) -> Vec<BlobIdentifier> {
-        let block_root = self.import_data.block_root;
-        self.block
-            .get_filtered_blob_ids(Some(block_root), |_, _| true)
-    }
-
-    pub fn get_filtered_blob_ids(
-        &self,
-        filter: impl Fn(usize, Hash256) -> bool,
-    ) -> Vec<BlobIdentifier> {
-        self.block
-            .get_filtered_blob_ids(Some(self.import_data.block_root), filter)
-    }
 }
 
 #[derive(Debug, PartialEq, Encode, Decode, Clone)]

beacon_node/beacon_chain/src/builder.rs

@@ -916,7 +916,7 @@ where
             validator_monitor: RwLock::new(validator_monitor),
             genesis_backfill_slot,
             data_availability_checker: Arc::new(
-                DataAvailabilityChecker::new(slot_clock, kzg.clone(), store, self.spec)
+                DataAvailabilityChecker::new(slot_clock, kzg.clone(), store, &log, self.spec)
                     .map_err(|e| format!("Error initializing DataAvailabiltyChecker: {:?}", e))?,
             ),
             kzg,