Gloas payload envelope processing [WIP]

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -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};
@@ -463,6 +464,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
@@ -1151,6 +1154,23 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         }
     }
 
+    /// Returns the full block at the given root, if it's available in the database.
+    ///
+    /// Should always return a full block for pre-merge and post-gloas blocks.
+    pub fn get_full_block(
+        &self,
+        block_root: &Hash256,
+    ) -> Result<Option<SignedBeaconBlock<T::EthSpec>>, Error> {
+        match self.store.try_get_full_block(block_root)? {
+            Some(DatabaseBlock::Full(block)) => Ok(Some(block)),
+            Some(DatabaseBlock::Blinded(_)) => {
+                // TODO(gloas) should we return None here?
+                Ok(None)
+            }
+            None => Ok(None),
+        }
+    }
+
     /// Returns the block at the given root, if any.
     ///
     /// ## Errors
@@ -4161,7 +4181,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     }
 
     /// Check block's consistentency with any configured weak subjectivity checkpoint.
-    fn check_block_against_weak_subjectivity_checkpoint(
+    pub(crate) fn check_block_against_weak_subjectivity_checkpoint(
         &self,
         block: BeaconBlockRef<T::EthSpec>,
         block_root: Hash256,
@@ -6468,6 +6488,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 {

beacon_node/beacon_chain/src/block_verification.rs

@@ -718,7 +718,8 @@ pub struct SignatureVerifiedBlock<T: BeaconChainTypes> {
 }
 
 /// Used to await the result of executing payload with an EE.
-type PayloadVerificationHandle = JoinHandle<Option<Result<PayloadVerificationOutcome, BlockError>>>;
+pub type PayloadVerificationHandle =
+    JoinHandle<Option<Result<PayloadVerificationOutcome, BlockError>>>;
 
 /// A wrapper around a `SignedBeaconBlock` that indicates that this block is fully verified and
 /// ready to import into the `BeaconChain`. The validation includes:

beacon_node/beacon_chain/src/builder.rs

@@ -1056,6 +1056,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(),

beacon_node/beacon_chain/src/envelope_times_cache.rs

@@ -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));
+    }
+}

beacon_node/beacon_chain/src/execution_payload.rs

@@ -112,12 +112,17 @@ impl<T: BeaconChainTypes> PayloadNotifier<T> {
         if let Some(precomputed_status) = self.payload_verification_status {
             Ok(precomputed_status)
         } else {
-            notify_new_payload(&self.chain, self.block.message()).await
+            notify_new_payload(
+                &self.chain,
+                self.block.message().tree_hash_root(),
+                self.block.message().try_into()?,
+            )
+            .await
         }
     }
 }
 
-/// Verify that `execution_payload` contained by `block` is considered valid by an execution
+/// Verify that `execution_payload` associated with `beacon_block_root` is considered valid by an execution
 /// engine.
 ///
 /// ## Specification
@@ -126,17 +131,20 @@ impl<T: BeaconChainTypes> PayloadNotifier<T> {
 /// contains a few extra checks by running `partially_verify_execution_payload` first:
 ///
 /// https://github.com/ethereum/consensus-specs/blob/v1.1.9/specs/bellatrix/beacon-chain.md#notify_new_payload
-async fn notify_new_payload<T: BeaconChainTypes>(
+pub async fn notify_new_payload<T: BeaconChainTypes>(
     chain: &Arc<BeaconChain<T>>,
-    block: BeaconBlockRef<'_, T::EthSpec>,
+    beacon_block_root: Hash256,
+    new_payload_request: NewPayloadRequest<'_, T::EthSpec>,
 ) -> Result<PayloadVerificationStatus, BlockError> {
     let execution_layer = chain
         .execution_layer
         .as_ref()
         .ok_or(ExecutionPayloadError::NoExecutionConnection)?;
 
-    let execution_block_hash = block.execution_payload()?.block_hash();
-    let new_payload_response = execution_layer.notify_new_payload(block.try_into()?).await;
+    let execution_block_hash = new_payload_request.execution_payload_ref().block_hash();
+    let new_payload_response = execution_layer
+        .notify_new_payload(new_payload_request.clone())
+        .await;
 
     match new_payload_response {
         Ok(status) => match status {
@@ -152,10 +160,11 @@ async fn notify_new_payload<T: BeaconChainTypes>(
                     ?validation_error,
                     ?latest_valid_hash,
                     ?execution_block_hash,
-                    root = ?block.tree_hash_root(),
-                    graffiti = block.body().graffiti().as_utf8_lossy(),
-                    proposer_index = block.proposer_index(),
-                    slot = %block.slot(),
+                    // TODO(gloas) are these other logs important?
+                    root = ?beacon_block_root,
+                    // graffiti = block.body().graffiti().as_utf8_lossy(),
+                    // proposer_index = block.proposer_index(),
+                    // slot = %block.slot(),
                     method = "new_payload",
                     "Invalid execution payload"
                 );
@@ -178,11 +187,11 @@ async fn notify_new_payload<T: BeaconChainTypes>(
                 {
                     // This block has not yet been applied to fork choice, so the latest block that was
                     // imported to fork choice was the parent.
-                    let latest_root = block.parent_root();
+                    let latest_root = new_payload_request.parent_beacon_block_root()?;
 
                     chain
                         .process_invalid_execution_payload(&InvalidationOperation::InvalidateMany {
-                            head_block_root: latest_root,
+                            head_block_root: *latest_root,
                             always_invalidate_head: false,
                             latest_valid_ancestor: latest_valid_hash,
                         })
@@ -197,10 +206,11 @@ async fn notify_new_payload<T: BeaconChainTypes>(
                 warn!(
                     ?validation_error,
                     ?execution_block_hash,
-                    root = ?block.tree_hash_root(),
-                    graffiti = block.body().graffiti().as_utf8_lossy(),
-                    proposer_index = block.proposer_index(),
-                    slot = %block.slot(),
+                    // TODO(gloas) are these other logs important?
+                    root = ?beacon_block_root,
+                    // graffiti = block.body().graffiti().as_utf8_lossy(),
+                    // proposer_index = block.proposer_index(),
+                    // slot = %block.slot(),
                     method = "new_payload",
                     "Invalid execution payload block hash"
                 );

beacon_node/beacon_chain/src/lib.rs

@@ -21,6 +21,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;
@@ -43,6 +44,7 @@ pub mod observed_block_producers;
 pub mod observed_data_sidecars;
 pub mod observed_operations;
 mod observed_slashable;
+pub mod payload_envelope_verification;
 pub mod pending_payload_envelopes;
 pub mod persisted_beacon_chain;
 pub mod persisted_custody;