Gloas payload cache

beacon_node/beacon_chain/benches/benches.rs

@@ -53,13 +53,21 @@ fn all_benches(c: &mut Criterion) {
         )
         .unwrap();
 
+        let kzg_commitments = signed_block
+            .message()
+            .body()
+            .blob_kzg_commitments()
+            .unwrap()
+            .clone();
+
         let spec = spec.clone();
 
         c.bench_function(&format!("reconstruct_{}", blob_count), |b| {
             b.iter(|| {
                 black_box(reconstruct_data_columns(
                     &kzg,
                     column_sidecars.iter().as_slice()[0..column_sidecars.len() / 2].to_vec(),
+                    &kzg_commitments,
                     spec.as_ref(),
                 ))
             })

beacon_node/beacon_chain/src/block_verification.rs

@@ -286,6 +286,10 @@ pub enum BlockError {
     /// TODO: We may need to penalize the peer that gave us a potentially invalid rpc blob.
     /// https://github.com/sigp/lighthouse/issues/4546
     AvailabilityCheck(AvailabilityCheckError),
+    /// The payload envelope's block root is unknown.
+    EnvelopeBlockRootUnknown(Hash256),
+    /// Optimistic sync is not supported for Gloas payload envelopes.
+    OptimisticSyncNotSupported { block_root: Hash256 },
     /// A Blob with a slot after PeerDAS is received and is not required to be imported.
     /// This can happen because we stay subscribed to the blob subnet after 2 epochs, as we could
     /// still receive valid blobs from a Deneb epoch after PeerDAS is activated.
@@ -624,7 +628,8 @@ pub fn signature_verify_chain_segment<T: BeaconChainTypes>(
             consensus_context,
         });
     }
-
+    // TODO(gloas) When implementing range and backfill sync for gloas
+    // we need a batch verify kzg function in the new da checker as well.
     chain
         .data_availability_checker
         .batch_verify_kzg_for_available_blocks(&available_blocks)?;

beacon_node/beacon_chain/src/builder.rs

@@ -12,6 +12,7 @@ use crate::kzg_utils::{build_data_column_sidecars_fulu, build_data_column_sideca
 use crate::light_client_server_cache::LightClientServerCache;
 use crate::migrate::{BackgroundMigrator, MigratorConfig};
 use crate::observed_data_sidecars::ObservedDataSidecars;
+use crate::pending_payload_cache::PendingPayloadCache;
 use crate::persisted_beacon_chain::PersistedBeaconChain;
 use crate::persisted_custody::load_custody_context;
 use crate::shuffling_cache::{BlockShufflingIds, ShufflingCache};
@@ -987,6 +988,7 @@ where
             )
         };
         debug!(?custody_context, "Loaded persisted custody context");
+        let custody_context = Arc::new(custody_context);
 
         let beacon_chain = BeaconChain {
             spec: self.spec.clone(),
@@ -1062,14 +1064,22 @@ where
             data_availability_checker: Arc::new(
                 DataAvailabilityChecker::new(
                     complete_blob_backfill,
-                    slot_clock,
+                    slot_clock.clone(),
                     self.kzg.clone(),
-                    Arc::new(custody_context),
-                    self.spec,
+                    custody_context.clone(),
+                    self.spec.clone(),
                     enable_partial_columns,
                 )
                 .map_err(|e| format!("Error initializing DataAvailabilityChecker: {:?}", e))?,
             ),
+            pending_payload_cache: Arc::new(
+                PendingPayloadCache::new(
+                    self.kzg.clone(),
+                    custody_context.clone(),
+                    self.spec.clone(),
+                )
+                .map_err(|e| format!("Error initializing PendingPayloadCache: {:?}", e))?,
+            ),
             kzg: self.kzg.clone(),
             rng: Arc::new(Mutex::new(rng)),
             gossip_verified_payload_bid_cache: <_>::default(),

beacon_node/beacon_chain/src/canonical_head.rs

@@ -988,6 +988,25 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 .start_slot(T::EthSpec::slots_per_epoch()),
         );
 
