Extend observed last blob to PeerDAS columns

beacon_node/beacon_chain/src/data_availability_checker.rs

@@ -5,6 +5,7 @@ use crate::block_verification_types::{
 use crate::data_availability_checker::overflow_lru_cache::{
     DataAvailabilityCheckerInner, ReconstructColumnsDecision,
 };
+use crate::validator_monitor::timestamp_now;
 use crate::{metrics, BeaconChain, BeaconChainTypes, BeaconStore};
 use kzg::Kzg;
 use slog::{debug, error, Logger};
@@ -230,6 +231,11 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
         block_root: Hash256,
         custody_columns: DataColumnSidecarList<T::EthSpec>,
     ) -> Result<Availability<T::EthSpec>, AvailabilityCheckError> {
+        let seen_timestamp = self
+            .slot_clock
+            .now_duration()
+            .ok_or(AvailabilityCheckError::SlotClockError)?;
+
         // TODO(das): report which column is invalid for proper peer scoring
         // TODO(das): batch KZG verification here, but fallback into checking each column
         // individually to report which column(s) are invalid.
@@ -238,7 +244,7 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
             .map(|column| {
                 let index = column.index;
                 Ok(KzgVerifiedCustodyDataColumn::from_asserted_custody(
-                    KzgVerifiedDataColumn::new(column, &self.kzg)
+                    KzgVerifiedDataColumn::new(column, &self.kzg, seen_timestamp)
                         .map_err(|e| AvailabilityCheckError::InvalidColumn(index, e))?,
                 ))
             })
@@ -565,6 +571,8 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
             &self.kzg,
             &pending_components.verified_data_columns,
             &self.spec,
+            // TODO(das): what timestamp to use when reconstructing columns?
+            timestamp_now(),
         )
         .map_err(|e| {
             error!(
@@ -732,7 +740,8 @@ where
     // If len > 1 at least one column MUST fail.
     if data_columns.len() > 1 {
         for data_column in data_columns {
-            if let Err(e) = verify_kzg_for_data_column(data_column.clone(), kzg) {
+            let seen_timestamp = Duration::ZERO; // not used
+            if let Err(e) = verify_kzg_for_data_column(data_column.clone(), kzg, seen_timestamp) {
                 return Err(AvailabilityCheckError::InvalidColumn(data_column.index, e));
             }
         }

beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs

@@ -283,17 +283,24 @@ impl<E: EthSpec> PendingComponents<E> {
             ..
         } = self;
 
-        let blobs_available_timestamp = verified_blobs
-            .iter()
-            .flatten()
-            .map(|blob| blob.seen_timestamp())
-            .max();
-
         let Some(diet_executed_block) = executed_block else {
             return Err(AvailabilityCheckError::Unexpected);
         };
 
         let is_peer_das_enabled = spec.is_peer_das_enabled_for_epoch(diet_executed_block.epoch());
+        let blobs_available_timestamp = if is_peer_das_enabled {
+            verified_data_columns
+                .iter()
+                .map(|data_column| data_column.seen_timestamp())
+                .max()
+        } else {
+            verified_blobs
+                .iter()
+                .flatten()
+                .map(|blob| blob.seen_timestamp())
+                .max()
+        };
+
         let (blobs, data_columns) = if is_peer_das_enabled {
             let data_columns = verified_data_columns
                 .into_iter()

beacon_node/beacon_chain/src/data_column_verification.rs

@@ -16,6 +16,7 @@ use ssz_derive::{Decode, Encode};
 use std::iter;
 use std::marker::PhantomData;
 use std::sync::Arc;
+use std::time::Duration;
 use types::data_column_sidecar::{ColumnIndex, DataColumnIdentifier};
 use types::{
     BeaconStateError, ChainSpec, DataColumnSidecar, DataColumnSubnetId, EthSpec, Hash256,
@@ -233,11 +234,17 @@ 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) -> Result<Self, KzgError> {
-        verify_kzg_for_data_column(data_column, kzg)
+    pub fn new(
+        data_column: Arc<DataColumnSidecar<E>>,
+        kzg: &Kzg,
+        seen_timestamp: Duration,
+    ) -> Result<Self, KzgError> {
+        verify_kzg_for_data_column(data_column, kzg, seen_timestamp)
     }
     pub fn to_data_column(self) -> Arc<DataColumnSidecar<E>> {
         self.data
@@ -293,29 +300,38 @@ 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(),
         }
     }
 
     /// Verify a column already marked as custody column
-    pub fn new(data_column: CustodyDataColumn<E>, kzg: &Kzg) -> Result<Self, KzgError> {
-        verify_kzg_for_data_column(data_column.clone_arc(), kzg)?;
+    pub fn new(
+        data_column: CustodyDataColumn<E>,
+        kzg: &Kzg,
+        seen_timestamp: Duration,
+    ) -> Result<Self, KzgError> {
+        verify_kzg_for_data_column(data_column.clone_arc(), kzg, seen_timestamp)?;
         Ok(Self {
             data: data_column.data,
+            seen_timestamp,
         })
     }
 
     pub fn reconstruct_columns(
         kzg: &Kzg,
         partial_set_of_columns: &[Self],
         spec: &ChainSpec,
+        seen_timestamp: Duration,
     ) -> Result<Vec<KzgVerifiedCustodyDataColumn<E>>, KzgError> {
         let all_data_columns = reconstruct_data_columns(
             kzg,
@@ -329,7 +345,10 @@ impl<E: EthSpec> KzgVerifiedCustodyDataColumn<E> {
         Ok(all_data_columns
             .into_iter()
             .map(|data| {
-                KzgVerifiedCustodyDataColumn::from_asserted_custody(KzgVerifiedDataColumn { data })
+                KzgVerifiedCustodyDataColumn::from_asserted_custody(KzgVerifiedDataColumn {
+                    data,
+                    seen_timestamp,
+                })
             })
             .collect::<Vec<_>>())
     }
@@ -347,6 +366,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`.
@@ -355,10 +378,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>, 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.
@@ -383,6 +410,7 @@ pub fn validate_data_column_sidecar_for_gossip<T: BeaconChainTypes, O: Observati
     subnet: u64,
     chain: &BeaconChain<T>,
 ) -> Result<GossipVerifiedDataColumn<T, O>, GossipDataColumnError> {
+    let seen_timestamp = chain.slot_clock.now_duration().unwrap_or_default();
     let column_slot = data_column.slot();
     verify_data_column_sidecar(&data_column, &chain.spec)?;
     verify_index_matches_subnet(&data_column, subnet, &chain.spec)?;
@@ -394,8 +422,9 @@ pub fn validate_data_column_sidecar_for_gossip<T: BeaconChainTypes, O: Observati
     verify_slot_higher_than_parent(&parent_block, column_slot)?;
     verify_proposer_and_signature(&data_column, &parent_block, chain)?;
     let kzg = &chain.kzg;
-    let kzg_verified_data_column = verify_kzg_for_data_column(data_column.clone(), kzg)
-        .map_err(GossipDataColumnError::InvalidKzgProof)?;
+    let kzg_verified_data_column =
+        verify_kzg_for_data_column(data_column.clone(), kzg, seen_timestamp)
+            .map_err(GossipDataColumnError::InvalidKzgProof)?;
 
     chain
         .observed_slashable