Devnet6

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -114,9 +114,8 @@ use store::{
 use task_executor::{ShutdownReason, TaskExecutor};
 use tokio_stream::Stream;
 use tree_hash::TreeHash;
-use types::beacon_block_body::KzgCommitments;
 use types::beacon_state::CloneConfig;
-use types::blob_sidecar::{BlobSidecarList, Blobs};
+use types::blob_sidecar::BlobSidecarList;
 use types::consts::deneb::MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUESTS;
 use types::*;
 
@@ -466,7 +465,7 @@ pub struct BeaconChain<T: BeaconChainTypes> {
     pub genesis_backfill_slot: Slot,
     pub proposal_blob_cache: BlobCache<T::EthSpec>,
     pub data_availability_checker: Arc<DataAvailabilityChecker<T>>,
-    pub kzg: Option<Arc<Kzg>>,
+    pub kzg: Option<Arc<Kzg<<T::EthSpec as EthSpec>::Kzg>>>,
 }
 
 type BeaconBlockAndState<T, Payload> = (BeaconBlock<T, Payload>, BeaconState<T>);
@@ -2914,7 +2913,14 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     ) -> Result<AvailabilityProcessingStatus, BlockError<T::EthSpec>> {
         let availability = cache_fn(self.clone())?;
         match availability {
-            Availability::Available(block) => self.import_available_block(block).await,
+            Availability::Available(block) => {
+                // This is the time since start of the slot where all the components of the block have become available
+                let delay =
+                    get_slot_delay_ms(timestamp_now(), block.block.slot(), &self.slot_clock);
+                metrics::observe_duration(&metrics::BLOCK_AVAILABILITY_DELAY, delay);
+                // Block is fully available, import into fork choice
+                self.import_available_block(block).await
+            }
             Availability::MissingComponents(block_root) => Ok(
                 AvailabilityProcessingStatus::MissingComponents(slot, block_root),
             ),
@@ -4903,7 +4909,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
         //FIXME(sean)
         // - add a new timer for processing here
-        if let Some(blobs) = blobs_opt {
+        if let (Some(blobs), Some(proofs)) = (blobs_opt, proofs_opt) {
             let kzg = self
                 .kzg
                 .as_ref()
@@ -4924,14 +4930,10 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 )));
             }
 
-            let kzg_proofs = if let Some(proofs) = proofs_opt {
-                Vec::from(proofs)
-            } else {
-                Self::compute_blob_kzg_proofs(kzg, &blobs, expected_kzg_commitments, slot)?
-            };
+            let kzg_proofs = Vec::from(proofs);
 
             kzg_utils::validate_blobs::<T::EthSpec>(
-                kzg,
+                kzg.as_ref(),
                 expected_kzg_commitments,
                 &blobs,
                 &kzg_proofs,
@@ -4982,29 +4984,6 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         Ok((block, state))
     }
 
-    fn compute_blob_kzg_proofs(
-        kzg: &Arc<Kzg>,
-        blobs: &Blobs<T::EthSpec>,
-        expected_kzg_commitments: &KzgCommitments<T::EthSpec>,
-        slot: Slot,
-    ) -> Result<Vec<KzgProof>, BlockProductionError> {
-        blobs
-            .iter()
-            .enumerate()
-            .map(|(blob_index, blob)| {
-                let kzg_commitment = expected_kzg_commitments.get(blob_index).ok_or(
-                    BlockProductionError::MissingKzgCommitment(format!(
-                        "Missing KZG commitment for slot {} blob index {}",
-                        slot, blob_index
-                    )),
-                )?;
-
-                kzg_utils::compute_blob_kzg_proof::<T::EthSpec>(kzg, blob, *kzg_commitment)
-                    .map_err(BlockProductionError::KzgError)
-            })
-            .collect::<Result<Vec<KzgProof>, BlockProductionError>>()
-    }
-
     /// This method must be called whenever an execution engine indicates that a payload is
     /// invalid.
     ///

beacon_node/beacon_chain/src/blob_verification.rs

