merge queue: embarking unstable (037b263) and [#9065 + #9024] together

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -3801,7 +3801,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             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,
@@ -3810,8 +3810,6 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             );
         }
 
-        // 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();

beacon_node/beacon_chain/src/canonical_head.rs

@@ -1379,8 +1379,8 @@ fn observe_head_block_delays<E: EthSpec, S: SlotClock>(
                 .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

beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs

@@ -282,8 +282,11 @@ impl<E: EthSpec> PendingComponents<E> {
                 .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 {

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<T: BeaconChainTypes, O: ObservationStrategy> GossipVerifiedDataColumn<T, O>
 #[ssz(struct_behaviour = "transparent")]
 pub struct KzgVerifiedDataColumn<E: EthSpec> {
     data: Arc<DataColumnSidecar<E>>,
+    #[ssz(skip_serializing, skip_deserializing)]
+    seen_timestamp: Duration,
 }
 
 impl<E: EthSpec> KzgVerifiedDataColumn<E> {
     pub fn new(
         data_column: Arc<DataColumnSidecar<E>>,
         kzg: &Kzg,
+        seen_timestamp: Duration,
     ) -> Result<Self, (Option<ColumnIndex>, 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<DataColumnSidecar<E>>) -> 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<DataColumnSidecar<E>>) -> Self {
-        Self { data: data_column }
+        Self {
+            data: data_column,
+            seen_timestamp: timestamp_now(),
+        }
     }
 
     pub fn from_batch_with_scoring(
@@ -348,7 +359,10 @@ impl<E: EthSpec> KzgVerifiedDataColumn<E> {
         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,13 +421,16 @@ impl<E: EthSpec> CustodyDataColumn<E> {
 #[ssz(struct_behaviour = "transparent")]
 pub struct KzgVerifiedCustodyDataColumn<E: EthSpec> {
     data: Arc<DataColumnSidecar<E>>,
+    #[ssz(skip_serializing, skip_deserializing)]
+    seen_timestamp: Duration,
 }
 
 impl<E: EthSpec> KzgVerifiedCustodyDataColumn<E> {
     /// Mark a column as custody column. Caller must ensure that our current custody requirements
     /// include this column
     pub fn from_asserted_custody(kzg_verified: KzgVerifiedDataColumn<E>) -> Self {
         Self {
+            seen_timestamp: kzg_verified.seen_timestamp,
             data: kzg_verified.to_data_column(),
         }
     }
@@ -422,10 +439,12 @@ impl<E: EthSpec> KzgVerifiedCustodyDataColumn<E> {
     pub fn new(
         data_column: CustodyDataColumn<E>,
         kzg: &Kzg,
+        seen_timestamp: Duration,
     ) -> Result<Self, (Option<ColumnIndex>, 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<E: EthSpec> KzgVerifiedCustodyDataColumn<E> {
             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::<Vec<_>>())
     }
@@ -464,6 +488,10 @@ impl<E: EthSpec> KzgVerifiedCustodyDataColumn<E> {
     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<E: EthSpec> KzgVerifiedCustodyDataColumn<E> {
 pub fn verify_kzg_for_data_column<E: EthSpec>(
     data_column: Arc<DataColumnSidecar<E>>,
     kzg: &Kzg,
+    seen_timestamp: Duration,
 ) -> Result<KzgVerifiedDataColumn<E>, (Option<ColumnIndex>, 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<T: BeaconChainTypes, O: Obse
     verify_slot_higher_than_parent(&parent_block, column_slot)?;
     verify_proposer_and_signature(data_column_fulu, &parent_block, chain)?;
     let kzg = &chain.kzg;
-    let kzg_verified_data_column = verify_kzg_for_data_column(data_column.clone(), kzg)
-        .map_err(|(_, e)| GossipDataColumnError::InvalidKzgProof(e))?;
+    let kzg_verified_data_column =
+        verify_kzg_for_data_column(data_column.clone(), kzg, timestamp_now())
+            .map_err(|(_, e)| GossipDataColumnError::InvalidKzgProof(e))?;
 
     chain
         .observed_slashable

beacon_node/beacon_chain/src/payload_envelope_verification/import.rs

@@ -1,11 +1,12 @@
 use std::sync::Arc;
 use std::time::Duration;
 
+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, Hash256, Slot};
+use types::{BeaconState, BlockImportSource, Hash256, SignedExecutionPayloadEnvelope};
 
 use super::{
     AvailableEnvelope, AvailableExecutedEnvelope, EnvelopeError, EnvelopeImportData,
@@ -225,7 +226,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         signed_envelope: AvailableEnvelope<T::EthSpec>,
         block_root: Hash256,
         state: BeaconState<T::EthSpec>,
-        _payload_verification_status: PayloadVerificationStatus,
+        payload_verification_status: PayloadVerificationStatus,
     ) -> Result<Hash256, EnvelopeError> {
         // 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
@@ -317,8 +318,9 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         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,
         );
 
@@ -327,10 +329,12 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
     fn import_envelope_update_metrics_and_events(
         &self,
+        signed_envelope: Arc<SignedExecutionPayloadEnvelope<T::EthSpec>>,
         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);
 
@@ -349,6 +353,17 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             );
         }
 
-        // TODO(gloas) emit SSE event for envelope import (similar to SseBlock for blocks).
+        if let Some(event_handler) = self.event_handler.as_ref()
+            && event_handler.has_execution_payload_subscribers()
+        {
+            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(),
+            }));
+        }
     }
 }