Gloas backfill

beacon_node/beacon_chain/src/data_availability_checker.rs

@@ -35,6 +35,7 @@ use crate::metrics::{
     KZG_DATA_COLUMN_RECONSTRUCTION_ATTEMPTS, KZG_DATA_COLUMN_RECONSTRUCTION_FAILURES,
 };
 use crate::observed_data_sidecars::ObservationStrategy;
+use crate::payload_envelope_verification::AvailableEnvelope;
 pub use error::{Error as AvailabilityCheckError, ErrorCategory as AvailabilityCheckErrorCategory};
 
 /// The LRU Cache stores `PendingComponents`, which store block and its associated blob data:
@@ -495,6 +496,30 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
         Ok(())
     }
 
+    /// Batch verify the KZG proofs for the data columns carried by Gloas payload envelopes.
+    ///
+    /// For Gloas blocks the data columns are coupled to the payload envelope rather than the
+    /// block itself, so they are not covered by `batch_verify_kzg_for_available_blocks` (which
+    /// sees `AvailableBlockData::NoData` for Gloas). Each envelope's columns are verified against
+    /// the `blob_kzg_commitments` in its corresponding block's execution payload bid.
+    ///
+    /// Used during Gloas backfill sync to verify envelope columns before persisting them.
+    pub fn batch_verify_kzg_for_available_envelopes<'a>(
+        &self,
+        blocks_and_envelopes: impl Iterator<
+            Item = (
+                &'a AvailableBlock<T::EthSpec>,
+                &'a AvailableEnvelope<T::EthSpec>,
+            ),
+        >,
+    ) -> Result<(), AvailabilityCheckError> {
+        for (available_block, envelope) in blocks_and_envelopes {
+            verify_columns_against_block(&self.kzg, available_block.block(), &envelope.columns)?;
+        }
+
+        Ok(())
+    }
+
     /// Collects metrics from the data availability checker.
     pub fn metrics(&self) -> DataAvailabilityCheckerMetrics {
         DataAvailabilityCheckerMetrics {

beacon_node/beacon_chain/src/historical_blocks.rs

@@ -1,4 +1,5 @@
 use crate::data_availability_checker::{AvailableBlock, AvailableBlockData};
+use crate::payload_envelope_verification::AvailableEnvelope;
 use crate::{BeaconChain, BeaconChainTypes, WhenSlotSkipped, metrics};
 use fixed_bytes::FixedBytesExtended;
 use itertools::Itertools;
@@ -15,6 +16,12 @@ use strum::IntoStaticStr;
 use tracing::{debug, debug_span, instrument};
 use types::{Hash256, Slot};
 
+/// An available block together with its optional Gloas payload envelope, as produced by
+/// `RangeSyncBlock::into_available_block` and consumed by `import_historical_block_batch`.
+///
+/// The envelope is `None` for pre-Gloas blocks (and for Gloas blocks with no payload data).
+pub type AvailableBlockWithEnvelope<E> = (AvailableBlock<E>, Option<AvailableEnvelope<E>>);
+
 /// Use a longer timeout on the pubkey cache.
 ///
 /// It's ok if historical sync is stalled due to writes from forwards block processing.
@@ -37,6 +44,9 @@ pub enum HistoricalBlockError {
     IndexOutOfBounds,
     /// Logic error: should never occur.
     MissingOldestBlockRoot { slot: Slot },
+    /// A Gloas block that expects blob data was provided without its payload envelope, so the
+    /// required data columns cannot be verified or persisted. Caller should retry/penalize.
+    MissingEnvelope { block_root: Hash256 },
     /// Internal store error
     StoreError(StoreError),
 }
@@ -70,7 +80,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     #[instrument(skip_all)]
     pub fn import_historical_block_batch(
         &self,
-        mut blocks: Vec<AvailableBlock<T::EthSpec>>,
+        mut blocks: Vec<AvailableBlockWithEnvelope<T::EthSpec>>,
     ) -> Result<usize, HistoricalBlockError> {
         let anchor_info = self.store.get_anchor_info();
         let blob_info = self.store.get_blob_info();
@@ -80,7 +90,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         //
         // This allows for reimport of the blobs/columns for the finalized block after checkpoint
         // sync.
-        let num_relevant = blocks.partition_point(|available_block| {
+        let num_relevant = blocks.partition_point(|(available_block, _envelope)| {
             available_block.block().slot() <= anchor_info.oldest_block_slot
         });
 
@@ -107,12 +117,42 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         let mut new_oldest_blob_slot = blob_info.oldest_blob_slot;
         let mut new_oldest_data_column_slot = data_column_info.oldest_data_column_slot;
 
+        // Track the child block's bid `parent_block_hash` while iterating backwards, in order to
+        // determine each Gloas block's payload status. Per `process_parent_execution_payload` in
+        // the spec, a block's payload was revealed on chain (the block is "full") iff its child's
+        // bid `parent_block_hash` equals its own bid `block_hash`. A withheld ("empty") payload
+        // never has an envelope or data columns, so none can be required during backfill.
+        //
+        // Initialize from the current anchor block, which is the child of the newest block in the
+        // batch. If the newest block *is* the anchor block being re-imported, this yields a
+        // self-comparison that is always false, making its envelope optional — its data was
+        // already stored during checkpoint sync.
+        let mut child_bid_parent_hash = blocks_to_import
+            .last()
+            .filter(|(available_block, _envelope)| {
+                available_block.block().fork_name_unchecked().gloas_enabled()
+            })
+            .and_then(|_| {
+                self.block_root_at_slot(anchor_info.oldest_block_slot, WhenSlotSkipped::None)
+                    .ok()
+                    .flatten()
+            })
+            .and_then(|root| self.get_blinded_block(&root).ok().flatten())
+            .and_then(|child_block| {
+                child_block
+                    .message()
+                    .body()
+                    .signed_execution_payload_bid()
+                    .ok()
+                    .map(|bid| bid.message.parent_block_hash)
+            });
+
         let mut blob_batch = Vec::<KeyValueStoreOp>::new();
         let mut cold_batch = Vec::with_capacity(blocks_to_import.len());
         let mut hot_batch = Vec::with_capacity(blocks_to_import.len());
         let mut signed_blocks = Vec::with_capacity(blocks_to_import.len());
 
-        for available_block in blocks_to_import.into_iter().rev() {
+        for (available_block, envelope) in blocks_to_import.into_iter().rev() {
             let (block_root, block, block_data) = available_block.deconstruct();
 
             if block.slot() == anchor_info.oldest_block_slot {
@@ -171,6 +211,54 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 blob_batch.extend(self.store.convert_to_kv_batch(vec![op])?);
             }
 
+            // Whether this block's execution payload was revealed on chain ("full" block). Only
+            // ever true for Gloas blocks, as earlier forks have no bid. See the comment on
+            // `child_bid_parent_hash` above.
+            let payload_revealed = block
+                .message()
+                .body()
+                .signed_execution_payload_bid()
+                .ok()
+                .zip(child_bid_parent_hash)
+                .is_some_and(|(bid, child_parent_hash)| {
+                    bid.message.block_hash == child_parent_hash
+                });
+
+            // Persist the Gloas payload envelope and its data columns. For Gloas blocks the data
+            // columns are carried by the envelope rather than the block's `block_data`, so they
+            // must be stored from here.
+            match envelope {
+                Some(envelope) => {
+                    let (signed_envelope, columns) = envelope.deconstruct();
+                    if !columns.is_empty() {
+                        new_oldest_data_column_slot = Some(block.slot());
+                        if let Some(op) = self.get_blobs_or_columns_store_op(
+                            block_root,
+                            block.slot(),
+                            AvailableBlockData::DataColumns(columns),
+                        ) {
+                            blob_batch.extend(self.store.convert_to_kv_batch(vec![op])?);
+                        }
+                    }
+                    self.store.payload_envelope_as_kv_store_ops(
+                        &block_root,
+                        &signed_envelope,
+                        &mut hot_batch,
+                    );
+                }
+                None => {
+                    // A Gloas block whose payload was revealed must be accompanied by its
+                    // envelope: the spec requires envelopes to be served over the same retention
+                    // window as blocks, and we must persist it to serve it over RPC and for
+                    // state reconstruction — even when it carries no blobs. Blocks with withheld
+                    // ("empty") payloads never have an envelope, and pre-Gloas blocks carry
+                    // their blob data in `block_data`.
+                    if payload_revealed {
+                        return Err(HistoricalBlockError::MissingEnvelope { block_root });
+                    }
+                }
+            }
+
             // Store block roots, including at all skip slots in the freezer DB.
             for slot in (block.slot().as_u64()..prev_block_slot.as_u64()).rev() {
                 debug!(%slot, ?block_root, "Storing frozen block to root mapping");
@@ -183,6 +271,13 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
             prev_block_slot = block.slot();
             expected_block_root = block.message().parent_root();
+            // This block is the child of the next (older) block in the iteration.
+            child_bid_parent_hash = block
+                .message()
+                .body()
+                .signed_execution_payload_bid()
+                .ok()
+                .map(|bid| bid.message.parent_block_hash);
             signed_blocks.push(block);
 
             // If we've reached genesis, add the genesis block root to the batch for all slots

beacon_node/beacon_chain/tests/store_tests.rs

@@ -3304,26 +3304,32 @@ async fn weak_subjectivity_sync_test(
             let range_sync_block = harness
                 .build_range_sync_block_from_store_blobs(Some(block_root), Arc::new(full_block));
 
-            let (fully_available_block, _envelope) =
+            let (fully_available_block, envelope) =
                 range_sync_block.into_available_block().unwrap();
             harness
                 .chain
                 .data_availability_checker
                 .verify_kzg_for_available_block(&fully_available_block)
                 .expect("should verify kzg");
-            available_blocks.push(fully_available_block);
+            available_blocks.push((fully_available_block, envelope));
         }
 
         // Corrupt the signature on the 1st block to ensure that the backfill processor is checking
         // signatures correctly. Regression test for https://github.com/sigp/lighthouse/pull/5120.
-        let mut batch_with_invalid_first_block =
-            available_blocks.iter().map(clone_block).collect::<Vec<_>>();
+        let mut batch_with_invalid_first_block = available_blocks
+            .iter()
+            .map(|(block, envelope)| (clone_block(block), envelope.clone()))
+            .collect::<Vec<_>>();
         batch_with_invalid_first_block[0] = {
-            let (_, block, data) = clone_block(&available_blocks[0]).deconstruct();
+            let (first_block, first_envelope) = &available_blocks[0];
+            let (_, block, data) = clone_block(first_block).deconstruct();
             let mut corrupt_block = (*block).clone();
             *corrupt_block.signature_mut() = Signature::empty();
-            AvailableBlock::new(Arc::new(corrupt_block), data, &beacon_chain.custody_context)
-                .expect("available block")
+            (
+                AvailableBlock::new(Arc::new(corrupt_block), data, &beacon_chain.custody_context)
+                    .expect("available block"),
+                first_envelope.clone(),
+            )
         };
 
         // Importing the invalid batch should error.
@@ -3340,7 +3346,7 @@ async fn weak_subjectivity_sync_test(
         for batch in available_blocks.rchunks(batch_size) {
             let available_blocks_slots = batch
                 .iter()
-                .map(|block| (block.block().slot(), block.block().canonical_root()))
+                .map(|(block, _envelope)| (block.block().slot(), block.block().canonical_root()))
                 .collect::<Vec<_>>();
             info!(
                 ?available_blocks_slots,
@@ -3349,15 +3355,18 @@ async fn weak_subjectivity_sync_test(
             );
 
             // Importing the batch with valid signatures should succeed.
-            let available_blocks_batch1 = batch.iter().map(clone_block).collect::<Vec<_>>();
+            let available_blocks_batch1 = batch
+                .iter()
+                .map(|(block, envelope)| (clone_block(block), envelope.clone()))
+                .collect::<Vec<_>>();
             beacon_chain
                 .import_historical_block_batch(available_blocks_batch1)
                 .unwrap();
 
             // We should be able to load the block root at the `oldest_block_slot`.
             //
             // This is a regression test for: https://github.com/sigp/lighthouse/issues/7690
-            let oldest_block_imported = &batch[0];
+            let (oldest_block_imported, _envelope) = &batch[0];
             let (oldest_block_slot, oldest_block_root) =
                 if oldest_block_imported.block().parent_root() == beacon_chain.genesis_block_root {
                     (Slot::new(0), beacon_chain.genesis_block_root)
@@ -3377,7 +3386,10 @@ async fn weak_subjectivity_sync_test(
             );
 
             // Resupplying the blocks should not fail, they can be safely ignored.
-            let available_blocks_batch2 = batch.iter().map(clone_block).collect::<Vec<_>>();
+            let available_blocks_batch2 = batch
+                .iter()
+                .map(|(block, envelope)| (clone_block(block), envelope.clone()))
+                .collect::<Vec<_>>();
             beacon_chain
                 .import_historical_block_batch(available_blocks_batch2)
                 .unwrap();