@@ -4,8 +4,8 @@ use state_processing::state_advance::partial_state_advance;
 use std::sync::Arc;
 
 use crate::beacon_chain::{
-    BeaconChain, BeaconChainTypes, MAXIMUM_GOSSIP_CLOCK_DISPARITY,
-    VALIDATOR_PUBKEY_CACHE_LOCK_TIMEOUT,
+    BeaconChain, BeaconChainTypes, BLOCK_PROCESSING_CACHE_LOCK_TIMEOUT,
+    MAXIMUM_GOSSIP_CLOCK_DISPARITY, VALIDATOR_PUBKEY_CACHE_LOCK_TIMEOUT,
 };
 use crate::data_availability_checker::{
     AvailabilityCheckError, AvailabilityPendingBlock, AvailableBlock,
@@ -20,9 +20,9 @@ use ssz_types::FixedVector;
 use std::borrow::Cow;
 use types::blob_sidecar::{BlobIdentifier, FixedBlobSidecarList};
 use types::{
-    BeaconBlockRef, BeaconState, BeaconStateError, BlobSidecar, ChainSpec, CloneConfig, Epoch,
-    EthSpec, FullPayload, Hash256, KzgCommitment, RelativeEpoch, SignedBeaconBlock,
-    SignedBeaconBlockHeader, SignedBlobSidecar, Slot,
+    BeaconBlockRef, BeaconState, BeaconStateError, BlobSidecar, BlobSidecarList, ChainSpec,
+    CloneConfig, Epoch, EthSpec, FullPayload, Hash256, KzgCommitment, RelativeEpoch,
+    SignedBeaconBlock, SignedBeaconBlockHeader, SignedBlobSidecar, Slot,
 };
 
 #[derive(Debug)]
@@ -240,36 +240,72 @@ pub fn validate_blob_sidecar_for_gossip<T: BeaconChainTypes>(
             "block_root" => %block_root,
             "index" => %blob_index,
         );
-        // The cached head state is in the same epoch as the blob or the state has already been
-        // advanced to the blob's epoch
-        let snapshot = &chain.canonical_head.cached_head().snapshot;
-        if snapshot.beacon_state.current_epoch() == blob_slot.epoch(T::EthSpec::slots_per_epoch()) {
-            (
-                snapshot
-                    .beacon_state
-                    .get_beacon_proposer_index(blob_slot, &chain.spec)?,
-                snapshot.beacon_state.fork(),
-            )
+        if let Some(mut snapshot) = chain
+            .snapshot_cache
+            .try_read_for(BLOCK_PROCESSING_CACHE_LOCK_TIMEOUT)
+            .and_then(|snapshot_cache| {
+                snapshot_cache.get_cloned(block_parent_root, CloneConfig::committee_caches_only())
+            })
+        {
+            if snapshot.beacon_state.slot() == blob_slot {
+                debug!(
+                    chain.log,
+                    "Cloning snapshot cache state for blob verification";
+                    "block_root" => %block_root,
+                    "index" => %blob_index,
+                );
+                (
+                    snapshot
+                        .beacon_state
+                        .get_beacon_proposer_index(blob_slot, &chain.spec)?,
+                    snapshot.beacon_state.fork(),
+                )
+            } else {
+                debug!(
+                    chain.log,
+                    "Cloning and advancing snapshot cache state for blob verification";
+                    "block_root" => %block_root,
+                    "index" => %blob_index,
+                );
+                let state = cheap_state_advance_to_obtain_committees(
+                    &mut snapshot.beacon_state,
+                    Some(snapshot.beacon_block_root),
+                    blob_slot,
+                    &chain.spec,
+                )?;
+                (
+                    state.get_beacon_proposer_index(blob_slot, &chain.spec)?,
+                    state.fork(),
+                )
+            }
         }
         // Need to advance the state to get the proposer index
         else {
-            // Reaching this condition too often might be an issue since we could theoretically have
-            // 5 threads (4 blob indices + 1 block) cloning the state.
-            // We shouldn't be seeing this condition a lot because we try to advance the state
-            // 3 seconds before the start of a slot. However, if this becomes an issue during testing, we should
-            // consider sending a blob for reprocessing to reduce the number of state clones.
             warn!(
                 chain.log,
-                "Cached head not advanced for blob verification";
+                "Snapshot cache miss for blob verification";
                 "block_root" => %block_root,
                 "index" => %blob_index,
-                "action" => "contact the devs if you see this msg too often"
             );
-            // The state produced is only valid for determining proposer/attester shuffling indices.
-            let mut cloned_state = snapshot.clone_with(CloneConfig::committee_caches_only());
+
+            let parent_block = chain
+                .get_blinded_block(&block_parent_root)
+                .map_err(BlobError::BeaconChainError)?
+                .ok_or_else(|| {
+                    BlobError::from(BeaconChainError::MissingBeaconBlock(block_parent_root))
+                })?;
+
+            let mut parent_state = chain
+                .get_state(&parent_block.state_root(), Some(parent_block.slot()))?
+                .ok_or_else(|| {
+                    BeaconChainError::DBInconsistent(format!(
+                        "Missing state {:?}",
+                        parent_block.state_root()
+                    ))
+                })?;
             let state = cheap_state_advance_to_obtain_committees(
-                &mut cloned_state.beacon_state,
-                None,
+                &mut parent_state,
+                Some(parent_block.state_root()),
                 blob_slot,
                 &chain.spec,
             )?;
@@ -449,8 +485,9 @@ impl<T: EthSpec> KzgVerifiedBlob<T> {
 /// Returns an error if the kzg verification check fails.
 pub fn verify_kzg_for_blob<T: EthSpec>(
     blob: Arc<BlobSidecar<T>>,
-    kzg: &Kzg,
+    kzg: &Kzg<T::Kzg>,
 ) -> Result<KzgVerifiedBlob<T>, AvailabilityCheckError> {
+    let _timer = crate::metrics::start_timer(&crate::metrics::KZG_VERIFICATION_SINGLE_TIMES);
     //TODO(sean) remove clone
     if validate_blob::<T>(kzg, blob.blob.clone(), blob.kzg_commitment, blob.kzg_proof)
         .map_err(AvailabilityCheckError::Kzg)?
@@ -468,8 +505,9 @@ pub fn verify_kzg_for_blob<T: EthSpec>(
 /// in a loop since this function kzg verifies a list of blobs more efficiently.
 pub fn verify_kzg_for_blob_list<T: EthSpec>(
     blob_list: Vec<Arc<BlobSidecar<T>>>,
-    kzg: &Kzg,
+    kzg: &Kzg<T::Kzg>,
 ) -> Result<KzgVerifiedBlobList<T>, AvailabilityCheckError> {
+    let _timer = crate::metrics::start_timer(&crate::metrics::KZG_VERIFICATION_BATCH_TIMES);
     let (blobs, (commitments, proofs)): (Vec<_>, (Vec<_>, Vec<_>)) = blob_list
         .clone()
         .into_iter()
@@ -612,6 +650,15 @@ pub enum BlockWrapper<E: EthSpec> {
 }
 
 impl<E: EthSpec> BlockWrapper<E> {
+    pub fn new(block: Arc<SignedBeaconBlock<E>>, blobs: Option<BlobSidecarList<E>>) -> Self {
+        match blobs {
+            Some(blobs) => {
+                let blobs = FixedVector::from(blobs.into_iter().map(Some).collect::<Vec<_>>());
+                BlockWrapper::BlockAndBlobs(block, blobs)
+            }
+            None => BlockWrapper::Block(block),
+        }
+    }
     pub fn deconstruct(self) -> (Arc<SignedBeaconBlock<E>>, Option<FixedBlobSidecarList<E>>) {
         match self {
             BlockWrapper::Block(block) => (block, None),

beacon_node/beacon_chain/src/data_availability_checker.rs

@@ -11,7 +11,6 @@ use kzg::Kzg;
 use slog::{debug, error};
 use slot_clock::SlotClock;
 use ssz_types::{Error, FixedVector, VariableList};
-use state_processing::per_block_processing::deneb::deneb::verify_kzg_commitments_against_transactions;
 use std::collections::HashSet;
 use std::sync::Arc;
 use strum::IntoStaticStr;
@@ -21,7 +20,7 @@ use types::blob_sidecar::{BlobIdentifier, BlobSidecar, FixedBlobSidecarList};
 use types::consts::deneb::MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUESTS;
 use types::ssz_tagged_signed_beacon_block;
 use types::{
-    BeaconBlockRef, BlobSidecarList, ChainSpec, Epoch, EthSpec, ExecPayload, FullPayload, Hash256,
+    BeaconBlockRef, BlobSidecarList, ChainSpec, Epoch, EthSpec, FullPayload, Hash256,
     SignedBeaconBlock, SignedBeaconBlockHeader, Slot,
 };
 
@@ -85,7 +84,7 @@ impl From<ssz::DecodeError> for AvailabilityCheckError {
 pub struct DataAvailabilityChecker<T: BeaconChainTypes> {
     availability_cache: Arc<OverflowLRUCache<T>>,
     slot_clock: T::SlotClock,
-    kzg: Option<Arc<Kzg>>,
+    kzg: Option<Arc<Kzg<<T::EthSpec as EthSpec>::Kzg>>>,
     spec: ChainSpec,
 }
 
@@ -114,7 +113,7 @@ impl<T: EthSpec> Availability<T> {
 impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
     pub fn new(
         slot_clock: T::SlotClock,
-        kzg: Option<Arc<Kzg>>,
+        kzg: Option<Arc<Kzg<<T::EthSpec as EthSpec>::Kzg>>>,
         store: BeaconStore<T>,
         spec: ChainSpec,
     ) -> Result<Self, AvailabilityCheckError> {
@@ -296,35 +295,20 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
         &self,
         block: &Arc<SignedBeaconBlock<T::EthSpec, FullPayload<T::EthSpec>>>,
     ) -> Result<BlobRequirements, AvailabilityCheckError> {
-        let verified_blobs = if let (Ok(block_kzg_commitments), Ok(payload)) = (
-            block.message().body().blob_kzg_commitments(),
-            block.message().body().execution_payload(),
-        ) {
-            if let Some(transactions) = payload.transactions() {
-                let verified = verify_kzg_commitments_against_transactions::<T::EthSpec>(
-                    transactions,
-                    block_kzg_commitments,
-                )
-                .map_err(|e| AvailabilityCheckError::TxKzgCommitmentMismatch(format!("{e:?}")))?;
-                if !verified {
-                    return Err(AvailabilityCheckError::TxKzgCommitmentMismatch(
-                        "a commitment and version didn't match".to_string(),
-                    ));
-                }
-            }
-
-            if self.da_check_required(block.epoch()) {
-                if block_kzg_commitments.is_empty() {
-                    BlobRequirements::EmptyBlobs
+        let verified_blobs =
+            if let Ok(block_kzg_commitments) = block.message().body().blob_kzg_commitments() {
+                if self.da_check_required(block.epoch()) {
+                    if block_kzg_commitments.is_empty() {
+                        BlobRequirements::EmptyBlobs
+                    } else {
+                        BlobRequirements::Required
+                    }
                 } else {
-                    BlobRequirements::Required
+                    BlobRequirements::NotRequired
                 }
             } else {
-                BlobRequirements::NotRequired
-            }
-        } else {
-            BlobRequirements::PreDeneb
-        };
+                BlobRequirements::PreDeneb
+            };
         Ok(verified_blobs)
     }
 

beacon_node/beacon_chain/src/execution_payload.rs

@@ -18,7 +18,7 @@ use proto_array::{Block as ProtoBlock, ExecutionStatus};
 use slog::{debug, warn};
 use slot_clock::SlotClock;
 use state_processing::per_block_processing::{
-    compute_timestamp_at_slot, get_expected_withdrawals, is_execution_enabled,
+    self, compute_timestamp_at_slot, get_expected_withdrawals, is_execution_enabled,
     is_merge_transition_complete, partially_verify_execution_payload,
 };
 use std::sync::Arc;
@@ -68,15 +68,16 @@ impl<T: BeaconChainTypes> PayloadNotifier<T> {
             // the block as optimistically imported. This is particularly relevant in the case
             // where we do not send the block to the EL at all.
             let block_message = block.message();
-            let payload = block_message.execution_payload()?;
             partially_verify_execution_payload::<_, FullPayload<_>>(
                 state,
                 block.slot(),
-                payload,
+                block_message.body(),
                 &chain.spec,
             )
             .map_err(BlockError::PerBlockProcessingError)?;
 
+            let payload = block_message.execution_payload()?;
+
             match notify_execution_layer {
                 NotifyExecutionLayer::No if chain.config.optimistic_finalized_sync => {
                     // Verify the block hash here in Lighthouse and immediately mark the block as
@@ -139,14 +140,22 @@ async fn notify_new_payload<'a, T: BeaconChainTypes>(
     block: BeaconBlockRef<'a, T::EthSpec>,
 ) -> Result<PayloadVerificationStatus, BlockError<T::EthSpec>> {
     let execution_payload = block.execution_payload()?;
+    let versioned_hashes = block.body().blob_kzg_commitments().ok().map(|commitments| {
+        commitments
+            .into_iter()
+            .map(|commitment| {
+                per_block_processing::deneb::deneb::kzg_commitment_to_versioned_hash(commitment)
+            })
+            .collect::<Vec<_>>()
+    });
 
     let execution_layer = chain
         .execution_layer
         .as_ref()
         .ok_or(ExecutionPayloadError::NoExecutionConnection)?;
 
     let new_payload_response = execution_layer
-        .notify_new_payload(&execution_payload.into())
+        .notify_new_payload(&execution_payload.into(), versioned_hashes)
         .await;
 
     match new_payload_response {