From 0c865937c5b5c0ba753edc4945ba76297783f3e3 Mon Sep 17 00:00:00 2001 From: dapplion <35266934+dapplion@users.noreply.github.com> Date: Sun, 17 Mar 2024 22:25:56 +0900 Subject: [PATCH 1/4] Move processing cache out of DA --- .../beacon_chain/src/beacon_block_streamer.rs | 11 +- beacon_node/beacon_chain/src/beacon_chain.rs | 18 +-- beacon_node/beacon_chain/src/builder.rs | 1 + .../src/data_availability_checker.rs | 108 +++--------------- .../availability_view.rs | 42 ------- .../overflow_lru_cache.rs | 25 ++++ .../processing_cache.rs | 78 ------------- beacon_node/beacon_chain/src/metrics.rs | 21 ++-- .../sync/block_lookups/single_block_lookup.rs | 15 +-- 9 files changed, 78 insertions(+), 241 deletions(-) delete mode 100644 beacon_node/beacon_chain/src/data_availability_checker/processing_cache.rs diff --git a/beacon_node/beacon_chain/src/beacon_block_streamer.rs b/beacon_node/beacon_chain/src/beacon_block_streamer.rs index 4f4f8ed1fe..0fc65e2d18 100644 --- a/beacon_node/beacon_chain/src/beacon_block_streamer.rs +++ b/beacon_node/beacon_chain/src/beacon_block_streamer.rs @@ -1,4 +1,4 @@ -use crate::{BeaconChain, BeaconChainError, BeaconChainTypes}; +use crate::{metrics, BeaconChain, BeaconChainError, BeaconChainTypes}; use execution_layer::{ExecutionLayer, ExecutionPayloadBodyV1}; use slog::{crit, debug, Logger}; use std::collections::HashMap; @@ -410,8 +410,13 @@ impl BeaconBlockStreamer { fn check_caches(&self, root: Hash256) -> Option>> { if self.check_caches == CheckCaches::Yes { self.beacon_chain - .data_availability_checker - .get_block(&root) + .reqresp_pre_import_cache + .read() + .get(&root) + .map(|block| { + metrics::inc_counter(&metrics::BEACON_REQRESP_PRE_IMPORT_CACHE_HITS); + block.clone() + }) .or(self.beacon_chain.early_attester_cache.get_block(root)) } else { None diff --git a/beacon_node/beacon_chain/src/beacon_chain.rs b/beacon_node/beacon_chain/src/beacon_chain.rs index ea1c5089a3..c15f7f349f 100644 --- a/beacon_node/beacon_chain/src/beacon_chain.rs +++ b/beacon_node/beacon_chain/src/beacon_chain.rs @@ -120,7 +120,6 @@ use store::{ use task_executor::{ShutdownReason, TaskExecutor}; use tokio_stream::Stream; use tree_hash::TreeHash; -use types::beacon_state::CloneConfig; use types::blob_sidecar::{BlobSidecarList, FixedBlobSidecarList}; use types::payload::BlockProductionVersion; use types::*; @@ -359,6 +358,9 @@ pub type BeaconStore = Arc< >, >; +/// Cache gossip verified blocks to serve over ReqResp before they are imported +type ReqRespPreImportCache = HashMap>>; + /// Represents the "Beacon Chain" component of Ethereum 2.0. Allows import of blocks and block /// operations and chooses a canonical head. pub struct BeaconChain { @@ -461,6 +463,8 @@ pub struct BeaconChain { pub(crate) attester_cache: Arc, /// A cache used when producing attestations whilst the head block is still being imported. pub early_attester_cache: EarlyAttesterCache, + /// Cache gossip verified blocks to serve over ReqResp before they are imported + pub reqresp_pre_import_cache: Arc>>, /// A cache used to keep track of various block timings. pub block_times_cache: Arc>, /// A cache used to track pre-finalization block roots for quick rejection. @@ -2891,8 +2895,6 @@ impl BeaconChain { } } - self.data_availability_checker - .notify_gossip_blob(blob.slot(), block_root, &blob); let r = self.check_gossip_blob_availability_and_import(blob).await; self.remove_notified(&block_root, r) } @@ -2925,8 +2927,6 @@ impl BeaconChain { } } - self.data_availability_checker - .notify_rpc_blobs(slot, block_root, &blobs); let r = self .check_rpc_blob_availability_and_import(slot, block_root, blobs) .await; @@ -2943,7 +2943,7 @@ impl BeaconChain { let has_missing_components = matches!(r, Ok(AvailabilityProcessingStatus::MissingComponents(_, _))); if !has_missing_components { - self.data_availability_checker.remove_notified(block_root); + self.reqresp_pre_import_cache.write().remove(block_root); } r } @@ -2956,8 +2956,10 @@ impl BeaconChain { unverified_block: B, notify_execution_layer: NotifyExecutionLayer, ) -> Result> { - self.data_availability_checker - .notify_block(block_root, unverified_block.block_cloned()); + self.reqresp_pre_import_cache + .write() + .insert(block_root, unverified_block.block_cloned()); + let r = self .process_block(block_root, unverified_block, notify_execution_layer, || { Ok(()) diff --git a/beacon_node/beacon_chain/src/builder.rs b/beacon_node/beacon_chain/src/builder.rs index dd4b612f60..8ca704292e 100644 --- a/beacon_node/beacon_chain/src/builder.rs +++ b/beacon_node/beacon_chain/src/builder.rs @@ -962,6 +962,7 @@ where validator_pubkey_cache: TimeoutRwLock::new(validator_pubkey_cache), attester_cache: <_>::default(), early_attester_cache: <_>::default(), + reqresp_pre_import_cache: <_>::default(), light_client_server_cache: LightClientServerCache::new(), light_client_server_tx: self.light_client_server_tx, shutdown_sender: self diff --git a/beacon_node/beacon_chain/src/data_availability_checker.rs b/beacon_node/beacon_chain/src/data_availability_checker.rs index f906032ecd..4269fc9fed 100644 --- a/beacon_node/beacon_chain/src/data_availability_checker.rs +++ b/beacon_node/beacon_chain/src/data_availability_checker.rs @@ -7,11 +7,8 @@ pub use crate::data_availability_checker::availability_view::{ }; pub use crate::data_availability_checker::child_components::ChildComponents; use crate::data_availability_checker::overflow_lru_cache::OverflowLRUCache; -use crate::data_availability_checker::processing_cache::ProcessingCache; use crate::{BeaconChain, BeaconChainTypes, BeaconStore}; use kzg::Kzg; -use parking_lot::RwLock; -pub use processing_cache::ProcessingComponents; use slasher::test_utils::E; use slog::{debug, error, Logger}; use slot_clock::SlotClock; @@ -20,15 +17,13 @@ use std::fmt::Debug; use std::num::NonZeroUsize; use std::sync::Arc; use task_executor::TaskExecutor; -use types::beacon_block_body::KzgCommitmentOpts; use types::blob_sidecar::{BlobIdentifier, BlobSidecar, FixedBlobSidecarList}; -use types::{BlobSidecarList, ChainSpec, Epoch, EthSpec, Hash256, SignedBeaconBlock, Slot}; +use types::{BlobSidecarList, ChainSpec, Epoch, EthSpec, Hash256, SignedBeaconBlock}; mod availability_view; mod child_components; mod error; mod overflow_lru_cache; -mod processing_cache; mod state_lru_cache; pub use error::{Error as AvailabilityCheckError, ErrorCategory as AvailabilityCheckErrorCategory}; @@ -49,7 +44,6 @@ pub const STATE_LRU_CAPACITY: usize = STATE_LRU_CAPACITY_NON_ZERO.get(); /// `DataAvailabilityChecker` is responsible for KZG verification of block components as well as /// checking whether a "availability check" is required at all. pub struct DataAvailabilityChecker { - processing_cache: RwLock>, availability_cache: Arc>, slot_clock: T::SlotClock, kzg: Option>, @@ -88,7 +82,6 @@ impl DataAvailabilityChecker { ) -> Result { let overflow_cache = OverflowLRUCache::new(OVERFLOW_LRU_CAPACITY, store, spec.clone())?; Ok(Self { - processing_cache: <_>::default(), availability_cache: Arc::new(overflow_cache), slot_clock, log: log.clone(), @@ -97,17 +90,17 @@ impl DataAvailabilityChecker { }) } - /// Checks if the given block root is cached. + /// Checks if the block root is currenlty in the availability cache awaiting processing because + /// of missing components. pub fn has_block(&self, block_root: &Hash256) -> bool { - self.processing_cache.read().has_block(block_root) + self.availability_cache.has_block(block_root) } - /// Get the processing info for a block. - pub fn get_processing_components( - &self, - block_root: Hash256, - ) -> Option> { - self.processing_cache.read().get(&block_root).cloned() + pub fn get_missing_blob_ids_with(&self, block_root: Hash256) -> MissingBlobs { + self.availability_cache + .with_pending_components(&block_root, |pending_components| { + self.get_missing_blob_ids(block_root, pending_components) + }) } /// A `None` indicates blobs are not required. @@ -117,7 +110,7 @@ impl DataAvailabilityChecker { pub fn get_missing_blob_ids>( &self, block_root: Hash256, - availability_view: &V, + availability_view: Option<&V>, ) -> MissingBlobs { let Some(current_slot) = self.slot_clock.now_or_genesis() else { error!( @@ -130,6 +123,12 @@ impl DataAvailabilityChecker { let current_epoch = current_slot.epoch(T::EthSpec::slots_per_epoch()); if self.da_check_required_for_epoch(current_epoch) { + let Some(availability_view) = availability_view else { + return MissingBlobs::PossibleMissing(BlobIdentifier::get_all_blob_ids::( + block_root, + )); + }; + match availability_view.get_cached_block() { Some(cached_block) => { let block_commitments = cached_block.get_commitments(); @@ -192,14 +191,6 @@ impl DataAvailabilityChecker { self.availability_cache.peek_blob(blob_id) } - /// Get a block from the availability cache. Includes any blocks we are currently processing. - pub fn get_block(&self, block_root: &Hash256) -> Option>> { - self.processing_cache - .read() - .get(block_root) - .and_then(|cached| cached.block.clone()) - } - /// Put a list of blobs received via RPC into the availability cache. This performs KZG /// verification on the blobs in the list. pub fn put_rpc_blobs( @@ -352,71 +343,6 @@ impl DataAvailabilityChecker { block.num_expected_blobs() > 0 && self.da_check_required_for_epoch(block.epoch()) } - /// Adds a block to the processing cache. This block's commitments are unverified but caching - /// them here is useful to avoid duplicate downloads of blocks, as well as understanding - /// our blob download requirements. We will also serve this over RPC. - pub fn notify_block(&self, block_root: Hash256, block: Arc>) { - let slot = block.slot(); - self.processing_cache - .write() - .entry(block_root) - .or_insert_with(|| ProcessingComponents::new(slot)) - .merge_block(block); - } - - /// Add a single blob commitment to the processing cache. This commitment is unverified but caching - /// them here is useful to avoid duplicate downloads of blobs, as well as understanding - /// our block and blob download requirements. - pub fn notify_gossip_blob( - &self, - slot: Slot, - block_root: Hash256, - blob: &GossipVerifiedBlob, - ) { - let index = blob.index(); - let commitment = blob.kzg_commitment(); - self.processing_cache - .write() - .entry(block_root) - .or_insert_with(|| ProcessingComponents::new(slot)) - .merge_single_blob(index as usize, commitment); - } - - /// Adds blob commitments to the processing cache. These commitments are unverified but caching - /// them here is useful to avoid duplicate downloads of blobs, as well as understanding - /// our block and blob download requirements. - pub fn notify_rpc_blobs( - &self, - slot: Slot, - block_root: Hash256, - blobs: &FixedBlobSidecarList, - ) { - let mut commitments = KzgCommitmentOpts::::default(); - for blob in blobs.iter().flatten() { - if let Some(commitment) = commitments.get_mut(blob.index as usize) { - *commitment = Some(blob.kzg_commitment); - } - } - self.processing_cache - .write() - .entry(block_root) - .or_insert_with(|| ProcessingComponents::new(slot)) - .merge_blobs(commitments); - } - - /// Clears the block and all blobs from the processing cache for a give root if they exist. - pub fn remove_notified(&self, block_root: &Hash256) { - self.processing_cache.write().remove(block_root) - } - - /// Gather all block roots for which we are not currently processing all components for the - /// given slot. - pub fn incomplete_processing_components(&self, slot: Slot) -> Vec { - self.processing_cache - .read() - .incomplete_processing_components(slot) - } - /// The epoch at which we require a data availability check in block processing. /// `None` if the `Deneb` fork is disabled. pub fn data_availability_boundary(&self) -> Option { @@ -458,7 +384,6 @@ impl DataAvailabilityChecker { /// Collects metrics from the data availability checker. pub fn metrics(&self) -> DataAvailabilityCheckerMetrics { DataAvailabilityCheckerMetrics { - processing_cache_size: self.processing_cache.read().len(), num_store_entries: self.availability_cache.num_store_entries(), state_cache_size: self.availability_cache.state_cache_size(), block_cache_size: self.availability_cache.block_cache_size(), @@ -468,7 +393,6 @@ impl DataAvailabilityChecker { /// Helper struct to group data availability checker metrics. pub struct DataAvailabilityCheckerMetrics { - pub processing_cache_size: usize, pub num_store_entries: usize, pub state_cache_size: usize, pub block_cache_size: usize, diff --git a/beacon_node/beacon_chain/src/data_availability_checker/availability_view.rs b/beacon_node/beacon_chain/src/data_availability_checker/availability_view.rs index 65093db26b..b69843f678 100644 --- a/beacon_node/beacon_chain/src/data_availability_checker/availability_view.rs +++ b/beacon_node/beacon_chain/src/data_availability_checker/availability_view.rs @@ -3,7 +3,6 @@ use super::state_lru_cache::DietAvailabilityPendingExecutedBlock; use crate::blob_verification::KzgVerifiedBlob; use crate::block_verification_types::AsBlock; use crate::data_availability_checker::overflow_lru_cache::PendingComponents; -use crate::data_availability_checker::ProcessingComponents; use kzg::KzgCommitment; use ssz_types::FixedVector; use std::sync::Arc; @@ -180,14 +179,6 @@ macro_rules! impl_availability_view { }; } -impl_availability_view!( - ProcessingComponents, - Arc>, - KzgCommitment, - block, - blob_commitments -); - impl_availability_view!( PendingComponents, DietAvailabilityPendingExecutedBlock, @@ -303,32 +294,6 @@ pub mod tests { (block, blobs, invalid_blobs) } - type ProcessingViewSetup = ( - Arc>, - FixedVector, ::MaxBlobsPerBlock>, - FixedVector, ::MaxBlobsPerBlock>, - ); - - pub fn setup_processing_components( - block: SignedBeaconBlock, - valid_blobs: FixedVector>>, ::MaxBlobsPerBlock>, - invalid_blobs: FixedVector>>, ::MaxBlobsPerBlock>, - ) -> ProcessingViewSetup { - let blobs = FixedVector::from( - valid_blobs - .iter() - .map(|blob_opt| blob_opt.as_ref().map(|blob| blob.kzg_commitment)) - .collect::>(), - ); - let invalid_blobs = FixedVector::from( - invalid_blobs - .iter() - .map(|blob_opt| blob_opt.as_ref().map(|blob| blob.kzg_commitment)) - .collect::>(), - ); - (Arc::new(block), blobs, invalid_blobs) - } - type PendingComponentsSetup = ( DietAvailabilityPendingExecutedBlock, FixedVector>, ::MaxBlobsPerBlock>, @@ -517,13 +482,6 @@ pub mod tests { }; } - generate_tests!( - processing_components_tests, - ProcessingComponents::, - kzg_commitments, - processing_blobs, - setup_processing_components - ); generate_tests!( pending_components_tests, PendingComponents, 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 80cbc6c899..32ec33244f 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 @@ -305,6 +305,11 @@ impl Critical { Ok(()) } + /// Returns true if the block root is known, without altering the LRU ordering + pub fn has_block(&self, block_root: &Hash256) -> bool { + self.in_memory.peek(block_root).is_some() || self.store_keys.get(block_root).is_some() + } + /// This only checks for the blobs in memory pub fn peek_blob( &self, @@ -322,6 +327,13 @@ impl Critical { } } + pub fn peek_pending_components( + &self, + block_root: &Hash256, + ) -> Option<&PendingComponents> { + self.in_memory.peek(block_root) + } + /// Puts the pending components in the LRU cache. If the cache /// is at capacity, the LRU entry is written to the store first pub fn put_pending_components( @@ -409,6 +421,11 @@ impl OverflowLRUCache { }) } + /// Returns true if the block root is known, without altering the LRU ordering + pub fn has_block(&self, block_root: &Hash256) -> bool { + self.critical.read().has_block(block_root) + } + /// Fetch a blob from the cache without affecting the LRU ordering pub fn peek_blob( &self, @@ -425,6 +442,14 @@ impl OverflowLRUCache { } } + pub fn with_pending_components>) -> R>( + &self, + block_root: &Hash256, + f: F, + ) -> R { + f(self.critical.read().peek_pending_components(block_root)) + } + pub fn put_kzg_verified_blobs>>( &self, block_root: Hash256, diff --git a/beacon_node/beacon_chain/src/data_availability_checker/processing_cache.rs b/beacon_node/beacon_chain/src/data_availability_checker/processing_cache.rs deleted file mode 100644 index af94803dcf..0000000000 --- a/beacon_node/beacon_chain/src/data_availability_checker/processing_cache.rs +++ /dev/null @@ -1,78 +0,0 @@ -use crate::data_availability_checker::AvailabilityView; -use std::collections::hash_map::Entry; -use std::collections::HashMap; -use std::sync::Arc; -use types::beacon_block_body::KzgCommitmentOpts; -use types::{EthSpec, Hash256, SignedBeaconBlock, Slot}; - -/// This cache is used only for gossip blocks/blobs and single block/blob lookups, to give req/resp -/// a view of what we have and what we require. This cache serves a slightly different purpose than -/// gossip caches because it allows us to process duplicate blobs that are valid in gossip. -/// See `AvailabilityView`'s trait definition. -#[derive(Default)] -pub struct ProcessingCache { - processing_cache: HashMap>, -} - -impl ProcessingCache { - pub fn get(&self, block_root: &Hash256) -> Option<&ProcessingComponents> { - self.processing_cache.get(block_root) - } - pub fn entry(&mut self, block_root: Hash256) -> Entry<'_, Hash256, ProcessingComponents> { - self.processing_cache.entry(block_root) - } - pub fn remove(&mut self, block_root: &Hash256) { - self.processing_cache.remove(block_root); - } - pub fn has_block(&self, block_root: &Hash256) -> bool { - self.processing_cache - .get(block_root) - .map_or(false, |b| b.block_exists()) - } - pub fn incomplete_processing_components(&self, slot: Slot) -> Vec { - let mut roots_missing_components = vec![]; - for (&block_root, info) in self.processing_cache.iter() { - if info.slot == slot && !info.is_available() { - roots_missing_components.push(block_root); - } - } - roots_missing_components - } - pub fn len(&self) -> usize { - self.processing_cache.len() - } -} - -#[derive(Debug, Clone)] -pub struct ProcessingComponents { - slot: Slot, - /// Blobs required for a block can only be known if we have seen the block. So `Some` here - /// means we've seen it, a `None` means we haven't. The `kzg_commitments` value helps us figure - /// out whether incoming blobs actually match the block. - pub block: Option>>, - /// `KzgCommitments` for blobs are always known, even if we haven't seen the block. See - /// `AvailabilityView`'s trait definition for more details. - pub blob_commitments: KzgCommitmentOpts, -} - -impl ProcessingComponents { - pub fn new(slot: Slot) -> Self { - Self { - slot, - block: None, - blob_commitments: KzgCommitmentOpts::::default(), - } - } -} - -// Not safe for use outside of tests as this always required a slot. -#[cfg(test)] -impl ProcessingComponents { - pub fn empty(_block_root: Hash256) -> Self { - Self { - slot: Slot::new(0), - block: None, - blob_commitments: KzgCommitmentOpts::::default(), - } - } -} diff --git a/beacon_node/beacon_chain/src/metrics.rs b/beacon_node/beacon_chain/src/metrics.rs index abac2c80e7..f8c6188bba 100644 --- a/beacon_node/beacon_chain/src/metrics.rs +++ b/beacon_node/beacon_chain/src/metrics.rs @@ -304,6 +304,14 @@ lazy_static! { "Count of times the early attester cache returns an attestation" ); + pub static ref BEACON_REQRESP_PRE_IMPORT_CACHE_SIZE: Result = try_create_int_gauge( + "beacon_reqresp_pre_import_cache_size", + "Current count of items of the reqresp pre import cache" + ); + pub static ref BEACON_REQRESP_PRE_IMPORT_CACHE_HITS: Result = try_create_int_counter( + "beacon_reqresp_pre_import_cache_hits", + "Count of times the reqresp pre import cache returns an item" + ); } // Second lazy-static block is used to account for macro recursion limit. @@ -1129,15 +1137,9 @@ lazy_static! { Ok(vec![0.1, 0.2, 0.3,0.4,0.5,0.75,1.0,1.25,1.5,1.75,2.0,2.5,3.0,3.5,4.0,5.0,6.0,7.0,8.0,9.0,10.0,15.0,20.0]) ); - /* * Data Availability cache metrics */ - pub static ref DATA_AVAILABILITY_PROCESSING_CACHE_SIZE: Result = - try_create_int_gauge( - "data_availability_processing_cache_size", - "Number of entries in the data availability processing cache." - ); pub static ref DATA_AVAILABILITY_OVERFLOW_MEMORY_BLOCK_CACHE_SIZE: Result = try_create_int_gauge( "data_availability_overflow_memory_block_cache_size", @@ -1196,11 +1198,12 @@ pub fn scrape_for_metrics(beacon_chain: &BeaconChain) { ) } - let da_checker_metrics = beacon_chain.data_availability_checker.metrics(); set_gauge_by_usize( - &DATA_AVAILABILITY_PROCESSING_CACHE_SIZE, - da_checker_metrics.processing_cache_size, + &BEACON_REQRESP_PRE_IMPORT_CACHE_SIZE, + beacon_chain.reqresp_pre_import_cache.read().len(), ); + + let da_checker_metrics = beacon_chain.data_availability_checker.metrics(); set_gauge_by_usize( &DATA_AVAILABILITY_OVERFLOW_MEMORY_BLOCK_CACHE_SIZE, da_checker_metrics.block_cache_size, diff --git a/beacon_node/network/src/sync/block_lookups/single_block_lookup.rs b/beacon_node/network/src/sync/block_lookups/single_block_lookup.rs index 8c60621f1c..c149a75abb 100644 --- a/beacon_node/network/src/sync/block_lookups/single_block_lookup.rs +++ b/beacon_node/network/src/sync/block_lookups/single_block_lookup.rs @@ -272,19 +272,16 @@ impl SingleBlockLookup { pub(crate) fn missing_blob_ids(&self) -> MissingBlobs { let block_root = self.block_root(); if let Some(components) = self.child_components.as_ref() { - self.da_checker.get_missing_blob_ids(block_root, components) - } else { - let Some(processing_availability_view) = - self.da_checker.get_processing_components(block_root) - else { - return MissingBlobs::new_without_block(block_root, self.da_checker.is_deneb()); - }; self.da_checker - .get_missing_blob_ids(block_root, &processing_availability_view) + .get_missing_blob_ids(block_root, Some(components)) + } else { + // TODO: the da_checker should look into the availability cache, not only the processing + // cache. + self.da_checker.get_missing_blob_ids_with(block_root) } } - /// Penalizes a blob peer if it should have blobs but didn't return them to us. + /// Penalizes a blob peer if it should have blobs but didn't return them to us. pub fn penalize_blob_peer(&mut self, cx: &SyncNetworkContext) { if let Ok(blob_peer) = self.blob_request_state.state.processing_peer() { cx.report_peer( From a6cd31469ee91ba49c00c6d7f10ed7fe3bb131b3 Mon Sep 17 00:00:00 2001 From: realbigsean Date: Tue, 9 Apr 2024 15:34:29 -0400 Subject: [PATCH 2/4] remove unused file, remove outdated TODO, add is_deneb check to missing blob id calculations --- .../src/data_availability_checker.rs | 4 +- .../processing_cache.rs | 63 ------------------- .../sync/block_lookups/single_block_lookup.rs | 2 - 3 files changed, 1 insertion(+), 68 deletions(-) delete mode 100644 beacon_node/beacon_chain/src/data_availability_checker/processing_cache.rs diff --git a/beacon_node/beacon_chain/src/data_availability_checker.rs b/beacon_node/beacon_chain/src/data_availability_checker.rs index 84aa61f6bf..1d4e7e57e9 100644 --- a/beacon_node/beacon_chain/src/data_availability_checker.rs +++ b/beacon_node/beacon_chain/src/data_availability_checker.rs @@ -108,9 +108,7 @@ impl DataAvailabilityChecker { .map(|b| b.as_block()), &pending_components.verified_blobs, ), - None => MissingBlobs::PossibleMissing( - BlobIdentifier::get_all_blob_ids::(block_root), - ), + None => MissingBlobs::new_without_block(block_root, self.is_deneb()), }) } diff --git a/beacon_node/beacon_chain/src/data_availability_checker/processing_cache.rs b/beacon_node/beacon_chain/src/data_availability_checker/processing_cache.rs deleted file mode 100644 index e09b3083be..0000000000 --- a/beacon_node/beacon_chain/src/data_availability_checker/processing_cache.rs +++ /dev/null @@ -1,63 +0,0 @@ -use crate::data_availability_checker::AvailabilityView; -use std::collections::hash_map::Entry; -use std::collections::HashMap; -use std::sync::Arc; -use types::beacon_block_body::KzgCommitmentOpts; -use types::{EthSpec, Hash256, SignedBeaconBlock}; - -/// This cache is used only for gossip blocks/blobs and single block/blob lookups, to give req/resp -/// a view of what we have and what we require. This cache serves a slightly different purpose than -/// gossip caches because it allows us to process duplicate blobs that are valid in gossip. -/// See `AvailabilityView`'s trait definition. -#[derive(Default)] -pub struct ProcessingCache { - processing_cache: HashMap>, -} - -impl ProcessingCache { - pub fn get(&self, block_root: &Hash256) -> Option<&ProcessingComponents> { - self.processing_cache.get(block_root) - } - pub fn entry(&mut self, block_root: Hash256) -> Entry<'_, Hash256, ProcessingComponents> { - self.processing_cache.entry(block_root) - } - pub fn remove(&mut self, block_root: &Hash256) { - self.processing_cache.remove(block_root); - } - pub fn has_block(&self, block_root: &Hash256) -> bool { - self.processing_cache - .get(block_root) - .map_or(false, |b| b.block_exists()) - } - pub fn len(&self) -> usize { - self.processing_cache.len() - } -} - -#[derive(Default, Debug, Clone)] -pub struct ProcessingComponents { - /// Blobs required for a block can only be known if we have seen the block. So `Some` here - /// means we've seen it, a `None` means we haven't. The `kzg_commitments` value helps us figure - /// out whether incoming blobs actually match the block. - pub block: Option>>, - /// `KzgCommitments` for blobs are always known, even if we haven't seen the block. See - /// `AvailabilityView`'s trait definition for more details. - pub blob_commitments: KzgCommitmentOpts, -} - -impl ProcessingComponents { - pub fn new() -> Self { - Self::default() - } -} - -// Not safe for use outside of tests as this always required a slot. -#[cfg(test)] -impl ProcessingComponents { - pub fn empty(_block_root: Hash256) -> Self { - Self { - block: None, - blob_commitments: KzgCommitmentOpts::::default(), - } - } -} diff --git a/beacon_node/network/src/sync/block_lookups/single_block_lookup.rs b/beacon_node/network/src/sync/block_lookups/single_block_lookup.rs index abee12905c..a312f6e970 100644 --- a/beacon_node/network/src/sync/block_lookups/single_block_lookup.rs +++ b/beacon_node/network/src/sync/block_lookups/single_block_lookup.rs @@ -282,8 +282,6 @@ impl SingleBlockLookup { &components.downloaded_blobs, ) } else { - // TODO: the da_checker should look into the availability cache, not only the processing - // cache. self.da_checker.get_missing_blob_ids_with(block_root) } } From 7ebfda91a2a3132ca5495ed08f4173349ebff2ae Mon Sep 17 00:00:00 2001 From: realbigsean Date: Tue, 9 Apr 2024 16:07:58 -0400 Subject: [PATCH 3/4] remove availability view trait --- .../beacon_chain/src/blob_verification.rs | 3 + .../src/data_availability_checker.rs | 4 - .../availability_view.rs | 465 ------------------ .../overflow_lru_cache.rs | 335 ++++++++++++- .../state_lru_cache.rs | 10 + 5 files changed, 347 insertions(+), 470 deletions(-) delete mode 100644 beacon_node/beacon_chain/src/data_availability_checker/availability_view.rs diff --git a/beacon_node/beacon_chain/src/blob_verification.rs b/beacon_node/beacon_chain/src/blob_verification.rs index a69f2b7452..1fb6170200 100644 --- a/beacon_node/beacon_chain/src/blob_verification.rs +++ b/beacon_node/beacon_chain/src/blob_verification.rs @@ -284,6 +284,9 @@ impl KzgVerifiedBlob { pub fn as_blob(&self) -> &BlobSidecar { &self.blob } + pub fn get_commitment(&self) -> &KzgCommitment { + &self.blob.kzg_commitment + } /// This is cheap as we're calling clone on an Arc pub fn clone_blob(&self) -> Arc> { self.blob.clone() diff --git a/beacon_node/beacon_chain/src/data_availability_checker.rs b/beacon_node/beacon_chain/src/data_availability_checker.rs index 1d4e7e57e9..3ef105c6d3 100644 --- a/beacon_node/beacon_chain/src/data_availability_checker.rs +++ b/beacon_node/beacon_chain/src/data_availability_checker.rs @@ -2,9 +2,6 @@ use crate::blob_verification::{verify_kzg_for_blob_list, GossipVerifiedBlob, Kzg use crate::block_verification_types::{ AvailabilityPendingExecutedBlock, AvailableExecutedBlock, RpcBlock, }; -pub use crate::data_availability_checker::availability_view::{ - AvailabilityView, GetCommitment, GetCommitments, -}; pub use crate::data_availability_checker::child_components::ChildComponents; use crate::data_availability_checker::overflow_lru_cache::OverflowLRUCache; use crate::{BeaconChain, BeaconChainTypes, BeaconStore}; @@ -21,7 +18,6 @@ use task_executor::TaskExecutor; use types::blob_sidecar::{BlobIdentifier, BlobSidecar, FixedBlobSidecarList}; use types::{BlobSidecarList, ChainSpec, Epoch, EthSpec, Hash256, SignedBeaconBlock}; -mod availability_view; mod child_components; mod error; mod overflow_lru_cache; diff --git a/beacon_node/beacon_chain/src/data_availability_checker/availability_view.rs b/beacon_node/beacon_chain/src/data_availability_checker/availability_view.rs deleted file mode 100644 index d4e5ca3449..0000000000 --- a/beacon_node/beacon_chain/src/data_availability_checker/availability_view.rs +++ /dev/null @@ -1,465 +0,0 @@ -use super::state_lru_cache::DietAvailabilityPendingExecutedBlock; -use crate::blob_verification::KzgVerifiedBlob; -use crate::block_verification_types::AsBlock; -use crate::data_availability_checker::overflow_lru_cache::PendingComponents; -use kzg::KzgCommitment; -use ssz_types::FixedVector; -use std::sync::Arc; -use types::beacon_block_body::KzgCommitments; -use types::{BlobSidecar, EthSpec, SignedBeaconBlock}; - -/// Defines an interface for managing data availability with two key invariants: -/// -/// 1. If we haven't seen a block yet, we will insert the first blob for a given (block_root, index) -/// but we won't insert subsequent blobs for the same (block_root, index) if they have a different -/// commitment. -/// 2. On block insertion, any non-matching blob commitments are evicted. -/// -/// Types implementing this trait can be used for validating and managing availability -/// of blocks and blobs in a cache-like data structure. -pub trait AvailabilityView { - /// The type representing a block in the implementation. - type BlockType: GetCommitments; - - /// The type representing a blob in the implementation. Must implement `Clone`. - type BlobType: Clone + GetCommitment; - - /// Returns an immutable reference to the cached block. - fn get_cached_block(&self) -> &Option; - - /// Returns an immutable reference to the fixed vector of cached blobs. - fn get_cached_blobs(&self) -> &FixedVector, E::MaxBlobsPerBlock>; - - /// Returns a mutable reference to the cached block. - fn get_cached_block_mut(&mut self) -> &mut Option; - - /// Returns a mutable reference to the fixed vector of cached blobs. - fn get_cached_blobs_mut( - &mut self, - ) -> &mut FixedVector, E::MaxBlobsPerBlock>; - - /// Checks if a block exists in the cache. - /// - /// Returns: - /// - `true` if a block exists. - /// - `false` otherwise. - fn block_exists(&self) -> bool { - self.get_cached_block().is_some() - } - - /// Checks if a blob exists at the given index in the cache. - /// - /// Returns: - /// - `true` if a blob exists at the given index. - /// - `false` otherwise. - fn blob_exists(&self, blob_index: usize) -> bool { - self.get_cached_blobs() - .get(blob_index) - .map(|b| b.is_some()) - .unwrap_or(false) - } - - /// Returns the number of blobs that are expected to be present. Returns `None` if we don't have a - /// block. - /// - /// This corresponds to the number of commitments that are present in a block. - fn num_expected_blobs(&self) -> Option { - self.get_cached_block() - .as_ref() - .map(|b| b.get_commitments().len()) - } - - /// Returns the number of blobs that have been received and are stored in the cache. - fn num_received_blobs(&self) -> usize { - self.get_cached_blobs().iter().flatten().count() - } - - /// Inserts a block into the cache. - fn insert_block(&mut self, block: Self::BlockType) { - *self.get_cached_block_mut() = Some(block) - } - - /// Inserts a blob at a specific index in the cache. - /// - /// Existing blob at the index will be replaced. - fn insert_blob_at_index(&mut self, blob_index: usize, blob: Self::BlobType) { - if let Some(b) = self.get_cached_blobs_mut().get_mut(blob_index) { - *b = Some(blob); - } - } - - /// Merges a given set of blobs into the cache. - /// - /// 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. - fn merge_blobs(&mut self, blobs: FixedVector, E::MaxBlobsPerBlock>) { - for (index, blob) in blobs.iter().cloned().enumerate() { - let Some(blob) = blob else { continue }; - self.merge_single_blob(index, blob); - } - } - - /// Merges a single blob into the cache. - /// - /// Blobs are only inserted if: - /// 1. The blob entry at the index is empty and no block exists, or - /// 2. The block exists and its commitment matches the blob's commitment. - fn merge_single_blob(&mut self, index: usize, blob: Self::BlobType) { - if let Some(cached_block) = self.get_cached_block() { - let block_commitment_opt = cached_block.get_commitments().get(index).copied(); - if let Some(block_commitment) = block_commitment_opt { - if block_commitment == *blob.get_commitment() { - self.insert_blob_at_index(index, blob) - } - } - } else if !self.blob_exists(index) { - self.insert_blob_at_index(index, blob) - } - } - - /// Inserts a new block and revalidates the existing blobs against it. - /// - /// Blobs that don't match the new block's commitments are evicted. - fn merge_block(&mut self, block: Self::BlockType) { - self.insert_block(block); - let reinsert = std::mem::take(self.get_cached_blobs_mut()); - self.merge_blobs(reinsert); - } - - /// Checks if the block and all of its expected blobs are available in the cache. - /// - /// Returns `true` if both the block exists and the number of received blobs matches the number - /// of expected blobs. - fn is_available(&self) -> bool { - if let Some(num_expected_blobs) = self.num_expected_blobs() { - num_expected_blobs == self.num_received_blobs() - } else { - false - } - } -} - -/// Implements the `AvailabilityView` trait for a given struct. -/// -/// - `$struct_name`: The name of the struct for which to implement `AvailabilityView`. -/// - `$block_type`: The type to use for `BlockType` in the `AvailabilityView` trait. -/// - `$blob_type`: The type to use for `BlobType` in the `AvailabilityView` trait. -/// - `$block_field`: The field name in the struct that holds the cached block. -/// - `$blob_field`: The field name in the struct that holds the cached blobs. -#[macro_export] -macro_rules! impl_availability_view { - ($struct_name:ident, $block_type:ty, $blob_type:ty, $block_field:ident, $blob_field:ident) => { - impl AvailabilityView for $struct_name { - type BlockType = $block_type; - type BlobType = $blob_type; - - fn get_cached_block(&self) -> &Option { - &self.$block_field - } - - fn get_cached_blobs( - &self, - ) -> &FixedVector, E::MaxBlobsPerBlock> { - &self.$blob_field - } - - fn get_cached_block_mut(&mut self) -> &mut Option { - &mut self.$block_field - } - - fn get_cached_blobs_mut( - &mut self, - ) -> &mut FixedVector, E::MaxBlobsPerBlock> { - &mut self.$blob_field - } - } - }; -} - -impl_availability_view!( - PendingComponents, - DietAvailabilityPendingExecutedBlock, - KzgVerifiedBlob, - executed_block, - verified_blobs -); - -pub trait GetCommitments { - fn get_commitments(&self) -> KzgCommitments; -} - -pub trait GetCommitment { - fn get_commitment(&self) -> &KzgCommitment; -} - -impl GetCommitment for KzgCommitment { - fn get_commitment(&self) -> &KzgCommitment { - self - } -} - -// These implementations are required to implement `AvailabilityView` for `PendingComponents`. -impl GetCommitments for DietAvailabilityPendingExecutedBlock { - fn get_commitments(&self) -> KzgCommitments { - self.as_block() - .message() - .body() - .blob_kzg_commitments() - .cloned() - .unwrap_or_default() - } -} - -impl GetCommitment for KzgVerifiedBlob { - fn get_commitment(&self) -> &KzgCommitment { - &self.as_blob().kzg_commitment - } -} - -// These implementations are required to implement `AvailabilityView` for `ChildComponents`. -impl GetCommitments for Arc> { - fn get_commitments(&self) -> KzgCommitments { - self.message() - .body() - .blob_kzg_commitments() - .ok() - .cloned() - .unwrap_or_default() - } -} -impl GetCommitment for Arc> { - fn get_commitment(&self) -> &KzgCommitment { - &self.kzg_commitment - } -} - -#[cfg(test)] -pub mod 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::AvailabilityPendingExecutedBlock; - use crate::PayloadVerificationOutcome; - use fork_choice::PayloadVerificationStatus; - use rand::rngs::StdRng; - use rand::SeedableRng; - use state_processing::ConsensusContext; - use types::test_utils::TestRandom; - use types::{BeaconState, ChainSpec, ForkName, MainnetEthSpec, Slot}; - - type E = MainnetEthSpec; - - type Setup = ( - SignedBeaconBlock, - FixedVector>>, ::MaxBlobsPerBlock>, - FixedVector>>, ::MaxBlobsPerBlock>, - ); - - pub fn pre_setup() -> Setup { - let mut rng = StdRng::seed_from_u64(0xDEADBEEF0BAD5EEDu64); - let (block, blobs_vec) = - generate_rand_block_and_blobs::(ForkName::Deneb, NumBlobs::Random, &mut rng); - let mut blobs: FixedVector<_, ::MaxBlobsPerBlock> = FixedVector::default(); - - for blob in blobs_vec { - if let Some(b) = blobs.get_mut(blob.index as usize) { - *b = Some(Arc::new(blob)); - } - } - - let mut invalid_blobs: FixedVector< - Option>>, - ::MaxBlobsPerBlock, - > = FixedVector::default(); - for (index, blob) in blobs.iter().enumerate() { - if let Some(invalid_blob) = blob { - let mut blob_copy = invalid_blob.as_ref().clone(); - blob_copy.kzg_commitment = KzgCommitment::random_for_test(&mut rng); - *invalid_blobs.get_mut(index).unwrap() = Some(Arc::new(blob_copy)); - } - } - - (block, blobs, invalid_blobs) - } - - type PendingComponentsSetup = ( - DietAvailabilityPendingExecutedBlock, - FixedVector>, ::MaxBlobsPerBlock>, - FixedVector>, ::MaxBlobsPerBlock>, - ); - - pub fn setup_pending_components( - block: SignedBeaconBlock, - valid_blobs: FixedVector>>, ::MaxBlobsPerBlock>, - invalid_blobs: FixedVector>>, ::MaxBlobsPerBlock>, - ) -> PendingComponentsSetup { - let blobs = FixedVector::from( - valid_blobs - .iter() - .map(|blob_opt| { - blob_opt - .as_ref() - .map(|blob| KzgVerifiedBlob::__assumed_valid(blob.clone())) - }) - .collect::>(), - ); - let invalid_blobs = FixedVector::from( - invalid_blobs - .iter() - .map(|blob_opt| { - blob_opt - .as_ref() - .map(|blob| KzgVerifiedBlob::__assumed_valid(blob.clone())) - }) - .collect::>(), - ); - let dummy_parent = block.clone_as_blinded(); - let block = AvailabilityPendingExecutedBlock { - block: Arc::new(block), - import_data: BlockImportData { - block_root: Default::default(), - state: BeaconState::new(0, Default::default(), &ChainSpec::minimal()), - parent_block: dummy_parent, - parent_eth1_finalization_data: Eth1FinalizationData { - eth1_data: Default::default(), - eth1_deposit_index: 0, - }, - confirmed_state_roots: vec![], - consensus_context: ConsensusContext::new(Slot::new(0)), - }, - payload_verification_outcome: PayloadVerificationOutcome { - payload_verification_status: PayloadVerificationStatus::Verified, - is_valid_merge_transition_block: false, - }, - }; - (block.into(), blobs, invalid_blobs) - } - - pub fn assert_cache_consistent>(cache: V) { - 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() { - 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()); - assert_eq!(block_commitment, blob_commitment); - } - } else { - panic!("No cached block") - } - } - - pub fn assert_empty_blob_cache>(cache: V) { - for blob in cache.get_cached_blobs().iter() { - assert!(blob.is_none()); - } - } - - #[macro_export] - macro_rules! generate_tests { - ($module_name:ident, $type_name:ty, $block_field:ident, $blob_field:ident, $setup_fn:ident) => { - mod $module_name { - use super::*; - use types::Hash256; - - #[test] - fn valid_block_invalid_blobs_valid_blobs() { - let (block_commitments, blobs, random_blobs) = pre_setup(); - let (block_commitments, blobs, random_blobs) = - $setup_fn(block_commitments, blobs, random_blobs); - let block_root = Hash256::zero(); - let mut cache = <$type_name>::empty(block_root); - cache.merge_block(block_commitments); - cache.merge_blobs(random_blobs); - cache.merge_blobs(blobs); - - assert_cache_consistent(cache); - } - - #[test] - fn invalid_blobs_block_valid_blobs() { - let (block_commitments, blobs, random_blobs) = pre_setup(); - let (block_commitments, blobs, random_blobs) = - $setup_fn(block_commitments, blobs, random_blobs); - let block_root = Hash256::zero(); - let mut cache = <$type_name>::empty(block_root); - cache.merge_blobs(random_blobs); - cache.merge_block(block_commitments); - cache.merge_blobs(blobs); - - assert_cache_consistent(cache); - } - - #[test] - fn invalid_blobs_valid_blobs_block() { - let (block_commitments, blobs, random_blobs) = pre_setup(); - let (block_commitments, blobs, random_blobs) = - $setup_fn(block_commitments, blobs, random_blobs); - - let block_root = Hash256::zero(); - let mut cache = <$type_name>::empty(block_root); - cache.merge_blobs(random_blobs); - cache.merge_blobs(blobs); - cache.merge_block(block_commitments); - - assert_empty_blob_cache(cache); - } - - #[test] - fn block_valid_blobs_invalid_blobs() { - let (block_commitments, blobs, random_blobs) = pre_setup(); - let (block_commitments, blobs, random_blobs) = - $setup_fn(block_commitments, blobs, random_blobs); - - let block_root = Hash256::zero(); - let mut cache = <$type_name>::empty(block_root); - cache.merge_block(block_commitments); - cache.merge_blobs(blobs); - cache.merge_blobs(random_blobs); - - assert_cache_consistent(cache); - } - - #[test] - fn valid_blobs_block_invalid_blobs() { - let (block_commitments, blobs, random_blobs) = pre_setup(); - let (block_commitments, blobs, random_blobs) = - $setup_fn(block_commitments, blobs, random_blobs); - - let block_root = Hash256::zero(); - let mut cache = <$type_name>::empty(block_root); - cache.merge_blobs(blobs); - cache.merge_block(block_commitments); - cache.merge_blobs(random_blobs); - - assert_cache_consistent(cache); - } - - #[test] - fn valid_blobs_invalid_blobs_block() { - let (block_commitments, blobs, random_blobs) = pre_setup(); - let (block_commitments, blobs, random_blobs) = - $setup_fn(block_commitments, blobs, random_blobs); - - let block_root = Hash256::zero(); - let mut cache = <$type_name>::empty(block_root); - cache.merge_blobs(blobs); - cache.merge_blobs(random_blobs); - cache.merge_block(block_commitments); - - assert_cache_consistent(cache); - } - } - }; - } - - generate_tests!( - pending_components_tests, - PendingComponents, - executed_block, - verified_blobs, - setup_pending_components - ); -} 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 db26d70982..4b4d696801 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 @@ -33,7 +33,6 @@ use crate::blob_verification::KzgVerifiedBlob; use crate::block_verification_types::{ AvailabilityPendingExecutedBlock, AvailableBlock, AvailableExecutedBlock, }; -use crate::data_availability_checker::availability_view::AvailabilityView; use crate::data_availability_checker::{Availability, AvailabilityCheckError}; use crate::store::{DBColumn, KeyValueStore}; use crate::BeaconChainTypes; @@ -59,6 +58,126 @@ pub struct PendingComponents { } impl PendingComponents { + /// Returns an immutable reference to the cached block. + pub fn get_cached_block(&self) -> &Option> { + &self.executed_block + } + + /// Returns an immutable reference to the fixed vector of cached blobs. + pub fn get_cached_blobs( + &self, + ) -> &FixedVector>, E::MaxBlobsPerBlock> { + &self.verified_blobs + } + + /// Returns a mutable reference to the cached block. + pub fn get_cached_block_mut(&mut self) -> &mut Option> { + &mut self.executed_block + } + + /// Returns a mutable reference to the fixed vector of cached blobs. + pub fn get_cached_blobs_mut( + &mut self, + ) -> &mut FixedVector>, E::MaxBlobsPerBlock> { + &mut self.verified_blobs + } + + /// Checks if a blob exists at the given index in the cache. + /// + /// Returns: + /// - `true` if a blob exists at the given index. + /// - `false` otherwise. + pub fn blob_exists(&self, blob_index: usize) -> bool { + self.get_cached_blobs() + .get(blob_index) + .map(|b| b.is_some()) + .unwrap_or(false) + } + + /// Returns the number of blobs that are expected to be present. Returns `None` if we don't have a + /// block. + /// + /// This corresponds to the number of commitments that are present in a block. + pub fn num_expected_blobs(&self) -> Option { + self.get_cached_block() + .as_ref() + .map(|b| b.get_commitments().len()) + } + + /// Returns the number of blobs that have been received and are stored in the cache. + pub fn num_received_blobs(&self) -> usize { + self.get_cached_blobs().iter().flatten().count() + } + + /// Inserts a block into the cache. + pub fn insert_block(&mut self, block: DietAvailabilityPendingExecutedBlock) { + *self.get_cached_block_mut() = Some(block) + } + + /// Inserts a blob at a specific index in the cache. + /// + /// Existing blob at the index will be replaced. + pub fn insert_blob_at_index(&mut self, blob_index: usize, blob: KzgVerifiedBlob) { + if let Some(b) = self.get_cached_blobs_mut().get_mut(blob_index) { + *b = Some(blob); + } + } + + /// Merges a given set of blobs into the cache. + /// + /// 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>, E::MaxBlobsPerBlock>, + ) { + for (index, blob) in blobs.iter().cloned().enumerate() { + let Some(blob) = blob else { continue }; + self.merge_single_blob(index, blob); + } + } + + /// Merges a single blob into the cache. + /// + /// Blobs are only inserted if: + /// 1. The blob entry at the index is empty and no block exists, or + /// 2. The block exists and its commitment matches the blob's commitment. + pub fn merge_single_blob(&mut self, index: usize, blob: KzgVerifiedBlob) { + if let Some(cached_block) = self.get_cached_block() { + let block_commitment_opt = cached_block.get_commitments().get(index).copied(); + if let Some(block_commitment) = block_commitment_opt { + if block_commitment == *blob.get_commitment() { + self.insert_blob_at_index(index, blob) + } + } + } else if !self.blob_exists(index) { + self.insert_blob_at_index(index, blob) + } + } + + /// Inserts a new block and revalidates the existing blobs against it. + /// + /// Blobs that don't match the new block's commitments are evicted. + pub fn merge_block(&mut self, block: DietAvailabilityPendingExecutedBlock) { + self.insert_block(block); + let reinsert = std::mem::take(self.get_cached_blobs_mut()); + self.merge_blobs(reinsert); + } + + /// Checks if the block and all of its expected blobs are available in the cache. + /// + /// Returns `true` if both the block exists and the number of received blobs matches the number + /// of expected blobs. + pub fn is_available(&self) -> bool { + if let Some(num_expected_blobs) = self.num_expected_blobs() { + num_expected_blobs == self.num_received_blobs() + } else { + false + } + } + + /// Returns an empty `PendingComponents` object with the given block root. pub fn empty(block_root: Hash256) -> Self { Self { block_root, @@ -118,6 +237,7 @@ impl PendingComponents { ))) } + /// Returns the epoch of the block if it is cached, otherwise returns the epoch of the first blob. pub fn epoch(&self) -> Option { self.executed_block .as_ref() @@ -1703,3 +1823,216 @@ mod test { ); } } + +#[cfg(test)] +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::AvailabilityPendingExecutedBlock; + use crate::PayloadVerificationOutcome; + use fork_choice::PayloadVerificationStatus; + use kzg::KzgCommitment; + use rand::rngs::StdRng; + use rand::SeedableRng; + use state_processing::ConsensusContext; + use types::test_utils::TestRandom; + use types::{BeaconState, ChainSpec, ForkName, MainnetEthSpec, SignedBeaconBlock, Slot}; + + type E = MainnetEthSpec; + + type Setup = ( + SignedBeaconBlock, + FixedVector>>, ::MaxBlobsPerBlock>, + FixedVector>>, ::MaxBlobsPerBlock>, + ); + + pub fn pre_setup() -> Setup { + let mut rng = StdRng::seed_from_u64(0xDEADBEEF0BAD5EEDu64); + let (block, blobs_vec) = + generate_rand_block_and_blobs::(ForkName::Deneb, NumBlobs::Random, &mut rng); + let mut blobs: FixedVector<_, ::MaxBlobsPerBlock> = FixedVector::default(); + + for blob in blobs_vec { + if let Some(b) = blobs.get_mut(blob.index as usize) { + *b = Some(Arc::new(blob)); + } + } + + let mut invalid_blobs: FixedVector< + Option>>, + ::MaxBlobsPerBlock, + > = FixedVector::default(); + for (index, blob) in blobs.iter().enumerate() { + if let Some(invalid_blob) = blob { + let mut blob_copy = invalid_blob.as_ref().clone(); + blob_copy.kzg_commitment = KzgCommitment::random_for_test(&mut rng); + *invalid_blobs.get_mut(index).unwrap() = Some(Arc::new(blob_copy)); + } + } + + (block, blobs, invalid_blobs) + } + + type PendingComponentsSetup = ( + DietAvailabilityPendingExecutedBlock, + FixedVector>, ::MaxBlobsPerBlock>, + FixedVector>, ::MaxBlobsPerBlock>, + ); + + pub fn setup_pending_components( + block: SignedBeaconBlock, + valid_blobs: FixedVector>>, ::MaxBlobsPerBlock>, + invalid_blobs: FixedVector>>, ::MaxBlobsPerBlock>, + ) -> PendingComponentsSetup { + let blobs = FixedVector::from( + valid_blobs + .iter() + .map(|blob_opt| { + blob_opt + .as_ref() + .map(|blob| KzgVerifiedBlob::__assumed_valid(blob.clone())) + }) + .collect::>(), + ); + let invalid_blobs = FixedVector::from( + invalid_blobs + .iter() + .map(|blob_opt| { + blob_opt + .as_ref() + .map(|blob| KzgVerifiedBlob::__assumed_valid(blob.clone())) + }) + .collect::>(), + ); + let dummy_parent = block.clone_as_blinded(); + let block = AvailabilityPendingExecutedBlock { + block: Arc::new(block), + import_data: BlockImportData { + block_root: Default::default(), + state: BeaconState::new(0, Default::default(), &ChainSpec::minimal()), + parent_block: dummy_parent, + parent_eth1_finalization_data: Eth1FinalizationData { + eth1_data: Default::default(), + eth1_deposit_index: 0, + }, + confirmed_state_roots: vec![], + consensus_context: ConsensusContext::new(Slot::new(0)), + }, + payload_verification_outcome: PayloadVerificationOutcome { + payload_verification_status: PayloadVerificationStatus::Verified, + is_valid_merge_transition_block: false, + }, + }; + (block.into(), blobs, invalid_blobs) + } + + pub fn assert_cache_consistent(cache: PendingComponents) { + 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() { + 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()); + assert_eq!(block_commitment, blob_commitment); + } + } else { + panic!("No cached block") + } + } + + pub fn assert_empty_blob_cache(cache: PendingComponents) { + for blob in cache.get_cached_blobs().iter() { + assert!(blob.is_none()); + } + } + + #[test] + fn valid_block_invalid_blobs_valid_blobs() { + let (block_commitments, blobs, random_blobs) = pre_setup(); + let (block_commitments, blobs, random_blobs) = + setup_pending_components(block_commitments, blobs, random_blobs); + let block_root = Hash256::zero(); + let mut cache = >::empty(block_root); + cache.merge_block(block_commitments); + cache.merge_blobs(random_blobs); + cache.merge_blobs(blobs); + + assert_cache_consistent(cache); + } + + #[test] + fn invalid_blobs_block_valid_blobs() { + let (block_commitments, blobs, random_blobs) = pre_setup(); + let (block_commitments, blobs, random_blobs) = + setup_pending_components(block_commitments, blobs, random_blobs); + let block_root = Hash256::zero(); + let mut cache = >::empty(block_root); + cache.merge_blobs(random_blobs); + cache.merge_block(block_commitments); + cache.merge_blobs(blobs); + + assert_cache_consistent(cache); + } + + #[test] + fn invalid_blobs_valid_blobs_block() { + let (block_commitments, blobs, random_blobs) = pre_setup(); + let (block_commitments, blobs, random_blobs) = + setup_pending_components(block_commitments, blobs, random_blobs); + + let block_root = Hash256::zero(); + let mut cache = >::empty(block_root); + cache.merge_blobs(random_blobs); + cache.merge_blobs(blobs); + cache.merge_block(block_commitments); + + assert_empty_blob_cache(cache); + } + + #[test] + fn block_valid_blobs_invalid_blobs() { + let (block_commitments, blobs, random_blobs) = pre_setup(); + let (block_commitments, blobs, random_blobs) = + setup_pending_components(block_commitments, blobs, random_blobs); + + let block_root = Hash256::zero(); + let mut cache = >::empty(block_root); + cache.merge_block(block_commitments); + cache.merge_blobs(blobs); + cache.merge_blobs(random_blobs); + + assert_cache_consistent(cache); + } + + #[test] + fn valid_blobs_block_invalid_blobs() { + let (block_commitments, blobs, random_blobs) = pre_setup(); + let (block_commitments, blobs, random_blobs) = + setup_pending_components(block_commitments, blobs, random_blobs); + + let block_root = Hash256::zero(); + let mut cache = >::empty(block_root); + cache.merge_blobs(blobs); + cache.merge_block(block_commitments); + cache.merge_blobs(random_blobs); + + assert_cache_consistent(cache); + } + + #[test] + fn valid_blobs_invalid_blobs_block() { + let (block_commitments, blobs, random_blobs) = pre_setup(); + let (block_commitments, blobs, random_blobs) = + setup_pending_components(block_commitments, blobs, random_blobs); + + let block_root = Hash256::zero(); + let mut cache = >::empty(block_root); + cache.merge_blobs(blobs); + cache.merge_blobs(random_blobs); + cache.merge_block(block_commitments); + + assert_cache_consistent(cache); + } +} diff --git a/beacon_node/beacon_chain/src/data_availability_checker/state_lru_cache.rs b/beacon_node/beacon_chain/src/data_availability_checker/state_lru_cache.rs index 35c114db54..c3492b53bd 100644 --- a/beacon_node/beacon_chain/src/data_availability_checker/state_lru_cache.rs +++ b/beacon_node/beacon_chain/src/data_availability_checker/state_lru_cache.rs @@ -10,6 +10,7 @@ use parking_lot::RwLock; use ssz_derive::{Decode, Encode}; use state_processing::{BlockReplayer, ConsensusContext, StateProcessingStrategy}; use std::sync::Arc; +use types::beacon_block_body::KzgCommitments; use types::{ssz_tagged_signed_beacon_block, ssz_tagged_signed_beacon_block_arc}; use types::{BeaconState, BlindedPayload, ChainSpec, Epoch, EthSpec, Hash256, SignedBeaconBlock}; @@ -42,6 +43,15 @@ impl DietAvailabilityPendingExecutedBlock { .blob_kzg_commitments() .map_or(0, |commitments| commitments.len()) } + + pub fn get_commitments(&self) -> KzgCommitments { + self.as_block() + .message() + .body() + .blob_kzg_commitments() + .cloned() + .unwrap_or_default() + } } /// This LRU cache holds BeaconStates used for block import. If the cache overflows, From 44b182362ae2ee19e0616e2b7eba417a091e38ac Mon Sep 17 00:00:00 2001 From: realbigsean Date: Wed, 10 Apr 2024 15:24:22 -0400 Subject: [PATCH 4/4] fix lints --- beacon_node/beacon_chain/src/beacon_chain.rs | 2 +- .../src/data_availability_checker/overflow_lru_cache.rs | 3 +-- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/beacon_node/beacon_chain/src/beacon_chain.rs b/beacon_node/beacon_chain/src/beacon_chain.rs index 9811ac7cf8..7c497e7458 100644 --- a/beacon_node/beacon_chain/src/beacon_chain.rs +++ b/beacon_node/beacon_chain/src/beacon_chain.rs @@ -121,7 +121,7 @@ use store::{ use task_executor::{ShutdownReason, TaskExecutor}; use tokio_stream::Stream; use tree_hash::TreeHash; -use types::blob_sidecar::{BlobSidecarList, FixedBlobSidecarList}; +use types::blob_sidecar::FixedBlobSidecarList; use types::payload::BlockProductionVersion; use types::*; 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 cc3871d604..edd981e6dd 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 @@ -1830,7 +1830,6 @@ mod pending_components_tests { use crate::block_verification_types::BlockImportData; use crate::eth1_finalization_cache::Eth1FinalizationData; use crate::test_utils::{generate_rand_block_and_blobs, NumBlobs}; - use crate::AvailabilityPendingExecutedBlock; use crate::PayloadVerificationOutcome; use fork_choice::PayloadVerificationStatus; use kzg::KzgCommitment; @@ -1838,7 +1837,7 @@ mod pending_components_tests { use rand::SeedableRng; use state_processing::ConsensusContext; use types::test_utils::TestRandom; - use types::{BeaconState, ChainSpec, ForkName, MainnetEthSpec, SignedBeaconBlock, Slot}; + use types::{BeaconState, ForkName, MainnetEthSpec, SignedBeaconBlock, Slot}; type E = MainnetEthSpec;