gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-03-11 18:04:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

Implement Overflow LRU Cache for Pending Blobs (#4203)

Browse Source 
* All Necessary Objects Implement Encode/Decode

* Major Components for LRUOverflowCache Implemented

* Finish Database Code

* Add Maintenance Methods

* Added Maintenance Service

* Persist Blobs on Shutdown / Reload on Startup

* Address Clippy Complaints

* Add (emum_behaviour = "tag") to ssz_derive

* Convert Encode/Decode Implementations to "tag"

* Started Adding Tests

* Added a ton of tests

* 1 character fix

* Feature Guard Minimal Spec Tests

* Update beacon_node/beacon_chain/src/data_availability_checker.rs

Co-authored-by: realbigsean <seananderson33@GMAIL.com>

* Address Sean's Comments

* Add iter_raw_keys method

* Remove TODOs

---------

Co-authored-by: realbigsean <seananderson33@GMAIL.com>
This commit is contained in:
ethDreamer
2023-05-12 09:08:24 -05:00
committed by GitHub
parent a22e4bf636
commit 46db30416d
23 changed files with 2364 additions and 253 deletions
Download Patch File Download Diff File Expand all files Collapse all files

450
beacon_node/beacon_chain/src/data_availability_checker.rs
View File

@@ -4,23 +4,29 @@ use crate::blob_verification::{
};
use crate::block_verification::{AvailabilityPendingExecutedBlock, AvailableExecutedBlock};
use crate::data_availability_checker::overflow_lru_cache::OverflowLRUCache;
use crate::{BeaconChain, BeaconChainTypes, BeaconStore};
use kzg::Error as KzgError;
use kzg::Kzg;
use parking_lot::RwLock;
use slog::{debug, error};
use slot_clock::SlotClock;
use ssz_types::{Error, FixedVector, VariableList};
use ssz_types::{Error, VariableList};
use state_processing::per_block_processing::deneb::deneb::verify_kzg_commitments_against_transactions;
use std::collections::hash_map::{Entry, OccupiedEntry};
use std::collections::HashMap;
use std::sync::Arc;
use task_executor::TaskExecutor;
use types::beacon_block_body::KzgCommitments;
use types::blob_sidecar::{BlobIdentifier, BlobSidecar};
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,
SignedBeaconBlock, SignedBeaconBlockHeader, Slot,
};
mod overflow_lru_cache;
pub const OVERFLOW_LRU_CAPACITY: usize = 1024;
#[derive(Debug)]
pub enum AvailabilityCheckError {
DuplicateBlob(Hash256),
@@ -39,6 +45,9 @@ pub enum AvailabilityCheckError {
},
Pending,
IncorrectFork,
BlobIndexInvalid(u64),
StoreError(store::Error),
DecodeError(ssz::DecodeError),
}
impl From<ssz_types::Error> for AvailabilityCheckError {
@@ -47,70 +56,35 @@ impl From<ssz_types::Error> for AvailabilityCheckError {
}
}
impl From<store::Error> for AvailabilityCheckError {
fn from(value: store::Error) -> Self {
Self::StoreError(value)
}
}
impl From<ssz::DecodeError> for AvailabilityCheckError {
fn from(value: ssz::DecodeError) -> Self {
Self::DecodeError(value)
}
}
/// This cache contains
/// - blobs that have been gossip verified
/// - commitments for blocks that have been gossip verified, but the commitments themselves
/// have not been verified against blobs
/// - blocks that have been fully verified and only require a data availability check
pub struct DataAvailabilityChecker<T: EthSpec, S: SlotClock> {
availability_cache: RwLock<HashMap<Hash256, ReceivedComponents<T>>>,
slot_clock: S,
pub struct DataAvailabilityChecker<T: BeaconChainTypes> {
availability_cache: Arc<OverflowLRUCache<T>>,
slot_clock: T::SlotClock,
kzg: Option<Arc<Kzg>>,
spec: ChainSpec,
}
/// Caches partially available blobs and execution verified blocks corresponding
/// to a given `block_hash` that are received over gossip.
///
/// The blobs are all gossip and kzg verified.
/// The block has completed all verifications except the availability check.
struct ReceivedComponents<T: EthSpec> {
verified_blobs: FixedVector<Option<KzgVerifiedBlob<T>>, T::MaxBlobsPerBlock>,
executed_block: Option<AvailabilityPendingExecutedBlock<T>>,
}
impl<T: EthSpec> ReceivedComponents<T> {
fn new_from_blob(blob: KzgVerifiedBlob<T>) -> Self {
let mut verified_blobs = FixedVector::<_, _>::default();
// TODO: verify that we've already ensured the blob index < T::MaxBlobsPerBlock
if let Some(mut_maybe_blob) = verified_blobs.get_mut(blob.blob_index() as usize) {
*mut_maybe_blob = Some(blob);
}
Self {
verified_blobs,
executed_block: None,
}
}
fn new_from_block(block: AvailabilityPendingExecutedBlock<T>) -> Self {
Self {
verified_blobs: <_>::default(),
executed_block: Some(block),
}
}
/// Returns `true` if the cache has all blobs corresponding to the
/// kzg commitments in the block.
fn has_all_blobs(&self, block: &AvailabilityPendingExecutedBlock<T>) -> bool {
for i in 0..block.num_blobs_expected() {
if self
.verified_blobs
.get(i)
.map(|maybe_blob| maybe_blob.is_none())
.unwrap_or(true)
{
return false;
}
}
true
}
}
/// This type is returned after adding a block / blob to the `DataAvailabilityChecker`.
///
/// Indicates if the block is fully `Available` or if we need blobs or blocks
/// to "complete" the requirements for an `AvailableBlock`.
#[derive(Debug, PartialEq)]
pub enum Availability<T: EthSpec> {
PendingBlobs(Vec<BlobIdentifier>),
PendingBlock(Hash256),
@@ -129,25 +103,28 @@ impl<T: EthSpec> Availability<T> {
}
}
impl<T: EthSpec, S: SlotClock> DataAvailabilityChecker<T, S> {
pub fn new(slot_clock: S, kzg: Option<Arc<Kzg>>, spec: ChainSpec) -> Self {
Self {
availability_cache: <_>::default(),
impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
pub fn new(
slot_clock: T::SlotClock,
kzg: Option<Arc<Kzg>>,
store: BeaconStore<T>,
spec: ChainSpec,
) -> Result<Self, AvailabilityCheckError> {
let overflow_cache = OverflowLRUCache::new(OVERFLOW_LRU_CAPACITY, store)?;
Ok(Self {
availability_cache: Arc::new(overflow_cache),
slot_clock,
kzg,
spec,
}
})
}
/// Get a blob from the availability cache.
pub fn get_blob(&self, blob_id: &BlobIdentifier) -> Option<Arc<BlobSidecar<T>>> {
self.availability_cache
.read()
.get(&blob_id.block_root)?
.verified_blobs
.get(blob_id.index as usize)?
.as_ref()
.map(|kzg_verified_blob| kzg_verified_blob.clone_blob())
pub fn get_blob(
&self,
blob_id: &BlobIdentifier,
) -> Result<Option<Arc<BlobSidecar<T::EthSpec>>>, AvailabilityCheckError> {
self.availability_cache.peek_blob(blob_id)
}
/// This first validates the KZG commitments included in the blob sidecar.
@@ -158,10 +135,8 @@ impl<T: EthSpec, S: SlotClock> DataAvailabilityChecker<T, S> {
/// This should only accept gossip verified blobs, so we should not have to worry about dupes.
pub fn put_gossip_blob(
&self,
gossip_blob: GossipVerifiedBlob<T>,
) -> Result<Availability<T>, AvailabilityCheckError> {
let block_root = gossip_blob.block_root();
gossip_blob: GossipVerifiedBlob<T::EthSpec>,
) -> Result<Availability<T::EthSpec>, AvailabilityCheckError> {
// Verify the KZG commitments.
let kzg_verified_blob = if let Some(kzg) = self.kzg.as_ref() {
verify_kzg_for_blob(gossip_blob, kzg)?
@@ -169,125 +144,26 @@ impl<T: EthSpec, S: SlotClock> DataAvailabilityChecker<T, S> {
return Err(AvailabilityCheckError::KzgNotInitialized);
};
let availability = match self
.availability_cache
.write()
.entry(kzg_verified_blob.block_root())
{
Entry::Occupied(mut occupied_entry) => {
// All blobs reaching this cache should be gossip verified and gossip verification
// should filter duplicates, as well as validate indices.
let received_components = occupied_entry.get_mut();
if let Some(maybe_verified_blob) = received_components
.verified_blobs
.get_mut(kzg_verified_blob.blob_index() as usize)
{
*maybe_verified_blob = Some(kzg_verified_blob)
}
if let Some(executed_block) = received_components.executed_block.take() {
self.check_block_availability_maybe_cache(occupied_entry, executed_block)?
} else {
Availability::PendingBlock(block_root)
}
}
Entry::Vacant(vacant_entry) => {
let block_root = kzg_verified_blob.block_root();
vacant_entry.insert(ReceivedComponents::new_from_blob(kzg_verified_blob));
Availability::PendingBlock(block_root)
}
};
Ok(availability)
self.availability_cache
.put_kzg_verified_blob(kzg_verified_blob)
}
/// Check if we have all the blobs for a block. If we do, return the Availability variant that
/// triggers import of the block.
pub fn put_pending_executed_block(
&self,
executed_block: AvailabilityPendingExecutedBlock<T>,
) -> Result<Availability<T>, AvailabilityCheckError> {
let availability = match self
.availability_cache
.write()
.entry(executed_block.import_data.block_root)
{
Entry::Occupied(occupied_entry) => {
self.check_block_availability_maybe_cache(occupied_entry, executed_block)?
}
Entry::Vacant(vacant_entry) => {
let all_blob_ids = executed_block.get_all_blob_ids();
vacant_entry.insert(ReceivedComponents::new_from_block(executed_block));
Availability::PendingBlobs(all_blob_ids)
}
};
Ok(availability)
}
/// Checks if the provided `executed_block` contains all required blobs to be considered an
/// `AvailableBlock` based on blobs that are cached.
///
/// Returns an error if there was an error when matching the block commitments against blob commitments.
///
/// Returns `Ok(Availability::Available(_))` if all blobs for the block are present in cache.
/// Returns `Ok(Availability::PendingBlobs(_))` if all corresponding blobs have not been received in the cache.
fn check_block_availability_maybe_cache(
&self,
mut occupied_entry: OccupiedEntry<Hash256, ReceivedComponents<T>>,
executed_block: AvailabilityPendingExecutedBlock<T>,
) -> Result<Availability<T>, AvailabilityCheckError> {
if occupied_entry.get().has_all_blobs(&executed_block) {
let num_blobs_expected = executed_block.num_blobs_expected();
let AvailabilityPendingExecutedBlock {
block,
import_data,
payload_verification_outcome,
} = executed_block;
let ReceivedComponents {
verified_blobs,
executed_block: _,
} = occupied_entry.remove();
let verified_blobs = Vec::from(verified_blobs)
.into_iter()
.take(num_blobs_expected)
.map(|maybe_blob| maybe_blob.ok_or(AvailabilityCheckError::MissingBlobs))
.collect::<Result<Vec<_>, _>>()?;
let available_block = self.make_available(block, verified_blobs)?;
Ok(Availability::Available(Box::new(
AvailableExecutedBlock::new(
available_block,
import_data,
payload_verification_outcome,
),
)))
} else {
let received_components = occupied_entry.get_mut();
let missing_blob_ids = executed_block.get_filtered_blob_ids(|index| {
received_components
.verified_blobs
.get(index as usize)
.map(|maybe_blob| maybe_blob.is_none())
.unwrap_or(true)
});
let _ = received_components.executed_block.insert(executed_block);
Ok(Availability::PendingBlobs(missing_blob_ids))
}
executed_block: AvailabilityPendingExecutedBlock<T::EthSpec>,
) -> Result<Availability<T::EthSpec>, AvailabilityCheckError> {
self.availability_cache
.put_pending_executed_block(executed_block)
}
/// Checks if a block is available, returns a `MaybeAvailableBlock` that may include the fully
/// available block.
pub fn check_availability(
&self,
block: BlockWrapper<T>,
) -> Result<MaybeAvailableBlock<T>, AvailabilityCheckError> {
block: BlockWrapper<T::EthSpec>,
) -> Result<MaybeAvailableBlock<T::EthSpec>, AvailabilityCheckError> {
match block {
BlockWrapper::Block(block) => self.check_availability_without_blobs(block),
BlockWrapper::BlockAndBlobs(block, blob_list) => {
@@ -308,8 +184,8 @@ impl<T: EthSpec, S: SlotClock> DataAvailabilityChecker<T, S> {
/// Does not access the gossip cache.
pub fn try_check_availability(
&self,
block: BlockWrapper<T>,
) -> Result<AvailableBlock<T>, AvailabilityCheckError> {
block: BlockWrapper<T::EthSpec>,
) -> Result<AvailableBlock<T::EthSpec>, AvailabilityCheckError> {
match block {
BlockWrapper::Block(block) => {
let blob_requirements = self.get_blob_requirements(&block)?;
@@ -329,13 +205,13 @@ impl<T: EthSpec, S: SlotClock> DataAvailabilityChecker<T, S> {
/// commitments are consistent with the provided verified blob commitments.
pub fn check_availability_with_blobs(
&self,
block: Arc<SignedBeaconBlock<T>>,
blobs: KzgVerifiedBlobList<T>,
) -> Result<AvailableBlock<T>, AvailabilityCheckError> {
block: Arc<SignedBeaconBlock<T::EthSpec>>,
blobs: KzgVerifiedBlobList<T::EthSpec>,
) -> Result<AvailableBlock<T::EthSpec>, AvailabilityCheckError> {
match self.check_availability_without_blobs(block)? {
MaybeAvailableBlock::Available(block) => Ok(block),
MaybeAvailableBlock::AvailabilityPending(pending_block) => {
self.make_available(pending_block, blobs)
pending_block.make_available(blobs)
}
}
}
@@ -344,8 +220,8 @@ impl<T: EthSpec, S: SlotClock> DataAvailabilityChecker<T, S> {
/// an AvailableBlock if no blobs are required. Otherwise this will return an AvailabilityPendingBlock.
pub fn check_availability_without_blobs(
&self,
block: Arc<SignedBeaconBlock<T>>,
) -> Result<MaybeAvailableBlock<T>, AvailabilityCheckError> {
block: Arc<SignedBeaconBlock<T::EthSpec>>,
) -> Result<MaybeAvailableBlock<T::EthSpec>, AvailabilityCheckError> {
let blob_requirements = self.get_blob_requirements(&block)?;
let blobs = match blob_requirements {
BlobRequirements::EmptyBlobs => VerifiedBlobs::EmptyBlobs,
@@ -363,50 +239,18 @@ impl<T: EthSpec, S: SlotClock> DataAvailabilityChecker<T, S> {
}))
}
/// Verifies an AvailabilityPendingBlock against a set of KZG verified blobs.
/// This does not check whether a block *should* have blobs, these checks should must have been
/// completed when producing the `AvailabilityPendingBlock`.
pub fn make_available(
&self,
block: AvailabilityPendingBlock<T>,
blobs: Vec<KzgVerifiedBlob<T>>,
) -> Result<AvailableBlock<T>, AvailabilityCheckError> {
let block_kzg_commitments = block.kzg_commitments()?;
if blobs.len() != block_kzg_commitments.len() {
return Err(AvailabilityCheckError::NumBlobsMismatch {
num_kzg_commitments: block_kzg_commitments.len(),
num_blobs: blobs.len(),
});
}
for (block_commitment, blob) in block_kzg_commitments.iter().zip(blobs.iter()) {
if *block_commitment != blob.kzg_commitment() {
return Err(AvailabilityCheckError::KzgCommitmentMismatch {
blob_index: blob.as_blob().index,
});
}
}
let blobs = VariableList::new(blobs.into_iter().map(|blob| blob.to_blob()).collect())?;
Ok(AvailableBlock {
block: block.block,
blobs: VerifiedBlobs::Available(blobs),
})
}
/// Determines the blob requirements for a block. Answers the question: "Does this block require
/// blobs?".
fn get_blob_requirements(
&self,
block: &Arc<SignedBeaconBlock<T, FullPayload<T>>>,
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>(
let verified = verify_kzg_commitments_against_transactions::<T::EthSpec>(
transactions,
block_kzg_commitments,
)
@@ -437,7 +281,7 @@ impl<T: EthSpec, S: SlotClock> DataAvailabilityChecker<T, S> {
self.spec.deneb_fork_epoch.and_then(|fork_epoch| {
self.slot_clock
.now()
.map(|slot| slot.epoch(T::slots_per_epoch()))
.map(|slot| slot.epoch(T::EthSpec::slots_per_epoch()))
.map(|current_epoch| {
std::cmp::max(
fork_epoch,
@@ -452,6 +296,96 @@ impl<T: EthSpec, S: SlotClock> DataAvailabilityChecker<T, S> {
self.data_availability_boundary()
.map_or(false, |da_epoch| block_epoch >= da_epoch)
}
/// Persist all in memory components to disk
pub fn persist_all(&self) -> Result<(), AvailabilityCheckError> {
self.availability_cache.write_all_to_disk()
}
}
pub fn start_availability_cache_maintenance_service<T: BeaconChainTypes>(
executor: TaskExecutor,
chain: Arc<BeaconChain<T>>,
) {
// this cache only needs to be maintained if deneb is configured
if chain.spec.deneb_fork_epoch.is_some() {
let overflow_cache = chain.data_availability_checker.availability_cache.clone();
executor.spawn(
async move { availability_cache_maintenance_service(chain, overflow_cache).await },
"availability_cache_service",
);
} else {
debug!(
chain.log,
"Deneb fork not configured, not starting availability cache maintenance service"
);
}
}
async fn availability_cache_maintenance_service<T: BeaconChainTypes>(
chain: Arc<BeaconChain<T>>,
overflow_cache: Arc<OverflowLRUCache<T>>,
) {
let epoch_duration = chain.slot_clock.slot_duration() * T::EthSpec::slots_per_epoch() as u32;
loop {
match chain
.slot_clock
.duration_to_next_epoch(T::EthSpec::slots_per_epoch())
{
Some(duration) => {
// this service should run 3/4 of the way through the epoch
let additional_delay = (epoch_duration * 3) / 4;
tokio::time::sleep(duration + additional_delay).await;
let deneb_fork_epoch = match chain.spec.deneb_fork_epoch {
Some(epoch) => epoch,
None => break, // shutdown service if deneb fork epoch not set
};
debug!(
chain.log,
"Availability cache maintenance service firing";
);
let current_epoch = match chain
.slot_clock
.now()
.map(|slot| slot.epoch(T::EthSpec::slots_per_epoch()))
{
Some(epoch) => epoch,
None => continue, // we'll have to try again next time I suppose..
};
if current_epoch < deneb_fork_epoch {
// we are not in deneb yet
continue;
}
let finalized_epoch = chain
.canonical_head
.fork_choice_read_lock()
.finalized_checkpoint()
.epoch;
// any data belonging to an epoch before this should be pruned
let cutoff_epoch = std::cmp::max(
finalized_epoch + 1,
std::cmp::max(
current_epoch.saturating_sub(*MIN_EPOCHS_FOR_BLOB_SIDECARS_REQUESTS),
deneb_fork_epoch,
),
);
if let Err(e) = overflow_cache.do_maintenance(cutoff_epoch) {
error!(chain.log, "Failed to maintain availability cache"; "error" => ?e);
}
}
None => {
error!(chain.log, "Failed to read slot clock");
// If we can't read the slot clock, just wait another slot.
tokio::time::sleep(chain.slot_clock.slot_duration()).await;
}
};
}
}
pub enum BlobRequirements {
@@ -493,6 +427,37 @@ impl<E: EthSpec> AvailabilityPendingBlock<E> {
.blob_kzg_commitments()
.map_err(|_| AvailabilityCheckError::IncorrectFork)
}
/// Verifies an AvailabilityPendingBlock against a set of KZG verified blobs.
/// This does not check whether a block *should* have blobs, these checks should must have been
/// completed when producing the `AvailabilityPendingBlock`.
pub fn make_available(
self,
blobs: Vec<KzgVerifiedBlob<E>>,
) -> Result<AvailableBlock<E>, AvailabilityCheckError> {
let block_kzg_commitments = self.kzg_commitments()?;
if blobs.len() != block_kzg_commitments.len() {
return Err(AvailabilityCheckError::NumBlobsMismatch {
num_kzg_commitments: block_kzg_commitments.len(),
num_blobs: blobs.len(),
});
}
for (block_commitment, blob) in block_kzg_commitments.iter().zip(blobs.iter()) {
if *block_commitment != blob.kzg_commitment() {
return Err(AvailabilityCheckError::KzgCommitmentMismatch {
blob_index: blob.as_blob().index,
});
}
}
let blobs = VariableList::new(blobs.into_iter().map(|blob| blob.to_blob()).collect())?;
Ok(AvailableBlock {
block: self.block,
blobs: VerifiedBlobs::Available(blobs),
})
}
}
#[derive(Clone, Debug, PartialEq)]
@@ -576,3 +541,44 @@ impl<E: EthSpec> AsBlock<E> for AvailableBlock<E> {
}
}
}
// The standard implementation of Encode for SignedBeaconBlock
// requires us to use ssz(enum_behaviour = "transparent"). This
// prevents us from implementing Decode. We need to use a
// custom Encode and Decode in this wrapper object that essentially
// encodes it as if it were ssz(enum_behaviour = "union")
impl<E: EthSpec> ssz::Encode for AvailabilityPendingBlock<E> {
fn is_ssz_fixed_len() -> bool {
ssz_tagged_signed_beacon_block::encode::is_ssz_fixed_len()
}
fn ssz_append(&self, buf: &mut Vec<u8>) {
ssz_tagged_signed_beacon_block::encode::ssz_append(self.block.as_ref(), buf);
}
fn ssz_bytes_len(&self) -> usize {
ssz_tagged_signed_beacon_block::encode::ssz_bytes_len(self.block.as_ref())
}
}
impl<E: EthSpec> ssz::Decode for AvailabilityPendingBlock<E> {
fn is_ssz_fixed_len() -> bool {
ssz_tagged_signed_beacon_block::decode::is_ssz_fixed_len()
}
fn from_ssz_bytes(bytes: &[u8]) -> Result<Self, ssz::DecodeError> {
Ok(Self {
block: Arc::new(ssz_tagged_signed_beacon_block::decode::from_ssz_bytes(
bytes,
)?),
})
}
}
#[cfg(test)]
mod test {
#[test]
fn check_encode_decode_availability_pending_block() {
// todo.. (difficult to create default beacon blocks to test)
}
}