gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-05-31 05:07:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

revert uneeded changes

Browse Source
This commit is contained in:
Eitan Seri-Levi
2026-04-28 16:33:47 +02:00
parent 269e474f49
commit 188f8271ec
1 changed files with 163 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

220
beacon_node/http_api/src/beacon/execution_payload_envelope.rs
View File

@@ -1,25 +1,24 @@
use crate::block_id::BlockId; use crate::block_id::BlockId;
use crate::publish_blocks::publish_column_sidecars;
use crate::task_spawner::{Priority, TaskSpawner}; use crate::task_spawner::{Priority, TaskSpawner};
use crate::utils::{ChainFilter, EthV1Filter, NetworkTxFilter, ResponseFilter, TaskSpawnerFilter}; use crate::utils::{ChainFilter, EthV1Filter, NetworkTxFilter, ResponseFilter, TaskSpawnerFilter};
use crate::version::{ use crate::version::{
ResponseIncludesVersion, add_consensus_version_header, add_ssz_content_type_header, ResponseIncludesVersion, add_consensus_version_header, add_ssz_content_type_header,
execution_optimistic_finalized_beacon_response, execution_optimistic_finalized_beacon_response,
}; };
use beacon_chain::payload_envelope_verification::gossip_verified_envelope::GossipVerifiedEnvelope; use beacon_chain::data_column_verification::{GossipDataColumnError, GossipVerifiedDataColumn};
use beacon_chain::{ use beacon_chain::{BeaconChain, BeaconChainTypes};
BeaconChain, BeaconChainTypes, NotifyExecutionLayer,
payload_envelope_verification::EnvelopeError,
};
use bytes::Bytes; use bytes::Bytes;
use eth2::types as api_types; use eth2::types as api_types;
use eth2::{CONTENT_TYPE_HEADER, SSZ_CONTENT_TYPE_HEADER}; use eth2::{CONTENT_TYPE_HEADER, SSZ_CONTENT_TYPE_HEADER};
use lighthouse_network::PubsubMessage; use lighthouse_network::PubsubMessage;
use network::NetworkMessage; use network::NetworkMessage;
use ssz::{Decode, Encode}; use ssz::{Decode, Encode};
use std::future::Future;
use std::sync::Arc; use std::sync::Arc;
use tokio::sync::mpsc::UnboundedSender; use tokio::sync::mpsc::UnboundedSender;
use tracing::{info, warn}; use tracing::{debug, error, info, warn};
use types::{BlockImportSource, SignedExecutionPayloadEnvelope}; use types::{EthSpec, SignedExecutionPayloadEnvelope};
use warp::{ use warp::{
Filter, Rejection, Reply, Filter, Rejection, Reply,
hyper::{Body, Response}, hyper::{Body, Response},
@@ -89,7 +88,9 @@ pub(crate) fn post_beacon_execution_payload_envelope<T: BeaconChainTypes>(
) )
.boxed() .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
/// <https://github.com/ethereum/beacon-APIs/pull/580>.
pub async fn publish_execution_payload_envelope<T: BeaconChainTypes>( pub async fn publish_execution_payload_envelope<T: BeaconChainTypes>(
envelope: SignedExecutionPayloadEnvelope<T::EthSpec>, envelope: SignedExecutionPayloadEnvelope<T::EthSpec>,
chain: Arc<BeaconChain<T>>, chain: Arc<BeaconChain<T>>,
@@ -97,70 +98,175 @@ pub async fn publish_execution_payload_envelope<T: BeaconChainTypes>(
) -> Result<Response<Body>, Rejection> { ) -> Result<Response<Body>, Rejection> {
let slot = envelope.slot(); let slot = envelope.slot();
let beacon_block_root = envelope.message.beacon_block_root; 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() { if !chain.spec.is_gloas_scheduled() {
return Err(warp_utils::reject::custom_bad_request( return Err(warp_utils::reject::custom_bad_request(
"Execution payload envelopes are not supported before the Gloas fork".into(), "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 let blobs_and_proofs = chain.pending_payload_envelopes.write().take_blobs(slot);
// 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 ctx = chain.payload_envelope_gossip_verification_context(); // Spawn the column-build task (CPU-bound KZG cell-and-proof computation) before
let gossip_verified_envelope = match GossipVerifiedEnvelope::new(signed_envelope, &ctx) { // publishing the envelope so it runs in parallel with envelope gossip, narrowing
Ok(envelope) => envelope, // the window in which peers see envelope-without-columns. If envelope publication
Err(e) => { // fails below, dropping this future drops the spawned `JoinHandle` (the running
warn!(%slot, %beacon_block_root, error = ?e, "Execution payload envelope rejected"); // closure on the blocking pool finishes and is then discarded — no work cancellation).
return Err(warp_utils::reject::custom_bad_request(format!( let column_build_future = match blobs_and_proofs {
"execution payload envelope rejected: {e:?}", Some(blobs) if !blobs.is_empty() => Some(spawn_build_gloas_data_columns_task(
))); &chain,
}
};
// Import the envelope locally (runs state transition and notifies the EL).
chain
.process_execution_payload_envelope(
beacon_block_root, beacon_block_root,
gossip_verified_envelope, slot,
NotifyExecutionLayer::Yes, blobs,
BlockImportSource::HttpApi, )?),
publish_fn, _ => 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"); // From here on the envelope is on the wire. `take_blobs` already consumed the cache
warp_utils::reject::custom_bad_request(format!( // entry, so a retry would not republish columns; returning Err would mislead the
"execution payload envelope rejected: {e:?}" // 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::<Vec<_>>();
// 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()) Ok(warp::reply().into_response())
} }
fn spawn_build_gloas_data_columns_task<T: BeaconChainTypes>(
chain: &Arc<BeaconChain<T>>,
beacon_block_root: types::Hash256,
slot: types::Slot,
blobs: types::BlobsList<T::EthSpec>,
) -> Result<impl Future<Output = Result<Vec<GossipVerifiedDataColumn<T>>, 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<T: BeaconChainTypes>(
chain: &BeaconChain<T>,
beacon_block_root: types::Hash256,
slot: types::Slot,
blobs: &types::BlobsList<T::EthSpec>,
) -> Result<Vec<GossipVerifiedDataColumn<T>>, 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::<Vec<_>>();
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 // TODO(gloas): add tests for this endpoint once we support importing payloads into the db
// GET beacon/execution_payload_envelope/{block_id} // GET beacon/execution_payload_envelope/{block_id}
pub(crate) fn get_beacon_execution_payload_envelope<T: BeaconChainTypes>( pub(crate) fn get_beacon_execution_payload_envelope<T: BeaconChainTypes>(