diff --git a/beacon_node/beacon_chain/src/beacon_chain.rs b/beacon_node/beacon_chain/src/beacon_chain.rs index 047610a4a7..723d64489e 100644 --- a/beacon_node/beacon_chain/src/beacon_chain.rs +++ b/beacon_node/beacon_chain/src/beacon_chain.rs @@ -130,7 +130,7 @@ use store::{ }; use task_executor::{RayonPoolType, ShutdownReason, TaskExecutor}; use tokio_stream::Stream; -use tracing::{Span, debug, debug_span, error, info, info_span, instrument, trace, warn}; +use tracing::{debug, debug_span, error, info, info_span, instrument, trace, warn}; use tree_hash::TreeHash; use types::data::{ColumnIndex, FixedBlobSidecarList}; use types::execution::BlockProductionVersion; @@ -2784,6 +2784,7 @@ impl BeaconChain { /// or already-known). /// /// This method is potentially long-running and should not run on the core executor. + #[instrument(skip_all, level = "debug")] pub fn filter_chain_segment( self: &Arc, chain_segment: Vec>, @@ -2919,12 +2920,8 @@ impl BeaconChain { // Filter uninteresting blocks from the chain segment in a blocking task. let chain = self.clone(); - let filter_chain_segment = debug_span!("filter_chain_segment"); let filtered_chain_segment_future = self.spawn_blocking_handle( - move || { - let _guard = filter_chain_segment.enter(); - chain.filter_chain_segment(chain_segment) - }, + move || chain.filter_chain_segment(chain_segment), "filter_chain_segment", ); let mut filtered_chain_segment = match filtered_chain_segment_future.await { @@ -2955,12 +2952,8 @@ impl BeaconChain { std::mem::swap(&mut blocks, &mut filtered_chain_segment); let chain = self.clone(); - let current_span = Span::current(); let signature_verification_future = self.spawn_blocking_handle( - move || { - let _guard = current_span.enter(); - signature_verify_chain_segment(blocks, &chain) - }, + move || signature_verify_chain_segment(blocks, &chain), "signature_verify_chain_segment", ); @@ -3083,12 +3076,10 @@ impl BeaconChain { block: Arc>, ) -> Result, BlockError> { let chain = self.clone(); - let span = Span::current(); self.task_executor .clone() .spawn_blocking_handle( move || { - let _guard = span.enter(); let slot = block.slot(); let graffiti_string = block.message().body().graffiti().as_utf8_lossy(); @@ -3396,11 +3387,9 @@ impl BeaconChain { let data_availability_checker = self.data_availability_checker.clone(); - let current_span = Span::current(); let result = self .task_executor .spawn_blocking_with_rayon_async(RayonPoolType::HighPriority, move || { - let _guard = current_span.enter(); data_availability_checker.reconstruct_data_columns(&block_root) }) .await @@ -3865,7 +3854,7 @@ impl BeaconChain { consensus_context, } = import_data; - // Record the time at which this block's blobs became available. + // Record the time at which this block's blobs/data columns became available. if let Some(blobs_available) = block.blobs_available_timestamp() { self.block_times_cache.write().set_time_blob_observed( block_root, @@ -3874,16 +3863,10 @@ impl BeaconChain { ); } - // TODO(das) record custody column available timestamp - let block_root = { - // Capture the current span before moving into the blocking task - let current_span = tracing::Span::current(); let chain = self.clone(); self.spawn_blocking_handle( move || { - // Enter the captured span in the blocking thread - let _guard = current_span.enter(); chain.import_block( block, block_root, @@ -4594,15 +4577,10 @@ impl BeaconChain { // // Load the parent state from disk. let chain = self.clone(); - let span = Span::current(); let (state, state_root_opt) = self .task_executor .spawn_blocking_handle( - move || { - let _guard = - debug_span!(parent: span, "load_state_for_block_production").entered(); - chain.load_state_for_block_production(slot) - }, + move || chain.load_state_for_block_production(slot), "load_state_for_block_production", ) .ok_or(BlockProductionError::ShuttingDown)? @@ -4918,8 +4896,8 @@ impl BeaconChain { return Err(Box::new(DoNotReOrg::NotProposing.into())); } - // TODO(gloas): reorg weight logic needs updating for GLOAS. For now use - // total weight which is correct for pre-GLOAS and conservative for post-GLOAS. + // TODO(gloas): reorg weight logic needs updating for Gloas. For now use + // total weight which is correct for pre-Gloas and conservative for post-Gloas. let head_weight = info.head_node.weight(); let parent_weight = info.parent_node.weight(); @@ -5036,13 +5014,10 @@ impl BeaconChain { .graffiti_calculator .get_graffiti(graffiti_settings) .await; - let span = Span::current(); let mut partial_beacon_block = self .task_executor .spawn_blocking_handle( move || { - let _guard = - debug_span!(parent: span, "produce_partial_beacon_block").entered(); chain.produce_partial_beacon_block( state, state_root_opt, @@ -5078,14 +5053,10 @@ impl BeaconChain { match block_contents_type { BlockProposalContentsType::Full(block_contents) => { let chain = self.clone(); - let span = Span::current(); let beacon_block_response = self .task_executor .spawn_blocking_handle( move || { - let _guard = - debug_span!(parent: span, "complete_partial_beacon_block") - .entered(); chain.complete_partial_beacon_block( partial_beacon_block, Some(block_contents), @@ -5102,14 +5073,10 @@ impl BeaconChain { } BlockProposalContentsType::Blinded(block_contents) => { let chain = self.clone(); - let span = Span::current(); let beacon_block_response = self .task_executor .spawn_blocking_handle( move || { - let _guard = - debug_span!(parent: span, "complete_partial_beacon_block") - .entered(); chain.complete_partial_beacon_block( partial_beacon_block, Some(block_contents), @@ -5127,13 +5094,10 @@ impl BeaconChain { } } else { let chain = self.clone(); - let span = Span::current(); let beacon_block_response = self .task_executor .spawn_blocking_handle( move || { - let _guard = - debug_span!(parent: span, "complete_partial_beacon_block").entered(); chain.complete_partial_beacon_block( partial_beacon_block, None, @@ -5151,6 +5115,7 @@ impl BeaconChain { } #[allow(clippy::too_many_arguments)] + #[instrument(skip_all, level = "debug")] fn produce_partial_beacon_block( self: &Arc, mut state: BeaconState, @@ -5395,6 +5360,7 @@ impl BeaconChain { }) } + #[instrument(skip_all, level = "debug")] fn complete_partial_beacon_block>( &self, partial_beacon_block: PartialBeaconBlock, diff --git a/beacon_node/beacon_chain/src/block_production/gloas.rs b/beacon_node/beacon_chain/src/block_production/gloas.rs index 2fc4fb51f7..51caf63b7a 100644 --- a/beacon_node/beacon_chain/src/block_production/gloas.rs +++ b/beacon_node/beacon_chain/src/block_production/gloas.rs @@ -19,7 +19,7 @@ use state_processing::{ }; use state_processing::{VerifyOperation, state_advance::complete_state_advance}; use task_executor::JoinHandle; -use tracing::{Instrument, Span, debug, debug_span, error, instrument, trace, warn}; +use tracing::{Instrument, debug, debug_span, error, instrument, trace, warn}; use tree_hash::TreeHash; use types::consts::gloas::BUILDER_INDEX_SELF_BUILD; use types::{ @@ -87,15 +87,10 @@ impl BeaconChain { // // Load the parent state from disk. let chain = self.clone(); - let span = Span::current(); let (state, state_root_opt) = self .task_executor .spawn_blocking_handle( - move || { - let _guard = - debug_span!(parent: span, "load_state_for_block_production").entered(); - chain.load_state_for_block_production(slot) - }, + move || chain.load_state_for_block_production(slot), "load_state_for_block_production", ) .ok_or(BlockProductionError::ShuttingDown)? @@ -135,13 +130,10 @@ impl BeaconChain { .graffiti_calculator .get_graffiti(graffiti_settings) .await; - let span = Span::current(); let (partial_beacon_block, state) = self .task_executor .spawn_blocking_handle( move || { - let _guard = - debug_span!(parent: span, "produce_partial_beacon_block_gloas").entered(); chain.produce_partial_beacon_block_gloas( state, state_root_opt, @@ -175,12 +167,9 @@ impl BeaconChain { // // Complete the block with the execution payload bid. let chain = self.clone(); - let span = Span::current(); self.task_executor .spawn_blocking_handle( move || { - let _guard = - debug_span!(parent: span, "complete_partial_beacon_block_gloas").entered(); chain.complete_partial_beacon_block_gloas( partial_beacon_block, execution_payload_bid, @@ -198,6 +187,7 @@ impl BeaconChain { #[allow(clippy::too_many_arguments)] #[allow(clippy::type_complexity)] + #[instrument(skip_all, level = "debug")] fn produce_partial_beacon_block_gloas( self: &Arc, mut state: BeaconState, @@ -432,6 +422,7 @@ impl BeaconChain { /// - `pending_state` is the state post block application (prior to payload application) /// - `block_value` is the consensus-layer rewards for `block` #[allow(clippy::type_complexity)] + #[instrument(skip_all, level = "debug")] fn complete_partial_beacon_block_gloas( &self, partial_beacon_block: PartialBeaconBlock, diff --git a/beacon_node/beacon_chain/src/block_production/mod.rs b/beacon_node/beacon_chain/src/block_production/mod.rs index 60cc3e919a..bf42923cbe 100644 --- a/beacon_node/beacon_chain/src/block_production/mod.rs +++ b/beacon_node/beacon_chain/src/block_production/mod.rs @@ -15,6 +15,7 @@ mod gloas; impl BeaconChain { /// Load a beacon state from the database for block production. This is a long-running process /// that should not be performed in an `async` context. + #[instrument(skip_all, level = "debug")] pub(crate) fn load_state_for_block_production( self: &Arc, slot: Slot, diff --git a/beacon_node/beacon_chain/src/block_verification.rs b/beacon_node/beacon_chain/src/block_verification.rs index 4da2562364..5e6d37cd4a 100644 --- a/beacon_node/beacon_chain/src/block_verification.rs +++ b/beacon_node/beacon_chain/src/block_verification.rs @@ -322,7 +322,7 @@ pub enum BlockError { bid_parent_root: Hash256, block_parent_root: Hash256, }, - /// The parent block is known but its execution payload envelope has not been received yet. + /// The child block is known but its parent execution payload envelope has not been received yet. /// /// ## Peer scoring /// @@ -331,6 +331,11 @@ pub enum BlockError { ParentEnvelopeUnknown { parent_root: Hash256 }, /// An error occurred while processing the execution payload envelope during range sync. EnvelopeError(Box), + + PayloadEnvelopeError { + e: Box, + penalize_peer: bool, + }, } /// Which specific signature(s) are invalid in a SignedBeaconBlock @@ -497,6 +502,36 @@ impl From for BlockError { } } +impl From for BlockError { + fn from(e: EnvelopeError) -> Self { + let penalize_peer = match &e { + // REJECT per spec: peer sent invalid envelope data + EnvelopeError::BadSignature + | EnvelopeError::BuilderIndexMismatch { .. } + | EnvelopeError::BlockHashMismatch { .. } + | EnvelopeError::SlotMismatch { .. } + | EnvelopeError::IncorrectBlockProposer { .. } => true, + // IGNORE per spec: not the peer's fault + EnvelopeError::BlockRootUnknown { .. } + | EnvelopeError::PriorToFinalization { .. } + | EnvelopeError::UnknownValidator { .. } => false, + // Internal errors: not the peer's fault + EnvelopeError::BeaconChainError(_) + | EnvelopeError::BeaconStateError(_) + | EnvelopeError::BlockProcessingError(_) + | EnvelopeError::EnvelopeProcessingError(_) + | EnvelopeError::ExecutionPayloadError(_) + | EnvelopeError::BlockError(_) + | EnvelopeError::InternalError(_) + | EnvelopeError::OptimisticSyncNotSupported { .. } => false, + }; + BlockError::PayloadEnvelopeError { + e: Box::new(e), + penalize_peer, + } + } +} + /// Stores information about verifying a payload against an execution engine. #[derive(Debug, PartialEq, Clone, Encode, Decode)] pub struct PayloadVerificationOutcome { @@ -1731,7 +1766,7 @@ impl ExecutionPendingBlock { indexed_payload_attestation, AttestationFromBlock::True, &ptc.0, - ) && !matches!(e, ForkChoiceError::InvalidAttestation(_)) + ) && !matches!(e, ForkChoiceError::InvalidPayloadAttestation(_)) { return Err(BlockError::BeaconChainError(Box::new(e.into()))); } @@ -2004,8 +2039,9 @@ fn load_parent>( } else if let Ok(parent_bid_block_hash) = parent_block.payload_bid_block_hash() && block.as_block().is_parent_block_full(parent_bid_block_hash) { - // Post-Gloas Full block case. - // TODO(gloas): loading the envelope here is not very efficient + // If the parent's execution payload envelope hasn't arrived yet, + // return an unknown parent error so the block gets sent to the + // reprocess queue. let envelope = chain .store .get_payload_envelope(&root)? diff --git a/beacon_node/beacon_chain/src/canonical_head.rs b/beacon_node/beacon_chain/src/canonical_head.rs index bfd3d79512..cd53d0ef7c 100644 --- a/beacon_node/beacon_chain/src/canonical_head.rs +++ b/beacon_node/beacon_chain/src/canonical_head.rs @@ -58,7 +58,6 @@ use store::{ Error as StoreError, KeyValueStore, KeyValueStoreOp, StoreConfig, iter::StateRootsIterator, }; use task_executor::{JoinHandle, ShutdownReason}; -use tracing::info_span; use tracing::{debug, error, info, instrument, warn}; use types::*; @@ -539,22 +538,15 @@ impl BeaconChain { /// such a case it's critical that the `BeaconChain` keeps importing blocks so that the /// situation can be rectified. We avoid returning an error here so that calling functions /// can't abort block import because an error is returned here. + #[instrument(name = "lh_recompute_head_at_slot", skip(self), level = "info", fields(slot = %current_slot))] pub async fn recompute_head_at_slot(self: &Arc, current_slot: Slot) { - let span = info_span!( - "lh_recompute_head_at_slot", - slot = %current_slot - ); - metrics::inc_counter(&metrics::FORK_CHOICE_REQUESTS); let _timer = metrics::start_timer(&metrics::FORK_CHOICE_TIMES); let chain = self.clone(); match self .spawn_blocking_handle( - move || { - let _guard = span.enter(); - chain.recompute_head_at_slot_internal(current_slot) - }, + move || chain.recompute_head_at_slot_internal(current_slot), "recompute_head_internal", ) .await @@ -1392,8 +1384,8 @@ fn observe_head_block_delays( .as_millis() as i64, ); - // The time from the start of the slot when all blobs have been observed. Technically this - // is the time we last saw a blob related to this block/slot. + // The time from the start of the slot when all blobs/data columns have been observed. Technically this + // is the time we last saw a blob/data column related to this block/slot. metrics::set_gauge( &metrics::BEACON_BLOB_DELAY_ALL_OBSERVED_SLOT_START, block_delays 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 f26e1d7ad4..3f6c6baa5b 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 @@ -282,8 +282,11 @@ impl PendingComponents { .flatten() .map(|blob| blob.seen_timestamp()) .max(), - // TODO(das): To be fixed with https://github.com/sigp/lighthouse/pull/6850 - AvailableBlockData::DataColumns(_) => None, + AvailableBlockData::DataColumns(_) => self + .verified_data_columns + .iter() + .map(|data_column| data_column.seen_timestamp()) + .max(), }; let AvailabilityPendingExecutedBlock { diff --git a/beacon_node/beacon_chain/src/data_column_verification.rs b/beacon_node/beacon_chain/src/data_column_verification.rs index dde9fad342..f47de01ddc 100644 --- a/beacon_node/beacon_chain/src/data_column_verification.rs +++ b/beacon_node/beacon_chain/src/data_column_verification.rs @@ -5,6 +5,7 @@ use crate::kzg_utils::{reconstruct_data_columns, validate_data_columns}; use crate::observed_data_sidecars::{ Error as ObservedDataSidecarsError, ObservationKey, ObservationStrategy, Observe, }; +use crate::validator_monitor::timestamp_now; use crate::{BeaconChain, BeaconChainError, BeaconChainTypes, metrics}; use educe::Educe; use fork_choice::ProtoBlock; @@ -16,6 +17,7 @@ use ssz_types::VariableList; use std::iter; use std::marker::PhantomData; use std::sync::Arc; +use std::time::Duration; use tracing::{debug, instrument}; use types::data::ColumnIndex; use types::{ @@ -320,25 +322,34 @@ impl GossipVerifiedDataColumn #[ssz(struct_behaviour = "transparent")] pub struct KzgVerifiedDataColumn { data: Arc>, + #[ssz(skip_serializing, skip_deserializing)] + seen_timestamp: Duration, } impl KzgVerifiedDataColumn { pub fn new( data_column: Arc>, kzg: &Kzg, + seen_timestamp: Duration, ) -> Result, KzgError)> { - verify_kzg_for_data_column(data_column, kzg) + verify_kzg_for_data_column(data_column, kzg, seen_timestamp) } /// Mark a data column as KZG verified. Caller must ONLY use this on columns constructed /// from EL blobs. pub fn from_execution_verified(data_column: Arc>) -> Self { - Self { data: data_column } + Self { + data: data_column, + seen_timestamp: timestamp_now(), + } } /// Create a `KzgVerifiedDataColumn` from `DataColumnSidecar` for testing ONLY. pub(crate) fn __new_for_testing(data_column: Arc>) -> Self { - Self { data: data_column } + Self { + data: data_column, + seen_timestamp: timestamp_now(), + } } pub fn from_batch_with_scoring( @@ -348,7 +359,10 @@ impl KzgVerifiedDataColumn { verify_kzg_for_data_column_list(data_columns.iter(), kzg)?; Ok(data_columns .into_iter() - .map(|column| Self { data: column }) + .map(|column| Self { + data: column, + seen_timestamp: timestamp_now(), + }) .collect()) } @@ -407,6 +421,8 @@ impl CustodyDataColumn { #[ssz(struct_behaviour = "transparent")] pub struct KzgVerifiedCustodyDataColumn { data: Arc>, + #[ssz(skip_serializing, skip_deserializing)] + seen_timestamp: Duration, } impl KzgVerifiedCustodyDataColumn { @@ -414,6 +430,7 @@ impl KzgVerifiedCustodyDataColumn { /// include this column pub fn from_asserted_custody(kzg_verified: KzgVerifiedDataColumn) -> Self { Self { + seen_timestamp: kzg_verified.seen_timestamp, data: kzg_verified.to_data_column(), } } @@ -422,10 +439,12 @@ impl KzgVerifiedCustodyDataColumn { pub fn new( data_column: CustodyDataColumn, kzg: &Kzg, + seen_timestamp: Duration, ) -> Result, KzgError)> { - verify_kzg_for_data_column(data_column.clone_arc(), kzg)?; + verify_kzg_for_data_column(data_column.clone_arc(), kzg, seen_timestamp)?; Ok(Self { data: data_column.data, + seen_timestamp, }) } @@ -443,10 +462,15 @@ impl KzgVerifiedCustodyDataColumn { spec, )?; + let seen_timestamp = timestamp_now(); + Ok(all_data_columns .into_iter() .map(|data| { - KzgVerifiedCustodyDataColumn::from_asserted_custody(KzgVerifiedDataColumn { data }) + KzgVerifiedCustodyDataColumn::from_asserted_custody(KzgVerifiedDataColumn { + data, + seen_timestamp, + }) }) .collect::>()) } @@ -464,6 +488,10 @@ impl KzgVerifiedCustodyDataColumn { pub fn index(&self) -> ColumnIndex { *self.data.index() } + + pub fn seen_timestamp(&self) -> Duration { + self.seen_timestamp + } } /// Complete kzg verification for a `DataColumnSidecar`. @@ -473,10 +501,14 @@ impl KzgVerifiedCustodyDataColumn { pub fn verify_kzg_for_data_column( data_column: Arc>, kzg: &Kzg, + seen_timestamp: Duration, ) -> Result, (Option, KzgError)> { let _timer = metrics::start_timer(&metrics::KZG_VERIFICATION_DATA_COLUMN_SINGLE_TIMES); validate_data_columns(kzg, iter::once(&data_column))?; - Ok(KzgVerifiedDataColumn { data: data_column }) + Ok(KzgVerifiedDataColumn { + data: data_column, + seen_timestamp, + }) } /// Complete kzg verification for a list of `DataColumnSidecar`s. @@ -538,8 +570,9 @@ pub fn validate_data_column_sidecar_for_gossip_fulu( let spec = chain_adapter.spec().clone(); let chain_adapter_cloned = chain_adapter.clone(); let custody_columns_indices = custody_columns_indices.to_vec(); - let current_span = Span::current(); chain_adapter .executor() .spawn_blocking_handle( move || { - let _guard = current_span.enter(); let mut timer = metrics::start_timer_vec( &metrics::DATA_COLUMN_SIDECAR_COMPUTATION, &[&blobs.len().to_string()], diff --git a/beacon_node/beacon_chain/src/payload_envelope_verification/gossip_verified_envelope.rs b/beacon_node/beacon_chain/src/payload_envelope_verification/gossip_verified_envelope.rs index 9a4ed2d044..4d40a29332 100644 --- a/beacon_node/beacon_chain/src/payload_envelope_verification/gossip_verified_envelope.rs +++ b/beacon_node/beacon_chain/src/payload_envelope_verification/gossip_verified_envelope.rs @@ -4,7 +4,7 @@ use educe::Educe; use eth2::types::{EventKind, SseExecutionPayloadGossip}; use parking_lot::{Mutex, RwLock}; use store::DatabaseBlock; -use tracing::{Span, debug}; +use tracing::debug; use types::{ ChainSpec, EthSpec, ExecutionPayloadBid, ExecutionPayloadEnvelope, Hash256, SignedBeaconBlock, SignedExecutionPayloadEnvelope, Slot, consts::gloas::BUILDER_INDEX_SELF_BUILD, @@ -270,12 +270,10 @@ impl BeaconChain { envelope: Arc>, ) -> Result, EnvelopeError> { let chain = self.clone(); - let span = Span::current(); self.task_executor .clone() .spawn_blocking_handle( move || { - let _guard = span.enter(); let slot = envelope.slot(); let beacon_block_root = envelope.message.beacon_block_root; diff --git a/beacon_node/beacon_chain/src/payload_envelope_verification/import.rs b/beacon_node/beacon_chain/src/payload_envelope_verification/import.rs index d07b7cde73..c09f5f9b4d 100644 --- a/beacon_node/beacon_chain/src/payload_envelope_verification/import.rs +++ b/beacon_node/beacon_chain/src/payload_envelope_verification/import.rs @@ -1,18 +1,16 @@ use std::sync::Arc; use std::time::Duration; -use eth2::types::{EventKind, SseExecutionPayloadAvailable}; +use eth2::types::{EventKind, SseExecutionPayload}; use fork_choice::PayloadVerificationStatus; use slot_clock::SlotClock; -use store::StoreOp; -use tracing::{debug, error, info, info_span, instrument, warn}; -use types::{BeaconState, BlockImportSource, EthSpec, Hash256, Slot}; - use state_processing::{ VerifySignatures, envelope_processing::{VerifyStateRoot, process_execution_payload_envelope}, }; -use store::DatabaseBlock; +use store::{DatabaseBlock, StoreOp}; +use tracing::{debug, error, info, info_span, instrument, warn}; +use types::{BeaconState, BlockImportSource, EthSpec, Hash256, SignedExecutionPayloadEnvelope}; use super::{ AvailableEnvelope, AvailableExecutedEnvelope, EnvelopeError, EnvelopeImportData, @@ -296,6 +294,16 @@ impl BeaconChain { .map_err(BeaconChainError::TokioJoin)? .ok_or(BeaconChainError::RuntimeShutdown)??; + // TODO(gloas): optimistic sync is not supported for Gloas, maybe we could re-add it + if payload_verification_outcome + .payload_verification_status + .is_optimistic() + { + return Err(EnvelopeError::OptimisticSyncNotSupported { + block_root: import_data.block_root, + }); + } + Ok(ExecutedEnvelope::new( signed_envelope, import_data, @@ -320,13 +328,9 @@ impl BeaconChain { } = import_data; let block_root = { - // Capture the current span before moving into the blocking task - let current_span = tracing::Span::current(); let chain = self.clone(); self.spawn_blocking_handle( move || { - // Enter the captured span in the blocking thread - let _guard = current_span.enter(); chain.import_execution_payload_envelope( envelope, block_root, @@ -354,7 +358,7 @@ impl BeaconChain { signed_envelope: AvailableEnvelope, block_root: Hash256, state: BeaconState, - _payload_verification_status: PayloadVerificationStatus, + payload_verification_status: PayloadVerificationStatus, ) -> Result { // Everything in this initial section is on the hot path for processing the envelope. // Take an upgradable read lock on fork choice so we can check if this block has already @@ -372,9 +376,10 @@ impl BeaconChain { // avoiding taking other locks whilst holding this lock. let mut fork_choice = parking_lot::RwLockUpgradableReadGuard::upgrade(fork_choice_reader); - // Update the node's payload_status from PENDING to FULL in fork choice. + // Update the block's payload to received in fork choice, which creates the `Full` virtual + // node which can be eligible for head. fork_choice - .on_execution_payload(block_root) + .on_valid_payload_envelope_received(block_root) .map_err(|e| EnvelopeError::InternalError(format!("{e:?}")))?; // TODO(gloas) emit SSE event if the payload became the new head payload @@ -443,8 +448,9 @@ impl BeaconChain { metrics::stop_timer(db_write_timer); self.import_envelope_update_metrics_and_events( + signed_envelope, block_root, - signed_envelope.slot(), + payload_verification_status, envelope_time_imported, ); @@ -453,10 +459,12 @@ impl BeaconChain { fn import_envelope_update_metrics_and_events( &self, + signed_envelope: Arc>, block_root: Hash256, - envelope_slot: Slot, + payload_verification_status: PayloadVerificationStatus, envelope_time_imported: Duration, ) { + let envelope_slot = signed_envelope.slot(); let envelope_delay_total = get_slot_delay_ms(envelope_time_imported, envelope_slot, &self.slot_clock); @@ -476,14 +484,16 @@ impl BeaconChain { } if let Some(event_handler) = self.event_handler.as_ref() - && event_handler.has_execution_payload_available_subscribers() + && event_handler.has_execution_payload_subscribers() { - event_handler.register(EventKind::ExecutionPayloadAvailable( - SseExecutionPayloadAvailable { - slot: envelope_slot, - block_root, - }, - )); + event_handler.register(EventKind::ExecutionPayload(SseExecutionPayload { + slot: envelope_slot, + builder_index: signed_envelope.message.builder_index, + block_hash: signed_envelope.block_hash(), + block_root, + state_root: signed_envelope.message.state_root, + execution_optimistic: payload_verification_status.is_optimistic(), + })); } } } diff --git a/beacon_node/beacon_chain/src/payload_envelope_verification/mod.rs b/beacon_node/beacon_chain/src/payload_envelope_verification/mod.rs index 1e7d00fce3..1581ab325c 100644 --- a/beacon_node/beacon_chain/src/payload_envelope_verification/mod.rs +++ b/beacon_node/beacon_chain/src/payload_envelope_verification/mod.rs @@ -202,6 +202,8 @@ pub enum EnvelopeError { payload_slot: Slot, latest_finalized_slot: Slot, }, + /// Optimistic sync is not supported for Gloas payload envelopes. + OptimisticSyncNotSupported { block_root: Hash256 }, /// Some Beacon Chain Error BeaconChainError(Arc), /// Some Beacon State error diff --git a/beacon_node/beacon_chain/src/persisted_fork_choice.rs b/beacon_node/beacon_chain/src/persisted_fork_choice.rs index 592ea9ecd7..8edccbbe98 100644 --- a/beacon_node/beacon_chain/src/persisted_fork_choice.rs +++ b/beacon_node/beacon_chain/src/persisted_fork_choice.rs @@ -22,27 +22,38 @@ pub struct PersistedForkChoice { pub fork_choice_store: PersistedForkChoiceStoreV28, } -macro_rules! impl_store_item { - ($type:ty) => { - impl store::StoreItem for $type { - fn db_column() -> DBColumn { - DBColumn::ForkChoice - } +impl PersistedForkChoiceV28 { + pub fn from_bytes(bytes: &[u8], store_config: &StoreConfig) -> Result { + let decompressed_bytes = store_config + .decompress_bytes(bytes) + .map_err(Error::Compression)?; + Self::from_ssz_bytes(&decompressed_bytes).map_err(Into::into) + } - fn as_store_bytes(&self) -> Vec { - self.as_ssz_bytes() - } + pub fn as_bytes(&self, store_config: &StoreConfig) -> Result, Error> { + let encode_timer = metrics::start_timer(&metrics::FORK_CHOICE_ENCODE_TIMES); + let ssz_bytes = self.as_ssz_bytes(); + drop(encode_timer); - fn from_store_bytes(bytes: &[u8]) -> std::result::Result { - Self::from_ssz_bytes(bytes).map_err(Into::into) - } - } - }; + let _compress_timer = metrics::start_timer(&metrics::FORK_CHOICE_COMPRESS_TIMES); + store_config + .compress_bytes(&ssz_bytes) + .map_err(Error::Compression) + } + + pub fn as_kv_store_op( + &self, + key: Hash256, + store_config: &StoreConfig, + ) -> Result { + Ok(KeyValueStoreOp::PutKeyValue( + DBColumn::ForkChoice, + key.as_slice().to_vec(), + self.as_bytes(store_config)?, + )) + } } -impl_store_item!(PersistedForkChoiceV28); -impl_store_item!(PersistedForkChoiceV29); - impl PersistedForkChoiceV29 { pub fn from_bytes(bytes: &[u8], store_config: &StoreConfig) -> Result { let decompressed_bytes = store_config @@ -83,3 +94,12 @@ impl From for PersistedForkChoiceV29 { } } } + +impl From for PersistedForkChoiceV28 { + fn from(v29: PersistedForkChoiceV29) -> Self { + Self { + fork_choice_v28: v29.fork_choice.into(), + fork_choice_store: v29.fork_choice_store, + } + } +} diff --git a/beacon_node/beacon_chain/src/schema_change.rs b/beacon_node/beacon_chain/src/schema_change.rs index fa2ab70d21..841f28e37d 100644 --- a/beacon_node/beacon_chain/src/schema_change.rs +++ b/beacon_node/beacon_chain/src/schema_change.rs @@ -22,13 +22,13 @@ pub fn migrate_schema( (_, _) if from == to && to == CURRENT_SCHEMA_VERSION => Ok(()), // Upgrade from v28 to v29. (SchemaVersion(28), SchemaVersion(29)) => { - upgrade_to_v29::(&db)?; - db.store_schema_version_atomically(to, vec![]) + let ops = upgrade_to_v29::(&db)?; + db.store_schema_version_atomically(to, ops) } // Downgrade from v29 to v28. (SchemaVersion(29), SchemaVersion(28)) => { - downgrade_from_v29::(&db)?; - db.store_schema_version_atomically(to, vec![]) + let ops = downgrade_from_v29::(&db)?; + db.store_schema_version_atomically(to, ops) } // Anything else is an error. (_, _) => Err(HotColdDBError::UnsupportedSchemaVersion { diff --git a/beacon_node/beacon_chain/src/schema_change/migration_schema_v29.rs b/beacon_node/beacon_chain/src/schema_change/migration_schema_v29.rs index 6c82e8a737..77d4be3443 100644 --- a/beacon_node/beacon_chain/src/schema_change/migration_schema_v29.rs +++ b/beacon_node/beacon_chain/src/schema_change/migration_schema_v29.rs @@ -1,12 +1,10 @@ -use crate::beacon_chain::BeaconChainTypes; +use crate::beacon_chain::{BeaconChainTypes, FORK_CHOICE_DB_KEY}; use crate::persisted_fork_choice::{PersistedForkChoiceV28, PersistedForkChoiceV29}; -use ssz::Decode; +use std::collections::HashMap; use store::hot_cold_store::HotColdDB; use store::{DBColumn, Error as StoreError, KeyValueStore, KeyValueStoreOp}; -use types::{EthSpec, Hash256}; - -/// The key used to store the fork choice in the database. -const FORK_CHOICE_DB_KEY: Hash256 = Hash256::ZERO; +use tracing::warn; +use types::EthSpec; /// Upgrade from schema v28 to v29. /// @@ -14,24 +12,25 @@ const FORK_CHOICE_DB_KEY: Hash256 = Hash256::ZERO; /// virtual tree walk). /// - Fails if the persisted fork choice contains any V17 (pre-Gloas) proto /// nodes at or after the Gloas fork slot. +/// +/// Returns a list of store ops to be applied atomically with the schema version write. pub fn upgrade_to_v29( db: &HotColdDB, -) -> Result<(), StoreError> { +) -> Result, StoreError> { let gloas_fork_slot = db .spec .gloas_fork_epoch .map(|epoch| epoch.start_slot(T::EthSpec::slots_per_epoch())); - // Load the persisted fork choice (v28 format, uncompressed SSZ). + // Load the persisted fork choice (v28 format). let Some(fc_bytes) = db .hot_db .get_bytes(DBColumn::ForkChoice, FORK_CHOICE_DB_KEY.as_slice())? else { - return Ok(()); + return Ok(vec![]); }; - let mut persisted_v28 = - PersistedForkChoiceV28::from_ssz_bytes(&fc_bytes).map_err(StoreError::SszDecodeError)?; + let persisted_v28 = PersistedForkChoiceV28::from_bytes(&fc_bytes, db.get_config())?; // Check for V17 nodes at/after the Gloas fork slot. if let Some(gloas_fork_slot) = gloas_fork_slot { @@ -52,39 +51,71 @@ pub fn upgrade_to_v29( } } - // Clear best_child/best_descendant — replaced by the virtual tree walk. - for node in &mut persisted_v28.fork_choice_v28.proto_array_v28.nodes { - node.best_child = None; - node.best_descendant = None; + // Read the previous proposer boost before converting to V29 (V29 no longer stores it). + let previous_proposer_boost = persisted_v28 + .fork_choice_v28 + .proto_array_v28 + .previous_proposer_boost; + + // Convert to v29. + let mut persisted_v29 = PersistedForkChoiceV29::from(persisted_v28); + + // Subtract the proposer boost from the boosted node and all its ancestors. + // + // In the V28 schema, `apply_score_changes` baked the proposer boost directly into node + // weights and back-propagated it up the parent chain. In V29, the boost is computed + // on-the-fly during the virtual tree walk. If we don't subtract the baked-in boost here, + // it will be double-counted after the upgrade. + if !previous_proposer_boost.root.is_zero() && previous_proposer_boost.score > 0 { + let score = previous_proposer_boost.score; + let indices: HashMap<_, _> = persisted_v29 + .fork_choice + .proto_array + .indices + .iter() + .cloned() + .collect(); + + if let Some(node_index) = indices.get(&previous_proposer_boost.root).copied() { + let nodes = &mut persisted_v29.fork_choice.proto_array.nodes; + let mut current = Some(node_index); + while let Some(idx) = current { + if let Some(node) = nodes.get_mut(idx) { + *node.weight_mut() = node.weight().saturating_sub(score); + current = node.parent(); + } else { + break; + } + } + } else { + warn!( + root = ?previous_proposer_boost.root, + "Proposer boost node missing from fork choice" + ); + } } - // Convert to v29 and write back. - let persisted_v29 = PersistedForkChoiceV29::from(persisted_v28); - let fc_bytes = persisted_v29 - .as_bytes(db.get_config()) - .map_err(|e| StoreError::MigrationError(format!("failed to encode v29: {:?}", e)))?; - db.hot_db.do_atomically(vec![KeyValueStoreOp::PutKeyValue( - DBColumn::ForkChoice, - FORK_CHOICE_DB_KEY.as_slice().to_vec(), - fc_bytes, - )])?; - - Ok(()) + Ok(vec![ + persisted_v29.as_kv_store_op(FORK_CHOICE_DB_KEY, db.get_config())?, + ]) } -/// Downgrade from schema v29 to v28 (no-op). +/// Downgrade from schema v29 to v28. /// +/// Converts the persisted fork choice from V29 format back to V28. /// Fails if the persisted fork choice contains any V29 proto nodes, as these contain /// payload-specific fields that cannot be losslessly converted back to V17 format. +/// +/// Returns a list of store ops to be applied atomically with the schema version write. pub fn downgrade_from_v29( db: &HotColdDB, -) -> Result<(), StoreError> { +) -> Result, StoreError> { // Load the persisted fork choice (v29 format, compressed). let Some(fc_bytes) = db .hot_db .get_bytes(DBColumn::ForkChoice, FORK_CHOICE_DB_KEY.as_slice())? else { - return Ok(()); + return Ok(vec![]); }; let persisted_v29 = @@ -111,5 +142,10 @@ pub fn downgrade_from_v29( )); } - Ok(()) + // Convert to v28 and encode. + let persisted_v28 = PersistedForkChoiceV28::from(persisted_v29); + + Ok(vec![ + persisted_v28.as_kv_store_op(FORK_CHOICE_DB_KEY, db.get_config())?, + ]) } diff --git a/beacon_node/beacon_chain/tests/payload_invalidation.rs b/beacon_node/beacon_chain/tests/payload_invalidation.rs index 13672bbb63..947024e8c2 100644 --- a/beacon_node/beacon_chain/tests/payload_invalidation.rs +++ b/beacon_node/beacon_chain/tests/payload_invalidation.rs @@ -1438,7 +1438,7 @@ async fn weights_after_resetting_optimistic_status() { .canonical_head .fork_choice_write_lock() .proto_array_mut() - .set_all_blocks_to_optimistic::(&rig.harness.chain.spec) + .set_all_blocks_to_optimistic::() .unwrap(); let new_weights = rig diff --git a/beacon_node/beacon_chain/tests/store_tests.rs b/beacon_node/beacon_chain/tests/store_tests.rs index c6e13bd160..a0b0b77ec4 100644 --- a/beacon_node/beacon_chain/tests/store_tests.rs +++ b/beacon_node/beacon_chain/tests/store_tests.rs @@ -2938,6 +2938,7 @@ async fn reproduction_unaligned_checkpoint_sync_pruned_payload() { wss_state, wss_block.clone(), wss_blobs_opt.clone(), + None, genesis_state, ) .unwrap() @@ -3088,6 +3089,7 @@ async fn weak_subjectivity_sync_test( } else { None }, + None, genesis_state, ) .unwrap() diff --git a/beacon_node/http_api/src/beacon/execution_payload_envelope.rs b/beacon_node/http_api/src/beacon/execution_payload_envelope.rs index ea8c0d4b8a..3479d62f6a 100644 --- a/beacon_node/http_api/src/beacon/execution_payload_envelope.rs +++ b/beacon_node/http_api/src/beacon/execution_payload_envelope.rs @@ -90,7 +90,6 @@ pub(crate) fn post_beacon_execution_payload_envelope( .boxed() } /// Publishes a signed execution payload envelope to the network. -/// TODO(gloas): Add gossip verification (BroadcastValidation::Gossip) before import. pub async fn publish_execution_payload_envelope( envelope: SignedExecutionPayloadEnvelope, chain: Arc>, @@ -132,18 +131,21 @@ pub async fn publish_execution_payload_envelope( }; let ctx = chain.gossip_verification_context(); - let Ok(gossip_verifed_envelope) = GossipVerifiedEnvelope::new(signed_envelope, &ctx) else { - warn!(%slot, %beacon_block_root, "Execution payload envelope rejected"); - return Err(warp_utils::reject::custom_bad_request( - "execution payload envelope rejected, gossip verification".to_string(), - )); + let gossip_verified_envelope = match GossipVerifiedEnvelope::new(signed_envelope, &ctx) { + Ok(envelope) => envelope, + Err(e) => { + warn!(%slot, %beacon_block_root, error = ?e, "Execution payload envelope rejected"); + return Err(warp_utils::reject::custom_bad_request(format!( + "execution payload envelope rejected: {e:?}", + ))); + } }; // Import the envelope locally (runs state transition and notifies the EL). chain .process_execution_payload_envelope( beacon_block_root, - gossip_verifed_envelope, + gossip_verified_envelope, NotifyExecutionLayer::Yes, BlockImportSource::HttpApi, publish_fn, diff --git a/beacon_node/http_api/src/lib.rs b/beacon_node/http_api/src/lib.rs index 5df1078617..0bb04888b7 100644 --- a/beacon_node/http_api/src/lib.rs +++ b/beacon_node/http_api/src/lib.rs @@ -2148,10 +2148,14 @@ pub fn serve( execution_status: execution_status_string, best_child: node .best_child() + .ok() + .flatten() .and_then(|index| proto_array.nodes.get(index)) .map(|child| child.root()), best_descendant: node .best_descendant() + .ok() + .flatten() .and_then(|index| proto_array.nodes.get(index)) .map(|descendant| descendant.root()), }, diff --git a/beacon_node/http_api/src/publish_blocks.rs b/beacon_node/http_api/src/publish_blocks.rs index 43dfbeb836..eb7e56e9cc 100644 --- a/beacon_node/http_api/src/publish_blocks.rs +++ b/beacon_node/http_api/src/publish_blocks.rs @@ -146,12 +146,8 @@ pub async fn publish_block>( let slot = block.message().slot(); let sender_clone = network_tx.clone(); - let build_sidecar_task_handle = spawn_build_data_sidecar_task( - chain.clone(), - block.clone(), - unverified_blobs, - current_span.clone(), - )?; + let build_sidecar_task_handle = + spawn_build_data_sidecar_task(chain.clone(), block.clone(), unverified_blobs)?; // Gossip verify the block and blobs/data columns separately. let gossip_verified_block_result = unverified_block.into_gossip_verified_block(&chain); @@ -358,7 +354,6 @@ fn spawn_build_data_sidecar_task( chain: Arc>, block: Arc>>, proofs_and_blobs: UnverifiedBlobs, - current_span: Span, ) -> Result>, Rejection> { chain .clone() @@ -368,7 +363,7 @@ fn spawn_build_data_sidecar_task( let Some((kzg_proofs, blobs)) = proofs_and_blobs else { return Ok((vec![], vec![])); }; - let _guard = debug_span!(parent: current_span, "build_data_sidecars").entered(); + let _span = debug_span!("build_data_sidecars").entered(); let peer_das_enabled = chain.spec.is_peer_das_enabled_for_epoch(block.epoch()); if !peer_das_enabled { diff --git a/beacon_node/http_api/src/validator/mod.rs b/beacon_node/http_api/src/validator/mod.rs index 3d96b85870..412851233e 100644 --- a/beacon_node/http_api/src/validator/mod.rs +++ b/beacon_node/http_api/src/validator/mod.rs @@ -671,7 +671,7 @@ pub fn post_validator_prepare_beacon_proposer( .await; // TODO(gloas): verify this is correct. We skip proposer preparation for - // GLOAS because the execution payload is no longer embedded in the beacon + // Gloas because the execution payload is no longer embedded in the beacon // block (it's in the payload envelope), so the head block's // execution_payload() is unavailable. let next_slot = current_slot + 1; diff --git a/beacon_node/http_api/tests/tests.rs b/beacon_node/http_api/tests/tests.rs index 14bfb5ce92..b28816302c 100644 --- a/beacon_node/http_api/tests/tests.rs +++ b/beacon_node/http_api/tests/tests.rs @@ -3179,10 +3179,14 @@ impl ApiTester { .unwrap_or_else(|| "irrelevant".to_string()), best_child: node .best_child() + .ok() + .flatten() .and_then(|index| expected_proto_array.nodes.get(index)) .map(|child| child.root()), best_descendant: node .best_descendant() + .ok() + .flatten() .and_then(|index| expected_proto_array.nodes.get(index)) .map(|descendant| descendant.root()), }, diff --git a/beacon_node/lighthouse_network/src/rpc/protocol.rs b/beacon_node/lighthouse_network/src/rpc/protocol.rs index 2c92e17c44..c949dfe17d 100644 --- a/beacon_node/lighthouse_network/src/rpc/protocol.rs +++ b/beacon_node/lighthouse_network/src/rpc/protocol.rs @@ -11,7 +11,7 @@ use std::io; use std::marker::PhantomData; use std::sync::{Arc, LazyLock}; use std::time::Duration; -use strum::{AsRefStr, Display, EnumString, IntoStaticStr}; +use strum::{AsRefStr, Display, EnumIter, EnumString, IntoStaticStr}; use tokio_util::{ codec::Framed, compat::{Compat, FuturesAsyncReadCompatExt}, @@ -329,7 +329,7 @@ pub enum Encoding { } /// All valid protocol name and version combinations. -#[derive(Debug, Clone, Copy, PartialEq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, EnumIter)] pub enum SupportedProtocol { StatusV1, StatusV2, @@ -499,6 +499,10 @@ impl UpgradeInfo for RPCProtocol { SupportedProtocol::LightClientFinalityUpdateV1, Encoding::SSZSnappy, )); + supported_protocols.push(ProtocolId::new( + SupportedProtocol::LightClientUpdatesByRangeV1, + Encoding::SSZSnappy, + )); } supported_protocols } @@ -1133,3 +1137,101 @@ impl RPCError { } } } + +#[cfg(test)] +mod tests { + use super::*; + use libp2p::core::UpgradeInfo; + use std::collections::HashSet; + use strum::IntoEnumIterator; + use types::{Hash256, Slot}; + + type E = MainnetEthSpec; + + /// Whether this protocol should appear in `currently_supported()` for the given context. + /// + /// Uses an exhaustive match so that adding a new `SupportedProtocol` variant + /// causes a compile error until this function is updated. + fn expected_in_currently_supported( + protocol: SupportedProtocol, + fork_context: &ForkContext, + ) -> bool { + use SupportedProtocol::*; + match protocol { + StatusV1 | StatusV2 | GoodbyeV1 | PingV1 | BlocksByRangeV1 | BlocksByRangeV2 + | BlocksByRootV1 | BlocksByRootV2 | MetaDataV1 | MetaDataV2 => true, + + BlobsByRangeV1 | BlobsByRootV1 => fork_context.fork_exists(ForkName::Deneb), + + DataColumnsByRootV1 | DataColumnsByRangeV1 | MetaDataV3 => { + fork_context.spec.is_peer_das_scheduled() + } + + PayloadEnvelopesByRangeV1 | PayloadEnvelopesByRootV1 => { + fork_context.fork_exists(ForkName::Gloas) + } + + // Light client protocols are not in currently_supported() + LightClientBootstrapV1 + | LightClientOptimisticUpdateV1 + | LightClientFinalityUpdateV1 + | LightClientUpdatesByRangeV1 => false, + } + } + + /// Whether this protocol should appear in `protocol_info()` when light client server is + /// enabled. + /// + /// Uses an exhaustive match so that adding a new `SupportedProtocol` variant + /// causes a compile error until this function is updated. + fn expected_in_protocol_info(protocol: SupportedProtocol, fork_context: &ForkContext) -> bool { + use SupportedProtocol::*; + match protocol { + LightClientBootstrapV1 + | LightClientOptimisticUpdateV1 + | LightClientFinalityUpdateV1 + | LightClientUpdatesByRangeV1 => true, + + _ => expected_in_currently_supported(protocol, fork_context), + } + } + + #[test] + fn all_protocols_registered() { + for fork in ForkName::list_all() { + let spec = fork.make_genesis_spec(E::default_spec()); + let fork_context = Arc::new(ForkContext::new::(Slot::new(0), Hash256::ZERO, &spec)); + + let currently_supported: HashSet = + SupportedProtocol::currently_supported(&fork_context) + .into_iter() + .map(|pid| pid.versioned_protocol) + .collect(); + + let rpc_protocol = RPCProtocol:: { + fork_context: fork_context.clone(), + max_rpc_size: spec.max_payload_size as usize, + enable_light_client_server: true, + phantom: PhantomData, + }; + let protocol_info: HashSet = rpc_protocol + .protocol_info() + .into_iter() + .map(|pid| pid.versioned_protocol) + .collect(); + + for protocol in SupportedProtocol::iter() { + assert_eq!( + currently_supported.contains(&protocol), + expected_in_currently_supported(protocol, &fork_context), + "{protocol:?} registration mismatch in currently_supported() at {fork:?}" + ); + assert_eq!( + protocol_info.contains(&protocol), + expected_in_protocol_info(protocol, &fork_context), + "{protocol:?} registration mismatch in protocol_info() at {fork:?}" + ); + } + } + } +} diff --git a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs index 8de90f991b..2d13ef8e2a 100644 --- a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs +++ b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs @@ -1393,7 +1393,8 @@ impl NetworkBeaconProcessor { // EnvelopeError is unreachable. Only constructed during range sync envelope processing. Err(e @ BlockError::InternalError(_)) | Err(e @ BlockError::BlobNotRequired(_)) - | Err(e @ BlockError::EnvelopeError(_)) => { + | Err(e @ BlockError::EnvelopeError(_)) + | Err(e @ BlockError::PayloadEnvelopeError { .. }) => { error!(error = %e, "Internal block gossip validation error"); return None; } diff --git a/beacon_node/network/src/network_beacon_processor/sync_methods.rs b/beacon_node/network/src/network_beacon_processor/sync_methods.rs index f6d4940121..57d3d7d220 100644 --- a/beacon_node/network/src/network_beacon_processor/sync_methods.rs +++ b/beacon_node/network/src/network_beacon_processor/sync_methods.rs @@ -109,9 +109,7 @@ impl NetworkBeaconProcessor { ); self.send_sync_message(SyncMessage::BlockComponentProcessed { process_type, - result: BlockProcessingResult::Err(BlockError::InternalError(format!( - "Envelope verification failed: {e:?}" - ))), + result: BlockProcessingResult::Err(e.into()), }); return; } @@ -138,9 +136,7 @@ impl NetworkBeaconProcessor { ?beacon_block_root, "RPC payload envelope processing failed" ); - BlockProcessingResult::Err(BlockError::InternalError(format!( - "Envelope processing failed: {e:?}" - ))) + BlockProcessingResult::Err(e.into()) } }; diff --git a/beacon_node/network/src/sync/block_lookups/common.rs b/beacon_node/network/src/sync/block_lookups/common.rs index edd99345b4..bb8d81cc6e 100644 --- a/beacon_node/network/src/sync/block_lookups/common.rs +++ b/beacon_node/network/src/sync/block_lookups/common.rs @@ -2,7 +2,7 @@ use crate::sync::block_lookups::single_block_lookup::{ LookupRequestError, SingleBlockLookup, SingleLookupRequestState, }; use crate::sync::block_lookups::{ - BlobRequestState, BlockRequestState, CustodyRequestState, PeerId, + BlobRequestState, BlockRequestState, CustodyRequestState, EnvelopeRequestState, PeerId, }; use crate::sync::manager::BlockProcessType; use crate::sync::network_context::{LookupRequestResult, SyncNetworkContext}; @@ -12,16 +12,17 @@ use parking_lot::RwLock; use std::collections::HashSet; use std::sync::Arc; use types::data::FixedBlobSidecarList; -use types::{DataColumnSidecarList, SignedBeaconBlock}; +use types::{DataColumnSidecarList, SignedBeaconBlock, SignedExecutionPayloadEnvelope}; use super::SingleLookupId; use super::single_block_lookup::{ComponentRequests, DownloadResult}; -#[derive(Debug, Copy, Clone)] +#[derive(Debug, Copy, Clone, PartialEq, Eq)] pub enum ResponseType { Block, Blob, CustodyColumn, + Envelope, } /// This trait unifies common single block lookup functionality across blocks and blobs. This @@ -151,6 +152,7 @@ impl RequestState for BlobRequestState { ComponentRequests::WaitingForBlock => Err("waiting for block"), ComponentRequests::ActiveBlobRequest(request, _) => Ok(request), ComponentRequests::ActiveCustodyRequest { .. } => Err("expecting custody request"), + ComponentRequests::ActiveEnvelopeRequest { .. } => Err("expecting envelope request"), ComponentRequests::NotNeeded { .. } => Err("not needed"), } } @@ -205,6 +207,7 @@ impl RequestState for CustodyRequestState { ComponentRequests::WaitingForBlock => Err("waiting for block"), ComponentRequests::ActiveBlobRequest { .. } => Err("expecting blob request"), ComponentRequests::ActiveCustodyRequest(request) => Ok(request), + ComponentRequests::ActiveEnvelopeRequest { .. } => Err("expecting envelope request"), ComponentRequests::NotNeeded { .. } => Err("not needed"), } } @@ -215,3 +218,52 @@ impl RequestState for CustodyRequestState { &mut self.state } } + +impl RequestState for EnvelopeRequestState { + type VerifiedResponseType = Arc>; + + fn make_request( + &self, + id: Id, + lookup_peers: Arc>>, + _: usize, + cx: &mut SyncNetworkContext, + ) -> Result { + cx.envelope_lookup_request(id, lookup_peers, self.block_root) + .map_err(LookupRequestError::SendFailedNetwork) + } + + fn send_for_processing( + id: Id, + download_result: DownloadResult, + cx: &SyncNetworkContext, + ) -> Result<(), LookupRequestError> { + let DownloadResult { + value, + block_root, + seen_timestamp, + .. + } = download_result; + cx.send_envelope_for_processing(id, value, seen_timestamp, block_root) + .map_err(LookupRequestError::SendFailedProcessor) + } + + fn response_type() -> ResponseType { + ResponseType::Envelope + } + + fn request_state_mut(request: &mut SingleBlockLookup) -> Result<&mut Self, &'static str> { + match &mut request.component_requests { + ComponentRequests::ActiveEnvelopeRequest(request) => Ok(request), + _ => Err("expecting envelope request"), + } + } + + fn get_state(&self) -> &SingleLookupRequestState { + &self.state + } + + fn get_state_mut(&mut self) -> &mut SingleLookupRequestState { + &mut self.state + } +} diff --git a/beacon_node/network/src/sync/block_lookups/mod.rs b/beacon_node/network/src/sync/block_lookups/mod.rs index 7b4e3ce753..27d96de51d 100644 --- a/beacon_node/network/src/sync/block_lookups/mod.rs +++ b/beacon_node/network/src/sync/block_lookups/mod.rs @@ -22,7 +22,9 @@ use self::parent_chain::{NodeChain, compute_parent_chains}; pub use self::single_block_lookup::DownloadResult; -use self::single_block_lookup::{LookupRequestError, LookupResult, SingleBlockLookup}; +use self::single_block_lookup::{ + AwaitingParent, LookupRequestError, LookupResult, SingleBlockLookup, +}; use super::manager::{BlockProcessType, BlockProcessingResult, SLOT_IMPORT_TOLERANCE}; use super::network_context::{PeerGroup, RpcResponseError, SyncNetworkContext}; use crate::metrics; @@ -39,7 +41,9 @@ use fnv::FnvHashMap; use lighthouse_network::service::api_types::SingleLookupReqId; use lighthouse_network::{PeerAction, PeerId}; use lru_cache::LRUTimeCache; -pub use single_block_lookup::{BlobRequestState, BlockRequestState, CustodyRequestState}; +pub use single_block_lookup::{ + BlobRequestState, BlockRequestState, CustodyRequestState, EnvelopeRequestState, +}; use std::collections::hash_map::Entry; use std::sync::Arc; use std::time::Duration; @@ -214,7 +218,7 @@ impl BlockLookups { self.new_current_lookup( block_root, Some(block_component), - Some(parent_root), + Some(AwaitingParent::Block(parent_root)), // On a `UnknownParentBlock` or `UnknownParentBlob` event the peer is not required // to have the rest of the block components (refer to decoupled blob gossip). Create // the lookup with zero peers to house the block components. @@ -226,7 +230,37 @@ impl BlockLookups { } } - /// Seach a block whose parent root is unknown. + /// A child block's parent envelope is missing. Create a child lookup (with the block component) + /// that waits for the parent envelope, and an envelope-only lookup for the parent. + /// + /// Returns true if both lookups are created or already exist. + #[must_use = "only reference the new lookup if returns true"] + pub fn search_child_and_parent_envelope( + &mut self, + block_root: Hash256, + block_component: BlockComponent, + parent_root: Hash256, + peer_id: PeerId, + cx: &mut SyncNetworkContext, + ) -> bool { + let envelope_lookup_exists = + self.search_parent_envelope_of_child(parent_root, &[peer_id], cx); + if envelope_lookup_exists { + // Create child lookup that waits for the parent envelope. + // The child block itself has already been seen, so we pass it as a component. + self.new_current_lookup( + block_root, + Some(block_component), + Some(AwaitingParent::Envelope(parent_root)), + &[], + cx, + ) + } else { + false + } + } + + /// Search a block whose parent root is unknown. /// /// Returns true if the lookup is created or already exists #[must_use = "only reference the new lookup if returns true"] @@ -344,6 +378,57 @@ impl BlockLookups { self.new_current_lookup(block_root_to_search, None, None, peers, cx) } + /// A block triggers the search of a parent envelope. + #[must_use = "only reference the new lookup if returns true"] + pub fn search_parent_envelope_of_child( + &mut self, + parent_root: Hash256, + peers: &[PeerId], + cx: &mut SyncNetworkContext, + ) -> bool { + // Check if there's already a lookup for this root (could be a block lookup or envelope + // lookup). If so, add peers and let it handle the envelope. + if let Some((&lookup_id, _lookup)) = self + .single_block_lookups + .iter_mut() + .find(|(_, lookup)| lookup.is_for_block(parent_root)) + { + if let Err(e) = self.add_peers_to_lookup_and_ancestors(lookup_id, peers, cx) { + warn!(error = ?e, "Error adding peers to envelope lookup"); + } + return true; + } + + if self.single_block_lookups.len() >= MAX_LOOKUPS { + warn!(?parent_root, "Dropping envelope lookup reached max"); + return false; + } + + let lookup = SingleBlockLookup::new_envelope_only(parent_root, peers, cx.next_id()); + let _guard = lookup.span.clone().entered(); + + let id = lookup.id; + let lookup = match self.single_block_lookups.entry(id) { + Entry::Vacant(entry) => entry.insert(lookup), + Entry::Occupied(_) => { + warn!(id, "Lookup exists with same id"); + return false; + } + }; + + debug!( + ?peers, + ?parent_root, + id = lookup.id, + "Created envelope-only lookup" + ); + metrics::inc_counter(&metrics::SYNC_LOOKUP_CREATED); + self.metrics.created_lookups += 1; + + let result = lookup.continue_requests(cx); + self.on_lookup_result(id, result, "new_envelope_lookup", cx) + } + /// Searches for a single block hash. If the blocks parent is unknown, a chain of blocks is /// constructed. /// Returns true if the lookup is created or already exists @@ -352,7 +437,7 @@ impl BlockLookups { &mut self, block_root: Hash256, block_component: Option>, - awaiting_parent: Option, + awaiting_parent: Option, peers: &[PeerId], cx: &mut SyncNetworkContext, ) -> bool { @@ -387,13 +472,14 @@ impl BlockLookups { } // Ensure that awaiting parent exists, otherwise this lookup won't be able to make progress - if let Some(awaiting_parent) = awaiting_parent + if let Some(AwaitingParent::Block(parent_root) | AwaitingParent::Envelope(parent_root)) = + awaiting_parent && !self .single_block_lookups .iter() - .any(|(_, lookup)| lookup.is_for_block(awaiting_parent)) + .any(|(_, lookup)| lookup.is_for_block(parent_root)) { - warn!(block_root = ?awaiting_parent, "Ignoring child lookup parent lookup not found"); + warn!(block_root = ?parent_root, "Ignoring child lookup parent lookup not found"); return false; } @@ -427,9 +513,7 @@ impl BlockLookups { debug!( ?peers, ?block_root, - awaiting_parent = awaiting_parent - .map(|root| root.to_string()) - .unwrap_or("none".to_owned()), + ?awaiting_parent, id = lookup.id, "Created block lookup" ); @@ -561,17 +645,13 @@ impl BlockLookups { self.on_processing_result_inner::>(id, result, cx) } BlockProcessType::SinglePayloadEnvelope { id, block_root } => { - match result { - BlockProcessingResult::Ok(_) => { - self.continue_envelope_child_lookups(block_root, cx); - } - BlockProcessingResult::Err(e) => { - debug!(%id, error = ?e, "Payload envelope processing failed"); - // TODO(EIP-7732): resolve awaiting_envelope on affected lookups so they can retry - } - _ => {} + let result = self + .on_processing_result_inner::>(id, result, cx); + // On successful envelope import, unblock child lookups waiting for this envelope + if matches!(&result, Ok(LookupResult::Completed)) { + self.continue_envelope_child_lookups(block_root, cx); } - return; + result } }; self.on_lookup_result(process_type.id(), lookup_result, "processing_result", cx); @@ -687,6 +767,26 @@ impl BlockLookups { // We opt to drop the lookup instead. Action::Drop(format!("{e:?}")) } + BlockError::PayloadEnvelopeError { e, penalize_peer } => { + debug!( + ?block_root, + error = ?e, + "Payload envelope processing error" + ); + if penalize_peer { + let peer_group = request_state.on_processing_failure()?; + for peer in peer_group.all() { + cx.report_peer( + *peer, + PeerAction::MidToleranceError, + "lookup_envelope_processing_failure", + ); + } + Action::Retry + } else { + Action::Drop(format!("{e:?}")) + } + } other => { debug!( ?block_root, @@ -721,6 +821,7 @@ impl BlockLookups { ResponseType::CustodyColumn => { "lookup_custody_column_processing_failure" } + ResponseType::Envelope => "lookup_envelope_processing_failure", }, ); } @@ -764,22 +865,21 @@ impl BlockLookups { } Action::ParentEnvelopeUnknown { parent_root } => { let peers = lookup.all_peers(); - lookup.set_awaiting_envelope(parent_root); - // Pick a peer to request the envelope from - let peer_id = peers.first().copied().ok_or_else(|| { - LookupRequestError::Failed("No peers available for envelope request".to_owned()) - })?; - match cx.envelope_lookup_request(lookup_id, peer_id, parent_root) { - Ok(_) => { - debug!( - id = lookup_id, - ?block_root, - ?parent_root, - "Requesting missing parent envelope" - ); - Ok(LookupResult::Pending) - } - Err(e) => Err(LookupRequestError::SendFailedNetwork(e)), + lookup.set_awaiting_parent_envelope(parent_root); + let envelope_lookup_exists = + self.search_parent_envelope_of_child(parent_root, &peers, cx); + if envelope_lookup_exists { + debug!( + id = lookup_id, + ?block_root, + ?parent_root, + "Marking lookup as awaiting parent envelope" + ); + Ok(LookupResult::Pending) + } else { + Err(LookupRequestError::Failed(format!( + "Envelope lookup could not be created for {parent_root:?}" + ))) } } Action::Drop(reason) => { @@ -831,7 +931,7 @@ impl BlockLookups { let mut lookup_results = vec![]; // < need to buffer lookup results to not re-borrow &mut self for (id, lookup) in self.single_block_lookups.iter_mut() { - if lookup.awaiting_parent() == Some(block_root) { + if lookup.awaiting_parent_block() == Some(block_root) { lookup.resolve_awaiting_parent(); debug!( parent_root = ?block_root, @@ -858,8 +958,8 @@ impl BlockLookups { let mut lookup_results = vec![]; for (id, lookup) in self.single_block_lookups.iter_mut() { - if lookup.awaiting_envelope() == Some(block_root) { - lookup.resolve_awaiting_envelope(); + if lookup.awaiting_parent_envelope() == Some(block_root) { + lookup.resolve_awaiting_parent(); debug!( envelope_root = ?block_root, id, @@ -891,10 +991,14 @@ impl BlockLookups { metrics::inc_counter_vec(&metrics::SYNC_LOOKUP_DROPPED, &[reason]); self.metrics.dropped_lookups += 1; + let dropped_root = dropped_lookup.block_root(); let child_lookups = self .single_block_lookups .iter() - .filter(|(_, lookup)| lookup.awaiting_parent() == Some(dropped_lookup.block_root())) + .filter(|(_, lookup)| { + lookup.awaiting_parent_block() == Some(dropped_root) + || lookup.awaiting_parent_envelope() == Some(dropped_root) + }) .map(|(id, _)| *id) .collect::>(); @@ -1062,17 +1166,15 @@ impl BlockLookups { &'a self, lookup: &'a SingleBlockLookup, ) -> Result<&'a SingleBlockLookup, String> { - if let Some(awaiting_parent) = lookup.awaiting_parent() { + if let Some(parent_root) = lookup.awaiting_parent_block() { if let Some(lookup) = self .single_block_lookups .values() - .find(|l| l.block_root() == awaiting_parent) + .find(|l| l.block_root() == parent_root) { self.find_oldest_ancestor_lookup(lookup) } else { - Err(format!( - "Lookup references unknown parent {awaiting_parent:?}" - )) + Err(format!("Lookup references unknown parent {parent_root:?}")) } } else { Ok(lookup) @@ -1105,7 +1207,7 @@ impl BlockLookups { } } - if let Some(parent_root) = lookup.awaiting_parent() { + if let Some(parent_root) = lookup.awaiting_parent_block() { if let Some((&child_id, _)) = self .single_block_lookups .iter() diff --git a/beacon_node/network/src/sync/block_lookups/parent_chain.rs b/beacon_node/network/src/sync/block_lookups/parent_chain.rs index 5deea1dd94..18363e9b8d 100644 --- a/beacon_node/network/src/sync/block_lookups/parent_chain.rs +++ b/beacon_node/network/src/sync/block_lookups/parent_chain.rs @@ -13,7 +13,7 @@ impl From<&SingleBlockLookup> for Node { fn from(value: &SingleBlockLookup) -> Self { Self { block_root: value.block_root(), - parent_root: value.awaiting_parent(), + parent_root: value.awaiting_parent_block(), } } } 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 51cc191056..6687a1ec75 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 @@ -16,7 +16,9 @@ use store::Hash256; use strum::IntoStaticStr; use tracing::{Span, debug_span}; use types::data::FixedBlobSidecarList; -use types::{DataColumnSidecarList, EthSpec, SignedBeaconBlock, Slot}; +use types::{ + DataColumnSidecarList, EthSpec, SignedBeaconBlock, SignedExecutionPayloadEnvelope, Slot, +}; // Dedicated enum for LookupResult to force its usage #[must_use = "LookupResult must be handled with on_lookup_result"] @@ -56,6 +58,14 @@ pub enum LookupRequestError { }, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AwaitingParent { + /// Waiting for the parent block to be imported. + Block(Hash256), + /// The parent block is imported but its execution payload envelope is missing. + Envelope(Hash256), +} + #[derive(Educe)] #[educe(Debug(bound(T: BeaconChainTypes)))] pub struct SingleBlockLookup { @@ -69,8 +79,7 @@ pub struct SingleBlockLookup { #[educe(Debug(method(fmt_peer_set_as_len)))] peers: Arc>>, block_root: Hash256, - awaiting_parent: Option, - awaiting_envelope: Option, + awaiting_parent: Option, created: Instant, pub(crate) span: Span, } @@ -80,6 +89,7 @@ pub(crate) enum ComponentRequests { WaitingForBlock, ActiveBlobRequest(BlobRequestState, usize), ActiveCustodyRequest(CustodyRequestState), + ActiveEnvelopeRequest(EnvelopeRequestState), // When printing in debug this state display the reason why it's not needed #[allow(dead_code)] NotNeeded(&'static str), @@ -90,7 +100,7 @@ impl SingleBlockLookup { requested_block_root: Hash256, peers: &[PeerId], id: Id, - awaiting_parent: Option, + awaiting_parent: Option, ) -> Self { let lookup_span = debug_span!( "lh_single_block_lookup", @@ -105,16 +115,38 @@ impl SingleBlockLookup { peers: Arc::new(RwLock::new(HashSet::from_iter(peers.iter().copied()))), block_root: requested_block_root, awaiting_parent, - awaiting_envelope: None, created: Instant::now(), span: lookup_span, } } + /// Create an envelope-only lookup. The block is already imported, we just need the envelope. + pub fn new_envelope_only(block_root: Hash256, peers: &[PeerId], id: Id) -> Self { + let mut lookup = Self::new(block_root, peers, id, None); + // Block is already imported, mark as completed + lookup + .block_request_state + .state + .on_completed_request("block already imported") + .expect("block state starts as AwaitingDownload"); + lookup.component_requests = + ComponentRequests::ActiveEnvelopeRequest(EnvelopeRequestState::new(block_root)); + lookup + } + /// Reset the status of all internal requests pub fn reset_requests(&mut self) { self.block_request_state = BlockRequestState::new(self.block_root); - self.component_requests = ComponentRequests::WaitingForBlock; + match &self.component_requests { + ComponentRequests::ActiveEnvelopeRequest(_) => { + self.component_requests = ComponentRequests::ActiveEnvelopeRequest( + EnvelopeRequestState::new(self.block_root), + ); + } + _ => { + self.component_requests = ComponentRequests::WaitingForBlock; + } + } } /// Return the slot of this lookup's block if it's currently cached as `AwaitingProcessing` @@ -130,34 +162,39 @@ impl SingleBlockLookup { self.block_root } - pub fn awaiting_parent(&self) -> Option { + pub fn awaiting_parent(&self) -> Option { self.awaiting_parent } - /// Mark this lookup as awaiting a parent lookup from being processed. Meanwhile don't send - /// components for processing. + /// Returns the parent root if awaiting a parent block. + pub fn awaiting_parent_block(&self) -> Option { + match self.awaiting_parent { + Some(AwaitingParent::Block(root)) => Some(root), + _ => None, + } + } + + /// Returns the parent root if awaiting a parent envelope. + pub fn awaiting_parent_envelope(&self) -> Option { + match self.awaiting_parent { + Some(AwaitingParent::Envelope(root)) => Some(root), + _ => None, + } + } + + /// Mark this lookup as awaiting a parent block to be imported before processing. pub fn set_awaiting_parent(&mut self, parent_root: Hash256) { - self.awaiting_parent = Some(parent_root) - } - - /// Mark this lookup as no longer awaiting a parent lookup. Components can be sent for - /// processing. - pub fn resolve_awaiting_parent(&mut self) { - self.awaiting_parent = None; - } - - pub fn awaiting_envelope(&self) -> Option { - self.awaiting_envelope + self.awaiting_parent = Some(AwaitingParent::Block(parent_root)); } /// Mark this lookup as awaiting a parent envelope to be imported before processing. - pub fn set_awaiting_envelope(&mut self, parent_root: Hash256) { - self.awaiting_envelope = Some(parent_root); + pub fn set_awaiting_parent_envelope(&mut self, parent_root: Hash256) { + self.awaiting_parent = Some(AwaitingParent::Envelope(parent_root)); } - /// Mark this lookup as no longer awaiting a parent envelope. - pub fn resolve_awaiting_envelope(&mut self) { - self.awaiting_envelope = None; + /// Mark this lookup as no longer awaiting any parent. + pub fn resolve_awaiting_parent(&mut self) { + self.awaiting_parent = None; } /// Returns the time elapsed since this lookup was created @@ -194,6 +231,7 @@ impl SingleBlockLookup { ComponentRequests::WaitingForBlock => false, ComponentRequests::ActiveBlobRequest(request, _) => request.state.is_processed(), ComponentRequests::ActiveCustodyRequest(request) => request.state.is_processed(), + ComponentRequests::ActiveEnvelopeRequest(request) => request.state.is_processed(), ComponentRequests::NotNeeded { .. } => true, } } @@ -201,7 +239,6 @@ impl SingleBlockLookup { /// Returns true if this request is expecting some event to make progress pub fn is_awaiting_event(&self) -> bool { self.awaiting_parent.is_some() - || self.awaiting_envelope.is_some() || self.block_request_state.state.is_awaiting_event() || match &self.component_requests { // If components are waiting for the block request to complete, here we should @@ -214,6 +251,9 @@ impl SingleBlockLookup { ComponentRequests::ActiveCustodyRequest(request) => { request.state.is_awaiting_event() } + ComponentRequests::ActiveEnvelopeRequest(request) => { + request.state.is_awaiting_event() + } ComponentRequests::NotNeeded { .. } => false, } } @@ -283,6 +323,9 @@ impl SingleBlockLookup { ComponentRequests::ActiveCustodyRequest(_) => { self.continue_request::>(cx, 0)? } + ComponentRequests::ActiveEnvelopeRequest(_) => { + self.continue_request::>(cx, 0)? + } ComponentRequests::NotNeeded { .. } => {} // do nothing } @@ -304,7 +347,7 @@ impl SingleBlockLookup { expected_blobs: usize, ) -> Result<(), LookupRequestError> { let id = self.id; - let awaiting_event = self.awaiting_parent.is_some() || self.awaiting_envelope.is_some(); + let awaiting_event = self.awaiting_parent.is_some(); let request = R::request_state_mut(self).map_err(|e| LookupRequestError::BadState(e.to_owned()))?; @@ -444,6 +487,26 @@ impl BlockRequestState { } } +/// The state of the envelope request component of a `SingleBlockLookup`. +/// Used for envelope-only lookups where the parent block is already imported +/// but its execution payload envelope is missing. +#[derive(Educe)] +#[educe(Debug)] +pub struct EnvelopeRequestState { + #[educe(Debug(ignore))] + pub block_root: Hash256, + pub state: SingleLookupRequestState>>, +} + +impl EnvelopeRequestState { + pub fn new(block_root: Hash256) -> Self { + Self { + block_root, + state: SingleLookupRequestState::new(), + } + } +} + #[derive(Debug, Clone)] pub struct DownloadResult { pub value: T, diff --git a/beacon_node/network/src/sync/manager.rs b/beacon_node/network/src/sync/manager.rs index 64c10e8f46..c1c1029446 100644 --- a/beacon_node/network/src/sync/manager.rs +++ b/beacon_node/network/src/sync/manager.rs @@ -45,6 +45,7 @@ use crate::service::NetworkMessage; use crate::status::ToStatusMessage; use crate::sync::block_lookups::{ BlobRequestState, BlockComponent, BlockRequestState, CustodyRequestState, DownloadResult, + EnvelopeRequestState, }; use crate::sync::custody_backfill_sync::CustodyBackFillSync; use crate::sync::network_context::{PeerGroup, RpcResponseResult}; @@ -937,9 +938,9 @@ impl SyncManager { debug!( %block_root, %parent_root, - "Parent envelope not yet available, creating lookup" + "Parent envelope not yet available, creating envelope lookup" ); - self.handle_unknown_parent( + self.handle_unknown_parent_envelope( peer_id, block_root, parent_root, @@ -1057,6 +1058,40 @@ impl SyncManager { } } + /// Handle a block whose parent block is known but parent envelope is missing. + /// Creates an envelope-only lookup for the parent and a child lookup that waits for it. + fn handle_unknown_parent_envelope( + &mut self, + peer_id: PeerId, + block_root: Hash256, + parent_root: Hash256, + slot: Slot, + block_component: BlockComponent, + ) { + match self.should_search_for_block(Some(slot), &peer_id) { + Ok(_) => { + if self.block_lookups.search_child_and_parent_envelope( + block_root, + block_component, + parent_root, + peer_id, + &mut self.network, + ) { + // Lookups created + } else { + debug!( + ?block_root, + ?parent_root, + "No lookup created for child and parent envelope" + ); + } + } + Err(reason) => { + debug!(%block_root, %parent_root, reason, "Ignoring unknown parent envelope request"); + } + } + } + fn handle_unknown_block_root(&mut self, peer_id: PeerId, block_root: Hash256) { match self.should_search_for_block(None, &peer_id) { Ok(_) => { @@ -1288,27 +1323,14 @@ impl SyncManager { .network .on_single_envelope_response(id, peer_id, rpc_event) { - match resp { - Ok((envelope, seen_timestamp)) => { - let block_root = envelope.beacon_block_root(); - debug!( - ?block_root, - %id, - "Downloaded payload envelope, sending for processing" - ); - if let Err(e) = self.network.send_envelope_for_processing( - id.req_id, - envelope, - seen_timestamp, - block_root, - ) { - error!(error = ?e, "Failed to send envelope for processing"); - } - } - Err(e) => { - debug!(error = ?e, %id, "Payload envelope download failed"); - } - } + self.block_lookups + .on_download_response::>( + id, + resp.map(|(value, seen_timestamp)| { + (value, PeerGroup::from_single(peer_id), seen_timestamp) + }), + &mut self.network, + ) } } diff --git a/beacon_node/network/src/sync/network_context.rs b/beacon_node/network/src/sync/network_context.rs index d512c9e24f..b840b12f61 100644 --- a/beacon_node/network/src/sync/network_context.rs +++ b/beacon_node/network/src/sync/network_context.rs @@ -1002,9 +1002,26 @@ impl SyncNetworkContext { pub fn envelope_lookup_request( &mut self, lookup_id: SingleLookupId, - peer_id: PeerId, + lookup_peers: Arc>>, block_root: Hash256, - ) -> Result { + ) -> Result { + let active_request_count_by_peer = self.active_request_count_by_peer(); + let Some(peer_id) = lookup_peers + .read() + .iter() + .map(|peer| { + ( + active_request_count_by_peer.get(peer).copied().unwrap_or(0), + rand::random::(), + peer, + ) + }) + .min() + .map(|(_, _, peer)| *peer) + else { + return Ok(LookupRequestResult::Pending("no peers")); + }; + let id = SingleLookupReqId { lookup_id, req_id: self.next_id(), @@ -1046,7 +1063,7 @@ impl SyncNetworkContext { request_span, ); - Ok(id.req_id) + Ok(LookupRequestResult::RequestSent(id.req_id)) } /// Request necessary blobs for `block_root`. Requests only the necessary blobs by checking: @@ -2015,6 +2032,10 @@ impl SyncNetworkContext { "data_columns_by_range", self.data_columns_by_range_requests.len(), ), + ( + "payload_envelopes_by_root", + self.payload_envelopes_by_root_requests.len(), + ), ( "payload_envelopes_by_range", self.payload_envelopes_by_range_requests.len(), diff --git a/common/task_executor/src/lib.rs b/common/task_executor/src/lib.rs index d3d862f96c..07716fa2e7 100644 --- a/common/task_executor/src/lib.rs +++ b/common/task_executor/src/lib.rs @@ -6,7 +6,7 @@ use futures::channel::mpsc::Sender; use futures::prelude::*; use std::sync::{Arc, Weak}; use tokio::runtime::{Handle, Runtime}; -use tracing::debug; +use tracing::{Span, debug}; use crate::rayon_pool_provider::RayonPoolProvider; pub use crate::rayon_pool_provider::RayonPoolType; @@ -225,9 +225,11 @@ impl TaskExecutor { F: FnOnce() + Send + 'static, { let thread_pool = self.rayon_pool_provider.get_thread_pool(rayon_pool_type); + let span = Span::current(); self.spawn_blocking( move || { thread_pool.install(|| { + let _guard = span.enter(); task(); }); }, @@ -247,8 +249,10 @@ impl TaskExecutor { { let thread_pool = self.rayon_pool_provider.get_thread_pool(rayon_pool_type); let (tx, rx) = tokio::sync::oneshot::channel(); + let span = Span::current(); thread_pool.spawn(move || { + let _guard = span.enter(); let result = task(); let _ = tx.send(result); }); @@ -320,8 +324,12 @@ impl TaskExecutor { let timer = metrics::start_timer_vec(&metrics::BLOCKING_TASKS_HISTOGRAM, &[name]); metrics::inc_gauge_vec(&metrics::BLOCKING_TASKS_COUNT, &[name]); + let span = Span::current(); let join_handle = if let Some(handle) = self.handle() { - handle.spawn_blocking(task) + handle.spawn_blocking(move || { + let _guard = span.enter(); + task() + }) } else { debug!("Couldn't spawn task. Runtime shutting down"); return None; diff --git a/consensus/fork_choice/src/fork_choice.rs b/consensus/fork_choice/src/fork_choice.rs index 3b13cd4429..92fd4c1faf 100644 --- a/consensus/fork_choice/src/fork_choice.rs +++ b/consensus/fork_choice/src/fork_choice.rs @@ -26,6 +26,7 @@ use types::{ #[derive(Debug)] pub enum Error { InvalidAttestation(InvalidAttestation), + InvalidPayloadAttestation(InvalidPayloadAttestation), InvalidAttesterSlashing(AttesterSlashingValidationError), InvalidBlock(InvalidBlock), ProtoArrayStringError(String), @@ -85,6 +86,12 @@ impl From for Error { } } +impl From for Error { + fn from(e: InvalidPayloadAttestation) -> Self { + Error::InvalidPayloadAttestation(e) + } +} + impl From for Error { fn from(e: AttesterSlashingValidationError) -> Self { Error::InvalidAttesterSlashing(e) @@ -138,10 +145,6 @@ pub enum InvalidBlock { finalized_root: Hash256, block_ancestor: Option, }, - MissingExecutionPayloadBid { - block_slot: Slot, - block_root: Hash256, - }, } #[derive(Debug)] @@ -174,21 +177,33 @@ pub enum InvalidAttestation { /// The attestation is attesting to a state that is later than itself. (Viz., attesting to the /// future). AttestsToFutureBlock { block: Slot, attestation: Slot }, - /// Post-GLOAS: attestation index must be 0 or 1. + /// Post-Gloas: attestation index must be 0 or 1. InvalidAttestationIndex { index: u64 }, - /// A same-slot attestation has a non-zero index, which is invalid post-GLOAS. + /// A same-slot attestation has a non-zero index, which is invalid post-Gloas. InvalidSameSlotAttestationIndex { slot: Slot }, - /// Post-GLOAS: attestation with index == 1 (payload_present) requires the block's + /// Post-Gloas: attestation with index == 1 (payload_present) requires the block's /// payload to have been received (`root in store.payload_states`). PayloadNotReceived { beacon_block_root: Hash256 }, - /// A payload attestation votes payload_present for a block in the current slot, which is - /// invalid because the payload cannot be known yet. - PayloadPresentDuringSameSlot { slot: Slot }, +} + +#[derive(Debug, Clone, PartialEq)] +pub enum InvalidPayloadAttestation { + /// The payload attestation's attesting indices were empty. + EmptyAggregationBitfield, + /// The `payload_attestation.data.beacon_block_root` block is unknown. + UnknownHeadBlock { beacon_block_root: Hash256 }, + /// The payload attestation is attesting to a block that is later than itself. + AttestsToFutureBlock { block: Slot, attestation: Slot }, /// A gossip payload attestation must be for the current slot. PayloadAttestationNotCurrentSlot { attestation_slot: Slot, current_slot: Slot, }, + /// One or more payload attesters are not part of the PTC. + PayloadAttestationAttestersNotInPtc { + attesting_indices_len: usize, + attesting_indices_in_ptc: usize, + }, } impl From for Error { @@ -260,10 +275,19 @@ pub struct QueuedAttestation { attesting_indices: Vec, block_root: Hash256, target_epoch: Epoch, - /// Per GLOAS spec: `payload_present = attestation.data.index == 1`. + /// Per Gloas spec: `payload_present = attestation.data.index == 1`. payload_present: bool, } +/// Legacy queued attestation without payload_present (pre-Gloas, schema V28). +#[derive(Clone, PartialEq, Encode, Decode)] +pub struct QueuedAttestationV28 { + slot: Slot, + attesting_indices: Vec, + block_root: Hash256, + target_epoch: Epoch, +} + impl<'a, E: EthSpec> From> for QueuedAttestation { fn from(a: IndexedAttestationRef<'a, E>) -> Self { Self { @@ -276,19 +300,6 @@ impl<'a, E: EthSpec> From> for QueuedAttestation { } } -/// Used for queuing payload attestations (PTC votes) from the current slot. -/// Payload attestations have different dequeue timing than regular attestations: -/// gossiped payload attestations need an extra slot of delay (slot + 1 < current_slot). -#[derive(Clone, PartialEq, Encode, Decode)] -pub struct QueuedPayloadAttestation { - slot: Slot, - /// Resolved PTC committee positions (not validator indices). - ptc_indices: Vec, - block_root: Hash256, - payload_present: bool, - blob_data_available: bool, -} - /// Returns all values in `self.queued_attestations` that have a slot that is earlier than the /// current slot. Also removes those values from `self.queued_attestations`. fn dequeue_attestations( @@ -310,22 +321,6 @@ fn dequeue_attestations( std::mem::replace(queued_attestations, remaining) } -/// Returns all values in `queued` that have `slot + 1 < current_slot`. -/// Payload attestations need an extra slot of delay compared to regular attestations. -fn dequeue_payload_attestations( - current_slot: Slot, - queued: &mut Vec, -) -> Vec { - let remaining = queued.split_off( - queued - .iter() - .position(|a| a.slot.saturating_add(1_u64) >= current_slot) - .unwrap_or(queued.len()), - ); - - std::mem::replace(queued, remaining) -} - /// Denotes whether an attestation we are processing was received from a block or from gossip. /// Equivalent to the `is_from_block` `bool` in: /// @@ -370,9 +365,6 @@ pub struct ForkChoice { proto_array: ProtoArrayForkChoice, /// Attestations that arrived at the current slot and must be queued for later processing. queued_attestations: Vec, - /// Payload attestations (PTC votes) that must be queued for later processing. - /// These have different dequeue timing than regular attestations. - queued_payload_attestations: Vec, /// Stores a cache of the values required to be sent to the execution layer. forkchoice_update_parameters: ForkchoiceUpdateParameters, _phantom: PhantomData, @@ -387,7 +379,6 @@ where self.fc_store == other.fc_store && self.proto_array == other.proto_array && self.queued_attestations == other.queued_attestations - && self.queued_payload_attestations == other.queued_payload_attestations } } @@ -423,22 +414,7 @@ where .map_err(Error::BeaconStateError)?; let (execution_status, execution_payload_parent_hash, execution_payload_block_hash) = - if let Ok(execution_payload) = anchor_block.message().execution_payload() { - // Pre-Gloas forks: hashes come from the execution payload. - if execution_payload.is_default_with_empty_roots() { - (ExecutionStatus::irrelevant(), None, None) - } else { - // Assume that this payload is valid, since the anchor should be a - // trusted block and state. - ( - ExecutionStatus::Valid(execution_payload.block_hash()), - Some(execution_payload.parent_hash()), - Some(execution_payload.block_hash()), - ) - } - } else if let Ok(signed_bid) = - anchor_block.message().body().signed_execution_payload_bid() - { + if let Ok(signed_bid) = anchor_block.message().body().signed_execution_payload_bid() { // Gloas: execution status is irrelevant post-Gloas; payload validation // is decoupled from beacon blocks. ( @@ -446,6 +422,19 @@ where Some(signed_bid.message.parent_block_hash), Some(signed_bid.message.block_hash), ) + } else if let Ok(execution_payload) = anchor_block.message().execution_payload() { + // Pre-Gloas forks: do not set payload hashes, they are only used post-Gloas. + if execution_payload.is_default_with_empty_roots() { + (ExecutionStatus::irrelevant(), None, None) + } else { + // Assume that this payload is valid, since the anchor should be a + // trusted block and state. + ( + ExecutionStatus::Valid(execution_payload.block_hash()), + None, + None, + ) + } } else { // Pre-merge: no execution payload at all. (ExecutionStatus::irrelevant(), None, None) @@ -465,6 +454,7 @@ where execution_status, execution_payload_parent_hash, execution_payload_block_hash, + anchor_block.message().proposer_index(), spec, )?; @@ -472,20 +462,19 @@ where fc_store, proto_array, queued_attestations: vec![], - queued_payload_attestations: vec![], // This will be updated during the next call to `Self::get_head`. forkchoice_update_parameters: ForkchoiceUpdateParameters { head_hash: None, justified_hash: None, finalized_hash: None, - // These will be updated during the next call to `Self::get_head`. + // This will be updated during the next call to `Self::get_head`. head_root: Hash256::zero(), }, _phantom: PhantomData, }; // Ensure that `fork_choice.forkchoice_update_parameters.head_root` is updated. - let _ = fork_choice.get_head(current_slot, spec)?; + fork_choice.get_head(current_slot, spec)?; Ok(fork_choice) } @@ -684,6 +673,20 @@ where } } + /// Mark a Gloas payload envelope as valid and received. + /// + /// This must only be called for valid Gloas payloads. + pub fn on_valid_payload_envelope_received( + &mut self, + block_root: Hash256, + ) -> Result<(), Error> { + self.proto_array + .on_valid_payload_envelope_received(block_root) + .map_err(Error::FailedToProcessValidExecutionPayload) + } + + /// Pre-Gloas only. + /// /// See `ProtoArrayForkChoice::process_execution_payload_validation` for documentation. pub fn on_valid_execution_payload( &mut self, @@ -694,6 +697,8 @@ where .map_err(Error::FailedToProcessValidExecutionPayload) } + /// Pre-Gloas only. + /// /// See `ProtoArrayForkChoice::process_execution_payload_invalidation` for documentation. pub fn on_invalid_execution_payload( &mut self, @@ -966,14 +971,6 @@ where Some(signed_bid.message.block_hash), ) } else { - if spec.fork_name_at_slot::(block.slot()).gloas_enabled() { - return Err(Error::InvalidBlock( - InvalidBlock::MissingExecutionPayloadBid { - block_slot: block.slot(), - block_root, - }, - )); - } (None, None) }; @@ -1015,12 +1012,6 @@ where Ok(()) } - pub fn on_execution_payload(&mut self, block_root: Hash256) -> Result<(), Error> { - self.proto_array - .on_execution_payload(block_root) - .map_err(Error::FailedToProcessValidExecutionPayload) - } - /// Update checkpoints in store if necessary fn update_checkpoints( &mut self, @@ -1163,7 +1154,7 @@ where { let index = indexed_attestation.data().index; - // Post-GLOAS: attestation index must be 0 or 1. + // Post-Gloas: attestation index must be 0 or 1. if index > 1 { return Err(InvalidAttestation::InvalidAttestationIndex { index }); } @@ -1176,6 +1167,7 @@ where } // index == 1 (payload_present) requires the block's payload to have been received. + // TODO(gloas): could optimise by adding `payload_received` to `Block` if index == 1 && !self .proto_array @@ -1195,50 +1187,48 @@ where &self, indexed_payload_attestation: &IndexedPayloadAttestation, is_from_block: AttestationFromBlock, - ) -> Result<(), InvalidAttestation> { + ) -> Result<(), InvalidPayloadAttestation> { + // This check is from `is_valid_indexed_payload_attestation`, but we do it immediately to + // avoid wasting time on junk attestations. if indexed_payload_attestation.attesting_indices.is_empty() { - return Err(InvalidAttestation::EmptyAggregationBitfield); + return Err(InvalidPayloadAttestation::EmptyAggregationBitfield); } + // PTC attestation must be for a known block. If block is unknown, delay consideration until + // the block is found (responsibility of caller). let block = self .proto_array .get_block(&indexed_payload_attestation.data.beacon_block_root) - .ok_or(InvalidAttestation::UnknownHeadBlock { + .ok_or(InvalidPayloadAttestation::UnknownHeadBlock { beacon_block_root: indexed_payload_attestation.data.beacon_block_root, })?; + // Not strictly part of the spec, but payload attestations to future slots are MORE INVALID + // than payload attestations to blocks at previous slots. if block.slot > indexed_payload_attestation.data.slot { - return Err(InvalidAttestation::AttestsToFutureBlock { + return Err(InvalidPayloadAttestation::AttestsToFutureBlock { block: block.slot, attestation: indexed_payload_attestation.data.slot, }); } - // Spec: `if data.slot != state.slot: return` — PTC votes can only - // change the vote for their assigned beacon block. + // PTC votes can only change the vote for their assigned beacon block, return early otherwise if block.slot != indexed_payload_attestation.data.slot { return Ok(()); } // Gossip payload attestations must be for the current slot. + // NOTE: signature is assumed to have been verified by caller. // https://github.com/ethereum/consensus-specs/blob/master/specs/gloas/fork-choice.md if matches!(is_from_block, AttestationFromBlock::False) && indexed_payload_attestation.data.slot != self.fc_store.get_current_slot() { - return Err(InvalidAttestation::PayloadAttestationNotCurrentSlot { - attestation_slot: indexed_payload_attestation.data.slot, - current_slot: self.fc_store.get_current_slot(), - }); - } - - // A payload attestation voting payload_present for a block in the current slot is - // invalid: the payload cannot be known yet. This only applies to gossip attestations; - // payload attestations from blocks have already been validated by the block producer. - if matches!(is_from_block, AttestationFromBlock::False) - && self.fc_store.get_current_slot() == block.slot - && indexed_payload_attestation.data.payload_present - { - return Err(InvalidAttestation::PayloadPresentDuringSameSlot { slot: block.slot }); + return Err( + InvalidPayloadAttestation::PayloadAttestationNotCurrentSlot { + attestation_slot: indexed_payload_attestation.data.slot, + current_slot: self.fc_store.get_current_slot(), + }, + ); } Ok(()) @@ -1334,46 +1324,40 @@ where ) -> Result<(), Error> { self.update_time(system_time_current_slot)?; - if attestation.data.beacon_block_root == Hash256::zero() { + if attestation.data.beacon_block_root.is_zero() { return Ok(()); } + // TODO(gloas): Should ignore wrong-slot payload attestations at the caller, they could + // have been processed at the correct slot when received on gossip, but then have the + // wrong-slot by the time they make it to here (TOCTOU). self.validate_on_payload_attestation(attestation, is_from_block)?; // Resolve validator indices to PTC committee positions. let ptc_indices: Vec = attestation - .attesting_indices_iter() + .attesting_indices + .iter() .filter_map(|vi| ptc.iter().position(|&p| p == *vi as usize)) .collect(); - let processing_slot = self.fc_store.get_current_slot(); - // Payload attestations from blocks can be applied in the next slot (S+1 for data.slot=S), - // while gossiped payload attestations are delayed one extra slot. - let should_process_now = match is_from_block { - AttestationFromBlock::True => attestation.data.slot < processing_slot, - AttestationFromBlock::False => { - attestation.data.slot.saturating_add(1_u64) < processing_slot - } - }; + // Check that all the attesters are in the PTC + if ptc_indices.len() != attestation.attesting_indices.len() { + return Err( + InvalidPayloadAttestation::PayloadAttestationAttestersNotInPtc { + attesting_indices_len: attestation.attesting_indices.len(), + attesting_indices_in_ptc: ptc_indices.len(), + } + .into(), + ); + } - if should_process_now { - for &ptc_index in &ptc_indices { - self.proto_array.process_payload_attestation( - attestation.data.beacon_block_root, - ptc_index, - attestation.data.payload_present, - attestation.data.blob_data_available, - )?; - } - } else { - self.queued_payload_attestations - .push(QueuedPayloadAttestation { - slot: attestation.data.slot, - ptc_indices, - block_root: attestation.data.beacon_block_root, - payload_present: attestation.data.payload_present, - blob_data_available: attestation.data.blob_data_available, - }); + for &ptc_index in &ptc_indices { + self.proto_array.process_payload_attestation( + attestation.data.beacon_block_root, + ptc_index, + attestation.data.payload_present, + attestation.data.blob_data_available, + )?; } Ok(()) @@ -1408,7 +1392,6 @@ where // Process any attestations that might now be eligible. self.process_attestation_queue()?; - self.process_payload_attestation_queue()?; Ok(self.fc_store.get_current_slot()) } @@ -1495,26 +1478,6 @@ where Ok(()) } - /// Processes and removes from the queue any queued payload attestations which may now be - /// eligible for processing. Payload attestations use `slot + 1 < current_slot` timing. - fn process_payload_attestation_queue(&mut self) -> Result<(), Error> { - let current_slot = self.fc_store.get_current_slot(); - for attestation in - dequeue_payload_attestations(current_slot, &mut self.queued_payload_attestations) - { - for &ptc_index in &attestation.ptc_indices { - self.proto_array.process_payload_attestation( - attestation.block_root, - ptc_index, - attestation.payload_present, - attestation.blob_data_available, - )?; - } - } - - Ok(()) - } - /// Returns `true` if the block is known **and** a descendant of the finalized root. pub fn contains_block(&self, block_root: &Hash256) -> bool { self.proto_array.contains_block(block_root) @@ -1670,11 +1633,6 @@ where &self.queued_attestations } - /// Returns a reference to the currently queued payload attestations. - pub fn queued_payload_attestations(&self) -> &[QueuedPayloadAttestation] { - &self.queued_payload_attestations - } - /// Returns the store's `proposer_boost_root`. pub fn proposer_boost_root(&self) -> Hash256 { self.fc_store.proposer_boost_root() @@ -1695,7 +1653,6 @@ where persisted_proto_array: proto_array::core::SszContainer, justified_balances: JustifiedBalances, reset_payload_statuses: ResetPayloadStatuses, - spec: &ChainSpec, ) -> Result> { let mut proto_array = ProtoArrayForkChoice::from_container( persisted_proto_array.clone(), @@ -1720,7 +1677,7 @@ where // Reset all blocks back to being "optimistic". This helps recover from an EL consensus // fault where an invalid payload becomes valid. - if let Err(e) = proto_array.set_all_blocks_to_optimistic::(spec) { + if let Err(e) = proto_array.set_all_blocks_to_optimistic::() { // If there is an error resetting the optimistic status then log loudly and revert // back to a proto-array which does not have the reset applied. This indicates a // significant error in Lighthouse and warrants detailed investigation. @@ -1750,7 +1707,6 @@ where persisted.proto_array, justified_balances, reset_payload_statuses, - spec, )?; let current_slot = fc_store.get_current_slot(); @@ -1758,8 +1714,7 @@ where let mut fork_choice = Self { fc_store, proto_array, - queued_attestations: persisted.queued_attestations, - queued_payload_attestations: persisted.queued_payload_attestations, + queued_attestations: vec![], // Will be updated in the following call to `Self::get_head`. forkchoice_update_parameters: ForkchoiceUpdateParameters { head_hash: None, @@ -1785,10 +1740,10 @@ where // get a different result. fork_choice .proto_array - .set_all_blocks_to_optimistic::(spec)?; + .set_all_blocks_to_optimistic::()?; // If the second attempt at finding a head fails, return an error since we do not // expect this scenario. - let _ = fork_choice.get_head(current_slot, spec)?; + fork_choice.get_head(current_slot, spec)?; } Ok(fork_choice) @@ -1799,8 +1754,6 @@ where pub fn to_persisted(&self) -> PersistedForkChoice { PersistedForkChoice { proto_array: self.proto_array().as_ssz_container(), - queued_attestations: self.queued_attestations().to_vec(), - queued_payload_attestations: self.queued_payload_attestations.clone(), } } @@ -1823,9 +1776,8 @@ pub struct PersistedForkChoice { pub proto_array_v28: proto_array::core::SszContainerV28, #[superstruct(only(V29))] pub proto_array: proto_array::core::SszContainerV29, - pub queued_attestations: Vec, - #[superstruct(only(V29))] - pub queued_payload_attestations: Vec, + #[superstruct(only(V28))] + pub queued_attestations_v28: Vec, } pub type PersistedForkChoice = PersistedForkChoiceV29; @@ -1834,8 +1786,15 @@ impl From for PersistedForkChoiceV29 { fn from(v28: PersistedForkChoiceV28) -> Self { Self { proto_array: v28.proto_array_v28.into(), - queued_attestations: v28.queued_attestations, - queued_payload_attestations: vec![], + } + } +} + +impl From for PersistedForkChoiceV28 { + fn from(v29: PersistedForkChoiceV29) -> Self { + Self { + proto_array_v28: v29.proto_array.into(), + queued_attestations_v28: vec![], } } } diff --git a/consensus/fork_choice/src/lib.rs b/consensus/fork_choice/src/lib.rs index 70f1dbc215..159eab0ec0 100644 --- a/consensus/fork_choice/src/lib.rs +++ b/consensus/fork_choice/src/lib.rs @@ -4,8 +4,8 @@ mod metrics; pub use crate::fork_choice::{ AttestationFromBlock, Error, ForkChoice, ForkChoiceView, ForkchoiceUpdateParameters, - InvalidAttestation, InvalidBlock, PayloadVerificationStatus, PersistedForkChoice, - PersistedForkChoiceV28, PersistedForkChoiceV29, QueuedAttestation, QueuedPayloadAttestation, + InvalidAttestation, InvalidBlock, InvalidPayloadAttestation, PayloadVerificationStatus, + PersistedForkChoice, PersistedForkChoiceV28, PersistedForkChoiceV29, QueuedAttestation, ResetPayloadStatuses, }; pub use fork_choice_store::ForkChoiceStore; diff --git a/consensus/fork_choice/tests/tests.rs b/consensus/fork_choice/tests/tests.rs index 839d0f4c5c..d6f937c0ca 100644 --- a/consensus/fork_choice/tests/tests.rs +++ b/consensus/fork_choice/tests/tests.rs @@ -11,7 +11,7 @@ use bls::AggregateSignature; use fixed_bytes::FixedBytesExtended; use fork_choice::{ AttestationFromBlock, ForkChoiceStore, InvalidAttestation, InvalidBlock, - PayloadVerificationStatus, QueuedAttestation, + InvalidPayloadAttestation, PayloadVerificationStatus, QueuedAttestation, }; use state_processing::state_advance::complete_state_advance; use std::fmt; @@ -73,9 +73,9 @@ impl ForkChoiceTest { Self { harness } } - /// Creates a new tester with the GLOAS fork active at epoch 1. + /// Creates a new tester with the Gloas fork active at epoch 1. /// Genesis is a standard Fulu block (epoch 0), so block production works normally. - /// Tests that need GLOAS semantics should advance the chain into epoch 1 first. + /// Tests that need Gloas semantics should advance the chain into epoch 1 first. /// Get a value from the `ForkChoice` instantiation. fn get(&self, func: T) -> U where @@ -969,8 +969,8 @@ async fn non_block_payload_attestation_for_previous_slot_is_rejected() { assert!( matches!( result, - Err(ForkChoiceError::InvalidAttestation( - InvalidAttestation::PayloadAttestationNotCurrentSlot { .. } + Err(ForkChoiceError::InvalidPayloadAttestation( + InvalidPayloadAttestation::PayloadAttestationNotCurrentSlot { .. } )) ), "gossip payload attestation for previous slot should be rejected, got: {:?}", diff --git a/consensus/proto_array/src/fork_choice_test_definition.rs b/consensus/proto_array/src/fork_choice_test_definition.rs index 34d7f2e48e..c9764d3e44 100644 --- a/consensus/proto_array/src/fork_choice_test_definition.rs +++ b/consensus/proto_array/src/fork_choice_test_definition.rs @@ -98,7 +98,7 @@ pub enum Operation { }, /// Simulate receiving and validating an execution payload for `block_root`. /// Sets `payload_received = true` on the V29 node via the live validation path. - ProcessExecutionPayload { + ProcessExecutionPayloadEnvelope { block_root: Hash256, }, AssertPayloadReceived { @@ -144,6 +144,7 @@ impl ForkChoiceTestDefinition { ExecutionStatus::Optimistic(ExecutionBlockHash::zero()), self.execution_payload_parent_hash, self.execution_payload_block_hash, + 0, &spec, ) .expect("should create fork choice struct"); @@ -499,9 +500,9 @@ impl ForkChoiceTestDefinition { // the payload to be in payload_states (payload_received). node_v29.payload_received = is_timely || is_data_available; } - Operation::ProcessExecutionPayload { block_root } => { + Operation::ProcessExecutionPayloadEnvelope { block_root } => { fork_choice - .on_execution_payload(block_root) + .on_valid_payload_envelope_received(block_root) .unwrap_or_else(|e| { panic!( "on_execution_payload op at index {} returned error: {}", diff --git a/consensus/proto_array/src/fork_choice_test_definition/gloas_payload.rs b/consensus/proto_array/src/fork_choice_test_definition/gloas_payload.rs index 0fb120328c..ea37780795 100644 --- a/consensus/proto_array/src/fork_choice_test_definition/gloas_payload.rs +++ b/consensus/proto_array/src/fork_choice_test_definition/gloas_payload.rs @@ -52,8 +52,8 @@ pub fn get_gloas_chain_following_test_definition() -> ForkChoiceTestDefinition { }); // Mark root_1 as having received its execution payload so that - // its FULL virtual node exists in the GLOAS fork choice tree. - ops.push(Operation::ProcessExecutionPayload { + // its FULL virtual node exists in the Gloas fork choice tree. + ops.push(Operation::ProcessExecutionPayloadEnvelope { block_root: get_root(1), }); @@ -262,8 +262,8 @@ pub fn get_gloas_find_head_vote_transition_test_definition() -> ForkChoiceTestDe }); // Mark root_1 as having received its execution payload so that - // its FULL virtual node exists in the GLOAS fork choice tree. - ops.push(Operation::ProcessExecutionPayload { + // its FULL virtual node exists in the Gloas fork choice tree. + ops.push(Operation::ProcessExecutionPayloadEnvelope { block_root: get_root(1), }); @@ -367,8 +367,8 @@ pub fn get_gloas_weight_priority_over_payload_preference_test_definition() }); // Mark root_1 as having received its execution payload so that - // its FULL virtual node exists in the GLOAS fork choice tree. - ops.push(Operation::ProcessExecutionPayload { + // its FULL virtual node exists in the Gloas fork choice tree. + ops.push(Operation::ProcessExecutionPayloadEnvelope { block_root: get_root(1), }); @@ -537,8 +537,8 @@ pub fn get_gloas_interleaved_attestations_test_definition() -> ForkChoiceTestDef }); // Mark root_1 as having received its execution payload so that - // its FULL virtual node exists in the GLOAS fork choice tree. - ops.push(Operation::ProcessExecutionPayload { + // its FULL virtual node exists in the Gloas fork choice tree. + ops.push(Operation::ProcessExecutionPayloadEnvelope { block_root: get_root(1), }); @@ -674,8 +674,8 @@ pub fn get_gloas_payload_received_interleaving_test_definition() -> ForkChoiceTe expected_payload_status: None, }); - // ProcessExecutionPayload on genesis is a no-op (already received at init). - ops.push(Operation::ProcessExecutionPayload { + // ProcessExecutionPayloadEnvelope on genesis is a no-op (already received at init). + ops.push(Operation::ProcessExecutionPayloadEnvelope { block_root: get_root(0), }); @@ -718,6 +718,137 @@ pub fn get_gloas_payload_received_interleaving_test_definition() -> ForkChoiceTe mod tests { use super::*; + fn gloas_fork_boundary_spec() -> ChainSpec { + let mut spec = MainnetEthSpec::default_spec(); + spec.proposer_score_boost = Some(50); + spec.gloas_fork_epoch = Some(Epoch::new(1)); + spec + } + + /// Gloas fork boundary: a chain starting pre-Gloas (V17 nodes) that crosses into + /// Gloas (V29 nodes). The head should advance through the fork boundary. + /// + /// Parameters: + /// - `skip_first_gloas_slot`: if true, there is no block at the first Gloas slot (slot 32); + /// the first V29 block appears at slot 33. + /// - `first_gloas_block_full`: if true, the first V29 block extends the parent V17 node's + /// EL chain (Full parent payload status). If false, it doesn't (Empty). + fn get_gloas_fork_boundary_test_definition( + skip_first_gloas_slot: bool, + first_gloas_block_full: bool, + ) -> ForkChoiceTestDefinition { + let mut ops = vec![]; + + // Block at slot 31 — last pre-Gloas slot. Created as a V17 node because + // gloas_fork_epoch = 1 → Gloas starts at slot 32. + // + // The test harness sets execution_status = Optimistic(ExecutionBlockHash::from_root(root)), + // so this V17 node's EL block hash = ExecutionBlockHash::from_root(get_root(1)). + ops.push(Operation::ProcessBlock { + slot: Slot::new(31), + root: get_root(1), + parent_root: get_root(0), + justified_checkpoint: get_checkpoint(0), + finalized_checkpoint: get_checkpoint(0), + execution_payload_parent_hash: None, + execution_payload_block_hash: None, + }); + + // First Gloas block (V29 node). + let gloas_slot = if skip_first_gloas_slot { 33 } else { 32 }; + + // The first Gloas block should always have the pre-Gloas block as its execution parent, + // although this is currently not checked anywhere (the spec doesn't mention this). + ops.push(Operation::ProcessBlock { + slot: Slot::new(gloas_slot), + root: get_root(2), + parent_root: get_root(1), + justified_checkpoint: get_checkpoint(0), + finalized_checkpoint: get_checkpoint(0), + execution_payload_parent_hash: Some(get_hash(1)), + execution_payload_block_hash: Some(get_hash(2)), + }); + + // Parent payload status of fork boundary block should always be Empty. + let expected_parent_status = PayloadStatus::Empty; + ops.push(Operation::AssertParentPayloadStatus { + block_root: get_root(2), + expected_status: expected_parent_status, + }); + + // Mark root 2's execution payload as received so the Full virtual child exists. + if first_gloas_block_full { + ops.push(Operation::ProcessExecutionPayloadEnvelope { + block_root: get_root(2), + }); + } + + // Extend the chain with another V29 block (Full child of root 2). + ops.push(Operation::ProcessBlock { + slot: Slot::new(gloas_slot + 1), + root: get_root(3), + parent_root: get_root(2), + justified_checkpoint: get_checkpoint(0), + finalized_checkpoint: get_checkpoint(0), + execution_payload_parent_hash: if first_gloas_block_full { + Some(get_hash(2)) + } else { + Some(get_hash(1)) + }, + execution_payload_block_hash: Some(get_hash(3)), + }); + + // Head should advance to the tip of the chain through the fork boundary. + ops.push(Operation::FindHead { + justified_checkpoint: get_checkpoint(0), + finalized_checkpoint: get_checkpoint(0), + justified_state_balances: vec![1], + expected_head: get_root(3), + current_slot: Slot::new(gloas_slot + 1), + expected_payload_status: None, + }); + + ops.push(Operation::AssertParentPayloadStatus { + block_root: get_root(3), + expected_status: if first_gloas_block_full { + PayloadStatus::Full + } else { + PayloadStatus::Empty + }, + }); + + ForkChoiceTestDefinition { + finalized_block_slot: Slot::new(0), + justified_checkpoint: get_checkpoint(0), + finalized_checkpoint: get_checkpoint(0), + operations: ops, + // Genesis is V17 (slot 0 < Gloas fork slot 32), these are unused for V17. + execution_payload_parent_hash: None, + execution_payload_block_hash: None, + spec: Some(gloas_fork_boundary_spec()), + } + } + + #[test] + fn fork_boundary_no_skip_full() { + get_gloas_fork_boundary_test_definition(false, true).run(); + } + + #[test] + fn fork_boundary_no_skip_empty() { + get_gloas_fork_boundary_test_definition(false, false).run(); + } + + #[test] + fn fork_boundary_skip_first_gloas_slot_full() { + get_gloas_fork_boundary_test_definition(true, true).run(); + } + + #[test] + fn fork_boundary_skip_first_gloas_slot_empty() { + get_gloas_fork_boundary_test_definition(true, false).run(); + } + #[test] fn chain_following() { let test = get_gloas_chain_following_test_definition(); diff --git a/consensus/proto_array/src/proto_array.rs b/consensus/proto_array/src/proto_array.rs index d806547cc0..1f7291b260 100644 --- a/consensus/proto_array/src/proto_array.rs +++ b/consensus/proto_array/src/proto_array.rs @@ -117,10 +117,10 @@ pub struct ProtoNode { pub finalized_checkpoint: Checkpoint, #[superstruct(getter(copy))] pub weight: u64, - #[superstruct(getter(copy))] + #[superstruct(only(V17), partial_getter(copy))] #[ssz(with = "four_byte_option_usize")] pub best_child: Option, - #[superstruct(getter(copy))] + #[superstruct(only(V17), partial_getter(copy))] #[ssz(with = "four_byte_option_usize")] pub best_descendant: Option, /// Indicates if an execution node has marked this block as valid. Also contains the execution @@ -143,6 +143,8 @@ pub struct ProtoNode { pub full_payload_weight: u64, #[superstruct(only(V29), partial_getter(copy))] pub execution_payload_block_hash: ExecutionBlockHash, + #[superstruct(only(V29), partial_getter(copy))] + pub execution_payload_parent_hash: ExecutionBlockHash, /// Equivalent to spec's `block_timeliness[root][ATTESTATION_TIMELINESS_INDEX]`. #[superstruct(only(V29), partial_getter(copy))] pub block_timeliness_attestation_threshold: bool, @@ -163,8 +165,6 @@ pub struct ProtoNode { pub payload_data_availability_votes: BitVector, /// Whether the execution payload for this block has been received and validated locally. /// Maps to `root in store.payload_states` in the spec. - /// When true, `is_payload_timely` and `is_payload_data_available` return true - /// regardless of PTC vote counts. #[superstruct(only(V29), partial_getter(copy))] pub payload_received: bool, /// The proposer index for this block, used by `should_apply_proposer_boost` @@ -181,7 +181,6 @@ pub struct ProtoNode { impl ProtoNode { /// Generic version of spec's `parent_payload_status` that works for pre-Gloas nodes by /// considering their parents Empty. - /// Pre-Gloas nodes have no ePBS, default to Empty. pub fn get_parent_payload_status(&self) -> PayloadStatus { self.parent_payload_status().unwrap_or(PayloadStatus::Empty) } @@ -212,7 +211,7 @@ impl ProtoNode { return false; } - node.payload_timeliness_votes.num_set_bits() > E::ptc_size() / 2 + node.payload_timeliness_votes.num_set_bits() > E::payload_timely_threshold() } pub fn is_payload_data_available(&self) -> bool { @@ -225,8 +224,8 @@ impl ProtoNode { return false; } - // TODO(gloas): add function on EthSpec for DATA_AVAILABILITY_TIMELY_THRESHOLD - node.payload_data_availability_votes.num_set_bits() > E::ptc_size() / 2 + node.payload_data_availability_votes.num_set_bits() + > E::data_availability_timely_threshold() } } @@ -368,7 +367,6 @@ pub struct ProtoArray { pub prune_threshold: usize, pub nodes: Vec, pub indices: HashMap, - pub previous_proposer_boost: ProposerBoost, } impl ProtoArray { @@ -491,20 +489,14 @@ impl ProtoArray { .ok_or(Error::DeltaOverflow(parent_index))?; } } else { - // V17 child of a V29 parent (fork transition): treat as FULL - // since V17 nodes always have execution payloads inline. - parent_delta.full_delta = parent_delta - .full_delta - .checked_add(delta) - .ok_or(Error::DeltaOverflow(parent_index))?; + // This is a v17 node with a v17 parent. + // There is no empty or full weight for v17 nodes, so nothing to propagate. + // In the tree walk, the v17 nodes have an empty child with 0 weight, which + // wins by default (it is the only child). } } } - // Proposer boost is now applied on-the-fly in `get_weight` during the - // walk, so clear any stale boost from a prior call. - self.previous_proposer_boost = ProposerBoost::default(); - Ok(()) } @@ -570,31 +562,31 @@ impl ProtoArray { block_root: block.root, })?; - let parent_payload_status: PayloadStatus = if let Some(parent_node) = - parent_index.and_then(|idx| self.nodes.get(idx)) - { - // Get the parent's execution block hash, handling both V17 and V29 nodes. - // V17 parents occur during the Gloas fork transition. - // TODO(gloas): the spec's `get_parent_payload_status` assumes all blocks are - // post-Gloas with bids. Revisit once the spec clarifies fork-transition behavior. - let parent_el_block_hash = match parent_node { - ProtoNode::V29(v29) => Some(v29.execution_payload_block_hash), - ProtoNode::V17(v17) => v17.execution_status.block_hash(), - }; - // Per spec's `is_parent_node_full`: if the child's EL parent hash - // matches the parent's EL block hash, the child extends the parent's - // payload chain, meaning the parent was Full. - if parent_el_block_hash.is_some_and(|hash| execution_payload_parent_hash == hash) { - PayloadStatus::Full + let parent_payload_status: PayloadStatus = + if let Some(parent_node) = parent_index.and_then(|idx| self.nodes.get(idx)) { + match parent_node { + ProtoNode::V29(v29) => { + // Both parent and child are Gloas blocks. The parent is full if the + // block hash in the parent node matches the parent block hash in the + // child bid. + if execution_payload_parent_hash == v29.execution_payload_block_hash { + PayloadStatus::Full + } else { + PayloadStatus::Empty + } + } + ProtoNode::V17(_) => { + // Parent is pre-Gloas, pre-Gloas blocks are treated as having Empty + // payload status. This case is reached during the fork transition. + PayloadStatus::Empty + } + } } else { - PayloadStatus::Empty - } - } else { - // Parent is missing (genesis or pruned due to finalization). Default to Full - // since this path should only be hit at Gloas genesis, and extending the payload - // chain is the safe default. - PayloadStatus::Full - }; + // TODO(gloas): re-assess this assumption + // Parent is missing (genesis or pruned due to finalization). Default to Full + // since this path should only be hit at Gloas genesis. + PayloadStatus::Full + }; // Per spec `get_forkchoice_store`: the anchor (genesis) block has // its payload state initialized (`payload_states = {anchor_root: ...}`). @@ -614,14 +606,13 @@ impl ProtoArray { justified_checkpoint: block.justified_checkpoint, finalized_checkpoint: block.finalized_checkpoint, weight: 0, - best_child: None, - best_descendant: None, unrealized_justified_checkpoint: block.unrealized_justified_checkpoint, unrealized_finalized_checkpoint: block.unrealized_finalized_checkpoint, parent_payload_status, empty_payload_weight: 0, full_payload_weight: 0, execution_payload_block_hash, + execution_payload_parent_hash, // Per spec `get_forkchoice_store`: the anchor block's PTC votes are // initialized to all-True, ensuring `is_payload_timely` and // `is_payload_data_available` return true for the anchor. @@ -641,11 +632,9 @@ impl ProtoArray { // Anchor gets [True, True]. Others computed from time_into_slot. block_timeliness_attestation_threshold: is_genesis || (is_current_slot - && time_into_slot < spec.get_unaggregated_attestation_due()), - // TODO(gloas): use GLOAS-specific PTC due threshold once - // `get_payload_attestation_due_ms` is on ChainSpec. + && time_into_slot < spec.get_attestation_due::(current_slot)), block_timeliness_ptc_threshold: is_genesis - || (is_current_slot && time_into_slot < spec.get_slot_duration() / 2), + || (is_current_slot && time_into_slot < spec.get_payload_attestation_due()), equivocating_attestation_score: 0, }) }; @@ -682,11 +671,17 @@ impl ProtoArray { } /// Spec: `is_head_weak`. - /// - /// The spec adds weight from equivocating validators in the head slot's - /// committees. We approximate this with `equivocating_attestation_score` - /// which tracks equivocating validators that voted for this block (close - /// but not identical to committee membership). + // TODO(gloas): the spec adds weight from equivocating validators in the + // head slot's *committees*, regardless of who they voted for. We approximate + // with `equivocating_attestation_score` which only tracks equivocating + // validators whose vote pointed at this block. This under-counts when an + // equivocating validator is in the committee but voted for a different fork, + // which could allow a re-org the spec wouldn't. In practice the deviation + // is small — it requires equivocating validators voting for competing forks + // AND the head weight to be exactly at the reorg threshold boundary. + // Fixing this properly requires committee computation from BeaconState, + // which is not available in proto_array. The fix would be to pass + // pre-computed equivocating committee weight from the beacon_chain caller. fn is_head_weak( &self, head_node: &ProtoNode, @@ -729,7 +724,6 @@ impl ProtoArray { .nodes .get(block_index) .ok_or(Error::InvalidNodeIndex(block_index))?; - // TODO(gloas): handle parent unknown case? let parent_index = block .parent() .ok_or(Error::NodeUnknown(proposer_boost_root))?; @@ -753,7 +747,6 @@ impl ProtoArray { // the parent's slot from the same proposer. let parent_slot = parent.slot(); let parent_root = parent.root(); - // TODO(gloas): handle proposer index for pre-Gloas blocks? let parent_proposer = parent.proposer_index(); let has_equivocation = self.nodes.iter().any(|node| { @@ -773,12 +766,10 @@ impl ProtoArray { Ok(!has_equivocation) } - /// Process an execution payload for a Gloas block. + /// Process a valid execution payload envelope for a Gloas block. /// - /// Sets `payload_received` to true, which makes `is_payload_timely` and - /// `is_payload_data_available` return true regardless of PTC votes. - /// This maps to `store.payload_states[root] = state` in the spec. - pub fn on_valid_execution_payload(&mut self, block_root: Hash256) -> Result<(), Error> { + /// Sets `payload_received` to true. + pub fn on_valid_payload_envelope_received(&mut self, block_root: Hash256) -> Result<(), Error> { let index = *self .indices .get(&block_root) @@ -814,6 +805,8 @@ impl ProtoArray { /// Updates the `verified_node_index` and all ancestors to have validated execution payloads. /// + /// This function is a no-op if called for a Gloas block. + /// /// Returns an error if: /// /// - The `verified_node_index` is unknown. @@ -857,18 +850,10 @@ impl ProtoArray { }); } }, - // Gloas nodes don't carry `ExecutionStatus`. Mark the validated - // block as payload-received so that `is_payload_timely` / - // `is_payload_data_available` and `index == 1` attestations work. - ProtoNode::V29(node) => { - if index == verified_node_index { - node.payload_received = true; - } - if let Some(parent_index) = node.parent { - parent_index - } else { - return Ok(()); - } + // Gloas nodes should not be marked valid by this function, which exists only + // for pre-Gloas fork choice. + ProtoNode::V29(_) => { + return Ok(()); } }; @@ -879,6 +864,7 @@ impl ProtoArray { /// Invalidate zero or more blocks, as specified by the `InvalidationOperation`. /// /// See the documentation of `InvalidationOperation` for usage. + // TODO(gloas): this needs some tests for the mixed Gloas/pre-Gloas case. pub fn propagate_execution_payload_invalidation( &mut self, op: &InvalidationOperation, @@ -978,7 +964,7 @@ impl ProtoArray { // This block is pre-merge, therefore it has no execution status. Nor do its // ancestors. Ok(ExecutionStatus::Irrelevant(_)) => break, - Err(_) => (), + Err(_) => break, } } @@ -1087,9 +1073,6 @@ impl ProtoArray { }); } - // In the post-Gloas world, always use a virtual tree walk. - // - // Best child/best descendant is dead. let best_fc_node = self.find_head_walk::( justified_index, current_slot, @@ -1125,26 +1108,6 @@ impl ProtoArray { Ok((best_fc_node.root, best_fc_node.payload_status)) } - /// Build a parent->children index. Invalid nodes are excluded - /// (they aren't in store.blocks in the spec). - fn build_children_index(&self) -> Vec> { - let mut children = vec![vec![]; self.nodes.len()]; - for (i, node) in self.nodes.iter().enumerate() { - if node - .execution_status() - .is_ok_and(|status| status.is_invalid()) - { - continue; - } - if let Some(parent) = node.parent() - && parent < children.len() - { - children[parent].push(i); - } - } - children - } - /// Spec: `get_filtered_block_tree`. /// /// Returns the set of node indices on viable branches — those with at least @@ -1155,7 +1118,6 @@ impl ProtoArray { current_slot: Slot, best_justified_checkpoint: Checkpoint, best_finalized_checkpoint: Checkpoint, - children_index: &[Vec], ) -> HashSet { let mut viable = HashSet::new(); self.filter_block_tree::( @@ -1163,7 +1125,6 @@ impl ProtoArray { current_slot, best_justified_checkpoint, best_finalized_checkpoint, - children_index, &mut viable, ); viable @@ -1176,17 +1137,25 @@ impl ProtoArray { current_slot: Slot, best_justified_checkpoint: Checkpoint, best_finalized_checkpoint: Checkpoint, - children_index: &[Vec], viable: &mut HashSet, ) -> bool { let Some(node) = self.nodes.get(node_index) else { return false; }; - let children = children_index - .get(node_index) - .map(|c| c.as_slice()) - .unwrap_or(&[]); + // Skip invalid children — they aren't in store.blocks in the spec. + let children: Vec = self + .nodes + .iter() + .enumerate() + .filter(|(_, child)| { + child.parent() == Some(node_index) + && !child + .execution_status() + .is_ok_and(|status| status.is_invalid()) + }) + .map(|(i, _)| i) + .collect(); if !children.is_empty() { // Evaluate ALL children (no short-circuit) to mark all viable branches. @@ -1198,7 +1167,6 @@ impl ProtoArray { current_slot, best_justified_checkpoint, best_finalized_checkpoint, - children_index, viable, ) }) @@ -1243,16 +1211,12 @@ impl ProtoArray { payload_status: PayloadStatus::Pending, }; - // Build parent->children index once for O(1) lookups. - let children_index = self.build_children_index(); - // Spec: `get_filtered_block_tree`. let viable_nodes = self.get_filtered_block_tree::( start_index, current_slot, best_justified_checkpoint, best_finalized_checkpoint, - &children_index, ); // Compute once rather than per-child per-level. @@ -1261,7 +1225,7 @@ impl ProtoArray { loop { let children: Vec<_> = self - .get_node_children(&head, &children_index)? + .get_node_children(&head)? .into_iter() .filter(|(fc_node, _)| viable_nodes.contains(&fc_node.proto_node_index)) .collect(); @@ -1272,11 +1236,7 @@ impl ProtoArray { head = children .into_iter() - .map(|(child, _)| -> Result<_, Error> { - let proto_node = self - .nodes - .get(child.proto_node_index) - .ok_or(Error::InvalidNodeIndex(child.proto_node_index))?; + .map(|(child, ref proto_node)| -> Result<_, Error> { let weight = self.get_weight::( &child, proto_node, @@ -1424,76 +1384,36 @@ impl ProtoArray { fn get_node_children( &self, node: &IndexedForkChoiceNode, - children_index: &[Vec], ) -> Result, Error> { if node.payload_status == PayloadStatus::Pending { let proto_node = self .nodes .get(node.proto_node_index) .ok_or(Error::InvalidNodeIndex(node.proto_node_index))?; - - // V17 (pre-GLOAS) nodes don't have payload_received or parent_payload_status. - // Skip the virtual Empty/Full split and return real children directly. - if proto_node.as_v17().is_ok() { - let child_indices = children_index - .get(node.proto_node_index) - .map(|c| c.as_slice()) - .unwrap_or(&[]); - return Ok(child_indices - .iter() - .filter_map(|&child_index| { - let child_node = self.nodes.get(child_index)?; - Some(( - IndexedForkChoiceNode { - root: child_node.root(), - proto_node_index: child_index, - payload_status: PayloadStatus::Pending, - }, - child_node.clone(), - )) - }) - .collect()); + let mut children = vec![(node.with_status(PayloadStatus::Empty), proto_node.clone())]; + // The FULL virtual child only exists if the payload has been received. + if proto_node.payload_received().is_ok_and(|received| received) { + children.push((node.with_status(PayloadStatus::Full), proto_node.clone())); } - - // TODO(gloas) this is the actual change we want to keep once PTC is implemented - // let mut children = vec![(node.with_status(PayloadStatus::Empty), proto_node.clone())]; - // // The FULL virtual child only exists if the payload has been received. - // if proto_node.payload_received().is_ok_and(|received| received) { - // children.push((node.with_status(PayloadStatus::Full), proto_node.clone())); - // } - - // TODO(gloas) remove this and uncomment the code above once we implement PTC - // Skip Empty/Full split: go straight to Full when payload received, - // giving full payload weight 100% without PTC votes. - let children = if proto_node.payload_received().is_ok_and(|received| received) { - vec![(node.with_status(PayloadStatus::Full), proto_node.clone())] - } else { - vec![(node.with_status(PayloadStatus::Empty), proto_node.clone())] - }; - // TODO(gloas) delete up to here - Ok(children) } else { - let child_indices = children_index - .get(node.proto_node_index) - .map(|c| c.as_slice()) - .unwrap_or(&[]); - Ok(child_indices + Ok(self + .nodes .iter() - .filter_map(|&child_index| { - let child_node = self.nodes.get(child_index)?; - // Skip parent_payload_status filter for V17 children (they don't have it) - if child_node.get_parent_payload_status() != node.payload_status { - return None; - } - Some(( + .enumerate() + .filter(|(_, child_node)| { + child_node.parent() == Some(node.proto_node_index) + && child_node.get_parent_payload_status() == node.payload_status + }) + .map(|(child_index, child_node)| { + ( IndexedForkChoiceNode { root: child_node.root(), proto_node_index: child_index, payload_status: PayloadStatus::Pending, }, child_node.clone(), - )) + ) }) .collect()) } diff --git a/consensus/proto_array/src/proto_array_fork_choice.rs b/consensus/proto_array/src/proto_array_fork_choice.rs index cb467f2531..0ecaea3971 100644 --- a/consensus/proto_array/src/proto_array_fork_choice.rs +++ b/consensus/proto_array/src/proto_array_fork_choice.rs @@ -2,8 +2,7 @@ use crate::{ JustifiedBalances, error::Error, proto_array::{ - InvalidationOperation, Iter, NodeDelta, ProposerBoost, ProtoArray, ProtoNode, - calculate_committee_fraction, + InvalidationOperation, Iter, NodeDelta, ProtoArray, ProtoNode, calculate_committee_fraction, }, ssz_container::SszContainer, }; @@ -33,6 +32,48 @@ pub struct VoteTracker { next_payload_present: bool, } +// Can be deleted once the V28 schema migration is buried. +// Matches the on-disk format from schema v28: current_root, next_root, next_epoch. +#[derive(Default, PartialEq, Clone, Encode, Decode)] +pub struct VoteTrackerV28 { + current_root: Hash256, + next_root: Hash256, + next_epoch: Epoch, +} + +// This impl is only used upon upgrade from pre-Gloas to Gloas with all pre-Gloas nodes. +// The payload status is `false` for pre-Gloas nodes. +impl From for VoteTracker { + fn from(v: VoteTrackerV28) -> Self { + VoteTracker { + current_root: v.current_root, + next_root: v.next_root, + // The v28 format stored next_epoch rather than slots. Default to 0 since the + // vote tracker will be updated on the next attestation. + current_slot: Slot::new(0), + next_slot: Slot::new(0), + current_payload_present: false, + next_payload_present: false, + } + } +} + +// This impl is only used upon downgrade from V29 to V28, with exclusively pre-Gloas nodes. +impl From for VoteTrackerV28 { + fn from(v: VoteTracker) -> Self { + // Drop the payload_present fields. This is safe because this is only called on pre-Gloas + // nodes. + VoteTrackerV28 { + current_root: v.current_root, + next_root: v.next_root, + // The v28 format stored next_epoch. Default to 0 since the vote tracker will be + // updated on the next attestation. + next_epoch: Epoch::new(0), + } + } +} + +/// Spec's `LatestMessage` type. Only used in tests. pub struct LatestMessage { pub slot: Slot, pub root: Hash256, @@ -479,13 +520,13 @@ impl ProtoArrayForkChoice { execution_status: ExecutionStatus, execution_payload_parent_hash: Option, execution_payload_block_hash: Option, + proposer_index: u64, spec: &ChainSpec, ) -> Result { let mut proto_array = ProtoArray { prune_threshold: DEFAULT_PRUNE_THRESHOLD, nodes: Vec::with_capacity(1), indices: HashMap::with_capacity(1), - previous_proposer_boost: ProposerBoost::default(), }; let block = Block { @@ -505,7 +546,7 @@ impl ProtoArrayForkChoice { unrealized_finalized_checkpoint: Some(finalized_checkpoint), execution_payload_parent_hash, execution_payload_block_hash, - proposer_index: Some(0), + proposer_index: Some(proposer_index), }; proto_array @@ -527,11 +568,18 @@ impl ProtoArrayForkChoice { }) } - pub fn on_execution_payload(&mut self, block_root: Hash256) -> Result<(), String> { + /// Mark a Gloas payload envelope as valid and received. + /// + /// This must only be called for valid Gloas payloads. + pub fn on_valid_payload_envelope_received( + &mut self, + block_root: Hash256, + ) -> Result<(), String> { self.proto_array - .on_valid_execution_payload(block_root) + .on_valid_payload_envelope_received(block_root) .map_err(|e| format!("Failed to process execution payload: {:?}", e)) } + /// See `ProtoArray::propagate_execution_payload_validation` for documentation. pub fn process_execution_payload_validation( &mut self, @@ -838,10 +886,7 @@ impl ProtoArrayForkChoice { /// status to be optimistic. /// /// In practice this means forgetting any `VALID` or `INVALID` statuses. - pub fn set_all_blocks_to_optimistic( - &mut self, - spec: &ChainSpec, - ) -> Result<(), String> { + pub fn set_all_blocks_to_optimistic(&mut self) -> Result<(), String> { // Iterate backwards through all nodes in the `proto_array`. Whilst it's not strictly // required to do this process in reverse, it seems natural when we consider how LMD votes // are counted. @@ -864,7 +909,7 @@ impl ProtoArrayForkChoice { // Restore the weight of the node, it would have been set to `0` in // `apply_score_changes` when it was invalidated. - let mut restored_weight: u64 = self + let restored_weight: u64 = self .votes .0 .iter() @@ -880,26 +925,6 @@ impl ProtoArrayForkChoice { }) .sum(); - // If the invalid root was boosted, apply the weight to it and - // ancestors. - if let Some(proposer_score_boost) = spec.proposer_score_boost - && self.proto_array.previous_proposer_boost.root == node.root() - { - // Compute the score based upon the current balances. We can't rely on - // the `previous_proposr_boost.score` since it is set to zero with an - // invalid node. - let proposer_score = - calculate_committee_fraction::(&self.balances, proposer_score_boost) - .ok_or("Failed to compute proposer boost")?; - // Store the score we've applied here so it can be removed in - // a later call to `apply_score_changes`. - self.proto_array.previous_proposer_boost.score = proposer_score; - // Apply this boost to this node. - restored_weight = restored_weight - .checked_add(proposer_score) - .ok_or("Overflow when adding boost to weight")?; - } - // Add the restored weight to the node and all ancestors. if restored_weight > 0 { let mut node_or_ancestor = node; @@ -988,7 +1013,7 @@ impl ProtoArrayForkChoice { .unwrap_or_else(|_| ExecutionStatus::irrelevant()), unrealized_justified_checkpoint: block.unrealized_justified_checkpoint(), unrealized_finalized_checkpoint: block.unrealized_finalized_checkpoint(), - execution_payload_parent_hash: None, + execution_payload_parent_hash: block.execution_payload_parent_hash().ok(), execution_payload_block_hash: block.execution_payload_block_hash().ok(), proposer_index: block.proposer_index().ok(), }) @@ -1005,7 +1030,8 @@ impl ProtoArrayForkChoice { } /// Returns whether the execution payload for a block has been received. - /// Returns `false` for pre-GLOAS (V17) nodes or unknown blocks. + /// + /// Returns `false` for pre-Gloas (V17) nodes or unknown blocks. pub fn is_payload_received(&self, block_root: &Hash256) -> bool { self.get_proto_node(block_root) .and_then(|node| node.payload_received().ok()) @@ -1039,10 +1065,9 @@ impl ProtoArrayForkChoice { .is_finalized_checkpoint_or_descendant::(descendant_root, best_finalized_checkpoint) } + /// NOTE: only used in tests. pub fn latest_message(&self, validator_index: usize) -> Option { - if validator_index < self.votes.0.len() { - let vote = &self.votes.0[validator_index]; - + if let Some(vote) = self.votes.0.get(validator_index) { if *vote == VoteTracker::default() { None } else { @@ -1317,6 +1342,7 @@ mod test_compute_deltas { execution_status, None, None, + 0, &spec, ) .unwrap(); @@ -1471,6 +1497,7 @@ mod test_compute_deltas { execution_status, None, None, + 0, &spec, ) .unwrap(); diff --git a/consensus/proto_array/src/ssz_container.rs b/consensus/proto_array/src/ssz_container.rs index 5edc1cd313..69efb35027 100644 --- a/consensus/proto_array/src/ssz_container.rs +++ b/consensus/proto_array/src/ssz_container.rs @@ -2,7 +2,7 @@ use crate::proto_array::ProposerBoost; use crate::{ Error, JustifiedBalances, proto_array::{ProtoArray, ProtoNode, ProtoNodeV17}, - proto_array_fork_choice::{ElasticList, ProtoArrayForkChoice, VoteTracker}, + proto_array_fork_choice::{ElasticList, ProtoArrayForkChoice, VoteTracker, VoteTrackerV28}, }; use ssz::{Encode, four_byte_option_impl}; use ssz_derive::{Decode, Encode}; @@ -22,6 +22,9 @@ pub type SszContainer = SszContainerV29; no_enum )] pub struct SszContainer { + #[superstruct(only(V28))] + pub votes_v28: Vec, + #[superstruct(only(V29))] pub votes: Vec, pub prune_threshold: usize, // Deprecated, remove in a future schema migration @@ -35,6 +38,7 @@ pub struct SszContainer { #[superstruct(only(V29))] pub nodes: Vec, pub indices: Vec<(Hash256, usize)>, + #[superstruct(only(V28))] pub previous_proposer_boost: ProposerBoost, } @@ -47,7 +51,6 @@ impl SszContainerV29 { prune_threshold: proto_array.prune_threshold, nodes: proto_array.nodes.clone(), indices: proto_array.indices.iter().map(|(k, v)| (*k, *v)).collect(), - previous_proposer_boost: proto_array.previous_proposer_boost, } } } @@ -60,7 +63,6 @@ impl TryFrom<(SszContainerV29, JustifiedBalances)> for ProtoArrayForkChoice { prune_threshold: from.prune_threshold, nodes: from.nodes, indices: from.indices.into_iter().collect::>(), - previous_proposer_boost: from.previous_proposer_boost, }; Ok(Self { @@ -75,11 +77,20 @@ impl TryFrom<(SszContainerV29, JustifiedBalances)> for ProtoArrayForkChoice { impl From for SszContainerV29 { fn from(v28: SszContainerV28) -> Self { Self { - votes: v28.votes, + votes: v28.votes_v28.into_iter().map(Into::into).collect(), prune_threshold: v28.prune_threshold, - nodes: v28.nodes.into_iter().map(ProtoNode::V17).collect(), + nodes: v28 + .nodes + .into_iter() + .map(|mut node| { + // best_child/best_descendant are no longer used (replaced by + // the virtual tree walk). Clear during conversion. + node.best_child = None; + node.best_descendant = None; + ProtoNode::V17(node) + }) + .collect(), indices: v28.indices, - previous_proposer_boost: v28.previous_proposer_boost, } } } @@ -88,7 +99,7 @@ impl From for SszContainerV29 { impl From for SszContainerV28 { fn from(v29: SszContainerV29) -> Self { Self { - votes: v29.votes, + votes_v28: v29.votes.into_iter().map(Into::into).collect(), prune_threshold: v29.prune_threshold, // These checkpoints are not consumed in v28 paths since the upgrade from v17, // we can safely default the values. @@ -103,7 +114,8 @@ impl From for SszContainerV28 { }) .collect(), indices: v29.indices, - previous_proposer_boost: v29.previous_proposer_boost, + // Proposer boost is not tracked in V29 (computed on-the-fly), so reset it. + previous_proposer_boost: ProposerBoost::default(), } } } diff --git a/consensus/types/src/attestation/indexed_payload_attestation.rs b/consensus/types/src/attestation/indexed_payload_attestation.rs index 4de805570c..bb2087e330 100644 --- a/consensus/types/src/attestation/indexed_payload_attestation.rs +++ b/consensus/types/src/attestation/indexed_payload_attestation.rs @@ -2,7 +2,6 @@ use crate::test_utils::TestRandom; use crate::{EthSpec, ForkName, PayloadAttestationData}; use bls::AggregateSignature; use context_deserialize::context_deserialize; -use core::slice::Iter; use serde::{Deserialize, Serialize}; use ssz_derive::{Decode, Encode}; use ssz_types::VariableList; @@ -21,12 +20,6 @@ pub struct IndexedPayloadAttestation { pub signature: AggregateSignature, } -impl IndexedPayloadAttestation { - pub fn attesting_indices_iter(&self) -> Iter<'_, u64> { - self.attesting_indices.iter() - } -} - #[cfg(test)] mod tests { use super::*; diff --git a/consensus/types/src/core/chain_spec.rs b/consensus/types/src/core/chain_spec.rs index cc79d3fc29..e612c8b6db 100644 --- a/consensus/types/src/core/chain_spec.rs +++ b/consensus/types/src/core/chain_spec.rs @@ -107,6 +107,8 @@ pub struct ChainSpec { pub shard_committee_period: u64, pub proposer_reorg_cutoff_bps: u64, pub attestation_due_bps: u64, + pub attestation_due_bps_gloas: u64, + pub payload_attestation_due_bps: u64, pub aggregate_due_bps: u64, pub sync_message_due_bps: u64, pub contribution_due_bps: u64, @@ -115,6 +117,8 @@ pub struct ChainSpec { * Derived time values (computed at startup via `compute_derived_values()`) */ pub unaggregated_attestation_due: Duration, + pub unaggregated_attestation_due_gloas: Duration, + pub payload_attestation_due: Duration, pub aggregate_attestation_due: Duration, pub sync_message_due: Duration, pub contribution_and_proof_due: Duration, @@ -877,6 +881,20 @@ impl ChainSpec { self.unaggregated_attestation_due } + /// Spec: `get_attestation_due_ms`. Returns the epoch-appropriate threshold. + pub fn get_attestation_due(&self, slot: Slot) -> Duration { + if self.fork_name_at_slot::(slot).gloas_enabled() { + self.unaggregated_attestation_due_gloas + } else { + self.unaggregated_attestation_due + } + } + + /// Spec: `get_payload_attestation_due_ms`. + pub fn get_payload_attestation_due(&self) -> Duration { + self.payload_attestation_due + } + /// Get the duration into a slot in which an aggregated attestation is due. /// Returns the pre-computed value from `compute_derived_values()`. pub fn get_aggregate_attestation_due(&self) -> Duration { @@ -949,6 +967,12 @@ impl ChainSpec { self.unaggregated_attestation_due = self .compute_slot_component_duration(self.attestation_due_bps) .expect("invalid chain spec: cannot compute unaggregated_attestation_due"); + self.unaggregated_attestation_due_gloas = self + .compute_slot_component_duration(self.attestation_due_bps_gloas) + .expect("invalid chain spec: cannot compute unaggregated_attestation_due_gloas"); + self.payload_attestation_due = self + .compute_slot_component_duration(self.payload_attestation_due_bps) + .expect("invalid chain spec: cannot compute payload_attestation_due"); self.aggregate_attestation_due = self .compute_slot_component_duration(self.aggregate_due_bps) .expect("invalid chain spec: cannot compute aggregate_attestation_due"); @@ -1079,6 +1103,8 @@ impl ChainSpec { shard_committee_period: 256, proposer_reorg_cutoff_bps: 1667, attestation_due_bps: 3333, + attestation_due_bps_gloas: 2500, + payload_attestation_due_bps: 7500, aggregate_due_bps: 6667, sync_message_due_bps: 3333, contribution_due_bps: 6667, @@ -1087,6 +1113,8 @@ impl ChainSpec { * Derived time values (set by `compute_derived_values()`) */ unaggregated_attestation_due: Duration::from_millis(3999), + unaggregated_attestation_due_gloas: Duration::from_millis(3000), + payload_attestation_due: Duration::from_millis(9000), aggregate_attestation_due: Duration::from_millis(8000), sync_message_due: Duration::from_millis(3999), contribution_and_proof_due: Duration::from_millis(8000), @@ -1390,6 +1418,8 @@ impl ChainSpec { * Precomputed for 6000ms slot: 3333 bps = 1999ms, 6667 bps = 4000ms */ unaggregated_attestation_due: Duration::from_millis(1999), + unaggregated_attestation_due_gloas: Duration::from_millis(1500), + payload_attestation_due: Duration::from_millis(4500), aggregate_attestation_due: Duration::from_millis(4000), sync_message_due: Duration::from_millis(1999), contribution_and_proof_due: Duration::from_millis(4000), @@ -1479,6 +1509,8 @@ impl ChainSpec { shard_committee_period: 256, proposer_reorg_cutoff_bps: 1667, attestation_due_bps: 3333, + attestation_due_bps_gloas: 2500, + payload_attestation_due_bps: 7500, aggregate_due_bps: 6667, /* @@ -1486,6 +1518,8 @@ impl ChainSpec { * Precomputed for 5000ms slot: 3333 bps = 1666ms, 6667 bps = 3333ms */ unaggregated_attestation_due: Duration::from_millis(1666), + unaggregated_attestation_due_gloas: Duration::from_millis(1250), + payload_attestation_due: Duration::from_millis(3750), aggregate_attestation_due: Duration::from_millis(3333), sync_message_due: Duration::from_millis(1666), contribution_and_proof_due: Duration::from_millis(3333), @@ -2062,6 +2096,12 @@ pub struct Config { #[serde(default = "default_attestation_due_bps")] #[serde(with = "serde_utils::quoted_u64")] attestation_due_bps: u64, + #[serde(default = "default_attestation_due_bps_gloas")] + #[serde(with = "serde_utils::quoted_u64")] + attestation_due_bps_gloas: u64, + #[serde(default = "default_payload_attestation_due_bps")] + #[serde(with = "serde_utils::quoted_u64")] + payload_attestation_due_bps: u64, #[serde(default = "default_aggregate_due_bps")] #[serde(with = "serde_utils::quoted_u64")] aggregate_due_bps: u64, @@ -2288,6 +2328,14 @@ const fn default_attestation_due_bps() -> u64 { 3333 } +const fn default_attestation_due_bps_gloas() -> u64 { + 2500 +} + +const fn default_payload_attestation_due_bps() -> u64 { + 7500 +} + const fn default_aggregate_due_bps() -> u64 { 6667 } @@ -2539,6 +2587,8 @@ impl Config { proposer_reorg_cutoff_bps: spec.proposer_reorg_cutoff_bps, attestation_due_bps: spec.attestation_due_bps, + attestation_due_bps_gloas: spec.attestation_due_bps_gloas, + payload_attestation_due_bps: spec.payload_attestation_due_bps, aggregate_due_bps: spec.aggregate_due_bps, sync_message_due_bps: spec.sync_message_due_bps, contribution_due_bps: spec.contribution_due_bps, @@ -2632,6 +2682,8 @@ impl Config { min_epochs_for_data_column_sidecars_requests, proposer_reorg_cutoff_bps, attestation_due_bps, + attestation_due_bps_gloas, + payload_attestation_due_bps, aggregate_due_bps, sync_message_due_bps, contribution_due_bps, @@ -2731,6 +2783,8 @@ impl Config { proposer_reorg_cutoff_bps, attestation_due_bps, + attestation_due_bps_gloas, + payload_attestation_due_bps, aggregate_due_bps, sync_message_due_bps, contribution_due_bps, @@ -3634,11 +3688,9 @@ mod yaml_tests { "EIP7928_FORK_VERSION", "EIP7928_FORK_EPOCH", // Gloas params not yet in Config - "ATTESTATION_DUE_BPS_GLOAS", "AGGREGATE_DUE_BPS_GLOAS", "SYNC_MESSAGE_DUE_BPS_GLOAS", "CONTRIBUTION_DUE_BPS_GLOAS", - "PAYLOAD_ATTESTATION_DUE_BPS", "MAX_REQUEST_PAYLOADS", // Gloas fork choice params not yet in Config "REORG_HEAD_WEIGHT_THRESHOLD", diff --git a/consensus/types/src/core/eth_spec.rs b/consensus/types/src/core/eth_spec.rs index 36d61fbbf9..4159091f5d 100644 --- a/consensus/types/src/core/eth_spec.rs +++ b/consensus/types/src/core/eth_spec.rs @@ -448,6 +448,11 @@ pub trait EthSpec: 'static + Default + Sync + Send + Clone + Debug + PartialEq + fn payload_timely_threshold() -> usize { Self::PTCSize::to_usize() / 2 } + + /// Returns the `DATA_AVAILABILITY_TIMELY_THRESHOLD` constant (PTC_SIZE / 2). + fn data_availability_timely_threshold() -> usize { + Self::PTCSize::to_usize() / 2 + } } /// Macro to inherit some type values from another EthSpec. diff --git a/testing/ef_tests/src/cases/fork_choice.rs b/testing/ef_tests/src/cases/fork_choice.rs index 22e8453e14..06f204ab01 100644 --- a/testing/ef_tests/src/cases/fork_choice.rs +++ b/testing/ef_tests/src/cases/fork_choice.rs @@ -1018,7 +1018,7 @@ impl Tester { .chain .canonical_head .fork_choice_write_lock() - .on_execution_payload(block_root); + .on_valid_payload_envelope_received(block_root); if valid { result.map_err(|e| {