Merge pull request #4091 from realbigsean/get-validator-block-and-blobs

Get validator block and blobs
This commit is contained in:
realbigsean 2023-03-15 12:22:03 -04:00 committed by GitHub
commit 2e075c0a80
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 262 additions and 53 deletions

View File

@ -73,6 +73,7 @@ use fork_choice::{
use futures::channel::mpsc::Sender;
use itertools::process_results;
use itertools::Itertools;
use kzg::Kzg;
use operation_pool::{AttestationRef, OperationPool, PersistedOperationPool, ReceivedPreCapella};
use parking_lot::{Mutex, RwLock};
use proto_array::{CountUnrealizedFull, DoNotReOrg, ProposerHeadError};
@ -107,6 +108,7 @@ use store::{
use task_executor::{ShutdownReason, TaskExecutor};
use tree_hash::TreeHash;
use types::beacon_state::CloneConfig;
use types::blobs_sidecar::KzgCommitments;
use types::consts::eip4844::MIN_EPOCHS_FOR_BLOBS_SIDECARS_REQUESTS;
use types::consts::merge::INTERVALS_PER_SLOT;
use types::*;
@ -4759,30 +4761,62 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
.kzg
.as_ref()
.ok_or(BlockProductionError::TrustedSetupNotInitialized)?;
let kzg_aggregated_proof =
kzg_utils::compute_aggregate_kzg_proof::<T::EthSpec>(kzg, &blobs)
.map_err(BlockProductionError::KzgError)?;
let beacon_block_root = block.canonical_root();
let expected_kzg_commitments = block.body().blob_kzg_commitments().map_err(|_| {
BlockProductionError::InvalidBlockVariant(
"EIP4844 block does not contain kzg commitments".to_string(),
)
})?;
let blobs_sidecar = BlobsSidecar {
beacon_block_slot: slot,
beacon_block_root,
blobs,
kzg_aggregated_proof,
};
kzg_utils::validate_blobs_sidecar(
let expected_kzg_commitments: &KzgCommitments<T::EthSpec> =
block.body().blob_kzg_commitments().map_err(|_| {
BlockProductionError::InvalidBlockVariant(
"EIP4844 block does not contain kzg commitments".to_string(),
)
})?;
if expected_kzg_commitments.len() != blobs.len() {
return Err(BlockProductionError::MissingKzgCommitment(format!(
"Missing KZG commitment for slot {}. Expected {}, got: {}",
slot,
blobs.len(),
expected_kzg_commitments.len()
)));
}
let kzg_proofs =
Self::compute_blob_kzg_proofs(kzg, &blobs, expected_kzg_commitments, slot)?;
kzg_utils::validate_blobs::<T::EthSpec>(
kzg,
slot,
beacon_block_root,
expected_kzg_commitments,
&blobs_sidecar,
&blobs,
&kzg_proofs,
)
.map_err(BlockProductionError::KzgError)?;
self.blob_cache.put(beacon_block_root, blobs_sidecar);
let blob_sidecars = VariableList::from(
blobs
.into_iter()
.enumerate()
.map(|(blob_index, blob)| {
let kzg_commitment = expected_kzg_commitments
.get(blob_index)
.expect("KZG commitment should exist for blob");
let kzg_proof = kzg_proofs
.get(blob_index)
.expect("KZG proof should exist for blob");
Ok(BlobSidecar {
block_root: beacon_block_root,
index: blob_index as u64,
slot,
block_parent_root: block.parent_root(),
proposer_index,
blob,
kzg_commitment: *kzg_commitment,
kzg_proof: *kzg_proof,
})
})
.collect::<Result<Vec<BlobSidecar<T::EthSpec>>, BlockProductionError>>()?,
);
self.blob_cache.put(beacon_block_root, blob_sidecars);
}
metrics::inc_counter(&metrics::BLOCK_PRODUCTION_SUCCESSES);
@ -4798,6 +4832,29 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
Ok((block, state))
}
fn compute_blob_kzg_proofs(
kzg: &Arc<Kzg>,
blobs: &Blobs<<T as BeaconChainTypes>::EthSpec>,
expected_kzg_commitments: &KzgCommitments<<T as BeaconChainTypes>::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.clone())
.map_err(BlockProductionError::KzgError)
})
.collect::<Result<Vec<KzgProof>, BlockProductionError>>()
}
/// This method must be called whenever an execution engine indicates that a payload is
/// invalid.
///

View File