+        // Prune the Gloas pending-payload cache. Anything older than the data-availability
+        // boundary cannot still be in flight; finalised entries are also safe to drop.
+        if self.spec.gloas_fork_epoch.is_some() {
+            let finalized_epoch = new_view.finalized_checkpoint.epoch;
+            let current_epoch = new_snapshot
+                .beacon_state
+                .slot()
+                .epoch(T::EthSpec::slots_per_epoch());
+            if let Some(min_epochs_for_blobs) = self
+                .spec
+                .min_epoch_data_availability_boundary(current_epoch)
+            {
+                let cutoff_epoch = std::cmp::max(finalized_epoch + 1, min_epochs_for_blobs);
+                if let Err(e) = self.pending_payload_cache.do_maintenance(cutoff_epoch) {
+                    error!(error = ?e, "Failed to prune pending payload cache on finalization");
+                }
+            }
+        }
+
         if let Some(event_handler) = self.event_handler.as_ref()
             && event_handler.has_finalized_subscribers()
         {

beacon_node/beacon_chain/src/data_availability_checker.rs

@@ -33,6 +33,7 @@ use crate::data_column_verification::{
     GossipVerifiedDataColumn, KzgVerifiedCustodyDataColumn, KzgVerifiedDataColumn,
     verify_kzg_for_data_column_list,
 };
+use crate::kzg_utils::validate_data_columns_with_commitments;
 use crate::metrics::{
     KZG_DATA_COLUMN_RECONSTRUCTION_ATTEMPTS, KZG_DATA_COLUMN_RECONSTRUCTION_FAILURES,
 };
@@ -490,8 +491,7 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
             AvailableBlockData::Blobs(blobs) => verify_kzg_for_blob_list(blobs.iter(), &self.kzg)
                 .map_err(AvailabilityCheckError::InvalidBlobs),
             AvailableBlockData::DataColumns(columns) => {
-                verify_kzg_for_data_column_list(columns.iter(), &self.kzg)
-                    .map_err(AvailabilityCheckError::InvalidColumn)
+                verify_columns_against_block(&self.kzg, available_block.block(), columns)
             }
         }
     }
@@ -504,13 +504,17 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
         available_blocks: &[AvailableBlock<T::EthSpec>],
     ) -> Result<(), AvailabilityCheckError> {
         let mut all_blobs = Vec::new();
-        let mut all_data_columns = Vec::new();
 
         for available_block in available_blocks {
-            match available_block.data().to_owned() {
+            match available_block.data() {
                 AvailableBlockData::NoData => {}
-                AvailableBlockData::Blobs(blobs) => all_blobs.extend(blobs),
-                AvailableBlockData::DataColumns(columns) => all_data_columns.extend(columns),
+                AvailableBlockData::Blobs(blobs) => all_blobs.extend(blobs.iter().cloned()),
+                AvailableBlockData::DataColumns(columns) => {
+                    // Each block has its own commitments. For Gloas they live in the bid; for
+                    // Fulu they live inline on the column. Verify per block and let the helper
+                    // pick the right path.
+                    verify_columns_against_block(&self.kzg, available_block.block(), columns)?;
+                }
             }
         }
 
@@ -519,11 +523,6 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
                 .map_err(AvailabilityCheckError::InvalidBlobs)?;
         }
 
-        if !all_data_columns.is_empty() {
-            verify_kzg_for_data_column_list(all_data_columns.iter(), &self.kzg)
-                .map_err(AvailabilityCheckError::InvalidColumn)?;
-        }
-
         Ok(())
     }
 
@@ -605,9 +604,21 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
         metrics::inc_counter(&KZG_DATA_COLUMN_RECONSTRUCTION_ATTEMPTS);
         let timer = metrics::start_timer(&metrics::DATA_AVAILABILITY_RECONSTRUCTION_TIME);
 
+        let columns: Vec<_> = verified_data_columns
+            .into_iter()
+            .map(|c| c.into_inner())
+            .collect();
+        // Fulu columns carry their commitments; reconstruction needs the count to drive the
+        // per-blob recovery loop.
+        let kzg_commitments = columns
+            .first()
+            .and_then(|c| c.kzg_commitments().ok().cloned())
+            .ok_or(AvailabilityCheckError::InvalidVariant)?;
+
         let all_data_columns = KzgVerifiedCustodyDataColumn::reconstruct_columns(
             &self.kzg,
-            &verified_data_columns,
+            columns,
+            &kzg_commitments,
             &self.spec,
         )
         .map_err(|e| {
@@ -676,6 +687,35 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
     }
 }
 
