diff --git a/beacon_node/http_api/src/beacon/execution_payload_envelope.rs b/beacon_node/http_api/src/beacon/execution_payload_envelope.rs index 161e091100..06a5915c08 100644 --- a/beacon_node/http_api/src/beacon/execution_payload_envelope.rs +++ b/beacon_node/http_api/src/beacon/execution_payload_envelope.rs @@ -1,25 +1,24 @@ use crate::block_id::BlockId; +use crate::publish_blocks::publish_column_sidecars; use crate::task_spawner::{Priority, TaskSpawner}; use crate::utils::{ChainFilter, EthV1Filter, NetworkTxFilter, ResponseFilter, TaskSpawnerFilter}; use crate::version::{ ResponseIncludesVersion, add_consensus_version_header, add_ssz_content_type_header, execution_optimistic_finalized_beacon_response, }; -use beacon_chain::payload_envelope_verification::gossip_verified_envelope::GossipVerifiedEnvelope; -use beacon_chain::{ - BeaconChain, BeaconChainTypes, NotifyExecutionLayer, - payload_envelope_verification::EnvelopeError, -}; +use beacon_chain::data_column_verification::{GossipDataColumnError, GossipVerifiedDataColumn}; +use beacon_chain::{BeaconChain, BeaconChainTypes}; use bytes::Bytes; use eth2::types as api_types; use eth2::{CONTENT_TYPE_HEADER, SSZ_CONTENT_TYPE_HEADER}; use lighthouse_network::PubsubMessage; use network::NetworkMessage; use ssz::{Decode, Encode}; +use std::future::Future; use std::sync::Arc; use tokio::sync::mpsc::UnboundedSender; -use tracing::{info, warn}; -use types::{BlockImportSource, SignedExecutionPayloadEnvelope}; +use tracing::{debug, error, info, warn}; +use types::{EthSpec, SignedExecutionPayloadEnvelope}; use warp::{ Filter, Rejection, Reply, hyper::{Body, Response}, @@ -89,7 +88,9 @@ pub(crate) fn post_beacon_execution_payload_envelope( ) .boxed() } -/// Publishes a signed execution payload envelope to the network. +/// Publishes a signed execution payload envelope to the network. Implements +/// `POST /eth/v1/beacon/execution_payload_envelope` per the in-flight beacon-APIs PR +/// . pub async fn publish_execution_payload_envelope( envelope: SignedExecutionPayloadEnvelope, chain: Arc>, @@ -97,70 +98,175 @@ pub async fn publish_execution_payload_envelope( ) -> Result, Rejection> { let slot = envelope.slot(); let beacon_block_root = envelope.message.beacon_block_root; - let builder_index = envelope.message.builder_index; + // TODO(gloas): Replace this check once we have gossip validation. if !chain.spec.is_gloas_scheduled() { return Err(warp_utils::reject::custom_bad_request( "Execution payload envelopes are not supported before the Gloas fork".into(), )); } - let signed_envelope = Arc::new(envelope); + // TODO(gloas): We should probably add validation here i.e. BroadcastValidation::Gossip + info!( + %slot, + %beacon_block_root, + builder_index = envelope.message.builder_index, + "Publishing signed execution payload envelope to network" + ); - // The publish_fn is called inside process_execution_payload_envelope after consensus - // verification but before the EL call. - let envelope_for_publish = signed_envelope.clone(); - let sender = network_tx.clone(); - let publish_fn = move || { - info!( - %slot, - %beacon_block_root, - builder_index, - "Publishing signed execution payload envelope to network" - ); - crate::utils::publish_pubsub_message( - &sender, - PubsubMessage::ExecutionPayload(Box::new((*envelope_for_publish).clone())), - ) - .map_err(|_| { - warn!(%slot, "Failed to publish execution payload envelope to network"); - EnvelopeError::InternalError( - "Unable to publish execution payload envelope to network".to_owned(), - ) - }) - }; + let blobs_and_proofs = chain.pending_payload_envelopes.write().take_blobs(slot); - let ctx = chain.payload_envelope_gossip_verification_context(); - let gossip_verified_envelope = match GossipVerifiedEnvelope::new(signed_envelope, &ctx) { - Ok(envelope) => envelope, - Err(e) => { - warn!(%slot, %beacon_block_root, error = ?e, "Execution payload envelope rejected"); - return Err(warp_utils::reject::custom_bad_request(format!( - "execution payload envelope rejected: {e:?}", - ))); - } - }; - - // Import the envelope locally (runs state transition and notifies the EL). - chain - .process_execution_payload_envelope( + // Spawn the column-build task (CPU-bound KZG cell-and-proof computation) before + // publishing the envelope so it runs in parallel with envelope gossip, narrowing + // the window in which peers see envelope-without-columns. If envelope publication + // fails below, dropping this future drops the spawned `JoinHandle` (the running + // closure on the blocking pool finishes and is then discarded — no work cancellation). + let column_build_future = match blobs_and_proofs { + Some(blobs) if !blobs.is_empty() => Some(spawn_build_gloas_data_columns_task( + &chain, beacon_block_root, - gossip_verified_envelope, - NotifyExecutionLayer::Yes, - BlockImportSource::HttpApi, - publish_fn, + slot, + blobs, + )?), + _ => None, + }; + + // Publish the envelope to the network. + crate::utils::publish_pubsub_message( + network_tx, + PubsubMessage::ExecutionPayload(Box::new(envelope)), + ) + .map_err(|_| { + warn!(%slot, "Failed to publish execution payload envelope to network"); + warp_utils::reject::custom_server_error( + "Unable to publish execution payload envelope to network".into(), ) - .await - .map_err(|e| { - warn!(%slot, %beacon_block_root, reason = ?e, "Execution payload envelope rejected"); - warp_utils::reject::custom_bad_request(format!( - "execution payload 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`. + if let Some(column_build_future) = column_build_future { + let gossip_verified_columns = match column_build_future.await { + Ok(columns) => columns, + 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.sampling_columns_for_epoch(epoch); + let sampling_columns = gossip_verified_columns + .into_iter() + .filter(|col| sampling_column_indices.contains(&col.index())) + .collect::>(); + + // Local processing only — envelope already broadcast, so log and fall through. + if !sampling_columns.is_empty() + && let Err(e) = + Box::pin(chain.process_gossip_data_columns(sampling_columns, || Ok(()))).await + { + error!( + %slot, + error = ?e, + "Failed to process sampling data columns during envelope publication" + ); + } + } + } Ok(warp::reply().into_response()) } +fn spawn_build_gloas_data_columns_task( + chain: &Arc>, + beacon_block_root: types::Hash256, + slot: types::Slot, + blobs: types::BlobsList, +) -> Result>, Rejection>>, Rejection> { + let chain_for_build = chain.clone(); + let handle = chain + .task_executor + .spawn_blocking_handle( + move || build_gloas_data_columns(&chain_for_build, beacon_block_root, slot, &blobs), + "build_gloas_data_columns", + ) + .ok_or_else(|| warp_utils::reject::custom_server_error("runtime shutdown".to_string()))?; + + Ok(async move { + handle + .await + .map_err(|_| warp_utils::reject::custom_server_error("join error".to_string()))? + }) +} + +fn build_gloas_data_columns( + chain: &BeaconChain, + beacon_block_root: types::Hash256, + slot: types::Slot, + blobs: &types::BlobsList, +) -> Result>, Rejection> { + let blob_refs: Vec<_> = blobs.iter().collect(); + let data_column_sidecars = beacon_chain::kzg_utils::blobs_to_data_column_sidecars_gloas( + &blob_refs, + beacon_block_root, + slot, + &chain.kzg, + &chain.spec, + ) + .map_err(|e| { + error!( + error = ?e, + %slot, + "Failed to build data column sidecars for envelope" + ); + warp_utils::reject::custom_server_error(format!("{e:?}")) + })?; + + let gossip_verified_columns = data_column_sidecars + .into_iter() + .filter_map(|col| { + let index = *col.index(); + match GossipVerifiedDataColumn::new_for_block_publishing(col, chain) { + Ok(verified) => Some(verified), + Err(GossipDataColumnError::PriorKnownUnpublished) => None, + Err(e) => { + warn!( + %slot, + column_index = index, + error = ?e, + "Locally-built data column failed gossip verification" + ); + None + } + } + }) + .collect::>(); + + debug!( + %slot, + column_count = gossip_verified_columns.len(), + "Built data columns for envelope publication" + ); + + Ok(gossip_verified_columns) +} + // TODO(gloas): add tests for this endpoint once we support importing payloads into the db // GET beacon/execution_payload_envelope/{block_id} pub(crate) fn get_beacon_execution_payload_envelope(