@ -1,12 +1,12 @@
use lru::LruCache;
use parking_lot::Mutex;
use types::{BlobsSidecar, EthSpec, Hash256};
use types::{BlobSidecarList, EthSpec, Hash256};
pub const DEFAULT_BLOB_CACHE_SIZE: usize = 10;
/// A cache blobs by beacon block root.
pub struct BlobCache<T: EthSpec> {
blobs: Mutex<LruCache<BlobCacheId, BlobsSidecar<T>>>,
blobs: Mutex<LruCache<BlobCacheId, BlobSidecarList<T>>>,
}
#[derive(Hash, PartialEq, Eq)]
@ -21,11 +21,15 @@ impl<T: EthSpec> Default for BlobCache<T> {
}
impl<T: EthSpec> BlobCache<T> {
pub fn put(&self, beacon_block: Hash256, blobs: BlobsSidecar<T>) -> Option<BlobsSidecar<T>> {
pub fn put(
&self,
beacon_block: Hash256,
blobs: BlobSidecarList<T>,
) -> Option<BlobSidecarList<T>> {
self.blobs.lock().put(BlobCacheId(beacon_block), blobs)
}
pub fn pop(&self, root: &Hash256) -> Option<BlobsSidecar<T>> {
pub fn pop(&self, root: &Hash256) -> Option<BlobSidecarList<T>> {
self.blobs.lock().pop(&BlobCacheId(*root))
}
}

View File

@ -135,18 +135,19 @@ fn verify_data_availability<T: BeaconChainTypes>(
.as_ref()
.ok_or(BlobError::TrustedSetupNotInitialized)?;
if !kzg_utils::validate_blobs_sidecar(
kzg,
block_slot,
block_root,
kzg_commitments,
blob_sidecar,
)
.map_err(BlobError::KzgError)?
{
return Err(BlobError::InvalidKzgProof);
}
Ok(())
todo!("use `kzg_utils::validate_blobs` once the function is updated")
// if !kzg_utils::validate_blobs_sidecar(
// kzg,
// block_slot,
// block_root,
// kzg_commitments,
// blob_sidecar,
// )
// .map_err(BlobError::KzgError)?
// {
// return Err(BlobError::InvalidKzgProof);
// }
// Ok(())
}
/// A wrapper over a [`SignedBeaconBlock`] or a [`SignedBeaconBlockAndBlobsSidecar`]. This makes no

View File

@ -266,12 +266,14 @@ pub enum BlockProductionError {
blob_block_hash: ExecutionBlockHash,
payload_block_hash: ExecutionBlockHash,
},
NoBlobsCached,
FailedToReadFinalizedBlock(store::Error),
MissingFinalizedBlock(Hash256),
BlockTooLarge(usize),
ShuttingDown,
MissingSyncAggregate,
MissingExecutionPayload,
MissingKzgCommitment(String),
TokioJoin(tokio::task::JoinError),
BeaconChain(BeaconChainError),
InvalidPayloadFork,

View File

@ -42,10 +42,11 @@ pub fn validate_blobs<T: EthSpec>(
/// Compute the kzg proof given an ssz blob and its kzg commitment.
pub fn compute_blob_kzg_proof<T: EthSpec>(
kzg: &Kzg,
blob: Blob<T>,
blob: &Blob<T>,
kzg_commitment: KzgCommitment,
) -> Result<KzgProof, KzgError> {
kzg.compute_blob_kzg_proof(ssz_blob_to_crypto_blob::<T>(blob), kzg_commitment)
// Avoid this blob clone
kzg.compute_blob_kzg_proof(ssz_blob_to_crypto_blob::<T>(blob.clone()), kzg_commitment)
}
/// Compute the kzg commitment for a given blob.

View File

@ -0,0 +1,33 @@
use beacon_chain::{BeaconChain, BeaconChainTypes, BlockProductionError};
use eth2::types::BlockContents;
use std::sync::Arc;
use types::{AbstractExecPayload, BeaconBlock, BeaconBlockAndBlobSidecars, ForkName};
type Error = warp::reject::Rejection;
pub fn build_block_contents<T: BeaconChainTypes, Payload: AbstractExecPayload<T::EthSpec>>(
fork_name: ForkName,
chain: Arc<BeaconChain<T>>,
block: BeaconBlock<T::EthSpec, Payload>,
) -> Result<BlockContents<T::EthSpec, Payload>, Error> {
match fork_name {
ForkName::Base | ForkName::Altair | ForkName::Merge | ForkName::Capella => {
Ok(BlockContents::Block(block))
}
ForkName::Eip4844 => {
let block_root = &block.canonical_root();
if let Some(blob_sidecars) = chain.blob_cache.pop(block_root) {
let block_and_blobs = BeaconBlockAndBlobSidecars {
block,
blob_sidecars,
};
Ok(BlockContents::BlockAndBlobSidecars(block_and_blobs))
} else {
return Err(warp_utils::reject::block_production_error(
BlockProductionError::NoBlobsCached,
));
}
}
}
}

View File

@ -11,6 +11,7 @@ mod attester_duties;
mod block_id;
mod block_packing_efficiency;
mod block_rewards;
mod build_block_contents;
mod database;
mod metrics;
mod proposer_duties;
@ -2421,7 +2422,10 @@ pub fn serve<T: BeaconChainTypes>(
.fork_name(&chain.spec)
.map_err(inconsistent_fork_rejection)?;
fork_versioned_response(endpoint_version, fork_name, block)
let block_contents =
build_block_contents::build_block_contents(fork_name, chain, block);
fork_versioned_response(endpoint_version, fork_name, block_contents?)
.map(|response| warp::reply::json(&response))
},
);

View File

@ -12,7 +12,7 @@ use tokio::sync::mpsc::UnboundedSender;
use tree_hash::TreeHash;
use types::{
AbstractExecPayload, BlindedPayload, EthSpec, ExecPayload, ExecutionBlockHash, FullPayload,
Hash256, SignedBeaconBlock, SignedBeaconBlockAndBlobsSidecar,
Hash256, SignedBeaconBlock,
};
use warp::Rejection;
@ -40,10 +40,11 @@ pub async fn publish_block<T: BeaconChainTypes>(
let wrapped_block: BlockWrapper<T::EthSpec> =
if matches!(block.as_ref(), &SignedBeaconBlock::Eip4844(_)) {
if let Some(sidecar) = chain.blob_cache.pop(&block_root) {
let block_and_blobs = SignedBeaconBlockAndBlobsSidecar {
beacon_block: block,
blobs_sidecar: Arc::new(sidecar),
};
// TODO: Needs to be adjusted
// let block_and_blobs = SignedBeaconBlockAndBlobsSidecar {
// beacon_block: block,
// blobs_sidecar: Arc::new(sidecar),
// };
unimplemented!("Needs to be adjusted")
} else {
//FIXME(sean): This should probably return a specific no-blob-cached error code, beacon API coordination required

View File

@ -314,7 +314,6 @@ mod tests {
VoluntaryExit,
ProposerSlashing,
AttesterSlashing,
BeaconBlocksAndBlobsSidecar,
]
.iter()
{

View File

@ -259,7 +259,7 @@ impl<E: EthSpec> HotColdDB<E, LevelDB<E>, LevelDB<E>> {
db.blobs_db = Some(LevelDB::open(path.as_path())?);
}
}
let blob_info = blob_info.unwrap_or(db.get_blob_info());
let blob_info = blob_info.unwrap_or_else(|| db.get_blob_info());
db.compare_and_set_blob_info_with_write(blob_info, new_blob_info)?;
info!(
db.log,
@ -1899,7 +1899,7 @@ impl<E: EthSpec, Hot: ItemStore<E>, Cold: ItemStore<E>> HotColdDB<E, Hot, Cold>
let blob_info = self.get_blob_info();
let oldest_blob_slot = blob_info
.oldest_blob_slot
.unwrap_or(eip4844_fork.start_slot(E::slots_per_epoch()));
.unwrap_or_else(|| eip4844_fork.start_slot(E::slots_per_epoch()));
// The last entirely pruned epoch, blobs sidecar pruning may have stopped early in the
// middle of an epoch otherwise the oldest blob slot is a start slot.

View File

@ -1388,7 +1388,7 @@ impl BeaconNodeHttpClient {
slot: Slot,
randao_reveal: &SignatureBytes,
graffiti: Option<&Graffiti>,
) -> Result<ForkVersionedResponse<BeaconBlock<T, Payload>>, Error> {
) -> Result<ForkVersionedResponse<BlockContents<T, Payload>>, Error> {
self.get_validator_blocks_modular(slot, randao_reveal, graffiti, SkipRandaoVerification::No)
.await
}
@ -1400,7 +1400,7 @@ impl BeaconNodeHttpClient {
randao_reveal: &SignatureBytes,
graffiti: Option<&Graffiti>,
skip_randao_verification: SkipRandaoVerification,
) -> Result<ForkVersionedResponse<BeaconBlock<T, Payload>>, Error> {
) -> Result<ForkVersionedResponse<BlockContents<T, Payload>>, Error> {
let mut path = self.eth_path(V2)?;
path.path_segments_mut()

View File

@ -1259,3 +1259,63 @@ mod tests {
)
}
}
/// A wrapper over a [`BeaconBlock`] or a [`BeaconBlockAndBlobSidecars`].
#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(untagged)]
#[serde(bound = "T: EthSpec")]
pub enum BlockContents<T: EthSpec, Payload: AbstractExecPayload<T>> {
BlockAndBlobSidecars(BeaconBlockAndBlobSidecars<T, Payload>),
Block(BeaconBlock<T, Payload>),
}
impl<T: EthSpec, Payload: AbstractExecPayload<T>> BlockContents<T, Payload> {
pub fn block(&self) -> &BeaconBlock<T, Payload> {
match self {
BlockContents::BlockAndBlobSidecars(block_and_sidecars) => &block_and_sidecars.block,
BlockContents::Block(block) => block,
}
}
pub fn deconstruct(self) -> (BeaconBlock<T, Payload>, Option<BlobSidecarList<T>>) {
match self {
BlockContents::BlockAndBlobSidecars(block_and_sidecars) => (
block_and_sidecars.block,
Some(block_and_sidecars.blob_sidecars),
),
BlockContents::Block(block) => (block, None),
}
}
}
impl<T: EthSpec, Payload: AbstractExecPayload<T>> ForkVersionDeserialize
for BlockContents<T, Payload>
{
fn deserialize_by_fork<'de, D: serde::Deserializer<'de>>(
value: serde_json::value::Value,
fork_name: ForkName,
) -> Result<Self, D::Error> {
match fork_name {
ForkName::Base | ForkName::Altair | ForkName::Merge | ForkName::Capella => {
Ok(BlockContents::Block(BeaconBlock::deserialize_by_fork::<
'de,
D,
>(value, fork_name)?))
}
ForkName::Eip4844 => Ok(BlockContents::BlockAndBlobSidecars(
BeaconBlockAndBlobSidecars::deserialize_by_fork::<'de, D>(value, fork_name)?,
)),
}
}
}
impl<T: EthSpec, Payload: AbstractExecPayload<T>> Into<BeaconBlock<T, Payload>>
for BlockContents<T, Payload>
{
fn into(self) -> BeaconBlock<T, Payload> {
match self {
Self::BlockAndBlobSidecars(block_and_sidecars) => block_and_sidecars.block,
Self::Block(block) => block,
}
}
}

View File

@ -0,0 +1,37 @@
use crate::{
AbstractExecPayload, BeaconBlock, BlobSidecarList, EthSpec, ForkName, ForkVersionDeserialize,
};
use derivative::Derivative;
use serde_derive::{Deserialize, Serialize};
use ssz_derive::Encode;
use tree_hash_derive::TreeHash;
#[derive(Debug, Clone, Serialize, Deserialize, Encode, TreeHash, Derivative)]
#[derivative(PartialEq, Hash(bound = "T: EthSpec"))]
#[serde(bound = "T: EthSpec, Payload: AbstractExecPayload<T>")]
pub struct BeaconBlockAndBlobSidecars<T: EthSpec, Payload: AbstractExecPayload<T>> {
pub block: BeaconBlock<T, Payload>,
pub blob_sidecars: BlobSidecarList<T>,
}
impl<T: EthSpec, Payload: AbstractExecPayload<T>> ForkVersionDeserialize
for BeaconBlockAndBlobSidecars<T, Payload>
{
fn deserialize_by_fork<'de, D: serde::Deserializer<'de>>(
value: serde_json::value::Value,
fork_name: ForkName,
) -> Result<Self, D::Error> {
#[derive(Deserialize)]
#[serde(bound = "T: EthSpec")]
struct Helper<T: EthSpec> {
block: serde_json::Value,
blob_sidecars: BlobSidecarList<T>,
}
let helper: Helper<T> = serde_json::from_value(value).map_err(serde::de::Error::custom)?;
Ok(Self {
block: BeaconBlock::deserialize_by_fork::<'de, D>(helper.block, fork_name)?,
blob_sidecars: helper.blob_sidecars,
})
}
}

View File

@ -5,6 +5,7 @@ use kzg::{KzgCommitment, KzgProof};
use serde_derive::{Deserialize, Serialize};
use ssz::Encode;
use ssz_derive::{Decode, Encode};
use ssz_types::VariableList;
use test_random_derive::TestRandom;
use tree_hash_derive::TreeHash;
@ -34,15 +35,20 @@ pub struct BlobIdentifier {
pub struct BlobSidecar<T: EthSpec> {
pub block_root: Hash256,
// TODO: fix the type, should fit in u8 as well
#[serde(with = "eth2_serde_utils::quoted_u64")]
pub index: u64,
pub slot: Slot,
pub block_parent_root: Hash256,
#[serde(with = "eth2_serde_utils::quoted_u64")]
pub proposer_index: u64,
#[serde(with = "ssz_types::serde_utils::hex_fixed_vec")]
pub blob: Blob<T>,
pub kzg_commitment: KzgCommitment,
pub kzg_proof: KzgProof,
}
pub type BlobSidecarList<T> = VariableList<BlobSidecar<T>, <T as EthSpec>::MaxBlobsPerBlock>;
impl<T: EthSpec> SignedRoot for BlobSidecar<T> {}
impl<T: EthSpec> BlobSidecar<T> {

View File

@ -99,6 +99,7 @@ pub mod slot_data;
#[cfg(feature = "sqlite")]
pub mod sqlite;
pub mod beacon_block_and_blob_sidecars;
pub mod blob_sidecar;
pub mod blobs_sidecar;
pub mod signed_blob;
@ -116,6 +117,7 @@ pub use crate::beacon_block::{
BeaconBlock, BeaconBlockAltair, BeaconBlockBase, BeaconBlockCapella, BeaconBlockEip4844,
BeaconBlockMerge, BeaconBlockRef, BeaconBlockRefMut, BlindedBeaconBlock, EmptyBlock,
};
pub use crate::beacon_block_and_blob_sidecars::BeaconBlockAndBlobSidecars;
pub use crate::beacon_block_body::{
BeaconBlockBody, BeaconBlockBodyAltair, BeaconBlockBodyBase, BeaconBlockBodyCapella,
BeaconBlockBodyEip4844, BeaconBlockBodyMerge, BeaconBlockBodyRef, BeaconBlockBodyRefMut,
@ -123,7 +125,7 @@ pub use crate::beacon_block_body::{
pub use crate::beacon_block_header::BeaconBlockHeader;
pub use crate::beacon_committee::{BeaconCommittee, OwnedBeaconCommittee};
pub use crate::beacon_state::{BeaconTreeHashCache, Error as BeaconStateError, *};
pub use crate::blob_sidecar::BlobSidecar;
pub use crate::blob_sidecar::{BlobSidecar, BlobSidecarList};
pub use crate::blobs_sidecar::{Blobs, BlobsSidecar};
pub use crate::bls_to_execution_change::BlsToExecutionChange;
pub use crate::chain_spec::{ChainSpec, Config, Domain};

View File

@ -8,7 +8,7 @@ use std::fmt::{Debug, Display, Formatter};
use std::str::FromStr;
use tree_hash::{PackedEncoding, TreeHash};
#[derive(Derivative, Clone, Encode, Decode)]
#[derive(Derivative, Clone, Copy, Encode, Decode)]
#[derivative(PartialEq, Eq, Hash)]
#[ssz(struct_behaviour = "transparent")]
pub struct KzgCommitment(pub [u8; BYTES_PER_COMMITMENT]);

View File

@ -15,8 +15,8 @@ use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::sleep;
use types::{
AbstractExecPayload, BlindedPayload, BlockType, EthSpec, FullPayload, Graffiti, PublicKeyBytes,
Slot,
AbstractExecPayload, BeaconBlock, BlindedPayload, BlockType, EthSpec, FullPayload, Graffiti,
PublicKeyBytes, Slot,
};
#[derive(Debug)]
@ -347,7 +347,7 @@ impl<T: SlotClock + 'static, E: EthSpec> BlockService<T, E> {
RequireSynced::No,
OfflineOnFailure::Yes,
|beacon_node| async move {
let block = match Payload::block_type() {
let block: BeaconBlock<E, Payload> = match Payload::block_type() {
BlockType::Full => {
let _get_timer = metrics::start_timer_vec(
&metrics::BLOCK_SERVICE_TIMES,
@ -367,6 +367,7 @@ impl<T: SlotClock + 'static, E: EthSpec> BlockService<T, E> {
))
})?
.data
.into()
}
BlockType::Blinded => {
let _get_timer = metrics::start_timer_vec(
@ -387,6 +388,7 @@ impl<T: SlotClock + 'static, E: EthSpec> BlockService<T, E> {
))
})?
.data
.into()
}
};