diff --git a/Cargo.lock b/Cargo.lock index af27fb11c84..cf7793fbb15 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1136,10 +1136,12 @@ name = "builder_client" version = "0.1.0" dependencies = [ "eth2", + "ethereum_ssz", "lighthouse_version", "reqwest", "sensitive_url", "serde", + "serde_json", ] [[package]] diff --git a/beacon_node/beacon_chain/src/data_availability_checker.rs b/beacon_node/beacon_chain/src/data_availability_checker.rs index aa4689121ce..f10d59ca1a5 100644 --- a/beacon_node/beacon_chain/src/data_availability_checker.rs +++ b/beacon_node/beacon_chain/src/data_availability_checker.rs @@ -27,8 +27,8 @@ mod overflow_lru_cache; mod state_lru_cache; use crate::data_column_verification::{ - verify_kzg_for_data_column, verify_kzg_for_data_column_list, CustodyDataColumn, - GossipVerifiedDataColumn, KzgVerifiedCustodyDataColumn, KzgVerifiedDataColumn, + verify_kzg_for_data_column_list_with_scoring, CustodyDataColumn, GossipVerifiedDataColumn, + KzgVerifiedCustodyDataColumn, KzgVerifiedDataColumn, }; use crate::metrics::{ KZG_DATA_COLUMN_RECONSTRUCTION_ATTEMPTS, KZG_DATA_COLUMN_RECONSTRUCTION_FAILURES, @@ -230,19 +230,14 @@ impl DataAvailabilityChecker { block_root: Hash256, custody_columns: DataColumnSidecarList, ) -> Result, AvailabilityCheckError> { - // TODO(das): report which column is invalid for proper peer scoring - // TODO(das): batch KZG verification here, but fallback into checking each column - // individually to report which column(s) are invalid. - let verified_custody_columns = custody_columns + // Attributes fault to the specific peer that sent an invalid column + let kzg_verified_columns = KzgVerifiedDataColumn::from_batch(custody_columns, &self.kzg) + .map_err(AvailabilityCheckError::InvalidColumn)?; + + let verified_custody_columns = kzg_verified_columns .into_iter() - .map(|column| { - let index = column.index; - Ok(KzgVerifiedCustodyDataColumn::from_asserted_custody( - KzgVerifiedDataColumn::new(column, &self.kzg) - .map_err(|e| AvailabilityCheckError::InvalidColumn(index, e))?, - )) - }) - .collect::, AvailabilityCheckError>>()?; + .map(KzgVerifiedCustodyDataColumn::from_asserted_custody) + .collect::>(); self.availability_cache.put_kzg_verified_data_columns( block_root, @@ -365,7 +360,8 @@ impl DataAvailabilityChecker { .iter() .map(|custody_column| custody_column.as_data_column()), &self.kzg, - )?; + ) + .map_err(AvailabilityCheckError::InvalidColumn)?; Ok(MaybeAvailableBlock::Available(AvailableBlock { block_root, block, @@ -432,8 +428,9 @@ impl DataAvailabilityChecker { // verify kzg for all data columns at once if !all_data_columns.is_empty() { - // TODO: Need to also attribute which specific block is faulty - verify_kzg_for_data_column_list_with_scoring(all_data_columns.iter(), &self.kzg)?; + // Attributes fault to the specific peer that sent an invalid column + verify_kzg_for_data_column_list_with_scoring(all_data_columns.iter(), &self.kzg) + .map_err(AvailabilityCheckError::InvalidColumn)?; } for block in blocks { @@ -716,32 +713,6 @@ async fn availability_cache_maintenance_service( } } -fn verify_kzg_for_data_column_list_with_scoring<'a, E: EthSpec, I>( - data_column_iter: I, - kzg: &'a Kzg, -) -> Result<(), AvailabilityCheckError> -where - I: Iterator>> + Clone, -{ - let Err(batch_err) = verify_kzg_for_data_column_list(data_column_iter.clone(), kzg) else { - return Ok(()); - }; - - let data_columns = data_column_iter.collect::>(); - // Find which column is invalid. If len is 1 or 0 continue to default case below. - // If len > 1 at least one column MUST fail. - if data_columns.len() > 1 { - for data_column in data_columns { - if let Err(e) = verify_kzg_for_data_column(data_column.clone(), kzg) { - return Err(AvailabilityCheckError::InvalidColumn(data_column.index, e)); - } - } - } - - // len 0 should never happen - Err(AvailabilityCheckError::InvalidColumn(0, batch_err)) -} - /// A fully available block that is ready to be imported into fork choice. #[derive(Clone, Debug, PartialEq)] pub struct AvailableBlock { diff --git a/beacon_node/beacon_chain/src/data_availability_checker/error.rs b/beacon_node/beacon_chain/src/data_availability_checker/error.rs index cfdb3cfe91e..1ab85ab1056 100644 --- a/beacon_node/beacon_chain/src/data_availability_checker/error.rs +++ b/beacon_node/beacon_chain/src/data_availability_checker/error.rs @@ -4,7 +4,7 @@ use types::{BeaconStateError, ColumnIndex, Hash256}; #[derive(Debug)] pub enum Error { InvalidBlobs(KzgError), - InvalidColumn(ColumnIndex, KzgError), + InvalidColumn(Vec<(ColumnIndex, KzgError)>), ReconstructColumnsError(KzgError), KzgCommitmentMismatch { blob_commitment: KzgCommitment, diff --git a/beacon_node/beacon_chain/src/data_column_verification.rs b/beacon_node/beacon_chain/src/data_column_verification.rs index 565e76704ef..1262fcdeb8e 100644 --- a/beacon_node/beacon_chain/src/data_column_verification.rs +++ b/beacon_node/beacon_chain/src/data_column_verification.rs @@ -239,6 +239,18 @@ impl KzgVerifiedDataColumn { pub fn new(data_column: Arc>, kzg: &Kzg) -> Result { verify_kzg_for_data_column(data_column, kzg) } + + pub fn from_batch( + data_columns: Vec>>, + kzg: &Kzg, + ) -> Result, Vec<(ColumnIndex, KzgError)>> { + verify_kzg_for_data_column_list_with_scoring(data_columns.iter(), kzg)?; + Ok(data_columns + .into_iter() + .map(|column| Self { data: column }) + .collect()) + } + pub fn to_data_column(self) -> Arc> { self.data } @@ -378,6 +390,38 @@ where Ok(()) } +/// Complete kzg verification for a list of `DataColumnSidecar`s. +/// +/// If there's at least one invalid column, it re-verifies all columns individually to identify the +/// first column that is invalid. This is necessary to attribute fault to the specific peer that +/// sent bad data. The re-verification cost should not be significant. If a peer sends invalid data it +/// will be quickly banned. +pub fn verify_kzg_for_data_column_list_with_scoring<'a, E: EthSpec, I>( + data_column_iter: I, + kzg: &'a Kzg, +) -> Result<(), Vec<(ColumnIndex, KzgError)>> +where + I: Iterator>> + Clone, +{ + if verify_kzg_for_data_column_list(data_column_iter.clone(), kzg).is_ok() { + return Ok(()); + }; + + // Find all columns that are invalid and identify by index. If we hit this condition there + // should be at least one invalid column + let errors = data_column_iter + .filter_map(|data_column| { + if let Err(e) = verify_kzg_for_data_column(data_column.clone(), kzg) { + Some((data_column.index, e)) + } else { + None + } + }) + .collect::>(); + + Err(errors) +} + pub fn validate_data_column_sidecar_for_gossip( data_column: Arc>, subnet: u64, diff --git a/beacon_node/builder_client/Cargo.toml b/beacon_node/builder_client/Cargo.toml index 3531e81c847..1920bd0ebb2 100644 --- a/beacon_node/builder_client/Cargo.toml +++ b/beacon_node/builder_client/Cargo.toml @@ -6,7 +6,9 @@ authors = ["Sean Anderson "] [dependencies] eth2 = { workspace = true } +ethereum_ssz = { workspace = true } lighthouse_version = { workspace = true } reqwest = { workspace = true } sensitive_url = { workspace = true } serde = { workspace = true } +serde_json = { workspace = true } diff --git a/beacon_node/builder_client/src/lib.rs b/beacon_node/builder_client/src/lib.rs index 91ee00a65f7..5f64ac7e43c 100644 --- a/beacon_node/builder_client/src/lib.rs +++ b/beacon_node/builder_client/src/lib.rs @@ -1,16 +1,24 @@ use eth2::types::builder_bid::SignedBuilderBid; +use eth2::types::fork_versioned_response::EmptyMetadata; use eth2::types::{ - EthSpec, ExecutionBlockHash, ForkVersionedResponse, PublicKeyBytes, - SignedValidatorRegistrationData, Slot, + ContentType, EthSpec, ExecutionBlockHash, ForkName, ForkVersionDecode, ForkVersionDeserialize, + ForkVersionedResponse, PublicKeyBytes, SignedValidatorRegistrationData, Slot, }; use eth2::types::{FullPayloadContents, SignedBlindedBeaconBlock}; pub use eth2::Error; -use eth2::{ok_or_error, StatusCode, CONSENSUS_VERSION_HEADER}; -use reqwest::header::{HeaderMap, HeaderValue}; +use eth2::{ + ok_or_error, StatusCode, CONSENSUS_VERSION_HEADER, CONTENT_TYPE_HEADER, + JSON_CONTENT_TYPE_HEADER, SSZ_CONTENT_TYPE_HEADER, +}; +use reqwest::header::{HeaderMap, HeaderValue, ACCEPT}; use reqwest::{IntoUrl, Response}; use sensitive_url::SensitiveUrl; use serde::de::DeserializeOwned; use serde::Serialize; +use ssz::Encode; +use std::str::FromStr; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; use std::time::Duration; pub const DEFAULT_TIMEOUT_MILLIS: u64 = 15000; @@ -49,6 +57,7 @@ pub struct BuilderHttpClient { server: SensitiveUrl, timeouts: Timeouts, user_agent: String, + ssz_enabled: Arc, } impl BuilderHttpClient { @@ -64,6 +73,7 @@ impl BuilderHttpClient { server, timeouts: Timeouts::new(builder_header_timeout), user_agent, + ssz_enabled: Arc::new(false.into()), }) } @@ -71,6 +81,78 @@ impl BuilderHttpClient { &self.user_agent } + fn fork_name_from_header(&self, headers: &HeaderMap) -> Result, String> { + headers + .get(CONSENSUS_VERSION_HEADER) + .map(|fork_name| { + fork_name + .to_str() + .map_err(|e| e.to_string()) + .and_then(ForkName::from_str) + }) + .transpose() + } + + fn content_type_from_header(&self, headers: &HeaderMap) -> ContentType { + let Some(content_type) = headers.get(CONTENT_TYPE_HEADER).map(|content_type| { + let content_type = content_type.to_str(); + match content_type { + Ok(SSZ_CONTENT_TYPE_HEADER) => ContentType::Ssz, + _ => ContentType::Json, + } + }) else { + return ContentType::Json; + }; + content_type + } + + async fn get_with_header< + T: DeserializeOwned + ForkVersionDecode + ForkVersionDeserialize, + U: IntoUrl, + >( + &self, + url: U, + timeout: Duration, + headers: HeaderMap, + ) -> Result, Error> { + let response = self + .get_response_with_header(url, Some(timeout), headers) + .await?; + + let headers = response.headers().clone(); + let response_bytes = response.bytes().await?; + + let Ok(Some(fork_name)) = self.fork_name_from_header(&headers) else { + // if no fork version specified, attempt to fallback to JSON + self.ssz_enabled.store(false, Ordering::SeqCst); + return serde_json::from_slice(&response_bytes).map_err(Error::InvalidJson); + }; + + let content_type = self.content_type_from_header(&headers); + + match content_type { + ContentType::Ssz => { + self.ssz_enabled.store(true, Ordering::SeqCst); + T::from_ssz_bytes_by_fork(&response_bytes, fork_name) + .map(|data| ForkVersionedResponse { + version: Some(fork_name), + metadata: EmptyMetadata {}, + data, + }) + .map_err(Error::InvalidSsz) + } + ContentType::Json => { + self.ssz_enabled.store(false, Ordering::SeqCst); + serde_json::from_slice(&response_bytes).map_err(Error::InvalidJson) + } + } + } + + /// Return `true` if the most recently received response from the builder had SSZ Content-Type. + pub fn is_ssz_enabled(&self) -> bool { + self.ssz_enabled.load(Ordering::SeqCst) + } + async fn get_with_timeout( &self, url: U, @@ -83,6 +165,21 @@ impl BuilderHttpClient { .map_err(Into::into) } + /// Perform a HTTP GET request, returning the `Response` for further processing. + async fn get_response_with_header( + &self, + url: U, + timeout: Option, + headers: HeaderMap, + ) -> Result { + let mut builder = self.client.get(url); + if let Some(timeout) = timeout { + builder = builder.timeout(timeout); + } + let response = builder.headers(headers).send().await.map_err(Error::from)?; + ok_or_error(response).await + } + /// Perform a HTTP GET request, returning the `Response` for further processing. async fn get_response_with_timeout( &self, @@ -112,6 +209,32 @@ impl BuilderHttpClient { ok_or_error(response).await } + async fn post_ssz_with_raw_response( + &self, + url: U, + ssz_body: Vec, + mut headers: HeaderMap, + timeout: Option, + ) -> Result { + let mut builder = self.client.post(url); + if let Some(timeout) = timeout { + builder = builder.timeout(timeout); + } + + headers.insert( + CONTENT_TYPE_HEADER, + HeaderValue::from_static(SSZ_CONTENT_TYPE_HEADER), + ); + + let response = builder + .headers(headers) + .body(ssz_body) + .send() + .await + .map_err(Error::from)?; + ok_or_error(response).await + } + async fn post_with_raw_response( &self, url: U, @@ -152,6 +275,42 @@ impl BuilderHttpClient { Ok(()) } + /// `POST /eth/v1/builder/blinded_blocks` with SSZ serialized request body + pub async fn post_builder_blinded_blocks_ssz( + &self, + blinded_block: &SignedBlindedBeaconBlock, + ) -> Result, Error> { + let mut path = self.server.full.clone(); + + let body = blinded_block.as_ssz_bytes(); + + path.path_segments_mut() + .map_err(|()| Error::InvalidUrl(self.server.clone()))? + .push("eth") + .push("v1") + .push("builder") + .push("blinded_blocks"); + + let mut headers = HeaderMap::new(); + if let Ok(value) = HeaderValue::from_str(&blinded_block.fork_name_unchecked().to_string()) { + headers.insert(CONSENSUS_VERSION_HEADER, value); + } + + let result = self + .post_ssz_with_raw_response( + path, + body, + headers, + Some(self.timeouts.post_blinded_blocks), + ) + .await? + .bytes() + .await?; + + FullPayloadContents::from_ssz_bytes_by_fork(&result, blinded_block.fork_name_unchecked()) + .map_err(Error::InvalidSsz) + } + /// `POST /eth/v1/builder/blinded_blocks` pub async fn post_builder_blinded_blocks( &self, @@ -202,7 +361,17 @@ impl BuilderHttpClient { .push(format!("{parent_hash:?}").as_str()) .push(pubkey.as_hex_string().as_str()); - let resp = self.get_with_timeout(path, self.timeouts.get_header).await; + let mut headers = HeaderMap::new(); + if let Ok(ssz_content_type_header) = HeaderValue::from_str(&format!( + "{}; q=1.0,{}; q=0.9", + SSZ_CONTENT_TYPE_HEADER, JSON_CONTENT_TYPE_HEADER + )) { + headers.insert(ACCEPT, ssz_content_type_header); + }; + + let resp = self + .get_with_header(path, self.timeouts.get_header, headers) + .await; if matches!(resp, Err(Error::StatusCode(StatusCode::NO_CONTENT))) { Ok(None) diff --git a/beacon_node/execution_layer/src/lib.rs b/beacon_node/execution_layer/src/lib.rs index d5fef4c5aaa..4e0fe1de16b 100644 --- a/beacon_node/execution_layer/src/lib.rs +++ b/beacon_node/execution_layer/src/lib.rs @@ -1900,11 +1900,18 @@ impl ExecutionLayer { if let Some(builder) = self.builder() { let (payload_result, duration) = timed_future(metrics::POST_BLINDED_PAYLOAD_BUILDER, async { - builder - .post_builder_blinded_blocks(block) - .await - .map_err(Error::Builder) - .map(|d| d.data) + if builder.is_ssz_enabled() { + builder + .post_builder_blinded_blocks_ssz(block) + .await + .map_err(Error::Builder) + } else { + builder + .post_builder_blinded_blocks(block) + .await + .map_err(Error::Builder) + .map(|d| d.data) + } }) .await; diff --git a/beacon_node/execution_layer/src/test_utils/mock_builder.rs b/beacon_node/execution_layer/src/test_utils/mock_builder.rs index 3540909fe46..f07ee7ac6f8 100644 --- a/beacon_node/execution_layer/src/test_utils/mock_builder.rs +++ b/beacon_node/execution_layer/src/test_utils/mock_builder.rs @@ -1,15 +1,20 @@ use crate::test_utils::{DEFAULT_BUILDER_PAYLOAD_VALUE_WEI, DEFAULT_JWT_SECRET}; use crate::{Config, ExecutionLayer, PayloadAttributes, PayloadParameters}; +use bytes::Bytes; use eth2::types::PublishBlockRequest; use eth2::types::{ BlobsBundle, BlockId, BroadcastValidation, EventKind, EventTopic, FullPayloadContents, ProposerData, StateId, ValidatorId, }; -use eth2::{BeaconNodeHttpClient, Timeouts, CONSENSUS_VERSION_HEADER}; +use eth2::{ + BeaconNodeHttpClient, Timeouts, CONSENSUS_VERSION_HEADER, CONTENT_TYPE_HEADER, + SSZ_CONTENT_TYPE_HEADER, +}; use fork_choice::ForkchoiceUpdateParameters; use parking_lot::RwLock; use sensitive_url::SensitiveUrl; use slog::{debug, error, info, warn, Logger}; +use ssz::Encode; use std::collections::HashMap; use std::fmt::Debug; use std::future::Future; @@ -26,11 +31,12 @@ use types::builder_bid::{ }; use types::{ Address, BeaconState, ChainSpec, Epoch, EthSpec, ExecPayload, ExecutionPayload, - ExecutionPayloadHeaderRefMut, ExecutionRequests, ForkName, ForkVersionedResponse, Hash256, - PublicKeyBytes, Signature, SignedBlindedBeaconBlock, SignedRoot, - SignedValidatorRegistrationData, Slot, Uint256, + ExecutionPayloadHeaderRefMut, ExecutionRequests, ForkName, ForkVersionDecode, + ForkVersionedResponse, Hash256, PublicKeyBytes, Signature, SignedBlindedBeaconBlock, + SignedRoot, SignedValidatorRegistrationData, Slot, Uint256, }; use types::{ExecutionBlockHash, SecretKey}; +use warp::reply::{self, Reply}; use warp::{Filter, Rejection}; pub const DEFAULT_FEE_RECIPIENT: Address = Address::repeat_byte(42); @@ -955,6 +961,33 @@ pub fn serve( ) .boxed(); + let blinded_block_ssz = prefix + .and(warp::path("blinded_blocks")) + .and(warp::body::bytes()) + .and(warp::header::header::(CONSENSUS_VERSION_HEADER)) + .and(warp::path::end()) + .and(ctx_filter.clone()) + .and_then( + |block_bytes: Bytes, fork_name: ForkName, builder: MockBuilder| async move { + let block = + SignedBlindedBeaconBlock::::from_ssz_bytes_by_fork(&block_bytes, fork_name) + .map_err(|e| warp::reject::custom(Custom(format!("{:?}", e))))?; + let payload = builder + .submit_blinded_block(block) + .await + .map_err(|e| warp::reject::custom(Custom(e)))?; + + Ok::<_, warp::reject::Rejection>( + warp::http::Response::builder() + .status(200) + .body(payload.as_ssz_bytes()) + .map(add_ssz_content_type_header) + .map(|res| add_consensus_version_header(res, fork_name)) + .unwrap(), + ) + }, + ); + let blinded_block = prefix .and(warp::path("blinded_blocks")) @@ -1007,35 +1040,47 @@ pub fn serve( ) .and(warp::path::end()) .and(ctx_filter.clone()) + .and(warp::header::optional::("accept")) .and_then( |slot: Slot, parent_hash: ExecutionBlockHash, pubkey: PublicKeyBytes, - builder: MockBuilder| async move { + builder: MockBuilder, + accept_header: Option| async move { let fork_name = builder.fork_name_at_slot(slot); let signed_bid = builder .get_header(slot, parent_hash, pubkey) .await .map_err(|e| warp::reject::custom(Custom(e)))?; - - let resp: ForkVersionedResponse<_> = ForkVersionedResponse { - version: Some(fork_name), - metadata: Default::default(), - data: signed_bid, - }; - let json_bid = serde_json::to_string(&resp) - .map_err(|_| reject("coudn't serialize signed bid"))?; - Ok::<_, Rejection>( - warp::http::Response::builder() - .status(200) - .body(json_bid) - .unwrap(), - ) + let accept_header = accept_header.unwrap_or(eth2::types::Accept::Any); + match accept_header { + eth2::types::Accept::Ssz => Ok::<_, Rejection>( + warp::http::Response::builder() + .status(200) + .body(signed_bid.as_ssz_bytes()) + .map(add_ssz_content_type_header) + .map(|res| add_consensus_version_header(res, fork_name)) + .unwrap(), + ), + eth2::types::Accept::Json | eth2::types::Accept::Any => { + let resp: ForkVersionedResponse<_> = ForkVersionedResponse { + version: Some(fork_name), + metadata: Default::default(), + data: signed_bid, + }; + Ok::<_, Rejection>(warp::reply::json(&resp).into_response()) + } + } }, ); let routes = warp::post() - .and(validators.or(blinded_block)) + // Routes which expect `application/octet-stream` go within this `and`. + .and( + warp::header::exact(CONTENT_TYPE_HEADER, SSZ_CONTENT_TYPE_HEADER) + .and(blinded_block_ssz), + ) + .or(validators.or(blinded_block)) .or(warp::get().and(status).or(header)) .map(|reply| warp::reply::with_header(reply, "Server", "lighthouse-mock-builder-server")); @@ -1048,3 +1093,13 @@ pub fn serve( fn reject(msg: &'static str) -> Rejection { warp::reject::custom(Custom(msg.to_string())) } + +/// Add the 'Content-Type application/octet-stream` header to a response. +fn add_ssz_content_type_header(reply: T) -> warp::reply::Response { + reply::with_header(reply, CONTENT_TYPE_HEADER, SSZ_CONTENT_TYPE_HEADER).into_response() +} + +/// Add the `Eth-Consensus-Version` header to a response. +fn add_consensus_version_header(reply: T, fork_name: ForkName) -> warp::reply::Response { + reply::with_header(reply, CONSENSUS_VERSION_HEADER, fork_name.to_string()).into_response() +} diff --git a/beacon_node/lighthouse_network/src/service/mod.rs b/beacon_node/lighthouse_network/src/service/mod.rs index 354def79b03..8586fd9cd36 100644 --- a/beacon_node/lighthouse_network/src/service/mod.rs +++ b/beacon_node/lighthouse_network/src/service/mod.rs @@ -708,11 +708,17 @@ impl Network { } // Subscribe to core topics for the new fork - for kind in fork_core_topics::(&new_fork, &self.fork_context.spec) { + for kind in fork_core_topics::( + &new_fork, + &self.fork_context.spec, + &self.network_globals.as_topic_config(), + ) { let topic = GossipTopic::new(kind, GossipEncoding::default(), new_fork_digest); self.subscribe(topic); } + // TODO(das): unsubscribe from blob topics at the Fulu fork + // Register the new topics for metrics let topics_to_keep_metrics_for = attestation_sync_committee_topics::() .map(|gossip_kind| { diff --git a/beacon_node/lighthouse_network/src/types/globals.rs b/beacon_node/lighthouse_network/src/types/globals.rs index c9e84e2dd11..2800b75133b 100644 --- a/beacon_node/lighthouse_network/src/types/globals.rs +++ b/beacon_node/lighthouse_network/src/types/globals.rs @@ -1,4 +1,5 @@ //! A collection of variables that are accessible outside of the network thread itself. +use super::TopicConfig; use crate::peer_manager::peerdb::PeerDB; use crate::rpc::{MetaData, MetaDataV3}; use crate::types::{BackFillState, SyncState}; @@ -183,6 +184,14 @@ impl NetworkGlobals { .collect::>() } + /// Returns the TopicConfig to compute the set of Gossip topics for a given fork + pub fn as_topic_config(&self) -> TopicConfig { + TopicConfig { + subscribe_all_data_column_subnets: self.config.subscribe_all_data_column_subnets, + sampling_subnets: &self.sampling_subnets, + } + } + /// TESTING ONLY. Build a dummy NetworkGlobals instance. pub fn new_test_globals( trusted_peers: Vec, diff --git a/beacon_node/lighthouse_network/src/types/mod.rs b/beacon_node/lighthouse_network/src/types/mod.rs index a1eedaef746..58ba7588b98 100644 --- a/beacon_node/lighthouse_network/src/types/mod.rs +++ b/beacon_node/lighthouse_network/src/types/mod.rs @@ -17,6 +17,6 @@ pub use subnet::{Subnet, SubnetDiscovery}; pub use sync_state::{BackFillState, SyncState}; pub use topics::{ attestation_sync_committee_topics, core_topics_to_subscribe, fork_core_topics, - subnet_from_topic_hash, GossipEncoding, GossipKind, GossipTopic, ALTAIR_CORE_TOPICS, - BASE_CORE_TOPICS, CAPELLA_CORE_TOPICS, LIGHT_CLIENT_GOSSIP_TOPICS, + subnet_from_topic_hash, GossipEncoding, GossipKind, GossipTopic, TopicConfig, + ALTAIR_CORE_TOPICS, BASE_CORE_TOPICS, CAPELLA_CORE_TOPICS, LIGHT_CLIENT_GOSSIP_TOPICS, }; diff --git a/beacon_node/lighthouse_network/src/types/topics.rs b/beacon_node/lighthouse_network/src/types/topics.rs index 2c79f934237..171dab09a35 100644 --- a/beacon_node/lighthouse_network/src/types/topics.rs +++ b/beacon_node/lighthouse_network/src/types/topics.rs @@ -1,5 +1,6 @@ use gossipsub::{IdentTopic as Topic, TopicHash}; use serde::{Deserialize, Serialize}; +use std::collections::HashSet; use strum::AsRefStr; use types::{ChainSpec, DataColumnSubnetId, EthSpec, ForkName, SubnetId, SyncSubnetId, Unsigned}; @@ -41,8 +42,18 @@ pub const LIGHT_CLIENT_GOSSIP_TOPICS: [GossipKind; 2] = [ GossipKind::LightClientOptimisticUpdate, ]; +#[derive(Debug)] +pub struct TopicConfig<'a> { + pub subscribe_all_data_column_subnets: bool, + pub sampling_subnets: &'a HashSet, +} + /// Returns the core topics associated with each fork that are new to the previous fork -pub fn fork_core_topics(fork_name: &ForkName, spec: &ChainSpec) -> Vec { +pub fn fork_core_topics( + fork_name: &ForkName, + spec: &ChainSpec, + topic_config: &TopicConfig, +) -> Vec { match fork_name { ForkName::Base => BASE_CORE_TOPICS.to_vec(), ForkName::Altair => ALTAIR_CORE_TOPICS.to_vec(), @@ -64,7 +75,21 @@ pub fn fork_core_topics(fork_name: &ForkName, spec: &ChainSpec) -> V } electra_blob_topics } - ForkName::Fulu => vec![], + ForkName::Fulu => { + let mut topics = vec![]; + if topic_config.subscribe_all_data_column_subnets { + for column_subnet in 0..spec.data_column_sidecar_subnet_count { + topics.push(GossipKind::DataColumnSidecar(DataColumnSubnetId::new( + column_subnet, + ))); + } + } else { + for column_subnet in topic_config.sampling_subnets { + topics.push(GossipKind::DataColumnSidecar(*column_subnet)); + } + } + topics + } } } @@ -84,10 +109,11 @@ pub fn attestation_sync_committee_topics() -> impl Iterator( mut current_fork: ForkName, spec: &ChainSpec, + topic_config: &TopicConfig, ) -> Vec { - let mut topics = fork_core_topics::(¤t_fork, spec); + let mut topics = fork_core_topics::(¤t_fork, spec, topic_config); while let Some(previous_fork) = current_fork.previous_fork() { - let previous_fork_topics = fork_core_topics::(&previous_fork, spec); + let previous_fork_topics = fork_core_topics::(&previous_fork, spec, topic_config); topics.extend(previous_fork_topics); current_fork = previous_fork; } @@ -475,8 +501,15 @@ mod tests { type E = MainnetEthSpec; let spec = E::default_spec(); let mut all_topics = Vec::new(); - let mut electra_core_topics = fork_core_topics::(&ForkName::Electra, &spec); - let mut deneb_core_topics = fork_core_topics::(&ForkName::Deneb, &spec); + let topic_config = TopicConfig { + subscribe_all_data_column_subnets: false, + sampling_subnets: &HashSet::from_iter([1, 2].map(DataColumnSubnetId::new)), + }; + let mut fulu_core_topics = fork_core_topics::(&ForkName::Fulu, &spec, &topic_config); + let mut electra_core_topics = + fork_core_topics::(&ForkName::Electra, &spec, &topic_config); + let mut deneb_core_topics = fork_core_topics::(&ForkName::Deneb, &spec, &topic_config); + all_topics.append(&mut fulu_core_topics); all_topics.append(&mut electra_core_topics); all_topics.append(&mut deneb_core_topics); all_topics.extend(CAPELLA_CORE_TOPICS); @@ -484,7 +517,7 @@ mod tests { all_topics.extend(BASE_CORE_TOPICS); let latest_fork = *ForkName::list_all().last().unwrap(); - let core_topics = core_topics_to_subscribe::(latest_fork, &spec); + let core_topics = core_topics_to_subscribe::(latest_fork, &spec, &topic_config); // Need to check all the topics exist in an order independent manner for topic in all_topics { assert!(core_topics.contains(&topic)); diff --git a/beacon_node/network/src/service.rs b/beacon_node/network/src/service.rs index 49f73bf9c8d..1b2a681c644 100644 --- a/beacon_node/network/src/service.rs +++ b/beacon_node/network/src/service.rs @@ -33,8 +33,8 @@ use task_executor::ShutdownReason; use tokio::sync::mpsc; use tokio::time::Sleep; use types::{ - ChainSpec, DataColumnSubnetId, EthSpec, ForkContext, Slot, SubnetId, SyncCommitteeSubscription, - SyncSubnetId, Unsigned, ValidatorSubscription, + ChainSpec, EthSpec, ForkContext, Slot, SubnetId, SyncCommitteeSubscription, SyncSubnetId, + Unsigned, ValidatorSubscription, }; mod tests; @@ -181,8 +181,6 @@ pub struct NetworkService { next_fork_subscriptions: Pin>>, /// A delay that expires when we need to unsubscribe from old fork topics. next_unsubscribe: Pin>>, - /// Subscribe to all the data column subnets. - subscribe_all_data_column_subnets: bool, /// Subscribe to all the subnets once synced. subscribe_all_subnets: bool, /// Shutdown beacon node after sync is complete. @@ -349,7 +347,6 @@ impl NetworkService { next_fork_update, next_fork_subscriptions, next_unsubscribe, - subscribe_all_data_column_subnets: config.subscribe_all_data_column_subnets, subscribe_all_subnets: config.subscribe_all_subnets, shutdown_after_sync: config.shutdown_after_sync, metrics_enabled: config.metrics_enabled, @@ -717,6 +714,7 @@ impl NetworkService { for topic_kind in core_topics_to_subscribe::( self.fork_context.current_fork(), &self.fork_context.spec, + &self.network_globals.as_topic_config(), ) { for fork_digest in self.required_gossip_fork_digests() { let topic = GossipTopic::new( @@ -751,10 +749,6 @@ impl NetworkService { } } - if self.fork_context.spec.is_peer_das_scheduled() { - self.subscribe_to_peer_das_topics(&mut subscribed_topics); - } - // If we are to subscribe to all subnets we do it here if self.subscribe_all_subnets { for subnet_id in 0..<::EthSpec as EthSpec>::SubnetBitfieldLength::to_u64() { @@ -801,37 +795,6 @@ impl NetworkService { } } - /// Keeping these separate from core topics because it has custom logic: - /// 1. Data column subscription logic depends on subscription configuration. - /// 2. Data column topic subscriptions will be dynamic based on validator balances due to - /// validator custody. - /// - /// TODO(das): The downside with not including it in core fork topic is - we subscribe to - /// PeerDAS topics on startup if Fulu is scheduled, rather than waiting until the fork. - /// If this is an issue we could potentially consider adding the logic to - /// `network.subscribe_new_fork_topics()`. - fn subscribe_to_peer_das_topics(&mut self, subscribed_topics: &mut Vec) { - let column_subnets_to_subscribe = if self.subscribe_all_data_column_subnets { - &(0..self.fork_context.spec.data_column_sidecar_subnet_count) - .map(DataColumnSubnetId::new) - .collect() - } else { - &self.network_globals.sampling_subnets - }; - - for column_subnet in column_subnets_to_subscribe.iter() { - for fork_digest in self.required_gossip_fork_digests() { - let gossip_kind = Subnet::DataColumn(*column_subnet).into(); - let topic = GossipTopic::new(gossip_kind, GossipEncoding::default(), fork_digest); - if self.libp2p.subscribe(topic.clone()) { - subscribed_topics.push(topic); - } else { - warn!(self.log, "Could not subscribe to topic"; "topic" => %topic); - } - } - } - } - /// Handle a message sent to the network service. async fn on_validator_subscription_msg(&mut self, msg: ValidatorSubscriptionMessage) { match msg { @@ -947,6 +910,7 @@ impl NetworkService { let core_topics = core_topics_to_subscribe::( self.fork_context.current_fork(), &self.fork_context.spec, + &self.network_globals.as_topic_config(), ); let core_topics: HashSet<&GossipKind> = HashSet::from_iter(&core_topics); let subscriptions = self.network_globals.gossipsub_subscriptions.read(); diff --git a/beacon_node/network/src/sync/block_lookups/mod.rs b/beacon_node/network/src/sync/block_lookups/mod.rs index ac4df42a4e8..2172c8dcd8d 100644 --- a/beacon_node/network/src/sync/block_lookups/mod.rs +++ b/beacon_node/network/src/sync/block_lookups/mod.rs @@ -36,6 +36,7 @@ use beacon_chain::data_availability_checker::{ use beacon_chain::{AvailabilityProcessingStatus, BeaconChainTypes, BlockError}; pub use common::RequestState; use fnv::FnvHashMap; +use itertools::Itertools; use lighthouse_network::service::api_types::SingleLookupReqId; use lighthouse_network::{PeerAction, PeerId}; use lru_cache::LRUTimeCache; @@ -644,8 +645,15 @@ impl BlockLookups { // but future errors may follow the same pattern. Generalize this // pattern with https://github.com/sigp/lighthouse/pull/6321 BlockError::AvailabilityCheck( - AvailabilityCheckError::InvalidColumn(index, _), - ) => peer_group.of_index(index as usize).collect(), + AvailabilityCheckError::InvalidColumn(errors), + ) => errors + .iter() + // Collect all peers that sent a column that was invalid. Must + // run .unique as a single peer can send multiple invalid + // columns. Penalize once to avoid insta-bans + .flat_map(|(index, _)| peer_group.of_index((*index) as usize)) + .unique() + .collect(), _ => peer_group.all().collect(), }; for peer in peers_to_penalize { diff --git a/beacon_node/store/src/hot_cold_store.rs b/beacon_node/store/src/hot_cold_store.rs index 134be9ec0dc..128c03f771b 100644 --- a/beacon_node/store/src/hot_cold_store.rs +++ b/beacon_node/store/src/hot_cold_store.rs @@ -666,7 +666,9 @@ impl, Cold: ItemStore> HotColdDB .hot_db .get_bytes(ExecutionPayload::::db_column(), key)? { - Some(bytes) => Ok(Some(ExecutionPayload::from_ssz_bytes(&bytes, fork_name)?)), + Some(bytes) => Ok(Some(ExecutionPayload::from_ssz_bytes_by_fork( + &bytes, fork_name, + )?)), None => Ok(None), } } diff --git a/common/eth2/src/lib.rs b/common/eth2/src/lib.rs index b86aa627657..73e9d57abc0 100644 --- a/common/eth2/src/lib.rs +++ b/common/eth2/src/lib.rs @@ -49,6 +49,7 @@ pub const CONSENSUS_BLOCK_VALUE_HEADER: &str = "Eth-Consensus-Block-Value"; pub const CONTENT_TYPE_HEADER: &str = "Content-Type"; pub const SSZ_CONTENT_TYPE_HEADER: &str = "application/octet-stream"; +pub const JSON_CONTENT_TYPE_HEADER: &str = "application/json"; #[derive(Debug)] pub enum Error { @@ -112,9 +113,9 @@ impl Error { Error::InvalidSignatureHeader => None, Error::MissingSignatureHeader => None, Error::InvalidJson(_) => None, + Error::InvalidSsz(_) => None, Error::InvalidServerSentEvent(_) => None, Error::InvalidHeaders(_) => None, - Error::InvalidSsz(_) => None, Error::TokenReadError(..) => None, Error::NoServerPubkey | Error::NoToken => None, } diff --git a/common/eth2/src/types.rs b/common/eth2/src/types.rs index c6e95e1ba30..59374f629d6 100644 --- a/common/eth2/src/types.rs +++ b/common/eth2/src/types.rs @@ -1664,7 +1664,7 @@ impl FullBlockContents { } /// SSZ decode with fork variant determined by slot. - pub fn from_ssz_bytes(bytes: &[u8], spec: &ChainSpec) -> Result { + pub fn from_ssz_bytes(bytes: &[u8], spec: &ChainSpec) -> Result { let slot_len = ::ssz_fixed_len(); let slot_bytes = bytes .get(0..slot_len) @@ -1678,10 +1678,7 @@ impl FullBlockContents { } /// SSZ decode with fork variant passed in explicitly. - pub fn from_ssz_bytes_for_fork( - bytes: &[u8], - fork_name: ForkName, - ) -> Result { + pub fn from_ssz_bytes_for_fork(bytes: &[u8], fork_name: ForkName) -> Result { if fork_name.deneb_enabled() { let mut builder = ssz::SszDecoderBuilder::new(bytes); @@ -1836,7 +1833,7 @@ impl PublishBlockRequest { } /// SSZ decode with fork variant determined by `fork_name`. - pub fn from_ssz_bytes(bytes: &[u8], fork_name: ForkName) -> Result { + pub fn from_ssz_bytes(bytes: &[u8], fork_name: ForkName) -> Result { if fork_name.deneb_enabled() { let mut builder = ssz::SszDecoderBuilder::new(bytes); builder.register_anonymous_variable_length_item()?; @@ -1845,7 +1842,7 @@ impl PublishBlockRequest { let mut decoder = builder.build()?; let block = decoder.decode_next_with(|bytes| { - SignedBeaconBlock::from_ssz_bytes_for_fork(bytes, fork_name) + SignedBeaconBlock::from_ssz_bytes_by_fork(bytes, fork_name) })?; let kzg_proofs = decoder.decode_next()?; let blobs = decoder.decode_next()?; @@ -1854,7 +1851,7 @@ impl PublishBlockRequest { Some((kzg_proofs, blobs)), )) } else { - SignedBeaconBlock::from_ssz_bytes_for_fork(bytes, fork_name) + SignedBeaconBlock::from_ssz_bytes_by_fork(bytes, fork_name) .map(|block| PublishBlockRequest::Block(Arc::new(block))) } } @@ -1946,6 +1943,24 @@ pub enum FullPayloadContents { PayloadAndBlobs(ExecutionPayloadAndBlobs), } +impl ForkVersionDecode for FullPayloadContents { + fn from_ssz_bytes_by_fork(bytes: &[u8], fork_name: ForkName) -> Result { + if fork_name.deneb_enabled() { + Ok(Self::PayloadAndBlobs( + ExecutionPayloadAndBlobs::from_ssz_bytes_by_fork(bytes, fork_name)?, + )) + } else if fork_name.bellatrix_enabled() { + Ok(Self::Payload(ExecutionPayload::from_ssz_bytes_by_fork( + bytes, fork_name, + )?)) + } else { + Err(ssz::DecodeError::BytesInvalid(format!( + "FullPayloadContents decoding for {fork_name} not implemented" + ))) + } + } +} + impl FullPayloadContents { pub fn new( execution_payload: ExecutionPayload, @@ -2012,6 +2027,36 @@ pub struct ExecutionPayloadAndBlobs { pub blobs_bundle: BlobsBundle, } +impl ForkVersionDecode for ExecutionPayloadAndBlobs { + fn from_ssz_bytes_by_fork(bytes: &[u8], fork_name: ForkName) -> Result { + let mut builder = ssz::SszDecoderBuilder::new(bytes); + builder.register_anonymous_variable_length_item()?; + builder.register_type::>()?; + let mut decoder = builder.build()?; + + if fork_name.deneb_enabled() { + let execution_payload = decoder.decode_next_with(|bytes| { + ExecutionPayload::from_ssz_bytes_by_fork(bytes, fork_name) + })?; + let blobs_bundle = decoder.decode_next()?; + Ok(Self { + execution_payload, + blobs_bundle, + }) + } else { + Err(DecodeError::BytesInvalid(format!( + "ExecutionPayloadAndBlobs decoding for {fork_name} not implemented" + ))) + } + } +} + +#[derive(Debug)] +pub enum ContentType { + Json, + Ssz, +} + #[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize, Encode, Decode)] #[serde(bound = "E: EthSpec")] pub struct BlobsBundle { diff --git a/consensus/types/src/builder_bid.rs b/consensus/types/src/builder_bid.rs index ac53c41216f..49911c39095 100644 --- a/consensus/types/src/builder_bid.rs +++ b/consensus/types/src/builder_bid.rs @@ -3,25 +3,37 @@ use crate::{ ChainSpec, EthSpec, ExecutionPayloadHeaderBellatrix, ExecutionPayloadHeaderCapella, ExecutionPayloadHeaderDeneb, ExecutionPayloadHeaderElectra, ExecutionPayloadHeaderFulu, ExecutionPayloadHeaderRef, ExecutionPayloadHeaderRefMut, ExecutionRequests, ForkName, - ForkVersionDeserialize, SignedRoot, Uint256, + ForkVersionDecode, ForkVersionDeserialize, SignedRoot, Uint256, }; use bls::PublicKeyBytes; use bls::Signature; use serde::{Deserialize, Deserializer, Serialize}; +use ssz::Decode; +use ssz_derive::{Decode, Encode}; use superstruct::superstruct; use tree_hash_derive::TreeHash; #[superstruct( variants(Bellatrix, Capella, Deneb, Electra, Fulu), variant_attributes( - derive(PartialEq, Debug, Serialize, Deserialize, TreeHash, Clone), + derive( + PartialEq, + Debug, + Encode, + Serialize, + Deserialize, + TreeHash, + Decode, + Clone + ), serde(bound = "E: EthSpec", deny_unknown_fields) ), map_ref_into(ExecutionPayloadHeaderRef), map_ref_mut_into(ExecutionPayloadHeaderRefMut) )] -#[derive(PartialEq, Debug, Serialize, Deserialize, TreeHash, Clone)] +#[derive(PartialEq, Debug, Encode, Serialize, Deserialize, TreeHash, Clone)] #[serde(bound = "E: EthSpec", deny_unknown_fields, untagged)] +#[ssz(enum_behaviour = "transparent")] #[tree_hash(enum_behaviour = "transparent")] pub struct BuilderBid { #[superstruct(only(Bellatrix), partial_getter(rename = "header_bellatrix"))] @@ -65,16 +77,54 @@ impl<'a, E: EthSpec> BuilderBidRefMut<'a, E> { } } +impl ForkVersionDecode for BuilderBid { + /// SSZ decode with explicit fork variant. + fn from_ssz_bytes_by_fork(bytes: &[u8], fork_name: ForkName) -> Result { + let builder_bid = match fork_name { + ForkName::Altair | ForkName::Base => { + return Err(ssz::DecodeError::BytesInvalid(format!( + "unsupported fork for ExecutionPayloadHeader: {fork_name}", + ))) + } + ForkName::Bellatrix => { + BuilderBid::Bellatrix(BuilderBidBellatrix::from_ssz_bytes(bytes)?) + } + ForkName::Capella => BuilderBid::Capella(BuilderBidCapella::from_ssz_bytes(bytes)?), + ForkName::Deneb => BuilderBid::Deneb(BuilderBidDeneb::from_ssz_bytes(bytes)?), + ForkName::Electra => BuilderBid::Electra(BuilderBidElectra::from_ssz_bytes(bytes)?), + ForkName::Fulu => BuilderBid::Fulu(BuilderBidFulu::from_ssz_bytes(bytes)?), + }; + Ok(builder_bid) + } +} + impl SignedRoot for BuilderBid {} /// Validator registration, for use in interacting with servers implementing the builder API. -#[derive(PartialEq, Debug, Serialize, Deserialize, Clone)] +#[derive(PartialEq, Debug, Encode, Serialize, Deserialize, Clone)] #[serde(bound = "E: EthSpec")] pub struct SignedBuilderBid { pub message: BuilderBid, pub signature: Signature, } +impl ForkVersionDecode for SignedBuilderBid { + /// SSZ decode with explicit fork variant. + fn from_ssz_bytes_by_fork(bytes: &[u8], fork_name: ForkName) -> Result { + let mut builder = ssz::SszDecoderBuilder::new(bytes); + + builder.register_anonymous_variable_length_item()?; + builder.register_type::()?; + + let mut decoder = builder.build()?; + let message = decoder + .decode_next_with(|bytes| BuilderBid::from_ssz_bytes_by_fork(bytes, fork_name))?; + let signature = decoder.decode_next()?; + + Ok(Self { message, signature }) + } +} + impl ForkVersionDeserialize for BuilderBid { fn deserialize_by_fork<'de, D: Deserializer<'de>>( value: serde_json::value::Value, diff --git a/consensus/types/src/execution_payload.rs b/consensus/types/src/execution_payload.rs index 2df66343af1..5d756c8529f 100644 --- a/consensus/types/src/execution_payload.rs +++ b/consensus/types/src/execution_payload.rs @@ -40,7 +40,7 @@ pub type Withdrawals = VariableList::MaxWithdrawal map_ref_into(ExecutionPayloadHeader) )] #[derive( - Debug, Clone, Serialize, Encode, Deserialize, TreeHash, Derivative, arbitrary::Arbitrary, + Debug, Clone, Serialize, Deserialize, Encode, TreeHash, Derivative, arbitrary::Arbitrary, )] #[derivative(PartialEq, Hash(bound = "E: EthSpec"))] #[serde(bound = "E: EthSpec", untagged)] @@ -102,8 +102,9 @@ impl<'a, E: EthSpec> ExecutionPayloadRef<'a, E> { } } -impl ExecutionPayload { - pub fn from_ssz_bytes(bytes: &[u8], fork_name: ForkName) -> Result { +impl ForkVersionDecode for ExecutionPayload { + /// SSZ decode with explicit fork variant. + fn from_ssz_bytes_by_fork(bytes: &[u8], fork_name: ForkName) -> Result { match fork_name { ForkName::Base | ForkName::Altair => Err(ssz::DecodeError::BytesInvalid(format!( "unsupported fork for ExecutionPayload: {fork_name}", @@ -117,7 +118,9 @@ impl ExecutionPayload { ForkName::Fulu => ExecutionPayloadFulu::from_ssz_bytes(bytes).map(Self::Fulu), } } +} +impl ExecutionPayload { #[allow(clippy::arithmetic_side_effects)] /// Returns the maximum size of an execution payload. pub fn max_execution_payload_bellatrix_size() -> usize { diff --git a/consensus/types/src/fork_versioned_response.rs b/consensus/types/src/fork_versioned_response.rs index cd78b5b3ca0..7e4efd05d66 100644 --- a/consensus/types/src/fork_versioned_response.rs +++ b/consensus/types/src/fork_versioned_response.rs @@ -4,6 +4,11 @@ use serde::{Deserialize, Deserializer, Serialize}; use serde_json::value::Value; use std::sync::Arc; +pub trait ForkVersionDecode: Sized { + /// SSZ decode with explicit fork variant. + fn from_ssz_bytes_by_fork(bytes: &[u8], fork_name: ForkName) -> Result; +} + pub trait ForkVersionDeserialize: Sized + DeserializeOwned { fn deserialize_by_fork<'de, D: Deserializer<'de>>( value: Value, diff --git a/consensus/types/src/lib.rs b/consensus/types/src/lib.rs index 11d1f5271b7..73a50b4ef3e 100644 --- a/consensus/types/src/lib.rs +++ b/consensus/types/src/lib.rs @@ -178,7 +178,9 @@ pub use crate::fork::Fork; pub use crate::fork_context::ForkContext; pub use crate::fork_data::ForkData; pub use crate::fork_name::{ForkName, InconsistentFork}; -pub use crate::fork_versioned_response::{ForkVersionDeserialize, ForkVersionedResponse}; +pub use crate::fork_versioned_response::{ + ForkVersionDecode, ForkVersionDeserialize, ForkVersionedResponse, +}; pub use crate::graffiti::{Graffiti, GRAFFITI_BYTES_LEN}; pub use crate::historical_batch::HistoricalBatch; pub use crate::indexed_attestation::{ diff --git a/consensus/types/src/signed_beacon_block.rs b/consensus/types/src/signed_beacon_block.rs index d9bf9bf55dd..eb5925a29b5 100644 --- a/consensus/types/src/signed_beacon_block.rs +++ b/consensus/types/src/signed_beacon_block.rs @@ -86,6 +86,17 @@ pub struct SignedBeaconBlock = FullP pub signature: Signature, } +impl> ForkVersionDecode + for SignedBeaconBlock +{ + /// SSZ decode with explicit fork variant. + fn from_ssz_bytes_by_fork(bytes: &[u8], fork_name: ForkName) -> Result { + Self::from_ssz_bytes_with(bytes, |bytes| { + BeaconBlock::from_ssz_bytes_for_fork(bytes, fork_name) + }) + } +} + pub type SignedBlindedBeaconBlock = SignedBeaconBlock>; impl> SignedBeaconBlock { @@ -108,16 +119,6 @@ impl> SignedBeaconBlock Self::from_ssz_bytes_with(bytes, |bytes| BeaconBlock::from_ssz_bytes(bytes, spec)) } - /// SSZ decode with explicit fork variant. - pub fn from_ssz_bytes_for_fork( - bytes: &[u8], - fork_name: ForkName, - ) -> Result { - Self::from_ssz_bytes_with(bytes, |bytes| { - BeaconBlock::from_ssz_bytes_for_fork(bytes, fork_name) - }) - } - /// SSZ decode which attempts to decode all variants (slow). pub fn any_from_ssz_bytes(bytes: &[u8]) -> Result { Self::from_ssz_bytes_with(bytes, BeaconBlock::any_from_ssz_bytes) diff --git a/testing/ef_tests/src/cases/operations.rs b/testing/ef_tests/src/cases/operations.rs index adb5bee7681..7178edb151c 100644 --- a/testing/ef_tests/src/cases/operations.rs +++ b/testing/ef_tests/src/cases/operations.rs @@ -25,9 +25,9 @@ use std::fmt::Debug; use types::{ Attestation, AttesterSlashing, BeaconBlock, BeaconBlockBody, BeaconBlockBodyBellatrix, BeaconBlockBodyCapella, BeaconBlockBodyDeneb, BeaconBlockBodyElectra, BeaconState, - BlindedPayload, ConsolidationRequest, Deposit, DepositRequest, ExecutionPayload, FullPayload, - ProposerSlashing, SignedBlsToExecutionChange, SignedVoluntaryExit, SyncAggregate, - WithdrawalRequest, + BlindedPayload, ConsolidationRequest, Deposit, DepositRequest, ExecutionPayload, + ForkVersionDecode, FullPayload, ProposerSlashing, SignedBlsToExecutionChange, + SignedVoluntaryExit, SyncAggregate, WithdrawalRequest, }; #[derive(Debug, Clone, Default, Deserialize)] @@ -398,7 +398,7 @@ impl Operation for WithdrawalsPayload { fn decode(path: &Path, fork_name: ForkName, _spec: &ChainSpec) -> Result { ssz_decode_file_with(path, |bytes| { - ExecutionPayload::from_ssz_bytes(bytes, fork_name) + ExecutionPayload::from_ssz_bytes_by_fork(bytes, fork_name) }) .map(|payload| WithdrawalsPayload { payload: payload.into(),