Merge branch 'gloas-range-sync-fix' into glamsterdam-devnet-6

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -20,6 +20,7 @@ use crate::custody_context::{CustodyContext, CustodyContextSsz};
 use crate::data_availability_checker::{
     Availability as BlockAvailability, AvailabilityCheckError, AvailableBlock, AvailableBlockData,
     DataColumnReconstructionResult as DataColumnReconstructionResultV1,
+    verify_columns_against_block,
 };
 
 use crate::data_availability_checker::DataAvailabilityChecker;
@@ -2998,9 +2999,25 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 }
             }
 
+            // The envelope needs import only if it's a Gloas block with an envelope and
+            // the envelope isn't already in fork choice.
+            let range_sync_envelope_needs_import = matches!(
+                block,
+                RangeSyncBlock::Gloas {
+                    envelope: Some(_),
+                    ..
+                }
+            ) && !self
+                .canonical_head
+                .fork_choice_read_lock()
+                .is_payload_received(&block_root);
+
             match check_block_relevancy(block.as_block(), block_root, self) {
                 // If the block is relevant, add it to the filtered chain segment.
                 Ok(_) => filtered_chain_segment.push((block_root, block)),
+                Err(BlockError::DuplicateFullyImported(_)) if range_sync_envelope_needs_import => {
+                    filtered_chain_segment.push((block_root, block));
+                }
                 // If the block is already known, simply ignore this block.
                 //
                 // Note that `check_block_relevancy` is incapable of returning
@@ -3016,12 +3033,27 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                 // 2. In some non-canonical chain at a slot that has been finalized already.
                 //
                 // In the case of (1), there's no need to re-import and later blocks in this
-                // segement might be useful.
+                // segment might be useful.
+                // This changes slightly post-gloas because the finalized block can be
+                // imported without its corresponding envelope. If the block we are processing is
+                // the finalized block then we still add it to the filtered chain segment so that
+                // its envelope can be processed.
                 //
                 // In the case of (2), skipping the block is valid since we should never import it.
                 // However, we will potentially get a `ParentUnknown` on a later block. The sync
                 // protocol will need to ensure this is handled gracefully.
-                Err(BlockError::WouldRevertFinalizedSlot { .. }) => continue,
+                Err(BlockError::WouldRevertFinalizedSlot { .. }) => {
+                    if range_sync_envelope_needs_import
+                        && self
+                            .canonical_head
+                            .cached_head()
+                            .finalized_checkpoint()
+                            .root
+                            == block_root
+                    {
+                        filtered_chain_segment.push((block_root, block));
+                    }
+                }
                 // The block has a known parent that does not descend from the finalized block.
                 // There is no need to process this block or any children.
                 Err(BlockError::NotFinalizedDescendant { block_parent_root }) => {
@@ -3106,6 +3138,48 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             let mut blocks = filtered_chain_segment.split_off(last_index);
             std::mem::swap(&mut blocks, &mut filtered_chain_segment);
 
+            // Here, we are special casing the checkpoint sync block's envelope processing.
+            // Post-gloas, if the first filtered block is the checkpoint block, range
+            // sync may still need to process its envelope so that the first post-checkpoint
+            // child can resolve its parent payload status.
+            // The block is an anchor, so there won't be a parent present in fork choice,
+            // so we need to avoid processing it.
+            if matches!(blocks.first(), Some((root, _)) if *root == self
+                .canonical_head
+                .cached_head()
+                .finalized_checkpoint()
+                .root)
+            {
+                let (block_root, block) = blocks.remove(0);
+                let block_slot = block.slot();
+
+                if let RangeSyncBlock::Gloas {
+                    block,
+                    envelope: Some(envelope),
+                } = block
+                {
+                    let chain = self.clone();
+                    if let Err(error) = async move {
+                        verify_columns_against_block(&chain.kzg, &block, &envelope.columns)
+                            .map_err(BlockError::AvailabilityCheck)?;
+
+                        self.process_range_sync_envelope(envelope, block_root, block)
+                            .await
+                            .map_err(BlockError::from)?;
+
+                        Ok::<(), BlockError>(())
+                    }
+                    .await
+                    {
+                        return ChainSegmentResult::Failed {
+                            imported_blocks,
+                            error,
+                        };
+                    }
+                }
+                imported_blocks.push((block_root, block_slot));
+            }
+
             // Extract envelopes before passing blocks to signature verification.
             let envelopes: Vec<_> = blocks
                 .iter()

beacon_node/beacon_chain/src/block_verification.rs

@@ -97,11 +97,10 @@ use store::{Error as DBError, KeyValueStore};
 use strum::{AsRefStr, IntoStaticStr};
 use task_executor::JoinHandle;
 use tracing::{Instrument, Span, debug, debug_span, error, info_span, instrument};
-use types::ExecutionBlockHash;
 use types::{
     BeaconBlockRef, BeaconState, BeaconStateError, BlobsList, ChainSpec, DataColumnSidecarList,
-    Epoch, EthSpec, FullPayload, Hash256, InconsistentFork, KzgProofs, RelativeEpoch,
-    SignedBeaconBlock, SignedBeaconBlockHeader, Slot, data::DataColumnSidecarError,
+    Epoch, EthSpec, ExecutionBlockHash, FullPayload, Hash256, InconsistentFork, KzgProofs,
+    RelativeEpoch, SignedBeaconBlock, SignedBeaconBlockHeader, Slot, data::DataColumnSidecarError,
 };
 
 /// Maximum block slot number. Block with slots bigger than this constant will NOT be processed.
@@ -622,6 +621,7 @@ pub(crate) fn process_block_slash_info<T: BeaconChainTypes, TErr: BlockBlobError
 /// signature in the block is invalid, an `Err` is returned (it is not possible to known _which_
 /// signature was invalid).
 ///
+/// Also performs kzg verification on columns if they exist.
 /// ## Errors
 ///
 /// The given `chain_segment` must contain only blocks from the same epoch, otherwise an error
@@ -652,17 +652,17 @@ pub fn signature_verify_chain_segment<T: BeaconChainTypes>(
         &chain.spec,
     )?;
 
-    let mut available_blocks = Vec::with_capacity(chain_segment.len());
-    let mut envelopes = Vec::with_capacity(chain_segment.len());
     let mut signature_verified_blocks = Vec::with_capacity(chain_segment.len());
 
     for (block_root, block) in chain_segment {
         let consensus_context =
             ConsensusContext::new(block.slot()).set_current_block_root(block_root);
-
-        let (available_block, envelope) = block.into_available_block()?;
-        available_blocks.push(available_block.clone());
-        envelopes.push(envelope);
+        // This gets columns from the block for pre-gloas and from the envelope for
+        // post gloas.
+        if let Some(columns) = block.data_columns() {
+            verify_columns_against_block(&chain.kzg, block.as_block(), &columns)?;
+        }
+        let (available_block, _envelope) = block.into_available_block()?;
         signature_verified_blocks.push(SignatureVerifiedBlock {
             block: MaybeAvailableBlock::Available(available_block),
             block_root,
@@ -671,16 +671,6 @@ pub fn signature_verify_chain_segment<T: BeaconChainTypes>(
         });
     }
 
-    chain
-        .data_availability_checker
-        .batch_verify_kzg_for_available_blocks(&available_blocks)?;
-
-    for (available_block, maybe_envelope) in available_blocks.iter().zip(envelopes.iter()) {
-        if let Some(envelope) = maybe_envelope {
-            verify_columns_against_block(&chain.kzg, available_block.block(), &envelope.columns)?;
-        }
-    }
-
     // verify signatures
     let pubkey_cache = get_validator_pubkey_cache(chain)?;
     let mut signature_verifier = get_signature_verifier(&state, &pubkey_cache, &chain.spec);

beacon_node/beacon_chain/tests/block_verification.rs

@@ -2419,6 +2419,149 @@ async fn range_sync_block_new_gloas_rejects_block_hash_mismatch() {
     );
 }
 
+/// Produces a Gloas block + envelope on top of the current head and imports the block (but not its
+/// envelope), so the block is known to fork choice with its payload not yet received.
+async fn import_gloas_block_without_envelope(
+    harness: &BeaconChainHarness<EphemeralHarnessType<E>>,
+) -> (
+    Arc<SignedBeaconBlock<E>>,
+    SignedExecutionPayloadEnvelope<E>,
+    Hash256,
+) {
+    harness.advance_slot();
+
+    let state = harness.get_current_state();
+    let slot = harness.get_current_slot();
+    let (block_contents, envelope, _) = harness.make_block_with_envelope(state, slot).await;
+    let block = block_contents.0.clone();
+    let block_root = block.canonical_root();
+    let envelope = envelope.expect("gloas block should have envelope");
+
+    harness
+        .process_block(slot, block_root, block_contents)
+        .await
+        .expect("block should import");
+
+    (block, envelope, block_root)
+}
+
+/// Retrying a range-sync batch can provide the envelope for a block that was previously imported
+/// without one. The duplicate block should be allowed through far enough to import the envelope.
+#[tokio::test]
+async fn process_chain_segment_imports_missing_envelope_for_duplicate_gloas_block() {
+    let spec = test_spec::<E>();
+    if !spec.fork_name_at_slot::<E>(Slot::new(1)).gloas_enabled() {
+        return;
+    }
+
+    let harness = BeaconChainHarness::builder(MainnetEthSpec)
+        .spec(spec.into())
+        .keypairs(KEYPAIRS[0..VALIDATOR_COUNT].to_vec())
+        .node_custody_type(NodeCustodyType::Supernode)
+        .fresh_ephemeral_store()
+        .mock_execution_layer()
+        .build();
+
+    let (block, envelope, block_root) = import_gloas_block_without_envelope(&harness).await;
+    let block_slot = block.slot();
+
+    assert!(
+        !harness
+            .chain
+            .canonical_head
+            .fork_choice_read_lock()
+            .is_payload_received(&block_root),
+        "payload should start missing"
+    );
+    assert!(
+        harness
+            .chain
+            .store
+            .get_payload_envelope(&block_root)
+            .expect("should read envelope from store")
+            .is_none(),
+        "envelope should start missing from the store"
+    );
+
+    let available_envelope = AvailableEnvelope::new(Arc::new(envelope), vec![]);
+    let segment = vec![RangeSyncBlock::new_gloas(block, Some(available_envelope)).unwrap()];
+
+    let result = harness
+        .chain
+        .process_chain_segment(segment, NotifyExecutionLayer::Yes)
+        .await;
+
+    let ChainSegmentResult::Successful { imported_blocks } = result else {
+        panic!("range sync should succeed");
+    };
+
+    assert_eq!(
+        imported_blocks,
+        vec![(block_root, block_slot)],
+        "the duplicate block should be reported as processed"
+    );
+    assert!(
+        harness
+            .chain
+            .canonical_head
+            .fork_choice_read_lock()
+            .is_payload_received(&block_root),
+        "range sync should mark the payload as received"
+    );
+    assert!(
+        harness
+            .chain
+            .store
+            .get_payload_envelope(&block_root)
+            .expect("should read envelope from store")
+            .is_some(),
+        "range sync should persist the envelope"
+    );
+}
+
+/// Once the payload has been received, retrying the same block and envelope is a no-op.
+#[tokio::test]
+async fn process_chain_segment_ignores_duplicate_gloas_block_when_payload_received() {
+    let spec = test_spec::<E>();
+    if !spec.fork_name_at_slot::<E>(Slot::new(1)).gloas_enabled() {
+        return;
+    }
+
+    let harness = BeaconChainHarness::builder(MainnetEthSpec)
+        .spec(spec.into())
+        .keypairs(KEYPAIRS[0..VALIDATOR_COUNT].to_vec())
+        .node_custody_type(NodeCustodyType::Supernode)
+        .fresh_ephemeral_store()
+        .mock_execution_layer()
+        .build();
+
+    let (block, envelope, block_root) = import_gloas_block_without_envelope(&harness).await;
+
+    harness
+        .chain
+        .canonical_head
+        .fork_choice_write_lock()
+        .on_valid_payload_envelope_received(block_root)
+        .expect("payload should be marked received");
+
+    let available_envelope = AvailableEnvelope::new(Arc::new(envelope), vec![]);
+    let segment = vec![RangeSyncBlock::new_gloas(block, Some(available_envelope)).unwrap()];
+
+    let result = harness
+        .chain
+        .process_chain_segment(segment, NotifyExecutionLayer::Yes)
+        .await;
+
+    let ChainSegmentResult::Successful { imported_blocks } = result else {
+        panic!("range sync should succeed");
+    };
+
+    assert!(
+        imported_blocks.is_empty(),
+        "block whose payload was already received should be ignored as a duplicate"
+    );
+}
+
 // Test that RpcBlock::new() rejects blocks when blob count doesn't match expected.
 #[tokio::test]
 async fn range_sync_block_construction_fails_with_wrong_blob_count() {