merge queue: checking unstable (3b39ffe) and #9276 together

beacon_node/http_api/src/beacon/execution_payload_envelopes.rs

@@ -1,5 +1,5 @@
 use crate::block_id::BlockId;
-use crate::publish_blocks::publish_column_sidecars;
+use crate::publish_blocks::{check_slashable, publish_column_sidecars};
 use crate::task_spawner::{Priority, TaskSpawner};
 use crate::utils::{ChainFilter, EthV1Filter, NetworkTxFilter, ResponseFilter, TaskSpawnerFilter};
 use crate::version::{
@@ -9,18 +9,23 @@ use crate::version::{
 use beacon_chain::data_column_verification::{GossipDataColumnError, GossipVerifiedDataColumn};
 use beacon_chain::payload_envelope_verification::EnvelopeError;
 use beacon_chain::{
-    AvailabilityProcessingStatus, BeaconChain, BeaconChainTypes, NotifyExecutionLayer,
+    AvailabilityProcessingStatus, BeaconChain, BeaconChainError, BeaconChainTypes, BlockError,
+    NotifyExecutionLayer,
 };
 use bytes::Bytes;
-use eth2::types as api_types;
+use eth2::{
+    BLOB_DATA_INCLUDED_HEADER, CONSENSUS_VERSION_HEADER,
+    types::{self as api_types, BroadcastValidation},
+};
 use lighthouse_network::PubsubMessage;
 use network::NetworkMessage;
 use ssz::{Decode, Encode};
 use std::future::Future;
 use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
 use tokio::sync::mpsc::UnboundedSender;
 use tracing::{debug, error, info, warn};
-use types::{BlockImportSource, EthSpec, SignedExecutionPayloadEnvelope};
+use types::{BlockImportSource, EthSpec, ForkName, SignedExecutionPayloadEnvelope};
 use warp::{
     Filter, Rejection,
     http::response::Builder,
@@ -37,23 +42,36 @@ pub(crate) fn post_beacon_execution_payload_envelopes_ssz<T: BeaconChainTypes>(
     eth_v1
         .and(warp::path("beacon"))
         .and(warp::path("execution_payload_envelopes"))
+        .and(warp::query::<api_types::BroadcastValidationQuery>())
         .and(warp::path::end())
+        .and(warp::header::<ForkName>(CONSENSUS_VERSION_HEADER))
+        .and(warp::header::<bool>(BLOB_DATA_INCLUDED_HEADER))
         .and(warp::body::bytes())
         .and(task_spawner_filter)
         .and(chain_filter)
         .and(network_tx_filter)
         .then(
-            |body_bytes: Bytes,
+            |validation_level: api_types::BroadcastValidationQuery,
+             _fork_name: ForkName,
+             blob_data_included: bool,
+             body_bytes: Bytes,
              task_spawner: TaskSpawner<T::EthSpec>,
              chain: Arc<BeaconChain<T>>,
              network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>| {
                 task_spawner.spawn_async_with_rejection(Priority::P0, async move {
+                    ensure_stateful_submission(blob_data_included)?;
                     let envelope =
                         SignedExecutionPayloadEnvelope::<T::EthSpec>::from_ssz_bytes(&body_bytes)
                             .map_err(|e| {
                             warp_utils::reject::custom_bad_request(format!("invalid SSZ: {e:?}"))
                         })?;
-                    publish_execution_payload_envelope(envelope, chain, &network_tx).await
+                    publish_execution_payload_envelope(
+                        envelope,
+                        validation_level.broadcast_validation,
+                        chain,
+                        &network_tx,
+                    )
+                    .await
                 })
             },
         )
@@ -70,28 +88,55 @@ pub(crate) fn post_beacon_execution_payload_envelopes<T: BeaconChainTypes>(
     eth_v1
         .and(warp::path("beacon"))
         .and(warp::path("execution_payload_envelopes"))
+        .and(warp::query::<api_types::BroadcastValidationQuery>())
         .and(warp::path::end())
-        .and(warp::body::json())
-        .and(task_spawner_filter.clone())
-        .and(chain_filter.clone())
-        .and(network_tx_filter.clone())
+        .and(warp::header::<ForkName>(CONSENSUS_VERSION_HEADER))
+        .and(warp::header::<bool>(BLOB_DATA_INCLUDED_HEADER))
+        .and(warp::body::bytes())
+        .and(task_spawner_filter)
+        .and(chain_filter)
+        .and(network_tx_filter)
         .then(
-            |envelope: SignedExecutionPayloadEnvelope<T::EthSpec>,
+            |validation_level: api_types::BroadcastValidationQuery,
+             _fork_name: ForkName,
+             blob_data_included: bool,
+             body_bytes: Bytes,
              task_spawner: TaskSpawner<T::EthSpec>,
              chain: Arc<BeaconChain<T>>,
              network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>| {
                 task_spawner.spawn_async_with_rejection(Priority::P0, async move {
-                    publish_execution_payload_envelope(envelope, chain, &network_tx).await
+                    ensure_stateful_submission(blob_data_included)?;
+                    let envelope = serde_json::from_slice(&body_bytes).map_err(|e| {
+                        warp_utils::reject::custom_bad_request(format!("invalid JSON: {e:?}"))
+                    })?;
+                    publish_execution_payload_envelope(
+                        envelope,
+                        validation_level.broadcast_validation,
+                        chain,
+                        &network_tx,
+                    )
+                    .await
                 })
             },
         )
         .boxed()
 }
-/// Publishes a signed execution payload envelope to the network. Implements
-/// `POST /eth/v1/beacon/execution_payload_envelopes` per the in-flight beacon-APIs PR
-/// <https://github.com/ethereum/beacon-APIs/pull/580>.
+
+// TODO(gloas): support Eth-Blob-Data-Included: true (stateless
+// SignedExecutionPayloadEnvelopeContents) instead of rejecting it (#9568 / #8828).
+fn ensure_stateful_submission(blob_data_included: bool) -> Result<(), Rejection> {
+    if blob_data_included {
+        return Err(warp_utils::reject::custom_bad_request(
+            "SignedExecutionPayloadEnvelopeContents is not supported".into(),
+        ));
+    }
+    Ok(())
+}
+
+/// Publishes a signed execution payload envelope to the network.
 pub async fn publish_execution_payload_envelope<T: BeaconChainTypes>(
     envelope: SignedExecutionPayloadEnvelope<T::EthSpec>,
+    validation_level: BroadcastValidation,
     chain: Arc<BeaconChain<T>>,
     network_tx: &UnboundedSender<NetworkMessage<T::EthSpec>>,
 ) -> Result<Response, Rejection> {
@@ -111,6 +156,9 @@ pub async fn publish_execution_payload_envelope<T: BeaconChainTypes>(
         "Publishing signed execution payload envelope to network"
     );
 
+    // TODO(gloas): if the block commits to blobs but none are cached, return 400 before
+    // publishing (beacon-APIs "no cached blobs and KZG proofs to attach"), rather than
+    // proceeding to a 202 via MissingComponents.
     let blobs_and_proofs = chain.pending_payload_envelopes.write().take_blobs(slot);
 
     // Spawn the column-build task (CPU-bound KZG cell-and-proof computation) before
@@ -140,11 +188,12 @@ pub async fn publish_execution_payload_envelope<T: BeaconChainTypes>(
         })?;
 
     let network_tx_clone = network_tx.clone();
-    let envelope_for_gossip = gossip_verified.signed_envelope.as_ref().clone();
-    let publish_fn = || {
+    let block_for_equivocation_check = gossip_verified.block.clone();
+    let envelope_for_gossip = gossip_verified.signed_envelope.clone();
+    let publish_envelope = || {
         crate::utils::publish_pubsub_message(
             &network_tx_clone,
-            PubsubMessage::ExecutionPayload(Box::new(envelope_for_gossip)),
+            PubsubMessage::ExecutionPayload(Box::new(envelope_for_gossip.as_ref().clone())),
         )
         .map_err(|_| {
             EnvelopeError::BeaconChainError(Box::new(
@@ -153,6 +202,35 @@ pub async fn publish_execution_payload_envelope<T: BeaconChainTypes>(
         })
     };
 
+    let published = AtomicBool::new(false);
+    if validation_level == BroadcastValidation::Gossip {
+        publish_envelope().map_err(|_| {
+            warp_utils::reject::custom_server_error(
+                "unable to publish to network channel".to_string(),
+            )
+        })?;
+        published.store(true, Ordering::SeqCst);
+    }
+
+    let publish_fn = || {
+        match validation_level {
+            BroadcastValidation::Gossip => return Ok(()),
+            BroadcastValidation::Consensus => {}
+            BroadcastValidation::ConsensusAndEquivocation => {
+                check_slashable(&chain, beacon_block_root, &block_for_equivocation_check).map_err(
+                    |e| match e {
+                        BlockError::BeaconChainError(e) => EnvelopeError::BeaconChainError(e),
+                        e => EnvelopeError::InternalError(format!("{e:?}")),
+                    },
+                )?;
+            }
+        }
+
+        publish_envelope()?;
+        published.store(true, Ordering::SeqCst);
+        Ok(())
+    };
+
     let import_result = chain
         .process_execution_payload_envelope(
             beacon_block_root,
@@ -163,73 +241,122 @@ pub async fn publish_execution_payload_envelope<T: BeaconChainTypes>(
         )
         .await;
 
-    let mut envelope_imported = match &import_result {
+    let mut envelope_imported = match import_result {
         Ok(AvailabilityProcessingStatus::Imported(_, _)) => true,
         Ok(AvailabilityProcessingStatus::MissingComponents(_, _)) => false,
         Err(e) => {
-            warn!(%slot, error = ?e, "Failed to import execution payload envelope");
-            return Err(warp_utils::reject::custom_server_error(format!(
-                "envelope import failed: {e}"
+            if is_unable_to_publish(&e) {
+                return Err(warp_utils::reject::custom_server_error(
+                    "unable to publish to network channel".to_string(),
+                ));
+            }
+
+            if published.load(Ordering::SeqCst) {
+                warn!(%slot, error = ?e, "Failed to import execution payload envelope after broadcast");
+                return Err(warp_utils::reject::broadcast_without_import(format!(
+                    "envelope import failed: {e:?}"
+                )));
+            }
+
+            warn!(%slot, error = ?e, "Rejecting execution payload envelope before broadcast");
+            return Err(warp_utils::reject::custom_bad_request(format!(
+                "envelope rejected: {e:?}"
             )));
         }
     };
 
-    // From here on the envelope is on the wire. `take_blobs` already consumed the cache
-    // entry, so a retry would not republish columns; returning Err would mislead the
-    // caller. Log column-build/publish failures and fall through to `Ok`.
+    // Column failures leave `envelope_imported` unchanged for the response below.
     if let Some(column_build_future) = column_build_future {
-        let gossip_verified_columns = match column_build_future.await {
-            Ok(columns) => columns,
+        match column_build_future.await {
+            Ok(columns) => {
+                if publish_and_import_columns(&chain, network_tx, slot, columns).await {
+                    envelope_imported = true;
+                }
+            }
             Err(e) => {
                 error!(
                     %slot,
                     error = ?e,
                     "Failed to build data columns after envelope publication"
                 );
-                return Ok(warp::reply().into_response());
-            }
-        };
-
-        if !gossip_verified_columns.is_empty() {
-            if let Err(e) = publish_column_sidecars(network_tx, &gossip_verified_columns, &chain) {
-                error!(
-                    %slot,
-                    error = ?e,
-                    "Failed to publish data column sidecars after envelope publication"
-                );
-                return Ok(warp::reply().into_response());
-            }
-
-            let epoch = slot.epoch(T::EthSpec::slots_per_epoch());
-            let sampling_column_indices = chain.custody_context.sampling_columns_for_epoch(epoch);
-            let sampling_columns = gossip_verified_columns
-                .into_iter()
-                .filter(|col| sampling_column_indices.contains(&col.index()))
-                .collect::<Vec<_>>();
-
-            // Local processing only — envelope already broadcast, so log and fall through.
-            if !sampling_columns.is_empty() {
-                match Box::pin(chain.process_gossip_data_columns(sampling_columns, || Ok(()))).await
-                {
-                    Ok(AvailabilityProcessingStatus::Imported(_, _)) => envelope_imported = true,
-                    Ok(AvailabilityProcessingStatus::MissingComponents(_, _)) => {}
-                    Err(e) => {
-                        error!(
-                            %slot,
-                            error = ?e,
-                            "Failed to process sampling data columns during envelope publication"
-                        );
-                    }
-                }
             }
         }
     }
 
+    // Return 202 when broadcast succeeds but integration does not. Unlike the block path,
+    // 202 applies at every validation level: missing components here are node-cached columns
+    // rather than request-supplied blobs, so incomplete import is not the submitter's fault.
     if envelope_imported {
         chain.recompute_head_at_current_slot().await;
+        Ok(warp::reply().into_response())
+    } else {
+        Err(warp_utils::reject::broadcast_without_import(format!(
+            "envelope for slot {slot} was broadcast but not fully imported"
+        )))
+    }
+}
+
+/// Publishes locally-built data columns and imports the node's sampling subset.
+///
+/// Returns `true` iff sampling-column processing completed envelope import. All failures are
+/// logged and swallowed: the envelope is already on the wire, so the caller's final
+/// 200-vs-202 decision reflects import state rather than column errors.
+async fn publish_and_import_columns<T: BeaconChainTypes>(
+    chain: &Arc<BeaconChain<T>>,
+    network_tx: &UnboundedSender<NetworkMessage<T::EthSpec>>,
+    slot: types::Slot,
+    gossip_verified_columns: Vec<GossipVerifiedDataColumn<T>>,
+) -> bool {
+    if gossip_verified_columns.is_empty() {
+        return false;
     }
 
-    Ok(warp::reply().into_response())
+    if let Err(e) = publish_column_sidecars(network_tx, &gossip_verified_columns, chain) {
+        error!(
+            %slot,
+            error = ?e,
+            "Failed to publish data column sidecars after envelope publication"
+        );
+        return false;
+    }
+
+    let epoch = slot.epoch(T::EthSpec::slots_per_epoch());
+    let sampling_column_indices = chain.custody_context.sampling_columns_for_epoch(epoch);
+    let sampling_columns = gossip_verified_columns
+        .into_iter()
+        .filter(|col| sampling_column_indices.contains(&col.index()))
+        .collect::<Vec<_>>();
+
+    if sampling_columns.is_empty() {
+        return false;
+    }
+
+    match Box::pin(chain.process_gossip_data_columns(sampling_columns, || Ok(()))).await {
+        Ok(AvailabilityProcessingStatus::Imported(_, _)) => true,
+        Ok(AvailabilityProcessingStatus::MissingComponents(_, _)) => false,
+        Err(e) => {
+            error!(
+                %slot,
+                error = ?e,
+                "Failed to process sampling data columns during envelope publication"
+            );
+            false
+        }
+    }
+}
+
+fn is_unable_to_publish(error: &BlockError) -> bool {
+    match error {
+        BlockError::BeaconChainError(error) => {
+            matches!(error.as_ref(), BeaconChainError::UnableToPublish)
+        }
+        BlockError::EnvelopeError(error) => matches!(
+            error.as_ref(),
+            EnvelopeError::BeaconChainError(error)
+                if matches!(error.as_ref(), BeaconChainError::UnableToPublish)
+        ),
+        _ => false,
+    }
 }
 
 fn spawn_build_gloas_data_columns_task<T: BeaconChainTypes>(

beacon_node/http_api/src/publish_blocks.rs

@@ -728,7 +728,7 @@ fn late_block_logging<T: BeaconChainTypes, P: AbstractExecPayload<T::EthSpec>>(
 }
 
 /// Check if any of the blobs or the block are slashable. Returns `BlockError::Slashable` if so.
-fn check_slashable<T: BeaconChainTypes>(
+pub(crate) fn check_slashable<T: BeaconChainTypes>(
     chain_clone: &BeaconChain<T>,
     block_root: Hash256,
     block_clone: &SignedBeaconBlock<T::EthSpec, FullPayload<T::EthSpec>>,

beacon_node/http_api/tests/gloas_reorg_tests.rs

@@ -727,7 +727,7 @@ pub async fn proposer_boost_re_org_test(
     let (block_c, block_c_blobs) = {
         let (response, _) = tester
             .client
-            .get_validator_blocks_v4::<E>(slot_c, &randao_reveal, None, None, None, None)
+            .get_validator_blocks_v4::<E>(slot_c, &randao_reveal, None, Some(false), None, None)
             .await
             .unwrap();
         (