+/// Verify a batch of data columns belonging to a single block, picking the right commitment
+/// source for the block's fork (Fulu: inline on column; Gloas: from the embedded payload bid).
+fn verify_columns_against_block<E: EthSpec>(
+    kzg: &Kzg,
+    block: &SignedBeaconBlock<E>,
+    columns: &[Arc<DataColumnSidecar<E>>],
+) -> Result<(), AvailabilityCheckError> {
+    if columns.is_empty() {
+        return Ok(());
+    }
+    if block.fork_name_unchecked().gloas_enabled() {
+        let commitments = block
+            .message()
+            .body()
+            .signed_execution_payload_bid()
+            .map(|bid| bid.message.blob_kzg_commitments.clone())
+            .map_err(|_| {
+                AvailabilityCheckError::Unexpected(
+                    "Gloas block missing signed_execution_payload_bid".to_string(),
+                )
+            })?;
+        validate_data_columns_with_commitments(kzg, columns.iter(), commitments.as_ref())
+            .map_err(AvailabilityCheckError::InvalidColumn)
+    } else {
+        verify_kzg_for_data_column_list(columns.iter(), kzg)
+            .map_err(AvailabilityCheckError::InvalidColumn)
+    }
+}
+
 /// Helper struct to group data availability checker metrics.
 pub struct DataAvailabilityCheckerMetrics {
     pub block_cache_size: usize,
@@ -874,10 +914,13 @@ impl<E: EthSpec> AvailableBlock<E> {
 
         match &block_data {
             AvailableBlockData::NoData => {
-                if columns_required {
-                    return Err(AvailabilityCheckError::MissingCustodyColumns);
-                } else if blobs_required {
-                    return Err(AvailabilityCheckError::MissingBlobs);
+                // For Gloas, DA is checked for the PayloadEnvelope, not for the block.
+                if !block.fork_name_unchecked().gloas_enabled() {
+                    if columns_required {
+                        return Err(AvailabilityCheckError::MissingCustodyColumns);
+                    } else if blobs_required {
+                        return Err(AvailabilityCheckError::MissingBlobs);
+                    }
                 }
             }
             AvailableBlockData::Blobs(blobs) => {

beacon_node/beacon_chain/src/data_availability_checker/error.rs

@@ -4,6 +4,7 @@ use types::{BeaconStateError, ColumnIndex, Hash256};
 #[derive(Debug)]
 pub enum Error {
     InvalidBlobs(KzgError),
+    MissingBid(Hash256),
     InvalidColumn((Option<ColumnIndex>, KzgError)),
     ReconstructColumnsError(KzgError),
     KzgCommitmentMismatch {
@@ -23,6 +24,7 @@ pub enum Error {
     RebuildingStateCaches(BeaconStateError),
     SlotClockError,
     InvalidAvailableBlockData,
+    InvalidVariant,
 }
 
 #[derive(PartialEq, Eq)]
@@ -38,6 +40,7 @@ impl Error {
         match self {
             Error::SszTypes(_)
             | Error::MissingBlobs
+            | Error::MissingBid(_)
             | Error::MissingCustodyColumns
             | Error::StoreError(_)
             | Error::DecodeError(_)
@@ -46,7 +49,8 @@ impl Error {
             | Error::BlockReplayError(_)
             | Error::RebuildingStateCaches(_)
             | Error::SlotClockError
-            | Error::InvalidAvailableBlockData => ErrorCategory::Internal,
+            | Error::InvalidAvailableBlockData
+            | Error::InvalidVariant => ErrorCategory::Internal,
             Error::InvalidBlobs { .. }
             | Error::InvalidColumn { .. }
             | Error::ReconstructColumnsError { .. }

beacon_node/beacon_chain/src/data_availability_checker/overflow_lru_cache.rs

@@ -109,7 +109,7 @@ impl<E: EthSpec> PendingComponents<E> {
             .unwrap_or(false)
     }
 
-    /// Returns the indices of cached custody columns
+    /// Returns the indices of cached sampling columns
     pub fn get_cached_data_columns_indices(&self) -> Vec<ColumnIndex> {
         self.verified_data_columns
             .iter()