FIX CHECKPOINT SYNC

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -3254,9 +3254,8 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                             AvailabilityProcessingStatus::Imported(block_root) => {
                                 // Import the envelope if one was provided (Gloas+).
                                 if let Some(envelope) = envelope
-                                    && let Err(e) = self.import_envelope_from_range_sync(
-                                        *envelope, block_root,
-                                    )
+                                    && let Err(e) =
+                                        self.import_envelope_from_range_sync(*envelope, block_root)
                                 {
                                     return ChainSegmentResult::Failed {
                                         imported_blocks,

beacon_node/beacon_chain/src/block_verification.rs

@@ -325,7 +325,7 @@ pub enum BlockError {
         bid_parent_root: Hash256,
         block_parent_root: Hash256,
     },
-    /// The child block is known but its parent execution payload envelope has not been received yet.
+    /// The block is known but its parent execution payload envelope has not been received yet.
     ///
     /// ## Peer scoring
     ///
@@ -521,9 +521,13 @@ impl From<EnvelopeError> for BlockError {
             // Internal errors: not the peer's fault
             EnvelopeError::BeaconChainError(_)
             | EnvelopeError::BeaconStateError(_)
+            | EnvelopeError::BlockProcessingError(_)
             | EnvelopeError::EnvelopeProcessingError(_)
             | EnvelopeError::ExecutionPayloadError(_)
-            | EnvelopeError::ImportError(_) => false,
+            | EnvelopeError::ImportError(_)
+            | EnvelopeError::BlockError(_)
+            | EnvelopeError::InternalError(_)
+            | EnvelopeError::OptimisticSyncNotSupported { .. } => false,
         };
         BlockError::PayloadEnvelopeError {
             e: Box::new(e),
@@ -966,12 +970,26 @@ impl<T: BeaconChainTypes> GossipVerifiedBlock<T> {
             });
         }
 
-        // TODO(gloas) The following validation can only be completed once fork choice has been implemented:
-        // The block's parent execution payload (defined by bid.parent_block_hash) has been seen
-        // (via gossip or non-gossip sources) (a client MAY queue blocks for processing
-        // once the parent payload is retrieved). If execution_payload verification of block's execution
-        // payload parent by an execution node is complete, verify the block's execution payload
-        // parent (defined by bid.parent_block_hash) passes all validation.
+        // Check that we've received the parent envelope. If not, issue a single envelope
+        // lookup for the parent and queue this block in the reprocess queue.
+        //
+        // The anchor block (proto-array root) is implicitly considered to have its payload
+        // received: there is no envelope to fetch for the anchor (per spec, the anchor is
+        // never added to `store.payloads`), and the anchor is trusted by definition.
+        let parent_is_gloas = chain
+            .spec
+            .fork_name_at_slot::<T::EthSpec>(parent_block.slot)
+            .gloas_enabled();
+        let parent_is_anchor = parent_block.parent_root.is_none();
+
+        if parent_is_gloas
+            && !parent_is_anchor
+            && !fork_choice_read_lock.is_payload_received(&block.message().parent_root())
+        {
+            return Err(BlockError::ParentEnvelopeUnknown {
+                parent_root: block.message().parent_root(),
+            });
+        }
 
         drop(fork_choice_read_lock);
 
@@ -2045,7 +2063,6 @@ fn load_parent<T: BeaconChainTypes, B: AsBlock<T::EthSpec>>(
         // Retrieve any state that is advanced through to at most `block.slot()`: this is
         // particularly important if `block` descends from the finalized/split block, but at a slot
         // prior to the finalized slot (which is invalid and inaccessible in our DB schema).
-        //
         let (parent_state_root, state) = chain
             .store
             .get_advanced_hot_state(root, block.slot(), parent_block.state_root())?

beacon_node/beacon_chain/src/payload_envelope_verification/import.rs

@@ -9,13 +9,14 @@ use tracing::{debug, error, info, info_span, instrument, warn};
 use types::{BlockImportSource, Hash256, SignedExecutionPayloadEnvelope};
 
 use super::{
-    AvailableEnvelope, AvailableExecutedEnvelope, EnvelopeError,
+    AvailableEnvelope, AvailableExecutedEnvelope, EnvelopeError, EnvelopeImportData,
     gossip_verified_envelope::GossipVerifiedEnvelope,
 };
 use crate::data_column_verification::load_gloas_payload_bid;
 use crate::{
     AvailabilityProcessingStatus, BeaconChain, BeaconChainError, BeaconChainTypes, BlockError,
     NotifyExecutionLayer,
+    block_verification::PayloadVerificationOutcome,
     block_verification_types::AvailableBlockData,
     metrics,
     payload_envelope_verification::{
@@ -206,6 +207,39 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         ))
     }
 
+    /// Import an envelope whose data column availability has not yet been satisfied.
+    ///
+    /// Marks the block's payload as received in fork choice and persists the envelope to the
+    /// store, but does not write data column ops. Columns are expected to arrive separately
+    /// (gossip, engineGetBlobs, or reconstruction).
+    #[instrument(skip_all)]
+    pub async fn import_pending_execution_payload_envelope(
+        self: &Arc<Self>,
+        signed_envelope: Arc<SignedExecutionPayloadEnvelope<T::EthSpec>>,
+        import_data: EnvelopeImportData<T::EthSpec>,
+        payload_verification_outcome: PayloadVerificationOutcome,
+    ) -> Result<AvailabilityProcessingStatus, EnvelopeError> {
+        let EnvelopeImportData {
+            block_root,
+            _phantom,
+        } = import_data;
+        let block_root = {
+            let chain = self.clone();
+            self.spawn_blocking_handle(
+                move || {
+                    chain.import_execution_payload_envelope_pending_columns(
+                        signed_envelope,
+                        block_root,
+                        payload_verification_outcome.payload_verification_status,
+                    )
+                },
+                "payload_verification_handle",
+            )
+            .await??
+        };
+        Ok(AvailabilityProcessingStatus::Imported(block_root))
+    }
+
     #[instrument(skip_all)]
     pub async fn import_available_execution_payload_envelope(
         self: &Arc<Self>,
@@ -235,6 +269,50 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         Ok(AvailabilityProcessingStatus::Imported(block_root))
     }
 
+    /// Same as `import_execution_payload_envelope` but for envelopes whose data columns
+    /// have not yet been received. Marks the payload as received in fork choice and
+    /// persists the envelope; columns are persisted separately as they arrive.
+    #[instrument(skip_all)]
+    fn import_execution_payload_envelope_pending_columns(
+        &self,
+        signed_envelope: Arc<SignedExecutionPayloadEnvelope<T::EthSpec>>,
+        block_root: Hash256,
+        payload_verification_status: PayloadVerificationStatus,
+    ) -> Result<Hash256, EnvelopeError> {
+        let fork_choice_reader = self.canonical_head.fork_choice_upgradable_read_lock();
+        if !fork_choice_reader.contains_block(&block_root) {
+            return Err(EnvelopeError::BlockRootUnknown { block_root });
+        }
+
+        let mut fork_choice = parking_lot::RwLockUpgradableReadGuard::upgrade(fork_choice_reader);
+        fork_choice
+            .on_valid_payload_envelope_received(block_root)
+            .map_err(|e| EnvelopeError::InternalError(format!("{e:?}")))?;
+
+        let db_write_timer = metrics::start_timer(&metrics::ENVELOPE_PROCESSING_DB_WRITE);
+        let ops = vec![StoreOp::PutPayloadEnvelope(
+            block_root,
+            signed_envelope.clone(),
+        )];
+        let db_span = info_span!("persist_envelope_pending_columns").entered();
+        if let Err(e) = self.store.do_atomically_with_block_and_blobs_cache(ops) {
+            error!(error = ?e, "Database write failed for pending-columns envelope");
+            return Err(e.into());
+        }
+        drop(db_span);
+        drop(fork_choice);
+
+        let envelope_time_imported = self.slot_clock.now_duration().unwrap_or(Duration::MAX);
+        metrics::stop_timer(db_write_timer);
+        self.import_envelope_update_metrics_and_events(
+            signed_envelope,
+            block_root,
+            payload_verification_status,
+            envelope_time_imported,
+        );
+        Ok(block_root)
+    }
+
     /// Accepts a fully-verified and available envelope and imports it into the chain without performing any
     /// additional verification.
     ///
@@ -426,10 +504,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             ops.push(blobs_or_columns_store_op);
         }
 
-        ops.push(StoreOp::PutPayloadEnvelope(
-            block_root,
-            signed_envelope,
-        ));
+        ops.push(StoreOp::PutPayloadEnvelope(block_root, signed_envelope));
 
         drop(fork_choice);
 

beacon_node/beacon_chain/src/payload_envelope_verification/mod.rs

@@ -18,6 +18,7 @@
 //!
 //! ```
 
+use state_processing::BlockProcessingError;
 use state_processing::envelope_processing::EnvelopeProcessingError;
 use std::marker::PhantomData;
 use std::sync::Arc;
@@ -107,7 +108,7 @@ pub struct EnvelopeProcessingSnapshot<E: EthSpec> {
     pub beacon_block_root: Hash256,
 }
 
-/// A payload ernvelope that has completed all envelope procesing checks, verification
+/// A payload envelope that has completed all envelope processing checks, verification
 /// by an EL client but does not have all requisite columns to get imported into
 /// fork choice.
 pub struct AvailabilityPendingExecutedEnvelope<E: EthSpec> {
@@ -116,6 +117,26 @@ pub struct AvailabilityPendingExecutedEnvelope<E: EthSpec> {
     pub payload_verification_outcome: PayloadVerificationOutcome,
 }
 
+/// A payload envelope that has gone through processing checks and execution by an EL client.
+/// This envelope hasn't necessarily completed data availability checks.
+///
+///
+/// It contains 2 variants:
+/// 1. `Available`: This envelope has been executed and also contains all data to consider it
+///    fully available.
+/// 2. `AvailabilityPending`: This envelope hasn't received all required blobs to consider it
+///    fully available. The envelope is still imported (fork-choice marks the block's payload
+///    as received and the envelope is persisted); column persistence is handled separately
+///    via gossip / engineGetBlobs as columns arrive.
+pub enum ExecutedEnvelope<E: EthSpec> {
+    Available(AvailableExecutedEnvelope<E>),
+    AvailabilityPending {
+        signed_envelope: Arc<SignedExecutionPayloadEnvelope<E>>,
+        import_data: EnvelopeImportData<E>,
+        payload_verification_outcome: PayloadVerificationOutcome,
+    },
+}
+
 impl<E: EthSpec> AvailabilityPendingExecutedEnvelope<E> {
     pub fn new(
         envelope: Arc<SignedExecutionPayloadEnvelope<E>>,
@@ -190,6 +211,14 @@ pub enum EnvelopeError {
     ExecutionPayloadError(ExecutionPayloadError),
     /// An error from importing the envelope.
     ImportError(BlockError),
+    /// A block processing error.
+    BlockProcessingError(BlockProcessingError),
+    /// A block error.
+    BlockError(BlockError),
+    /// An internal error.
+    InternalError(String),
+    /// Optimistic sync is not supported.
+    OptimisticSyncNotSupported { block_root: Hash256 },
 }
 
 impl std::fmt::Display for EnvelopeError {