Try pushing blob timing down into verification

Author: michaelsproul

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -2891,15 +2891,6 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             return Err(BlockError::BlockIsAlreadyKnown(blob.block_root()));
         }
 
-        // Record the delay for receiving this blob
-        if let Some(seen_timestamp) = self.slot_clock.now_duration() {
-            self.block_times_cache.write().set_time_blob_observed(
-                block_root,
-                blob.slot(),
-                seen_timestamp,
-            );
-        }
-
         if let Some(event_handler) = self.event_handler.as_ref() {
             if event_handler.has_blob_sidecar_subscribers() {
                 event_handler.register(EventKind::BlobSidecar(SseBlobSidecar::from_blob_sidecar(
@@ -3268,17 +3259,14 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             consensus_context,
         } = import_data;
 
-        // This is the time since start of the slot where all the components of the block have become available
-        metrics::set_gauge(
-            &metrics::BLOCK_AVAILABILITY_DELAY,
-            block.available_timestamp().as_millis() as i64,
-        );
-        // Record the time at which this block became available.
-        self.block_times_cache.write().set_time_available(
-            block_root,
-            block.slot(),
-            block.available_timestamp(),
-        );
+        // Record the time at which this block's blobs became available.
+        if let Some(blobs_available) = block.blobs_available_timestamp() {
+            self.block_times_cache.write().set_time_blob_observed(
+                block_root,
+                block.slot(),
+                blobs_available,
+            );
+        }
 
         // import
         let chain = self.clone();

beacon_node/beacon_chain/src/blob_verification.rs

@@ -14,6 +14,7 @@ use merkle_proof::MerkleTreeError;
 use slog::{debug, warn};
 use ssz_derive::{Decode, Encode};
 use ssz_types::VariableList;
+use std::time::Duration;
 use tree_hash::TreeHash;
 use types::blob_sidecar::BlobIdentifier;
 use types::{
@@ -214,7 +215,10 @@ impl<T: BeaconChainTypes> GossipVerifiedBlob<T> {
     pub fn __assumed_valid(blob: Arc<BlobSidecar<T::EthSpec>>) -> Self {
         Self {
             block_root: blob.block_root(),
-            blob: KzgVerifiedBlob { blob },
+            blob: KzgVerifiedBlob {
+                blob,
+                seen_timestamp: Duration::from_secs(0),
+            },
         }
     }
     pub fn id(&self) -> BlobIdentifier {
@@ -260,6 +264,8 @@ impl<T: BeaconChainTypes> GossipVerifiedBlob<T> {
 #[ssz(struct_behaviour = "transparent")]
 pub struct KzgVerifiedBlob<E: EthSpec> {
     blob: Arc<BlobSidecar<E>>,
+    #[ssz(skip_serializing, skip_deserializing)]
+    seen_timestamp: Duration,
 }
 
 impl<E: EthSpec> PartialOrd for KzgVerifiedBlob<E> {
@@ -275,8 +281,12 @@ impl<E: EthSpec> Ord for KzgVerifiedBlob<E> {
 }
 
 impl<E: EthSpec> KzgVerifiedBlob<E> {
-    pub fn new(blob: Arc<BlobSidecar<E>>, kzg: &Kzg) -> Result<Self, KzgError> {
-        verify_kzg_for_blob(blob, kzg)
+    pub fn new(
+        blob: Arc<BlobSidecar<E>>,
+        kzg: &Kzg,
+        seen_timestamp: Duration,
+    ) -> Result<Self, KzgError> {
+        verify_kzg_for_blob(blob, kzg, seen_timestamp)
     }
     pub fn to_blob(self) -> Arc<BlobSidecar<E>> {
         self.blob
@@ -294,12 +304,18 @@ impl<E: EthSpec> KzgVerifiedBlob<E> {
     pub fn blob_index(&self) -> u64 {
         self.blob.index
     }
+    pub fn seen_timestamp(&self) -> Duration {
+        self.seen_timestamp
+    }
     /// Construct a `KzgVerifiedBlob` that is assumed to be valid.
     ///
     /// This should ONLY be used for testing.
     #[cfg(test)]
     pub fn __assumed_valid(blob: Arc<BlobSidecar<E>>) -> Self {
-        Self { blob }
+        Self {
+            blob,
+            seen_timestamp: Duration::from_secs(0),
+        }
     }
 }
 
@@ -309,9 +325,13 @@ impl<E: EthSpec> KzgVerifiedBlob<E> {
 pub fn verify_kzg_for_blob<E: EthSpec>(
     blob: Arc<BlobSidecar<E>>,
     kzg: &Kzg,
+    seen_timestamp: Duration,
 ) -> Result<KzgVerifiedBlob<E>, KzgError> {
     validate_blob::<E>(kzg, &blob.blob, blob.kzg_commitment, blob.kzg_proof)?;
-    Ok(KzgVerifiedBlob { blob })
+    Ok(KzgVerifiedBlob {
+        blob,
+        seen_timestamp,
+    })
 }
 
 pub struct KzgVerifiedBlobList<E: EthSpec> {
@@ -322,13 +342,17 @@ impl<E: EthSpec> KzgVerifiedBlobList<E> {
     pub fn new<I: IntoIterator<Item = Arc<BlobSidecar<E>>>>(
         blob_list: I,
         kzg: &Kzg,
+        seen_timestamp: Duration,
     ) -> Result<Self, KzgError> {
         let blobs = blob_list.into_iter().collect::<Vec<_>>();
         verify_kzg_for_blob_list(blobs.iter(), kzg)?;
         Ok(Self {
             verified_blobs: blobs
                 .into_iter()
-                .map(|blob| KzgVerifiedBlob { blob })
+                .map(|blob| KzgVerifiedBlob {
+                    blob,
+                    seen_timestamp,
+                })
                 .collect(),
         })
     }
@@ -374,6 +398,8 @@ pub fn validate_blob_sidecar_for_gossip<T: BeaconChainTypes>(
     let blob_epoch = blob_slot.epoch(T::EthSpec::slots_per_epoch());
     let signed_block_header = &blob_sidecar.signed_block_header;
 
+    let seen_timestamp = chain.slot_clock.now_duration().unwrap_or_default();
+
     // This condition is not possible if we have received the blob from the network
     // since we only subscribe to `MaxBlobsPerBlock` subnets over gossip network.
     // We include this check only for completeness.
@@ -641,8 +667,8 @@ pub fn validate_blob_sidecar_for_gossip<T: BeaconChainTypes>(
         .kzg
         .as_ref()
         .ok_or(GossipBlobError::KzgNotInitialized)?;
-    let kzg_verified_blob =
-        KzgVerifiedBlob::new(blob_sidecar, kzg).map_err(GossipBlobError::KzgError)?;
+    let kzg_verified_blob = KzgVerifiedBlob::new(blob_sidecar, kzg, seen_timestamp)
+        .map_err(GossipBlobError::KzgError)?;
 
     Ok(GossipVerifiedBlob {
         block_root,

beacon_node/beacon_chain/src/block_times_cache.rs

@@ -20,7 +20,6 @@ pub struct Timestamps {
     pub observed: Option<Duration>,
     pub all_blobs_observed: Option<Duration>,
     pub execution_time: Option<Duration>,
-    pub available: Option<Duration>,
     pub attestable: Option<Duration>,
     pub imported: Option<Duration>,
     pub set_as_head: Option<Duration>,
@@ -35,7 +34,9 @@ pub struct BlockDelays {
     pub all_blobs_observed: Option<Duration>,
     /// The time it took to get verification from the EL for the block.
     pub execution_time: Option<Duration>,
-    /// Time after max(`observed + execution_time`, `all_blobs_observed`) the block became available.
+    /// The delay from the start of the slot before the block became available
+    ///
+    /// Equal to max(`observed + execution_time`, `all_blobs_observed`).
     pub available: Option<Duration>,
     /// Time after `available`.
     pub attestable: Option<Duration>,
@@ -60,23 +61,31 @@ impl BlockDelays {
         let execution_time = times
             .execution_time
             .and_then(|execution_time| execution_time.checked_sub(times.observed?));
-        let available = times
-            .available
-            .and_then(|available_time| available_time.checked_sub(slot_start_time));
+        // Duration since UNIX epoch at which block became available.
+        let available_time = times.execution_time.and_then(|execution_time| {
+            if let Some(all_blobs_observed) = times.all_blobs_observed {
+                Some(std::cmp::max(execution_time, all_blobs_observed))
+            } else {
+                Some(execution_time)
+            }
+        });
+        // Duration from the start of the slot until the block became available.
+        let available_delay =
+            available_time.and_then(|available_time| available_time.checked_sub(slot_start_time));
         let attestable = times
             .attestable
             .and_then(|attestable_time| attestable_time.checked_sub(slot_start_time));
         let imported = times
             .imported
-            .and_then(|imported_time| imported_time.checked_sub(times.available?));
+            .and_then(|imported_time| imported_time.checked_sub(available_time?));
         let set_as_head = times
             .set_as_head
             .and_then(|set_as_head_time| set_as_head_time.checked_sub(times.imported?));
         BlockDelays {
             observed,
             all_blobs_observed,
             execution_time,
-            available,
+            available: available_delay,
             attestable,
             imported,
             set_as_head,
@@ -177,20 +186,6 @@ impl BlockTimesCache {
         }
     }
 
-    pub fn set_time_available(&mut self, block_root: BlockRoot, slot: Slot, timestamp: Duration) {
-        let block_times = self
-            .cache
-            .entry(block_root)
-            .or_insert_with(|| BlockTimesCacheValue::new(slot));
-        if block_times
-            .timestamps
-            .available
-            .map_or(true, |prev| timestamp < prev)
-        {
-            block_times.timestamps.available = Some(timestamp);
-        }
-    }
-
     pub fn set_time_attestable(&mut self, block_root: BlockRoot, slot: Slot, timestamp: Duration) {
         let block_times = self
             .cache

beacon_node/beacon_chain/src/data_availability_checker.rs

@@ -177,11 +177,17 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
             return Err(AvailabilityCheckError::KzgNotInitialized);
         };
 
-        let verified_blobs = KzgVerifiedBlobList::new(Vec::from(blobs).into_iter().flatten(), kzg)
-            .map_err(AvailabilityCheckError::Kzg)?;
+        let seen_timestamp = self
+            .slot_clock
+            .now_duration()
+            .ok_or(AvailabilityCheckError::SlotClockError)?;
+
+        let verified_blobs =
+            KzgVerifiedBlobList::new(Vec::from(blobs).into_iter().flatten(), kzg, seen_timestamp)
+                .map_err(AvailabilityCheckError::Kzg)?;
 
         self.availability_cache
-            .put_kzg_verified_blobs(block_root, verified_blobs, &self.slot_clock)
+            .put_kzg_verified_blobs(block_root, verified_blobs)
     }
 
     /// Check if we've cached other blobs for this block. If it completes a set and we also
@@ -193,11 +199,8 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
         &self,
         gossip_blob: GossipVerifiedBlob<T>,
     ) -> Result<Availability<T::EthSpec>, AvailabilityCheckError> {
-        self.availability_cache.put_kzg_verified_blobs(
-            gossip_blob.block_root(),
-            vec![gossip_blob.into_inner()],
-            &self.slot_clock,
-        )
+        self.availability_cache
+            .put_kzg_verified_blobs(gossip_blob.block_root(), vec![gossip_blob.into_inner()])
     }
 
     /// Check if we have all the blobs for a block. Returns `Availability` which has information
@@ -207,7 +210,7 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
         executed_block: AvailabilityPendingExecutedBlock<T::EthSpec>,
     ) -> Result<Availability<T::EthSpec>, AvailabilityCheckError> {
         self.availability_cache
-            .put_pending_executed_block(executed_block, &self.slot_clock)
+            .put_pending_executed_block(executed_block)
     }
 
     /// Verifies kzg commitments for an RpcBlock, returns a `MaybeAvailableBlock` that may
@@ -225,15 +228,11 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
                 if self.blobs_required_for_block(&block) {
                     Ok(MaybeAvailableBlock::AvailabilityPending { block_root, block })
                 } else {
-                    let available_timestamp = self
-                        .slot_clock
-                        .now_duration()
-                        .ok_or(AvailabilityCheckError::SlotClockError)?;
                     Ok(MaybeAvailableBlock::Available(AvailableBlock {
                         block_root,
                         block,
                         blobs: None,
-                        available_timestamp,
+                        blobs_available_timestamp: None,
                     }))
                 }
             }
@@ -249,15 +248,11 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
                 } else {
                     None
                 };
-                let available_timestamp = self
-                    .slot_clock
-                    .now_duration()
-                    .ok_or(AvailabilityCheckError::SlotClockError)?;
                 Ok(MaybeAvailableBlock::Available(AvailableBlock {
                     block_root,
                     block,
                     blobs: verified_blobs,
-                    available_timestamp,
+                    blobs_available_timestamp: None,
                 }))
             }
         }
@@ -299,15 +294,11 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
                     if self.blobs_required_for_block(&block) {
                         results.push(MaybeAvailableBlock::AvailabilityPending { block_root, block })
                     } else {
-                        let available_timestamp = self
-                            .slot_clock
-                            .now_duration()
-                            .ok_or(AvailabilityCheckError::SlotClockError)?;
                         results.push(MaybeAvailableBlock::Available(AvailableBlock {
                             block_root,
                             block,
                             blobs: None,
-                            available_timestamp,
+                            blobs_available_timestamp: None,
                         }))
                     }
                 }
@@ -318,15 +309,11 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
                         None
                     };
                     // already verified kzg for all blobs
-                    let available_timestamp = self
-                        .slot_clock
-                        .now_duration()
-                        .ok_or(AvailabilityCheckError::SlotClockError)?;
                     results.push(MaybeAvailableBlock::Available(AvailableBlock {
                         block_root,
                         block,
                         blobs: verified_blobs,
-                        available_timestamp,
+                        blobs_available_timestamp: None,
                     }))
                 }
             }
@@ -487,7 +474,7 @@ pub struct AvailableBlock<E: EthSpec> {
     block: Arc<SignedBeaconBlock<E>>,
     blobs: Option<BlobSidecarList<E>>,
     /// Timestamp at which this block first became available (UNIX timestamp, time since 1970).
-    available_timestamp: Duration,
+    blobs_available_timestamp: Option<Duration>,
 }
 
 impl<E: EthSpec> AvailableBlock<E> {
@@ -500,7 +487,7 @@ impl<E: EthSpec> AvailableBlock<E> {
             block_root,
             block,
             blobs,
-            available_timestamp: Duration::from_millis(0),
+            blobs_available_timestamp: None,
         }
     }
 
@@ -515,8 +502,8 @@ impl<E: EthSpec> AvailableBlock<E> {
         self.blobs.as_ref()
     }
 
-    pub fn available_timestamp(&self) -> Duration {
-        self.available_timestamp
+    pub fn blobs_available_timestamp(&self) -> Option<Duration> {
+        self.blobs_available_timestamp
     }
 
     pub fn deconstruct(
@@ -530,7 +517,7 @@ impl<E: EthSpec> AvailableBlock<E> {
             block_root,
             block,
             blobs,
-            available_timestamp: _,
+            blobs_available_timestamp: _,
         } = self;
         (block_root, block, blobs)
     }