Parent envelope unknown lookup

beacon_node/beacon_chain/src/block_verification.rs

@@ -60,6 +60,7 @@ use crate::execution_payload::{
 };
 use crate::kzg_utils::blobs_to_data_column_sidecars;
 use crate::observed_block_producers::SeenBlock;
+use crate::payload_envelope_verification::EnvelopeError;
 use crate::validator_monitor::HISTORIC_EPOCHS as VALIDATOR_MONITOR_HISTORIC_EPOCHS;
 use crate::validator_pubkey_cache::ValidatorPubkeyCache;
 use crate::{
@@ -320,6 +321,18 @@ pub enum BlockError {
         bid_parent_root: Hash256,
         block_parent_root: Hash256,
     },
+    /// The block is known but its parent execution payload envelope has not been received yet.
+    ///
+    /// ## Peer scoring
+    ///
+    /// It's unclear if this block is valid, but it cannot be fully verified without the parent's
+    /// execution payload envelope.
+    ParentEnvelopeUnknown { parent_root: Hash256 },
+
+    PayloadEnvelopeError {
+        e: Box<EnvelopeError>,
+        penalize_peer: bool,
+    },
 }
 
 /// Which specific signature(s) are invalid in a SignedBeaconBlock
@@ -486,6 +499,36 @@ impl From<ArithError> for BlockError {
     }
 }
 
+impl From<EnvelopeError> for BlockError {
+    fn from(e: EnvelopeError) -> Self {
+        let penalize_peer = match &e {
+            // REJECT per spec: peer sent invalid envelope data
+            EnvelopeError::BadSignature
+            | EnvelopeError::BuilderIndexMismatch { .. }
+            | EnvelopeError::BlockHashMismatch { .. }
+            | EnvelopeError::SlotMismatch { .. }
+            | EnvelopeError::IncorrectBlockProposer { .. } => true,
+            // IGNORE per spec: not the peer's fault
+            EnvelopeError::BlockRootUnknown { .. }
+            | EnvelopeError::PriorToFinalization { .. }
+            | EnvelopeError::UnknownValidator { .. } => false,
+            // Internal errors: not the peer's fault
+            EnvelopeError::BeaconChainError(_)
+            | EnvelopeError::BeaconStateError(_)
+            | EnvelopeError::BlockProcessingError(_)
+            | EnvelopeError::EnvelopeProcessingError(_)
+            | EnvelopeError::ExecutionPayloadError(_)
+            | EnvelopeError::BlockError(_)
+            | EnvelopeError::InternalError(_)
+            | EnvelopeError::OptimisticSyncNotSupported { .. } => false,
+        };
+        BlockError::PayloadEnvelopeError {
+            e: Box::new(e),
+            penalize_peer,
+        }
+    }
+}
+
 /// Stores information about verifying a payload against an execution engine.
 #[derive(Debug, PartialEq, Clone, Encode, Decode)]
 pub struct PayloadVerificationOutcome {
@@ -897,12 +940,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);
 
@@ -1951,7 +2008,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

@@ -14,7 +14,8 @@ use super::{
 };
 use crate::{
     AvailabilityProcessingStatus, BeaconChain, BeaconChainError, BeaconChainTypes,
-    NotifyExecutionLayer, block_verification_types::AvailableBlockData, metrics,
+    NotifyExecutionLayer, block_verification::PayloadVerificationOutcome,
+    block_verification_types::AvailableBlockData, metrics,
     payload_envelope_verification::ExecutionPendingEnvelope, validator_monitor::get_slot_delay_ms,
 };
 
@@ -99,9 +100,22 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                     self.import_available_execution_payload_envelope(Box::new(envelope))
                         .await
                 }
-                ExecutedEnvelope::AvailabilityPending() => Err(EnvelopeError::InternalError(
-                    "Pending payload envelope not yet implemented".to_owned(),
-                )),
+                ExecutedEnvelope::AvailabilityPending {
+                    signed_envelope,
+                    import_data,
+                    payload_verification_outcome: _,
+                } => {
+                    // The envelope has been executed but data columns have not yet arrived.
+                    // Do not import — return MissingComponents so callers know to fetch columns.
+                    // TODO(gloas): once an envelope DA checker exists, cache the envelope here
+                    // (analogous to `data_availability_checker.put_executed_block`) so that
+                    // import is driven automatically when columns arrive.
+                    let slot = signed_envelope.slot();
+                    let block_root = import_data.block_root;
+                    Ok(AvailabilityProcessingStatus::MissingComponents(
+                        slot, block_root,
+                    ))
+                }
             }
         };
 
@@ -182,10 +196,42 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             signed_envelope,
             import_data,
             payload_verification_outcome,
-            self.spec.clone(),
         ))
     }
 
+    /// 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>,
@@ -220,6 +266,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.
     ///

beacon_node/beacon_chain/src/payload_envelope_verification/mod.rs

