diff --git a/beacon_node/http_api/src/beacon/execution_payload_envelopes.rs b/beacon_node/http_api/src/beacon/execution_payload_envelopes.rs index bad05ec9191..914432b965f 100644 --- a/beacon_node/http_api/src/beacon/execution_payload_envelopes.rs +++ b/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( eth_v1 .and(warp::path("beacon")) .and(warp::path("execution_payload_envelopes")) + .and(warp::query::()) .and(warp::path::end()) + .and(warp::header::(CONSENSUS_VERSION_HEADER)) + .and(warp::header::(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, chain: Arc>, network_tx: UnboundedSender>| { task_spawner.spawn_async_with_rejection(Priority::P0, async move { + ensure_stateful_submission(blob_data_included)?; let envelope = SignedExecutionPayloadEnvelope::::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( eth_v1 .and(warp::path("beacon")) .and(warp::path("execution_payload_envelopes")) + .and(warp::query::()) .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::(CONSENSUS_VERSION_HEADER)) + .and(warp::header::(BLOB_DATA_INCLUDED_HEADER)) + .and(warp::body::bytes()) + .and(task_spawner_filter) + .and(chain_filter) + .and(network_tx_filter) .then( - |envelope: SignedExecutionPayloadEnvelope, + |validation_level: api_types::BroadcastValidationQuery, + _fork_name: ForkName, + blob_data_included: bool, + body_bytes: Bytes, task_spawner: TaskSpawner, chain: Arc>, network_tx: UnboundedSender>| { 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 -/// . + +// 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( envelope: SignedExecutionPayloadEnvelope, + validation_level: BroadcastValidation, chain: Arc>, network_tx: &UnboundedSender>, ) -> Result { @@ -111,6 +156,9 @@ pub async fn publish_execution_payload_envelope( "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( })?; 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( }) }; + 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( ) .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::>(); - - // 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( + chain: &Arc>, + network_tx: &UnboundedSender>, + slot: types::Slot, + gossip_verified_columns: Vec>, +) -> 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::>(); + + 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( diff --git a/beacon_node/http_api/src/publish_blocks.rs b/beacon_node/http_api/src/publish_blocks.rs index 636911b0663..0368a874934 100644 --- a/beacon_node/http_api/src/publish_blocks.rs +++ b/beacon_node/http_api/src/publish_blocks.rs @@ -728,7 +728,7 @@ fn late_block_logging>( } /// Check if any of the blobs or the block are slashable. Returns `BlockError::Slashable` if so. -fn check_slashable( +pub(crate) fn check_slashable( chain_clone: &BeaconChain, block_root: Hash256, block_clone: &SignedBeaconBlock>, diff --git a/beacon_node/http_api/tests/gloas_reorg_tests.rs b/beacon_node/http_api/tests/gloas_reorg_tests.rs index e91413dcdf6..bed7a35b024 100644 --- a/beacon_node/http_api/tests/gloas_reorg_tests.rs +++ b/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::(slot_c, &randao_reveal, None, None, None, None) + .get_validator_blocks_v4::(slot_c, &randao_reveal, None, Some(false), None, None) .await .unwrap(); ( diff --git a/beacon_node/http_api/tests/tests.rs b/beacon_node/http_api/tests/tests.rs index b5d4b65c35b..9b4dfbba873 100644 --- a/beacon_node/http_api/tests/tests.rs +++ b/beacon_node/http_api/tests/tests.rs @@ -9,9 +9,10 @@ use beacon_chain::{ }; use bls::{AggregateSignature, Keypair, PublicKeyBytes, SecretKey, Signature, SignatureBytes}; use eth2::{ - BeaconNodeHttpClient, Error, + BLOB_DATA_INCLUDED_HEADER, BeaconNodeHttpClient, CONSENSUS_VERSION_HEADER, CONTENT_TYPE_HEADER, + Error, Error::ServerMessage, - Timeouts, + JSON_CONTENT_TYPE_HEADER, Timeouts, mixin::{RequestAccept, ResponseForkName, ResponseOptional}, types::{ BlockId as CoreBlockId, ForkChoiceNode, ProduceBlockV3Response, ProduceBlockV4Metadata, @@ -4407,6 +4408,172 @@ impl ApiTester { } } + fn advance_to_gloas_slot(&self) -> Option<(Slot, Epoch, ForkName)> { + for _ in 0..E::slots_per_epoch() * 3 { + let slot = self.chain.slot().unwrap(); + let fork_name = self.chain.spec.fork_name_at_slot::(slot); + if fork_name.gloas_enabled() { + return Some((slot, self.chain.epoch().unwrap(), fork_name)); + } + self.chain.slot_clock.set_slot(slot.as_u64() + 1); + } + None + } + + async fn build_and_post_block_for_envelope( + &self, + slot: Slot, + epoch: Epoch, + fork: &Fork, + genesis_validators_root: Hash256, + ) -> (SecretKey, u64, ExecutionPayloadEnvelope) { + let (sk, randao_reveal) = self + .proposer_setup(slot, epoch, fork, genesis_validators_root) + .await; + + let (response, _metadata) = self + .client + .get_validator_blocks_v4::(slot, &randao_reveal, None, Some(false), None, None) + .await + .unwrap(); + let block = response.data; + let block_root = block.tree_hash_root(); + let proposer_index = block.proposer_index(); + + let signed_block = block.sign(&sk, fork, genesis_validators_root, &self.chain.spec); + let signed_block_request = PublishBlockRequest::try_from(Arc::new(signed_block)).unwrap(); + self.client + .post_beacon_blocks_v2(&signed_block_request, None) + .await + .unwrap(); + + let envelope = self + .client + .get_validator_execution_payload_envelopes::(slot, block_root) + .await + .unwrap() + .data; + + (sk, proposer_index, envelope) + } + + pub async fn test_envelope_post_consensus_invalid_returns_400_no_broadcast(mut self) -> Self { + if !self.chain.spec.is_gloas_scheduled() { + return self; + } + + let fork = self.chain.canonical_head.cached_head().head_fork(); + let genesis_validators_root = self.chain.genesis_validators_root; + let Some((slot, epoch, fork_name)) = self.advance_to_gloas_slot() else { + return self; + }; + + let (sk, _, mut envelope) = self + .build_and_post_block_for_envelope(slot, epoch, &fork, genesis_validators_root) + .await; + envelope.payload.gas_limit = envelope.payload.gas_limit.saturating_add(1); + let signed_envelope = + self.sign_envelope(envelope, &sk, epoch, &fork, genesis_validators_root); + + while self.network_rx.network_recv.recv().now_or_never().is_some() {} + + let err = self + .client + .post_beacon_execution_payload_envelopes( + &signed_envelope, + fork_name, + Some(BroadcastValidation::Consensus), + ) + .await + .expect_err("consensus-invalid envelope should fail"); + + assert_eq!(err.status(), Some(StatusCode::BAD_REQUEST)); + assert!(self.network_rx.network_recv.recv().now_or_never().is_none()); + + self + } + + pub async fn test_envelope_post_gossip_partial_pass_returns_202(mut self) -> Self { + if !self.chain.spec.is_gloas_scheduled() { + return self; + } + + let fork = self.chain.canonical_head.cached_head().head_fork(); + let genesis_validators_root = self.chain.genesis_validators_root; + let Some((slot, epoch, fork_name)) = self.advance_to_gloas_slot() else { + return self; + }; + + let (sk, _, mut envelope) = self + .build_and_post_block_for_envelope(slot, epoch, &fork, genesis_validators_root) + .await; + envelope.payload.gas_limit = envelope.payload.gas_limit.saturating_add(1); + let signed_envelope = + self.sign_envelope(envelope, &sk, epoch, &fork, genesis_validators_root); + + while self.network_rx.network_recv.recv().now_or_never().is_some() {} + + let url = self + .client + .post_beacon_execution_payload_envelopes_path(Some(BroadcastValidation::Gossip)) + .unwrap(); + let response = reqwest::Client::new() + .post(url) + .header(CONTENT_TYPE_HEADER, JSON_CONTENT_TYPE_HEADER) + .header(CONSENSUS_VERSION_HEADER, fork_name.to_string()) + .header(BLOB_DATA_INCLUDED_HEADER, "false") + .json(&signed_envelope) + .send() + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::ACCEPTED); + assert!(self.network_rx.network_recv.recv().now_or_never().is_some()); + + self + } + + pub async fn test_envelope_post_equivocation_returns_400(mut self) -> Self { + if !self.chain.spec.is_gloas_scheduled() { + return self; + } + + let fork = self.chain.canonical_head.cached_head().head_fork(); + let genesis_validators_root = self.chain.genesis_validators_root; + let Some((slot, epoch, fork_name)) = self.advance_to_gloas_slot() else { + return self; + }; + + let (sk, proposer_index, envelope) = self + .build_and_post_block_for_envelope(slot, epoch, &fork, genesis_validators_root) + .await; + let signed_envelope = + self.sign_envelope(envelope, &sk, epoch, &fork, genesis_validators_root); + + self.chain + .observed_slashable + .write() + .observe_slashable(slot, proposer_index, Hash256::repeat_byte(0xee)) + .unwrap(); + + while self.network_rx.network_recv.recv().now_or_never().is_some() {} + + let err = self + .client + .post_beacon_execution_payload_envelopes( + &signed_envelope, + fork_name, + Some(BroadcastValidation::ConsensusAndEquivocation), + ) + .await + .expect_err("equivocating envelope should fail"); + + assert_eq!(err.status(), Some(StatusCode::BAD_REQUEST)); + assert!(self.network_rx.network_recv.recv().now_or_never().is_none()); + + self + } + /// Test V4 block production (JSON). Only runs if Gloas is scheduled. pub async fn test_block_production_v4(self) -> Self { if !self.chain.spec.is_gloas_scheduled() { @@ -4432,7 +4599,7 @@ impl ApiTester { let (response, metadata) = self .client - .get_validator_blocks_v4::(slot, &randao_reveal, None, None, None, None) + .get_validator_blocks_v4::(slot, &randao_reveal, None, Some(false), None, None) .await .unwrap(); let block = response.data; @@ -4471,7 +4638,7 @@ impl ApiTester { let signed_envelope = self.sign_envelope(envelope, &sk, epoch, &fork, genesis_validators_root); self.client - .post_beacon_execution_payload_envelopes(&signed_envelope, fork_name) + .post_beacon_execution_payload_envelopes(&signed_envelope, fork_name, None) .await .unwrap(); @@ -4506,7 +4673,14 @@ impl ApiTester { let (block, metadata) = self .client - .get_validator_blocks_v4_ssz::(slot, &randao_reveal, None, None, None, None) + .get_validator_blocks_v4_ssz::( + slot, + &randao_reveal, + None, + Some(false), + None, + None, + ) .await .unwrap(); @@ -4532,7 +4706,7 @@ impl ApiTester { let signed_envelope = self.sign_envelope(envelope, &sk, epoch, &fork, genesis_validators_root); self.client - .post_beacon_execution_payload_envelopes_ssz(&signed_envelope, fork_name) + .post_beacon_execution_payload_envelopes_ssz(&signed_envelope, fork_name, None) .await .unwrap(); @@ -4957,7 +5131,7 @@ impl ApiTester { // Produce and publish a block. let (response, _metadata) = self .client - .get_validator_blocks_v4::(slot, &randao_reveal, None, None, None, None) + .get_validator_blocks_v4::(slot, &randao_reveal, None, Some(false), None, None) .await .unwrap(); let block = response.data; @@ -4982,7 +5156,7 @@ impl ApiTester { let signed_envelope = self.sign_envelope(envelope, &sk, epoch, &fork, genesis_validators_root); self.client - .post_beacon_execution_payload_envelopes(&signed_envelope, fork_name) + .post_beacon_execution_payload_envelopes(&signed_envelope, fork_name, None) .await .unwrap(); @@ -5040,7 +5214,7 @@ impl ApiTester { // Produce and publish a block, but withhold its envelope. let (response, _metadata) = self .client - .get_validator_blocks_v4::(slot, &randao_reveal, None, None, None, None) + .get_validator_blocks_v4::(slot, &randao_reveal, None, Some(false), None, None) .await .unwrap(); let block = response.data; @@ -8871,6 +9045,39 @@ async fn payload_attestation_present_after_envelope_publish() { .await; } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn envelope_post_consensus_invalid_returns_400_no_broadcast() { + if !fork_name_from_env().is_some_and(|f| f.gloas_enabled()) { + return; + } + ApiTester::new_with_hard_forks() + .await + .test_envelope_post_consensus_invalid_returns_400_no_broadcast() + .await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn envelope_post_gossip_partial_pass_returns_202() { + if !fork_name_from_env().is_some_and(|f| f.gloas_enabled()) { + return; + } + ApiTester::new_with_hard_forks() + .await + .test_envelope_post_gossip_partial_pass_returns_202() + .await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn envelope_post_equivocation_returns_400() { + if !fork_name_from_env().is_some_and(|f| f.gloas_enabled()) { + return; + } + ApiTester::new_with_hard_forks() + .await + .test_envelope_post_equivocation_returns_400() + .await; +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn payload_attestation_unavailable_without_envelope() { ApiTester::new_with_hard_forks() diff --git a/common/eth2/src/lib.rs b/common/eth2/src/lib.rs index 2154c524441..3bda02935fb 100644 --- a/common/eth2/src/lib.rs +++ b/common/eth2/src/lib.rs @@ -61,6 +61,7 @@ pub const EXECUTION_PAYLOAD_BLINDED_HEADER: &str = "Eth-Execution-Payload-Blinde pub const EXECUTION_PAYLOAD_VALUE_HEADER: &str = "Eth-Execution-Payload-Value"; pub const EXECUTION_PAYLOAD_INCLUDED_HEADER: &str = "Eth-Execution-Payload-Included"; pub const CONSENSUS_BLOCK_VALUE_HEADER: &str = "Eth-Consensus-Block-Value"; +pub const BLOB_DATA_INCLUDED_HEADER: &str = "Eth-Blob-Data-Included"; pub const CONTENT_TYPE_HEADER: &str = "Content-Type"; pub const SSZ_CONTENT_TYPE_HEADER: &str = "application/octet-stream"; @@ -2803,12 +2804,10 @@ impl BeaconNodeHttpClient { ExecutionPayloadEnvelope::from_ssz_bytes(&response_bytes).map_err(Error::InvalidSsz) } - /// `POST v1/beacon/execution_payload_envelopes` - pub async fn post_beacon_execution_payload_envelopes( + pub fn post_beacon_execution_payload_envelopes_path( &self, - envelope: &SignedExecutionPayloadEnvelope, - fork_name: ForkName, - ) -> Result<(), Error> { + validation_level: Option, + ) -> Result { let mut path = self.eth_path(V1)?; path.path_segments_mut() @@ -2816,13 +2815,31 @@ impl BeaconNodeHttpClient { .push("beacon") .push("execution_payload_envelopes"); - self.post_generic_with_consensus_version( - path, - envelope, - Some(self.timeouts.proposal), - fork_name, - ) - .await?; + if let Some(validation_level) = validation_level { + path.query_pairs_mut() + .append_pair("broadcast_validation", &validation_level.to_string()); + } + + Ok(path) + } + + /// `POST v1/beacon/execution_payload_envelopes` + pub async fn post_beacon_execution_payload_envelopes( + &self, + envelope: &SignedExecutionPayloadEnvelope, + fork_name: ForkName, + validation_level: Option, + ) -> Result<(), Error> { + let response = self + .client + .post(self.post_beacon_execution_payload_envelopes_path(validation_level)?) + .timeout(self.timeouts.proposal) + .header(CONSENSUS_VERSION_HEADER, fork_name.to_string()) + .header(BLOB_DATA_INCLUDED_HEADER, "false") + .json(envelope) + .send() + .await?; + success_or_error(response).await?; Ok(()) } @@ -2832,21 +2849,19 @@ impl BeaconNodeHttpClient { &self, envelope: &SignedExecutionPayloadEnvelope, fork_name: ForkName, + validation_level: Option, ) -> Result<(), Error> { - let mut path = self.eth_path(V1)?; - - path.path_segments_mut() - .map_err(|()| Error::InvalidUrl(self.server.clone()))? - .push("beacon") - .push("execution_payload_envelopes"); - - self.post_generic_with_consensus_version_and_ssz_body( - path, - envelope.as_ssz_bytes(), - Some(self.timeouts.proposal), - fork_name, - ) - .await?; + let response = self + .client + .post(self.post_beacon_execution_payload_envelopes_path(validation_level)?) + .timeout(self.timeouts.proposal) + .header(CONSENSUS_VERSION_HEADER, fork_name.to_string()) + .header(BLOB_DATA_INCLUDED_HEADER, "false") + .header(CONTENT_TYPE_HEADER, SSZ_CONTENT_TYPE_HEADER) + .body(envelope.as_ssz_bytes()) + .send() + .await?; + success_or_error(response).await?; Ok(()) } diff --git a/testing/validator_test_rig/src/mock_beacon_node.rs b/testing/validator_test_rig/src/mock_beacon_node.rs index 255831da502..ca9fb547fff 100644 --- a/testing/validator_test_rig/src/mock_beacon_node.rs +++ b/testing/validator_test_rig/src/mock_beacon_node.rs @@ -1,5 +1,5 @@ use eth2::types::{GenericResponse, PublishBlockRequest, SyncingData}; -use eth2::{BeaconNodeHttpClient, Timeouts}; +use eth2::{BLOB_DATA_INCLUDED_HEADER, BeaconNodeHttpClient, CONSENSUS_VERSION_HEADER, Timeouts}; use mockito::{Matcher, Mock, Server, ServerGuard}; use regex::Regex; use reqwest::StatusCode; @@ -121,6 +121,10 @@ impl MockBeaconNode { self.server .mock("GET", Matcher::Regex(path_pattern.to_string())) + .match_query(Matcher::UrlEncoded( + "include_payload".into(), + "false".into(), + )) .match_header("accept", "application/json") .with_status(200) .with_header("Eth-Consensus-Version", &fork_name.to_string()) @@ -144,6 +148,10 @@ impl MockBeaconNode { self.server .mock("GET", Matcher::Regex(path_pattern.to_string())) + .match_query(Matcher::UrlEncoded( + "include_payload".into(), + "false".into(), + )) .match_header("accept", "application/octet-stream") .with_status(200) // These headers are required for v4 get validator block endpoint: https://github.com/ethereum/beacon-APIs/pull/580 @@ -161,6 +169,10 @@ impl MockBeaconNode { self.server .mock("GET", Matcher::Regex(path_pattern.to_string())) + .match_query(Matcher::UrlEncoded( + "include_payload".into(), + "false".into(), + )) .match_header("accept", "application/octet-stream") .with_status(500) .with_body(r#"{"message":"Internal server error"}"#) @@ -384,6 +396,8 @@ impl MockBeaconNode { self.server .mock("POST", Matcher::Regex(path_pattern.to_string())) .match_header("content-type", "application/octet-stream") + .match_header(CONSENSUS_VERSION_HEADER, "gloas") + .match_header(BLOB_DATA_INCLUDED_HEADER, "false") .with_status(200) .with_body_from_request(move |request| { let body = request.body().expect("Failed to get request body"); @@ -402,6 +416,8 @@ impl MockBeaconNode { self.server .mock("POST", Matcher::Regex(path_pattern.to_string())) .match_header("content-type", "application/octet-stream") + .match_header(CONSENSUS_VERSION_HEADER, "gloas") + .match_header(BLOB_DATA_INCLUDED_HEADER, "false") .with_status(500) .with_body(r#"{"message":"Internal server error"}"#) .create() diff --git a/validator_client/validator_services/src/block_service.rs b/validator_client/validator_services/src/block_service.rs index 97708b21f7a..1cf9c8cb1f5 100644 --- a/validator_client/validator_services/src/block_service.rs +++ b/validator_client/validator_services/src/block_service.rs @@ -478,7 +478,7 @@ impl BlockService { slot, randao_reveal_ref, graffiti.as_ref(), - None, + Some(false), builder_boost_factor, self_ref.graffiti_policy, ) @@ -506,7 +506,7 @@ impl BlockService { slot, randao_reveal_ref, graffiti.as_ref(), - None, + Some(false), builder_boost_factor, self_ref.graffiti_policy, ) @@ -712,7 +712,11 @@ impl BlockService { let signed_envelope = signed_envelope.clone(); async move { beacon_node - .post_beacon_execution_payload_envelopes_ssz(&signed_envelope, fork_name) + .post_beacon_execution_payload_envelopes_ssz( + &signed_envelope, + fork_name, + None, + ) .await .map_err(|e| { BlockError::Recoverable(format!(