mirror of
https://github.com/sigp/lighthouse.git
synced 2026-04-30 19:23:50 +00:00
delay cache, and remove some todos
This commit is contained in:
Eitan Seri- Levi
@@ -27,6 +27,7 @@ use crate::data_availability_checker::{
|
||||
};
|
||||
use crate::data_column_verification::{GossipDataColumnError, GossipVerifiedDataColumn};
|
||||
use crate::early_attester_cache::EarlyAttesterCache;
|
||||
use crate::envelope_times_cache::EnvelopeTimesCache;
|
||||
use crate::errors::{BeaconChainError as Error, BlockProductionError};
|
||||
use crate::events::ServerSentEventHandler;
|
||||
use crate::execution_payload::{NotifyExecutionLayer, PreparePayloadHandle, get_execution_payload};
|
||||
@@ -56,7 +57,9 @@ use crate::observed_block_producers::ObservedBlockProducers;
|
||||
use crate::observed_data_sidecars::ObservedDataSidecars;
|
||||
use crate::observed_operations::{ObservationOutcome, ObservedOperations};
|
||||
use crate::observed_slashable::ObservedSlashable;
|
||||
use crate::payload_envelope_verification::{ExecutedEnvelope, ExecutionPendingEnvelope};
|
||||
use crate::payload_envelope_verification::{
|
||||
EnvelopeError, ExecutedEnvelope, ExecutionPendingEnvelope,
|
||||
};
|
||||
use crate::persisted_beacon_chain::PersistedBeaconChain;
|
||||
use crate::persisted_custody::persist_custody_context;
|
||||
use crate::persisted_fork_choice::PersistedForkChoice;
|
||||
@@ -460,6 +463,8 @@ pub struct BeaconChain<T: BeaconChainTypes> {
|
||||
pub early_attester_cache: EarlyAttesterCache<T::EthSpec>,
|
||||
/// A cache used to keep track of various block timings.
|
||||
pub block_times_cache: Arc<RwLock<BlockTimesCache>>,
|
||||
/// A cache used to keep track of various envelope timings.
|
||||
pub envelope_times_cache: Arc<RwLock<EnvelopeTimesCache>>,
|
||||
/// A cache used to track pre-finalization block roots for quick rejection.
|
||||
pub pre_finalization_block_cache: PreFinalizationBlockCache,
|
||||
/// A cache used to produce light_client server messages
|
||||
@@ -1158,8 +1163,8 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
|
||||
match self.store.try_get_full_block(block_root)? {
|
||||
Some(DatabaseBlock::Full(block)) => Ok(Some(block)),
|
||||
Some(DatabaseBlock::Blinded(_)) => {
|
||||
// TODO(gloas) this should error out
|
||||
todo!()
|
||||
// TODO(gloas) should we return None here?
|
||||
Ok(None)
|
||||
}
|
||||
None => Ok(None),
|
||||
}
|
||||
@@ -3556,7 +3561,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
|
||||
pub async fn into_executed_payload_envelope(
|
||||
self: Arc<Self>,
|
||||
pending_envelope: ExecutionPendingEnvelope<T::EthSpec>,
|
||||
) -> Result<ExecutedEnvelope<T::EthSpec>, BlockError> {
|
||||
) -> Result<ExecutedEnvelope<T::EthSpec>, EnvelopeError> {
|
||||
let ExecutionPendingEnvelope {
|
||||
signed_envelope,
|
||||
import_data,
|
||||
@@ -4168,7 +4173,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
|
||||
Ok(block_root)
|
||||
}
|
||||
|
||||
pub(crate) fn handle_import_block_db_write_error(
|
||||
fn handle_import_block_db_write_error(
|
||||
&self,
|
||||
// We don't actually need this value, however it's always present when we call this function
|
||||
// and it needs to be dropped to prevent a dead-lock. Requiring it to be passed here is
|
||||
@@ -6698,6 +6703,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
|
||||
// sync anyway).
|
||||
self.naive_aggregation_pool.write().prune(slot);
|
||||
self.block_times_cache.write().prune(slot);
|
||||
self.envelope_times_cache.write().prune(slot);
|
||||
|
||||
// Don't run heavy-weight tasks during sync.
|
||||
if self.best_slot() + MAX_PER_SLOT_FORK_CHOICE_DISTANCE < slot {
|
||||
|
||||
@@ -1055,6 +1055,7 @@ where
|
||||
)),
|
||||
beacon_proposer_cache,
|
||||
block_times_cache: <_>::default(),
|
||||
envelope_times_cache: <_>::default(),
|
||||
pre_finalization_block_cache: <_>::default(),
|
||||
validator_pubkey_cache: RwLock::new(validator_pubkey_cache),
|
||||
early_attester_cache: <_>::default(),
|
||||
|
||||
197
beacon_node/beacon_chain/src/envelope_times_cache.rs
Normal file
197
beacon_node/beacon_chain/src/envelope_times_cache.rs
Normal file
@@ -0,0 +1,197 @@
|
||||
//! This module provides the `EnvelopeTimesCache` which contains information regarding payload
|
||||
//! envelope timings.
|
||||
//!
|
||||
//! This provides `BeaconChain` and associated functions with access to the timestamps of when a
|
||||
//! payload envelope was observed, verified, executed, and imported.
|
||||
//! This allows for better traceability and allows us to determine the root cause for why an
|
||||
//! envelope was imported late.
|
||||
//! This allows us to distinguish between the following scenarios:
|
||||
//! - The envelope was observed late.
|
||||
//! - Consensus verification was slow.
|
||||
//! - Execution verification was slow.
|
||||
//! - The DB write was slow.
|
||||
|
||||
use eth2::types::{Hash256, Slot};
|
||||
use std::collections::HashMap;
|
||||
use std::time::Duration;
|
||||
|
||||
type BlockRoot = Hash256;
|
||||
|
||||
#[derive(Clone, Default)]
|
||||
pub struct EnvelopeTimestamps {
|
||||
/// When the envelope was first observed (gossip or RPC).
|
||||
pub observed: Option<Duration>,
|
||||
/// When consensus verification (state transition) completed.
|
||||
pub consensus_verified: Option<Duration>,
|
||||
/// When execution layer verification started.
|
||||
pub started_execution: Option<Duration>,
|
||||
/// When execution layer verification completed.
|
||||
pub executed: Option<Duration>,
|
||||
/// When the envelope was imported into the DB.
|
||||
pub imported: Option<Duration>,
|
||||
}
|
||||
|
||||
/// Delay data for envelope processing, computed relative to the slot start time.
|
||||
#[derive(Debug, Default)]
|
||||
pub struct EnvelopeDelays {
|
||||
/// Time after start of slot we saw the envelope.
|
||||
pub observed: Option<Duration>,
|
||||
/// The time it took to complete consensus verification of the envelope.
|
||||
pub consensus_verification_time: Option<Duration>,
|
||||
/// The time it took to complete execution verification of the envelope.
|
||||
pub execution_time: Option<Duration>,
|
||||
/// Time after execution until the envelope was imported.
|
||||
pub imported: Option<Duration>,
|
||||
}
|
||||
|
||||
impl EnvelopeDelays {
|
||||
fn new(times: EnvelopeTimestamps, slot_start_time: Duration) -> EnvelopeDelays {
|
||||
let observed = times
|
||||
.observed
|
||||
.and_then(|observed_time| observed_time.checked_sub(slot_start_time));
|
||||
let consensus_verification_time = times
|
||||
.consensus_verified
|
||||
.and_then(|consensus_verified| consensus_verified.checked_sub(times.observed?));
|
||||
let execution_time = times
|
||||
.executed
|
||||
.and_then(|executed| executed.checked_sub(times.started_execution?));
|
||||
let imported = times
|
||||
.imported
|
||||
.and_then(|imported_time| imported_time.checked_sub(times.executed?));
|
||||
EnvelopeDelays {
|
||||
observed,
|
||||
consensus_verification_time,
|
||||
execution_time,
|
||||
imported,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct EnvelopeTimesCacheValue {
|
||||
pub slot: Slot,
|
||||
pub timestamps: EnvelopeTimestamps,
|
||||
pub peer_id: Option<String>,
|
||||
}
|
||||
|
||||
impl EnvelopeTimesCacheValue {
|
||||
fn new(slot: Slot) -> Self {
|
||||
EnvelopeTimesCacheValue {
|
||||
slot,
|
||||
timestamps: Default::default(),
|
||||
peer_id: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
pub struct EnvelopeTimesCache {
|
||||
pub cache: HashMap<BlockRoot, EnvelopeTimesCacheValue>,
|
||||
}
|
||||
|
||||
impl EnvelopeTimesCache {
|
||||
/// Set the observation time for `block_root` to `timestamp` if `timestamp` is less than
|
||||
/// any previous timestamp at which this envelope was observed.
|
||||
pub fn set_time_observed(
|
||||
&mut self,
|
||||
block_root: BlockRoot,
|
||||
slot: Slot,
|
||||
timestamp: Duration,
|
||||
peer_id: Option<String>,
|
||||
) {
|
||||
let entry = self
|
||||
.cache
|
||||
.entry(block_root)
|
||||
.or_insert_with(|| EnvelopeTimesCacheValue::new(slot));
|
||||
match entry.timestamps.observed {
|
||||
Some(existing) if existing <= timestamp => {
|
||||
// Existing timestamp is earlier, do nothing.
|
||||
}
|
||||
_ => {
|
||||
entry.timestamps.observed = Some(timestamp);
|
||||
entry.peer_id = peer_id;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Set the timestamp for `field` if that timestamp is less than any previously known value.
|
||||
fn set_time_if_less(
|
||||
&mut self,
|
||||
block_root: BlockRoot,
|
||||
slot: Slot,
|
||||
field: impl Fn(&mut EnvelopeTimestamps) -> &mut Option<Duration>,
|
||||
timestamp: Duration,
|
||||
) {
|
||||
let entry = self
|
||||
.cache
|
||||
.entry(block_root)
|
||||
.or_insert_with(|| EnvelopeTimesCacheValue::new(slot));
|
||||
let existing_timestamp = field(&mut entry.timestamps);
|
||||
if existing_timestamp.is_none_or(|prev| timestamp < prev) {
|
||||
*existing_timestamp = Some(timestamp);
|
||||
}
|
||||
}
|
||||
|
||||
pub fn set_time_consensus_verified(
|
||||
&mut self,
|
||||
block_root: BlockRoot,
|
||||
slot: Slot,
|
||||
timestamp: Duration,
|
||||
) {
|
||||
self.set_time_if_less(
|
||||
block_root,
|
||||
slot,
|
||||
|timestamps| &mut timestamps.consensus_verified,
|
||||
timestamp,
|
||||
)
|
||||
}
|
||||
|
||||
pub fn set_time_started_execution(
|
||||
&mut self,
|
||||
block_root: BlockRoot,
|
||||
slot: Slot,
|
||||
timestamp: Duration,
|
||||
) {
|
||||
self.set_time_if_less(
|
||||
block_root,
|
||||
slot,
|
||||
|timestamps| &mut timestamps.started_execution,
|
||||
timestamp,
|
||||
)
|
||||
}
|
||||
|
||||
pub fn set_time_executed(&mut self, block_root: BlockRoot, slot: Slot, timestamp: Duration) {
|
||||
self.set_time_if_less(
|
||||
block_root,
|
||||
slot,
|
||||
|timestamps| &mut timestamps.executed,
|
||||
timestamp,
|
||||
)
|
||||
}
|
||||
|
||||
pub fn set_time_imported(&mut self, block_root: BlockRoot, slot: Slot, timestamp: Duration) {
|
||||
self.set_time_if_less(
|
||||
block_root,
|
||||
slot,
|
||||
|timestamps| &mut timestamps.imported,
|
||||
timestamp,
|
||||
)
|
||||
}
|
||||
|
||||
pub fn get_envelope_delays(
|
||||
&self,
|
||||
block_root: BlockRoot,
|
||||
slot_start_time: Duration,
|
||||
) -> EnvelopeDelays {
|
||||
if let Some(entry) = self.cache.get(&block_root) {
|
||||
EnvelopeDelays::new(entry.timestamps.clone(), slot_start_time)
|
||||
} else {
|
||||
EnvelopeDelays::default()
|
||||
}
|
||||
}
|
||||
|
||||
/// Prune the cache to only store the most recent 2 epochs.
|
||||
pub fn prune(&mut self, current_slot: Slot) {
|
||||
self.cache
|
||||
.retain(|_, entry| entry.slot > current_slot.saturating_sub(64_u64));
|
||||
}
|
||||
}
|
||||
@@ -20,6 +20,7 @@ pub mod custody_context;
|
||||
pub mod data_availability_checker;
|
||||
pub mod data_column_verification;
|
||||
mod early_attester_cache;
|
||||
pub mod envelope_times_cache;
|
||||
mod errors;
|
||||
pub mod events;
|
||||
pub mod execution_payload;
|
||||
@@ -42,7 +43,6 @@ pub mod observed_block_producers;
|
||||
pub mod observed_data_sidecars;
|
||||
pub mod observed_operations;
|
||||
mod observed_slashable;
|
||||
pub mod payload_envelope_import;
|
||||
pub mod payload_envelope_verification;
|
||||
pub mod persisted_beacon_chain;
|
||||
pub mod persisted_custody;
|
||||
|
||||
@@ -21,6 +21,44 @@ pub const VALIDATOR_MONITOR_ATTESTATION_SIMULATOR_SOURCE_ATTESTER_HIT_TOTAL: &st
|
||||
pub const VALIDATOR_MONITOR_ATTESTATION_SIMULATOR_SOURCE_ATTESTER_MISS_TOTAL: &str =
|
||||
"validator_monitor_attestation_simulator_source_attester_miss_total";
|
||||
|
||||
/*
|
||||
* Execution Payload Envelope Procsesing
|
||||
*/
|
||||
|
||||
pub static ENVELOPE_PROCESSING_REQUESTS: LazyLock<Result<IntCounter>> = LazyLock::new(|| {
|
||||
try_create_int_counter(
|
||||
"payload_envelope_processing_requests_total",
|
||||
"Count of payload envelopes submitted for processing",
|
||||
)
|
||||
});
|
||||
pub static ENVELOPE_PROCESSING_SUCCESSES: LazyLock<Result<IntCounter>> = LazyLock::new(|| {
|
||||
try_create_int_counter(
|
||||
"payload_envelope_processing_successes_total",
|
||||
"Count of payload envelopes processed without error",
|
||||
)
|
||||
});
|
||||
pub static ENVELOPE_PROCESSING_TIMES: LazyLock<Result<Histogram>> = LazyLock::new(|| {
|
||||
try_create_histogram(
|
||||
"payload_envelope_processing_seconds",
|
||||
"Full runtime of payload envelope processing",
|
||||
)
|
||||
});
|
||||
pub static ENVELOPE_PROCESSING_DB_WRITE: LazyLock<Result<Histogram>> = LazyLock::new(|| {
|
||||
try_create_histogram(
|
||||
"payload_envelope_processing_db_write_seconds",
|
||||
"Time spent writing a newly processed payload envelope and state to DB",
|
||||
)
|
||||
});
|
||||
pub static ENVELOPE_PROCESSING_POST_EXEC_PROCESSING: LazyLock<Result<Histogram>> =
|
||||
LazyLock::new(|| {
|
||||
try_create_histogram_with_buckets(
|
||||
"payload_envelope_processing_post_exec_pre_attestable_seconds",
|
||||
"Time between finishing execution processing and the payload envelope
|
||||
becoming attestable",
|
||||
linear_buckets(0.01, 0.01, 15),
|
||||
)
|
||||
});
|
||||
|
||||
/*
|
||||
* Block Processing
|
||||
*/
|
||||
|
||||
@@ -10,7 +10,7 @@ use types::{
|
||||
};
|
||||
|
||||
use crate::{
|
||||
BeaconChain, BeaconChainError, BeaconChainTypes, BlockError, NotifyExecutionLayer,
|
||||
BeaconChain, BeaconChainError, BeaconChainTypes, NotifyExecutionLayer,
|
||||
PayloadVerificationOutcome,
|
||||
payload_envelope_verification::{
|
||||
EnvelopeError, EnvelopeImportData, EnvelopeProcessingSnapshot, ExecutionPendingEnvelope,
|
||||
@@ -151,8 +151,7 @@ impl<T: BeaconChainTypes> GossipVerifiedEnvelope<T> {
|
||||
);
|
||||
(is_valid, opt_snapshot)
|
||||
} else {
|
||||
// TODO(gloas) we should probably introduce a builder cache or some type of
|
||||
// global cache.
|
||||
// TODO(gloas) if we implement a builder pubkey cache, we'll need to use it here.
|
||||
// External builder: must load the state to get the builder pubkey.
|
||||
let snapshot = load_snapshot(signed_envelope.as_ref(), chain)?;
|
||||
let is_valid =
|
||||
@@ -181,7 +180,7 @@ impl<T: BeaconChainTypes> IntoExecutionPendingEnvelope<T> for GossipVerifiedEnve
|
||||
self,
|
||||
chain: &Arc<BeaconChain<T>>,
|
||||
notify_execution_layer: NotifyExecutionLayer,
|
||||
) -> Result<ExecutionPendingEnvelope<T::EthSpec>, BlockError> {
|
||||
) -> Result<ExecutionPendingEnvelope<T::EthSpec>, EnvelopeError> {
|
||||
let signed_envelope = self.signed_envelope;
|
||||
let envelope = &signed_envelope.message;
|
||||
let payload = &envelope.payload;
|
||||
@@ -198,13 +197,11 @@ impl<T: BeaconChainTypes> IntoExecutionPendingEnvelope<T> for GossipVerifiedEnve
|
||||
|
||||
let payload_verification_future = async move {
|
||||
let chain = payload_notifier.chain.clone();
|
||||
// TODO:(gloas): timing metrics
|
||||
if let Some(started_execution) = chain.slot_clock.now_duration() {
|
||||
chain.block_times_cache.write().set_time_started_execution(
|
||||
block_root,
|
||||
slot,
|
||||
started_execution,
|
||||
);
|
||||
chain
|
||||
.envelope_times_cache
|
||||
.write()
|
||||
.set_time_started_execution(block_root, slot, started_execution);
|
||||
}
|
||||
|
||||
let payload_verification_status = payload_notifier.notify_new_payload().await?;
|
||||
|
||||
@@ -1,20 +1,23 @@
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use fork_choice::PayloadVerificationStatus;
|
||||
use logging::crit;
|
||||
use slot_clock::SlotClock;
|
||||
use store::StoreOp;
|
||||
use tracing::{debug, error, info_span, instrument};
|
||||
use types::{BeaconState, BlockImportSource, EthSpec, Hash256, SignedBeaconBlock};
|
||||
use types::{BeaconState, BlockImportSource, Hash256, SignedBeaconBlock, Slot};
|
||||
|
||||
use super::{
|
||||
AvailableEnvelope, AvailableExecutedEnvelope, EnvelopeError, EnvelopeImportData,
|
||||
ExecutedEnvelope, IntoExecutionPendingEnvelope,
|
||||
};
|
||||
use crate::{
|
||||
AvailabilityProcessingStatus, BeaconChain, BeaconChainError, BeaconChainTypes, BlockError,
|
||||
AvailabilityProcessingStatus, BeaconChain, BeaconChainError, BeaconChainTypes,
|
||||
NotifyExecutionLayer,
|
||||
block_verification_types::{AsBlock, AvailableBlockData},
|
||||
payload_envelope_verification::{
|
||||
AvailableEnvelope, AvailableExecutedEnvelope, EnvelopeImportData, ExecutedEnvelope,
|
||||
IntoExecutionPendingEnvelope,
|
||||
},
|
||||
validator_monitor::timestamp_now,
|
||||
metrics,
|
||||
validator_monitor::{get_slot_delay_ms, timestamp_now},
|
||||
};
|
||||
|
||||
impl<T: BeaconChainTypes> BeaconChain<T> {
|
||||
@@ -36,21 +39,26 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
|
||||
unverified_envelope: P,
|
||||
notify_execution_layer: NotifyExecutionLayer,
|
||||
block_source: BlockImportSource,
|
||||
publish_fn: impl FnOnce() -> Result<(), BlockError>,
|
||||
) -> Result<AvailabilityProcessingStatus, BlockError> {
|
||||
publish_fn: impl FnOnce() -> Result<(), EnvelopeError>,
|
||||
) -> Result<AvailabilityProcessingStatus, EnvelopeError> {
|
||||
let block_slot = unverified_envelope.envelope().slot();
|
||||
|
||||
// TODO(gloas) Set observed time if not already set. Usually this should be set by gossip or RPC,
|
||||
// Set observed time if not already set. Usually this should be set by gossip or RPC,
|
||||
// but just in case we set it again here (useful for tests).
|
||||
if let Some(seen_timestamp) = self.slot_clock.now_duration() {
|
||||
self.envelope_times_cache.write().set_time_observed(
|
||||
block_root,
|
||||
block_slot,
|
||||
seen_timestamp,
|
||||
None,
|
||||
);
|
||||
}
|
||||
|
||||
// TODO(gloas) if we decide to insert the payload envelope into the new DA checker
|
||||
// we should insert the pre executed envelope here.
|
||||
// TODO(gloas) insert the pre-executed envelope into some type of cache.
|
||||
|
||||
// TODO(gloas) Start the Prometheus timer.
|
||||
// let _full_timer = metrics::start_timer(&metrics::BLOCK_PROCESSING_TIMES);
|
||||
let _full_timer = metrics::start_timer(&metrics::ENVELOPE_PROCESSING_TIMES);
|
||||
|
||||
// TODO(gloas) Increment the Prometheus counter for envelope processing requests.
|
||||
// metrics::inc_counter(&metrics::BLOCK_PROCESSING_REQUESTS);
|
||||
metrics::inc_counter(&metrics::ENVELOPE_PROCESSING_REQUESTS);
|
||||
|
||||
// A small closure to group the verification and import errors.
|
||||
let chain = self.clone();
|
||||
@@ -59,12 +67,16 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
|
||||
.into_execution_pending_envelope(&chain, notify_execution_layer)?;
|
||||
publish_fn()?;
|
||||
|
||||
// TODO(gloas) Record the time it took to complete consensus verification.
|
||||
// if let Some(timestamp) = self.slot_clock.now_duration() {
|
||||
// self.block_times_cache
|
||||
// .write()
|
||||
// .set_time_consensus_verified(block_root, block_slot, timestamp)
|
||||
// }
|
||||
// Record the time it took to complete consensus verification.
|
||||
if let Some(timestamp) = chain.slot_clock.now_duration() {
|
||||
chain
|
||||
.envelope_times_cache
|
||||
.write()
|
||||
.set_time_consensus_verified(block_root, block_slot, timestamp);
|
||||
}
|
||||
|
||||
let envelope_times_cache = chain.envelope_times_cache.clone();
|
||||
let slot_clock = chain.slot_clock.clone();
|
||||
|
||||
let executed_envelope = chain
|
||||
.into_executed_payload_envelope(execution_pending)
|
||||
@@ -75,28 +87,23 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
|
||||
// reprocess this block until the block is evicted from DA checker, causing the
|
||||
// chain to get stuck temporarily if the block is canonical. Therefore we remove
|
||||
// it from the cache if execution fails.
|
||||
|
||||
//self.data_availability_checker
|
||||
// .remove_block_on_execution_error(&block_root);
|
||||
})?;
|
||||
|
||||
// TODO(gloas) Record the *additional* time it took to wait for execution layer verification.
|
||||
// if let Some(timestamp) = self.slot_clock.now_duration() {
|
||||
// self.block_times_cache
|
||||
// .write()
|
||||
// .set_time_executed(block_root, block_slot, timestamp)
|
||||
// }
|
||||
// Record the time it took to wait for execution layer verification.
|
||||
if let Some(timestamp) = slot_clock.now_duration() {
|
||||
envelope_times_cache
|
||||
.write()
|
||||
.set_time_executed(block_root, block_slot, timestamp);
|
||||
}
|
||||
|
||||
match executed_envelope {
|
||||
ExecutedEnvelope::Available(envelope) => {
|
||||
self.import_available_execution_payload_envelope(Box::new(envelope))
|
||||
.await
|
||||
}
|
||||
ExecutedEnvelope::AvailabilityPending() => {
|
||||
return Err(BlockError::InternalError(
|
||||
"Pending payload envelope not yet implemented".to_owned(),
|
||||
));
|
||||
}
|
||||
ExecutedEnvelope::AvailabilityPending() => Err(EnvelopeError::InternalError(
|
||||
"Pending payload envelope not yet implemented".to_owned(),
|
||||
)),
|
||||
}
|
||||
};
|
||||
|
||||
@@ -111,8 +118,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
|
||||
"Envelope imported"
|
||||
);
|
||||
|
||||
// TODO(gloas) Increment the Prometheus counter for block processing successes.
|
||||
// metrics::inc_counter(&metrics::BLOCK_PROCESSING_SUCCESSES);
|
||||
metrics::inc_counter(&metrics::ENVELOPE_PROCESSING_SUCCESSES);
|
||||
|
||||
Ok(status)
|
||||
}
|
||||
@@ -121,7 +127,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
|
||||
|
||||
Ok(status)
|
||||
}
|
||||
Err(BlockError::BeaconChainError(e)) => {
|
||||
Err(EnvelopeError::BeaconChainError(e)) => {
|
||||
match e.as_ref() {
|
||||
BeaconChainError::TokioJoin(e) => {
|
||||
debug!(
|
||||
@@ -138,7 +144,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
|
||||
);
|
||||
}
|
||||
};
|
||||
Err(BlockError::BeaconChainError(e))
|
||||
Err(EnvelopeError::BeaconChainError(e))
|
||||
}
|
||||
// The block failed verification.
|
||||
Err(other) => {
|
||||
@@ -152,7 +158,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
|
||||
pub async fn import_available_execution_payload_envelope(
|
||||
self: &Arc<Self>,
|
||||
envelope: Box<AvailableExecutedEnvelope<T::EthSpec>>,
|
||||
) -> Result<AvailabilityProcessingStatus, BlockError> {
|
||||
) -> Result<AvailabilityProcessingStatus, EnvelopeError> {
|
||||
let AvailableExecutedEnvelope {
|
||||
envelope,
|
||||
import_data,
|
||||
@@ -165,8 +171,6 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
|
||||
post_state,
|
||||
} = import_data;
|
||||
|
||||
// TODO(gloas) Record the time at which this block's blobs became available.
|
||||
|
||||
let block_root = {
|
||||
// Capture the current span before moving into the blocking task
|
||||
let current_span = tracing::Span::current();
|
||||
@@ -202,21 +206,18 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
|
||||
&self,
|
||||
signed_envelope: AvailableEnvelope<T::EthSpec>,
|
||||
block_root: Hash256,
|
||||
mut state: BeaconState<T::EthSpec>,
|
||||
payload_verification_status: PayloadVerificationStatus,
|
||||
state: BeaconState<T::EthSpec>,
|
||||
_payload_verification_status: PayloadVerificationStatus,
|
||||
parent_block: Arc<SignedBeaconBlock<T::EthSpec>>,
|
||||
) -> Result<Hash256, BlockError> {
|
||||
) -> Result<Hash256, EnvelopeError> {
|
||||
// ----------------------------- ENVELOPE NOT YET ATTESTABLE ----------------------------------
|
||||
// Everything in this initial section is on the hot path between processing the envelope and
|
||||
// being able to attest to it. DO NOT add any extra processing in this initial section
|
||||
// unless it must run before fork choice.
|
||||
// -----------------------------------------------------------------------------------------
|
||||
let current_slot = self.slot()?;
|
||||
let current_epoch = current_slot.epoch(T::EthSpec::slots_per_epoch());
|
||||
let envelope = signed_envelope.message();
|
||||
|
||||
// TODO(gloas) implement metrics
|
||||
// let post_exec_timer = metrics::start_timer(&metrics::BLOCK_PROCESSING_POST_EXEC_PROCESSING);
|
||||
let post_exec_timer =
|
||||
metrics::start_timer(&metrics::ENVELOPE_PROCESSING_POST_EXEC_PROCESSING);
|
||||
|
||||
// Check the payloads parent block against weak subjectivity checkpoint.
|
||||
self.check_block_against_weak_subjectivity_checkpoint(
|
||||
@@ -228,26 +229,24 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
|
||||
// Take an upgradable read lock on fork choice so we can check if this block has already
|
||||
// been imported. We don't want to repeat work importing a block that is already imported.
|
||||
let fork_choice_reader = self.canonical_head.fork_choice_upgradable_read_lock();
|
||||
if fork_choice_reader.contains_block(&block_root) {
|
||||
return Err(BlockError::DuplicateFullyImported(block_root));
|
||||
if !fork_choice_reader.contains_block(&block_root) {
|
||||
return Err(EnvelopeError::BlockRootUnknown { block_root });
|
||||
}
|
||||
|
||||
// TODO(gloas) no fork choice logic yet
|
||||
// Take an exclusive write-lock on fork choice. It's very important to prevent deadlocks by
|
||||
// avoiding taking other locks whilst holding this lock.
|
||||
let mut fork_choice = parking_lot::RwLockUpgradableReadGuard::upgrade(fork_choice_reader);
|
||||
// let fork_choice = parking_lot::RwLockUpgradableReadGuard::upgrade(fork_choice_reader);
|
||||
|
||||
// TODO(gloas) Do we need this check? Do not import a block that doesn't descend from the finalized root.
|
||||
// let signed_block = check_block_is_finalized_checkpoint_or_descendant(self, &fork_choice, signed_block)?;
|
||||
|
||||
// TODO(gloas)Do we want to use an early attester cache like mechanism for payload enevelopes?
|
||||
// TODO(gloas) Do we want to use an early attester cache like mechanism for payload enevelopes?
|
||||
// TODO(gloas) emit SSE event if the payload became the new head payload
|
||||
// drop(post_exec_timer);
|
||||
drop(post_exec_timer);
|
||||
|
||||
// ---------------------------- BLOCK PROBABLY ATTESTABLE ----------------------------------
|
||||
// Most blocks are now capable of being attested to thanks to the `early_attester_cache`
|
||||
// cache above. Resume non-essential processing.
|
||||
//
|
||||
// It is important NOT to return errors here before the database commit, because the block
|
||||
// ---------------------------- ENVELOPE PROBABLY ATTESTABLE ----------------------------------
|
||||
// It is important NOT to return errors here before the database commit, because the envelope
|
||||
// has already been added to fork choice and the database would be left in an inconsistent
|
||||
// state if we returned early without committing. In other words, an error here would
|
||||
// corrupt the node's database permanently.
|
||||
@@ -278,15 +277,13 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
|
||||
?block_root,
|
||||
"Failed to store data columns into the database"
|
||||
);
|
||||
return Err(self
|
||||
.handle_import_block_db_write_error(fork_choice)
|
||||
.err()
|
||||
.unwrap_or(BlockError::InternalError(e)));
|
||||
// TODO(gloas) implement failed write handling to fork choice
|
||||
// let _ = self.handle_import_block_db_write_error(fork_choice);
|
||||
return Err(EnvelopeError::InternalError(e));
|
||||
}
|
||||
}
|
||||
|
||||
// TODO(gloas) metrics
|
||||
// let db_write_timer = metrics::start_timer(&metrics::BLOCK_PROCESSING_DB_WRITE);
|
||||
let db_write_timer = metrics::start_timer(&metrics::ENVELOPE_PROCESSING_DB_WRITE);
|
||||
|
||||
ops.push(StoreOp::PutPayloadEnvelope(
|
||||
block_root,
|
||||
@@ -299,7 +296,6 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
|
||||
|
||||
let db_span = info_span!("persist_payloads_and_blobs").entered();
|
||||
|
||||
// TODO(gloas) do i need this
|
||||
if let Err(e) = self.store.do_atomically_with_block_and_blobs_cache(ops) {
|
||||
error!(
|
||||
msg = "Restoring fork choice from disk",
|
||||
@@ -315,25 +311,49 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
|
||||
|
||||
drop(db_span);
|
||||
|
||||
// TODO(gloas) drop fork choice lock
|
||||
// The fork choice write-lock is dropped *after* the on-disk database has been updated.
|
||||
// This prevents inconsistency between the two at the expense of concurrency.
|
||||
drop(fork_choice);
|
||||
// drop(fork_choice);
|
||||
|
||||
// We're declaring the envelope "imported" at this point, since fork choice and the DB know
|
||||
// about it.
|
||||
let envelope_time_imported = timestamp_now();
|
||||
|
||||
// TODO(gloas) depending on what happens with light clients
|
||||
// we might need to do some computations here
|
||||
// we might need to do some light client related computations here
|
||||
|
||||
// TODO(gloas) metrics
|
||||
// metrics::stop_timer(db_write_timer);
|
||||
metrics::stop_timer(db_write_timer);
|
||||
metrics::inc_counter(&metrics::ENVELOPE_PROCESSING_SUCCESSES);
|
||||
|
||||
// TODO(gloas) metrics
|
||||
// metrics::inc_counter(&metrics::BLOCK_PROCESSING_SUCCESSES);
|
||||
self.import_envelope_update_metrics_and_events(
|
||||
block_root,
|
||||
signed_envelope.slot(),
|
||||
envelope_time_imported,
|
||||
);
|
||||
|
||||
// TODO(gloas) we might want to implement something similar
|
||||
// to `import_block_update_metrics_and_events`
|
||||
Ok(block_root)
|
||||
}
|
||||
|
||||
fn import_envelope_update_metrics_and_events(
|
||||
&self,
|
||||
block_root: Hash256,
|
||||
envelope_slot: Slot,
|
||||
envelope_time_imported: Duration,
|
||||
) {
|
||||
let envelope_delay_total =
|
||||
get_slot_delay_ms(envelope_time_imported, envelope_slot, &self.slot_clock);
|
||||
|
||||
// Do not write to the cache for envelopes older than 2 epochs, this helps reduce writes
|
||||
// to the cache during sync.
|
||||
if envelope_delay_total < self.slot_clock.slot_duration().saturating_mul(64) {
|
||||
self.envelope_times_cache.write().set_time_imported(
|
||||
block_root,
|
||||
envelope_slot,
|
||||
envelope_time_imported,
|
||||
);
|
||||
}
|
||||
|
||||
// TODO(gloas) emit SSE event for envelope import (similar to SseBlock for blocks).
|
||||
}
|
||||
}
|
||||
@@ -1,8 +1,9 @@
|
||||
//! The incremental processing steps (e.g., signatures verified but not the state transition) is
|
||||
//! represented as a sequence of wrapper-types around the block. There is a linear progression of
|
||||
//! types, starting at a `SignedBeaconBlock` and finishing with a `Fully VerifiedBlock` (see
|
||||
//! types, starting at a `SignedExecutionPayloadEnvelope` and finishing with an `AvailableExecutedEnvelope` (see
|
||||
//! diagram below).
|
||||
//!
|
||||
//! // TODO(gloas) we might want to update this diagram to include `AvailabelExecutedEnvelope`
|
||||
//! ```ignore
|
||||
//! START
|
||||
//! |
|
||||
@@ -28,9 +29,9 @@
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use state_processing::{
|
||||
BlockProcessingError, ConsensusContext, envelope_processing::EnvelopeProcessingError,
|
||||
};
|
||||
use store::Error as DBError;
|
||||
|
||||
use state_processing::{BlockProcessingError, envelope_processing::EnvelopeProcessingError};
|
||||
use tracing::instrument;
|
||||
use types::{
|
||||
BeaconState, BeaconStateError, ChainSpec, DataColumnSidecarList, EthSpec, ExecutionBlockHash,
|
||||
@@ -45,15 +46,15 @@ use crate::{
|
||||
};
|
||||
|
||||
pub mod gossip_verified_envelope;
|
||||
pub mod import;
|
||||
mod payload_notifier;
|
||||
mod tests;
|
||||
|
||||
pub trait IntoExecutionPendingEnvelope<T: BeaconChainTypes>: Sized {
|
||||
fn into_execution_pending_envelope(
|
||||
self,
|
||||
chain: &Arc<BeaconChain<T>>,
|
||||
notify_execution_layer: NotifyExecutionLayer,
|
||||
) -> Result<ExecutionPendingEnvelope<T::EthSpec>, BlockError>;
|
||||
) -> Result<ExecutionPendingEnvelope<T::EthSpec>, EnvelopeError>;
|
||||
|
||||
fn envelope(&self) -> &Arc<SignedExecutionPayloadEnvelope<T::EthSpec>>;
|
||||
}
|
||||
@@ -74,8 +75,7 @@ pub struct EnvelopeImportData<E: EthSpec> {
|
||||
#[derive(Debug)]
|
||||
#[allow(dead_code)]
|
||||
pub struct AvailableEnvelope<E: EthSpec> {
|
||||
// TODO(EIP-7732): rename to execution_block_hash
|
||||
block_hash: ExecutionBlockHash,
|
||||
execution_block_hash: ExecutionBlockHash,
|
||||
envelope: Arc<SignedExecutionPayloadEnvelope<E>>,
|
||||
columns: DataColumnSidecarList<E>,
|
||||
/// Timestamp at which this block first became available (UNIX timestamp, time since 1970).
|
||||
@@ -222,6 +222,10 @@ pub enum EnvelopeError {
|
||||
EnvelopeProcessingError(EnvelopeProcessingError),
|
||||
// Error verifying the execution payload
|
||||
ExecutionPayloadError(ExecutionPayloadError),
|
||||
// An error from block-level checks reused during envelope import
|
||||
BlockError(BlockError),
|
||||
// Internal error
|
||||
InternalError(String),
|
||||
}
|
||||
|
||||
impl std::fmt::Display for EnvelopeError {
|
||||
@@ -248,6 +252,18 @@ impl From<BeaconStateError> for EnvelopeError {
|
||||
}
|
||||
}
|
||||
|
||||
impl From<DBError> for EnvelopeError {
|
||||
fn from(e: DBError) -> Self {
|
||||
EnvelopeError::BeaconChainError(Arc::new(BeaconChainError::DBError(e)))
|
||||
}
|
||||
}
|
||||
|
||||
impl From<BlockError> for EnvelopeError {
|
||||
fn from(e: BlockError) -> Self {
|
||||
EnvelopeError::BlockError(e)
|
||||
}
|
||||
}
|
||||
|
||||
/// Pull errors up from EnvelopeProcessingError to EnvelopeError
|
||||
impl From<EnvelopeProcessingError> for EnvelopeError {
|
||||
fn from(e: EnvelopeProcessingError) -> Self {
|
||||
@@ -274,14 +290,14 @@ impl From<EnvelopeProcessingError> for EnvelopeError {
|
||||
pub(crate) fn load_snapshot<T: BeaconChainTypes>(
|
||||
envelope: &SignedExecutionPayloadEnvelope<T::EthSpec>,
|
||||
chain: &BeaconChain<T>,
|
||||
) -> Result<EnvelopeProcessingSnapshot<T::EthSpec>, BlockError> {
|
||||
// Reject any block if its block is not known to fork choice.
|
||||
) -> Result<EnvelopeProcessingSnapshot<T::EthSpec>, EnvelopeError> {
|
||||
// Reject any envelope if its block is not known to fork choice.
|
||||
//
|
||||
// A block that is not in fork choice is either:
|
||||
//
|
||||
// - Not yet imported: we should reject this block because we should only import a child
|
||||
// envelope after its parent has been fully imported.
|
||||
// - Pre-finalized: if the block is _prior_ to finalization, we should ignore the envelope
|
||||
// - Not yet imported: we should reject this envelope because we should only import it after its parent block
|
||||
// has been fully imported.
|
||||
// - Pre-finalized: if the parent block is _prior_ to finalization, we should ignore the envelope
|
||||
// because it will revert finalization. Note that the finalized block is stored in fork
|
||||
// choice, so we will not reject any child of the finalized block (this is relevant during
|
||||
// genesis).
|
||||
@@ -289,8 +305,8 @@ pub(crate) fn load_snapshot<T: BeaconChainTypes>(
|
||||
let fork_choice_read_lock = chain.canonical_head.fork_choice_read_lock();
|
||||
let beacon_block_root = envelope.beacon_block_root();
|
||||
let Some(proto_beacon_block) = fork_choice_read_lock.get_block(&beacon_block_root) else {
|
||||
return Err(BlockError::ParentUnknown {
|
||||
parent_root: beacon_block_root,
|
||||
return Err(EnvelopeError::BlockRootUnknown {
|
||||
block_root: beacon_block_root,
|
||||
});
|
||||
};
|
||||
drop(fork_choice_read_lock);
|
||||
@@ -304,7 +320,7 @@ pub(crate) fn load_snapshot<T: BeaconChainTypes>(
|
||||
let state = chain
|
||||
.store
|
||||
.get_hot_state(&block_state_root, cache_state)
|
||||
.map_err(|e| BlockError::BeaconChainError(Box::new(e.into())))?
|
||||
.map_err(EnvelopeError::from)?
|
||||
.ok_or_else(|| {
|
||||
BeaconChainError::DBInconsistent(format!(
|
||||
"Missing state for envelope block {block_state_root:?}",
|
||||
@@ -325,8 +341,7 @@ impl<T: BeaconChainTypes> IntoExecutionPendingEnvelope<T>
|
||||
self,
|
||||
chain: &Arc<BeaconChain<T>>,
|
||||
notify_execution_layer: NotifyExecutionLayer,
|
||||
) -> Result<ExecutionPendingEnvelope<T::EthSpec>, BlockError> {
|
||||
// TODO(EIP-7732): figure out how this should be refactored..
|
||||
) -> Result<ExecutionPendingEnvelope<T::EthSpec>, EnvelopeError> {
|
||||
GossipVerifiedEnvelope::new(self, chain)?
|
||||
.into_execution_pending_envelope(chain, notify_execution_layer)
|
||||
}
|
||||
@@ -335,10 +350,3 @@ impl<T: BeaconChainTypes> IntoExecutionPendingEnvelope<T>
|
||||
self
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
pub struct PayloadEnvelopeImportData<E: EthSpec> {
|
||||
pub block_root: Hash256,
|
||||
pub state: BeaconState<E>,
|
||||
pub consensus_context: ConsensusContext<E>,
|
||||
}
|
||||
|
||||
@@ -7,7 +7,7 @@ use tracing::warn;
|
||||
use types::{SignedBeaconBlock, SignedExecutionPayloadEnvelope};
|
||||
|
||||
use crate::{
|
||||
BeaconChain, BeaconChainTypes, BlockError, ExecutionPayloadError, NotifyExecutionLayer,
|
||||
BeaconChain, BeaconChainTypes, BlockError, NotifyExecutionLayer,
|
||||
execution_payload::notify_new_payload, payload_envelope_verification::EnvelopeError,
|
||||
};
|
||||
|
||||
@@ -25,15 +25,13 @@ impl<T: BeaconChainTypes> PayloadNotifier<T> {
|
||||
envelope: Arc<SignedExecutionPayloadEnvelope<T::EthSpec>>,
|
||||
block: Arc<SignedBeaconBlock<T::EthSpec>>,
|
||||
notify_execution_layer: NotifyExecutionLayer,
|
||||
) -> Result<Self, ExecutionPayloadError> {
|
||||
) -> Result<Self, EnvelopeError> {
|
||||
let payload_verification_status = {
|
||||
let payload_message = &envelope.message;
|
||||
|
||||
match notify_execution_layer {
|
||||
NotifyExecutionLayer::No if chain.config.optimistic_finalized_sync => {
|
||||
// TODO(gloas) unwrap
|
||||
let new_payload_request =
|
||||
Self::build_new_payload_request(&envelope, &block).unwrap();
|
||||
let new_payload_request = Self::build_new_payload_request(&envelope, &block)?;
|
||||
if let Err(e) = new_payload_request.perform_optimistic_sync_verifications() {
|
||||
warn!(
|
||||
block_number = ?payload_message.payload.block_number,
|
||||
|
||||
@@ -1,23 +0,0 @@
|
||||
use std::sync::Arc;
|
||||
|
||||
use crate::{
|
||||
AvailabilityProcessingStatus, BeaconChain, BeaconChainTypes,
|
||||
payload_envelope_verification::{ExecutedEnvelope, ExecutionPendingEnvelope},
|
||||
};
|
||||
|
||||
async fn import_execution_pending_envelope<T: BeaconChainTypes>(
|
||||
chain: Arc<BeaconChain<T>>,
|
||||
execution_pending_envelope: ExecutionPendingEnvelope<T::EthSpec>,
|
||||
) -> Result<AvailabilityProcessingStatus, String> {
|
||||
match chain
|
||||
.clone()
|
||||
.into_executed_payload_envelope(execution_pending_envelope)
|
||||
.await
|
||||
.unwrap()
|
||||
{
|
||||
ExecutedEnvelope::Available(envelope) => todo!(),
|
||||
ExecutedEnvelope::AvailabilityPending() => {
|
||||
Err("AvailabilityPending not expected in this test. Block not imported.".to_string())
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -30,7 +30,7 @@ use types::{
|
||||
Attestation, AttestationData, AttesterSlashingRef, BeaconBlockRef, BeaconState,
|
||||
BeaconStateError, ChainSpec, Epoch, EthSpec, Hash256, IndexedAttestation,
|
||||
IndexedAttestationRef, ProposerSlashing, SignedAggregateAndProof, SignedContributionAndProof,
|
||||
SignedExecutionPayloadEnvelope, Slot, SyncCommitteeMessage, VoluntaryExit,
|
||||
Slot, SyncCommitteeMessage, VoluntaryExit,
|
||||
};
|
||||
|
||||
/// Used for Prometheus labels.
|
||||
|
||||
Reference in New Issue
Block a user