@@ -119,20 +119,23 @@ pub struct EnvelopeProcessingSnapshot<E: EthSpec> {
 /// 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.
-#[allow(dead_code)]
+///    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>),
-    // TODO(gloas): check data column availability via DA checker
-    AvailabilityPending(),
+    AvailabilityPending {
+        signed_envelope: Arc<SignedExecutionPayloadEnvelope<E>>,
+        import_data: EnvelopeImportData<E>,
+        payload_verification_outcome: PayloadVerificationOutcome,
+    },
 }
 
 impl<E: EthSpec> ExecutedEnvelope<E> {
     pub fn new(
         envelope: MaybeAvailableEnvelope<E>,
         import_data: EnvelopeImportData<E>,
         payload_verification_outcome: PayloadVerificationOutcome,
-        spec: Arc<ChainSpec>,
     ) -> Self {
         match envelope {
             MaybeAvailableEnvelope::Available(available_envelope) => {
@@ -142,15 +145,14 @@ impl<E: EthSpec> ExecutedEnvelope<E> {
                     payload_verification_outcome,
                 ))
             }
-            // TODO(gloas): check data column availability via DA checker
             MaybeAvailableEnvelope::AvailabilityPending {
-                block_hash,
-                envelope,
-            } => Self::Available(AvailableExecutedEnvelope::new(
-                AvailableEnvelope::new(block_hash, envelope, vec![], None, spec),
+                block_hash: _,
+                envelope: signed_envelope,
+            } => Self::AvailabilityPending {
+                signed_envelope,
                 import_data,
                 payload_verification_outcome,
-            )),
+            },
         }
     }
 }

beacon_node/beacon_processor/src/lib.rs

@@ -417,6 +417,9 @@ pub enum Work<E: EthSpec> {
     RpcBlobs {
         process_fn: AsyncFn,
     },
+    RpcPayloadEnvelope {
+        process_fn: AsyncFn,
+    },
     RpcCustodyColumn(AsyncFn),
     ColumnReconstruction(AsyncFn),
     IgnoredRpcBlock {
@@ -483,6 +486,7 @@ pub enum WorkType {
     GossipLightClientOptimisticUpdate,
     RpcBlock,
     RpcBlobs,
+    RpcPayloadEnvelope,
     RpcCustodyColumn,
     ColumnReconstruction,
     IgnoredRpcBlock,
@@ -545,6 +549,7 @@ impl<E: EthSpec> Work<E> {
             Work::GossipProposerPreferences(_) => WorkType::GossipProposerPreferences,
             Work::RpcBlock { .. } => WorkType::RpcBlock,
             Work::RpcBlobs { .. } => WorkType::RpcBlobs,
+            Work::RpcPayloadEnvelope { .. } => WorkType::RpcPayloadEnvelope,
             Work::RpcCustodyColumn { .. } => WorkType::RpcCustodyColumn,
             Work::ColumnReconstruction(_) => WorkType::ColumnReconstruction,
             Work::IgnoredRpcBlock { .. } => WorkType::IgnoredRpcBlock,
@@ -1183,7 +1188,9 @@ impl<E: EthSpec> BeaconProcessor<E> {
                             Work::GossipLightClientOptimisticUpdate { .. } => work_queues
                                 .lc_gossip_optimistic_update_queue
                                 .push(work, work_id),
-                            Work::RpcBlock { .. } | Work::IgnoredRpcBlock { .. } => {
+                            Work::RpcBlock { .. }
+                            | Work::IgnoredRpcBlock { .. }
+                            | Work::RpcPayloadEnvelope { .. } => {
                                 work_queues.rpc_block_queue.push(work, work_id)
                             }
                             Work::RpcBlobs { .. } => work_queues.rpc_blob_queue.push(work, work_id),
@@ -1318,7 +1325,9 @@ impl<E: EthSpec> BeaconProcessor<E> {
                         WorkType::GossipLightClientOptimisticUpdate => {
                             work_queues.lc_gossip_optimistic_update_queue.len()
                         }
-                        WorkType::RpcBlock => work_queues.rpc_block_queue.len(),
+                        WorkType::RpcBlock | WorkType::RpcPayloadEnvelope => {
+                            work_queues.rpc_block_queue.len()
+                        }
                         WorkType::RpcBlobs | WorkType::IgnoredRpcBlock => {
                             work_queues.rpc_blob_queue.len()
                         }
@@ -1513,6 +1522,7 @@ impl<E: EthSpec> BeaconProcessor<E> {
                 beacon_block_root: _,
             }
             | Work::RpcBlobs { process_fn }
+            | Work::RpcPayloadEnvelope { process_fn }
             | Work::RpcCustodyColumn(process_fn)
             | Work::ColumnReconstruction(process_fn) => task_spawner.spawn_async(process_fn),
             Work::IgnoredRpcBlock { process_fn } => task_spawner.spawn_blocking(process_fn),

beacon_node/lighthouse_network/src/service/api_types.rs

@@ -31,6 +31,8 @@ pub enum SyncRequestId {
     BlobsByRange(BlobsByRangeRequestId),
     /// Data columns by range request
     DataColumnsByRange(DataColumnsByRangeRequestId),
+    /// Request searching for an execution payload envelope given a block root.
+    SinglePayloadEnvelope { id: SingleLookupReqId },
 }
 
 /// Request ID for data_columns_by_root requests. Block lookups do not issue this request directly.