diff --git a/beacon_node/beacon_chain/src/block_verification.rs b/beacon_node/beacon_chain/src/block_verification.rs index 2cc909f28b..6109537bff 100644 --- a/beacon_node/beacon_chain/src/block_verification.rs +++ b/beacon_node/beacon_chain/src/block_verification.rs @@ -321,6 +321,13 @@ pub enum BlockError { bid_parent_root: Hash256, block_parent_root: Hash256, }, + /// The parent block is known but its 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 }, } /// Which specific signature(s) are invalid in a SignedBeaconBlock @@ -1954,7 +1961,7 @@ fn load_parent>( let envelope = chain .store .get_payload_envelope(&root)? - .ok_or(BlockError::ParentUnknown { parent_root: root })?; + .ok_or(BlockError::ParentEnvelopeUnknown { parent_root: root })?; (StatePayloadStatus::Full, envelope.message.state_root) } else { (StatePayloadStatus::Pending, parent_block.state_root()) diff --git a/beacon_node/beacon_processor/src/lib.rs b/beacon_node/beacon_processor/src/lib.rs index 724c41cfc9..229816ba77 100644 --- a/beacon_node/beacon_processor/src/lib.rs +++ b/beacon_node/beacon_processor/src/lib.rs @@ -416,6 +416,9 @@ pub enum Work { RpcBlobs { process_fn: AsyncFn, }, + RpcPayloadEnvelope { + process_fn: AsyncFn, + }, RpcCustodyColumn(AsyncFn), ColumnReconstruction(AsyncFn), IgnoredRpcBlock { @@ -477,6 +480,7 @@ pub enum WorkType { GossipLightClientOptimisticUpdate, RpcBlock, RpcBlobs, + RpcPayloadEnvelope, RpcCustodyColumn, ColumnReconstruction, IgnoredRpcBlock, @@ -538,6 +542,7 @@ impl Work { 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, @@ -1169,7 +1174,9 @@ impl BeaconProcessor { 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), @@ -1301,7 +1308,9 @@ impl BeaconProcessor { 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() } @@ -1496,6 +1505,7 @@ impl BeaconProcessor { 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), diff --git a/beacon_node/beacon_processor/src/scheduler/work_reprocessing_queue.rs b/beacon_node/beacon_processor/src/scheduler/work_reprocessing_queue.rs index 6cb82a2906..151c6009c7 100644 --- a/beacon_node/beacon_processor/src/scheduler/work_reprocessing_queue.rs +++ b/beacon_node/beacon_processor/src/scheduler/work_reprocessing_queue.rs @@ -892,15 +892,13 @@ impl ReprocessQueue { InboundEvent::ReadyEnvelope(block_root) => { if let Some((envelope, _delay_key)) = self.awaiting_envelopes_per_root.remove(&block_root) - { - if self + && self .ready_work_tx .try_send(ReadyWork::Envelope(envelope)) .is_err() { error!(?block_root, "Failed to send envelope after timeout"); } - } } InboundEvent::ReadyAttestation(queued_id) => { metrics::inc_counter( 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 6a04274bac..c8e7267129 100644 --- a/beacon_node/http_api/src/beacon/execution_payload_envelope.rs +++ b/beacon_node/http_api/src/beacon/execution_payload_envelope.rs @@ -132,9 +132,7 @@ pub async fn publish_execution_payload_envelope( let ctx = chain.gossip_verification_context(); let Ok(gossip_verifed_envelope) = GossipVerifiedEnvelope::new(signed_envelope, &ctx) else { warn!(%slot, %beacon_block_root, "Execution payload envelope rejected"); - return Err(warp_utils::reject::custom_bad_request(format!( - "execution payload envelope rejected, gossip verification" - ))); + return Err(warp_utils::reject::custom_bad_request("execution payload envelope rejected, gossip verification".to_string())); }; // Import the envelope locally (runs state transition and notifies the EL). diff --git a/beacon_node/lighthouse_network/src/service/api_types.rs b/beacon_node/lighthouse_network/src/service/api_types.rs index 486a443857..a190a42a80 100644 --- a/beacon_node/lighthouse_network/src/service/api_types.rs +++ b/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. diff --git a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs index 3ef5daf7ed..0fda9760b7 100644 --- a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs +++ b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs @@ -1288,6 +1288,19 @@ impl NetworkBeaconProcessor { self.send_sync_message(SyncMessage::UnknownParentBlock(peer_id, block, block_root)); return None; } + Err(BlockError::ParentEnvelopeUnknown { parent_root }) => { + // The parent block is known but its execution payload envelope has not + // been received yet. Queue this block for reprocessing and trigger an + // envelope lookup. + debug!( + ?block_root, + ?parent_root, + "Parent envelope not yet available for gossip block" + ); + // TODO(gloas): trigger an envelope lookup for `parent_root` and queue + // this block for reprocessing once the envelope arrives. + return None; + } Err(e @ BlockError::BeaconChainError(_)) => { debug!( error = ?e, @@ -1576,6 +1589,17 @@ impl NetworkBeaconProcessor { "Block with unknown parent attempted to be processed" ); } + Err(BlockError::ParentEnvelopeUnknown { parent_root }) => { + // The parent block is known but its execution payload envelope has not + // been received yet. + debug!( + %block_root, + ?parent_root, + "Parent envelope not yet available, need envelope lookup" + ); + // TODO(gloas): trigger an envelope lookup for `parent_root` and queue + // this block for reprocessing once the envelope arrives. + } Err(e @ BlockError::ExecutionPayloadError(epe)) if !epe.penalize_peer() => { debug!( error = %e, diff --git a/beacon_node/network/src/network_beacon_processor/mod.rs b/beacon_node/network/src/network_beacon_processor/mod.rs index 6fd1d37bfc..8f45097408 100644 --- a/beacon_node/network/src/network_beacon_processor/mod.rs +++ b/beacon_node/network/src/network_beacon_processor/mod.rs @@ -540,6 +540,22 @@ impl NetworkBeaconProcessor { }) } + /// Create a new `Work` event for an RPC payload envelope. + pub fn send_rpc_payload_envelope( + self: &Arc, + envelope: Arc>, + seen_timestamp: Duration, + process_type: BlockProcessType, + ) -> Result<(), Error> { + let process_fn = + self.clone() + .generate_rpc_envelope_process_fn(envelope, seen_timestamp, process_type); + self.try_send(BeaconWorkEvent { + drop_during_sync: false, + work: Work::RpcPayloadEnvelope { process_fn }, + }) + } + /// Create a new `Work` event for some blobs, where the result from computation (if any) is /// sent to the other side of `result_tx`. pub fn send_rpc_blobs( diff --git a/beacon_node/network/src/network_beacon_processor/sync_methods.rs b/beacon_node/network/src/network_beacon_processor/sync_methods.rs index 629a42c688..a90ad0b750 100644 --- a/beacon_node/network/src/network_beacon_processor/sync_methods.rs +++ b/beacon_node/network/src/network_beacon_processor/sync_methods.rs @@ -4,7 +4,7 @@ use crate::sync::BatchProcessResult; use crate::sync::manager::CustodyBatchProcessResult; use crate::sync::{ ChainId, - manager::{BlockProcessType, SyncMessage}, + manager::{BlockProcessType, BlockProcessingResult, SyncMessage}, }; use beacon_chain::block_verification_types::{AsBlock, RpcBlock}; use beacon_chain::data_availability_checker::AvailabilityCheckError; @@ -27,7 +27,9 @@ use store::KzgCommitment; use tracing::{debug, debug_span, error, info, instrument, warn}; use types::data::FixedBlobSidecarList; use types::kzg_ext::format_kzg_commitments; -use types::{BlockImportSource, DataColumnSidecarList, Epoch, Hash256}; +use types::{ + BlockImportSource, DataColumnSidecarList, Epoch, Hash256, SignedExecutionPayloadEnvelope, +}; /// Id associated to a batch processing request, either a sync batch or a parent lookup. #[derive(Clone, Debug, PartialEq)] @@ -72,6 +74,80 @@ impl NetworkBeaconProcessor { Box::pin(process_fn) } + /// Returns an async closure which processes a payload envelope received via RPC. + pub fn generate_rpc_envelope_process_fn( + self: Arc, + envelope: Arc>, + seen_timestamp: Duration, + process_type: BlockProcessType, + ) -> AsyncFn { + let process_fn = async move { + self.process_rpc_envelope(envelope, seen_timestamp, process_type) + .await; + }; + Box::pin(process_fn) + } + + /// Process an execution payload envelope received via RPC. + async fn process_rpc_envelope( + self: Arc, + envelope: Arc>, + _seen_timestamp: Duration, + process_type: BlockProcessType, + ) { + let beacon_block_root = envelope.beacon_block_root(); + + // Verify the envelope using the gossip verification path (same checks apply to RPC) + let verified_envelope = match self.chain.verify_envelope_for_gossip(envelope).await { + Ok(verified) => verified, + Err(e) => { + debug!( + error = ?e, + ?beacon_block_root, + "RPC payload envelope failed verification" + ); + self.send_sync_message(SyncMessage::BlockComponentProcessed { + process_type, + result: BlockProcessingResult::Err(BlockError::InternalError(format!( + "Envelope verification failed: {e:?}" + ))), + }); + return; + } + }; + + // Process the verified envelope + let result = self + .chain + .process_execution_payload_envelope( + beacon_block_root, + verified_envelope, + NotifyExecutionLayer::Yes, + BlockImportSource::Lookup, + || Ok(()), + ) + .await; + + let processing_result = match result { + Ok(status) => BlockProcessingResult::Ok(status), + Err(e) => { + debug!( + error = ?e, + ?beacon_block_root, + "RPC payload envelope processing failed" + ); + BlockProcessingResult::Err(BlockError::InternalError(format!( + "Envelope processing failed: {e:?}" + ))) + } + }; + + self.send_sync_message(SyncMessage::BlockComponentProcessed { + process_type, + result: processing_result, + }); + } + /// Returns the `process_fn` and `ignore_fn` required when requeuing an RPC block. pub fn generate_rpc_beacon_block_fns( self: Arc, diff --git a/beacon_node/network/src/router.rs b/beacon_node/network/src/router.rs index 369ca80f28..3fb2196975 100644 --- a/beacon_node/network/src/router.rs +++ b/beacon_node/network/src/router.rs @@ -24,7 +24,10 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio::sync::mpsc; use tokio_stream::wrappers::UnboundedReceiverStream; use tracing::{debug, error, trace, warn}; -use types::{BlobSidecar, DataColumnSidecar, EthSpec, ForkContext, SignedBeaconBlock}; +use types::{ + BlobSidecar, DataColumnSidecar, EthSpec, ForkContext, SignedBeaconBlock, + SignedExecutionPayloadEnvelope, +}; /// Handles messages from the network and routes them to the appropriate service to be handled. pub struct Router { @@ -327,9 +330,12 @@ impl Router { Response::DataColumnsByRange(data_column) => { self.on_data_columns_by_range_response(peer_id, app_request_id, data_column); } - // TODO(EIP-7732): implement outgoing payload envelopes by range and root - // responses once sync manager requests them. - Response::PayloadEnvelopesByRoot(_) | Response::PayloadEnvelopesByRange(_) => { + Response::PayloadEnvelopesByRoot(envelope) => { + self.on_payload_envelopes_by_root_response(peer_id, app_request_id, envelope); + } + // TODO(EIP-7732): implement outgoing payload envelopes by range responses once + // range sync requests them. + Response::PayloadEnvelopesByRange(_) => { unreachable!() } // Light client responses should not be received @@ -703,6 +709,40 @@ impl Router { }); } + /// Handle a `PayloadEnvelopesByRoot` response from the peer. + pub fn on_payload_envelopes_by_root_response( + &mut self, + peer_id: PeerId, + app_request_id: AppRequestId, + envelope: Option>>, + ) { + let sync_request_id = match app_request_id { + AppRequestId::Sync(sync_id) => match sync_id { + id @ SyncRequestId::SinglePayloadEnvelope { .. } => id, + other => { + crit!(request = ?other, "PayloadEnvelopesByRoot response on incorrect request"); + return; + } + }, + AppRequestId::Router => { + crit!(%peer_id, "All PayloadEnvelopesByRoot requests belong to sync"); + return; + } + AppRequestId::Internal => unreachable!("Handled internally"), + }; + + trace!( + %peer_id, + "Received PayloadEnvelopesByRoot Response" + ); + self.send_to_sync(SyncMessage::RpcPayloadEnvelope { + peer_id, + sync_request_id, + envelope, + seen_timestamp: timestamp_now(), + }); + } + /// Handle a `BlobsByRoot` response from the peer. pub fn on_blobs_by_root_response( &mut self, diff --git a/beacon_node/network/src/sync/block_lookups/mod.rs b/beacon_node/network/src/sync/block_lookups/mod.rs index 394f2fc37d..539445ef50 100644 --- a/beacon_node/network/src/sync/block_lookups/mod.rs +++ b/beacon_node/network/src/sync/block_lookups/mod.rs @@ -559,6 +559,16 @@ impl BlockLookups { BlockProcessType::SingleCustodyColumn(id) => { self.on_processing_result_inner::>(id, result, cx) } + BlockProcessType::SinglePayloadEnvelope { id } => { + // TODO(EIP-7732): Wire into lookup state machine once envelope lookups are + // fully integrated. For now, just log the result. + debug!( + %id, + ?result, + "Payload envelope processing result (not yet wired to lookups)" + ); + return; + } }; self.on_lookup_result(process_type.id(), lookup_result, "processing_result", cx); } diff --git a/beacon_node/network/src/sync/manager.rs b/beacon_node/network/src/sync/manager.rs index 7e618d8980..98053a049f 100644 --- a/beacon_node/network/src/sync/manager.rs +++ b/beacon_node/network/src/sync/manager.rs @@ -74,7 +74,8 @@ use strum::IntoStaticStr; use tokio::sync::mpsc; use tracing::{debug, error, info, trace}; use types::{ - BlobSidecar, DataColumnSidecar, EthSpec, ForkContext, Hash256, SignedBeaconBlock, Slot, + BlobSidecar, DataColumnSidecar, EthSpec, ForkContext, Hash256, SignedBeaconBlock, + SignedExecutionPayloadEnvelope, Slot, }; /// The number of slots ahead of us that is allowed before requesting a long-range (batch) Sync @@ -133,6 +134,14 @@ pub enum SyncMessage { seen_timestamp: Duration, }, + /// An execution payload envelope has been received from the RPC. + RpcPayloadEnvelope { + sync_request_id: SyncRequestId, + peer_id: PeerId, + envelope: Option>>, + seen_timestamp: Duration, + }, + /// A block with an unknown parent has been received. UnknownParentBlock(PeerId, Arc>, Hash256), @@ -184,6 +193,7 @@ pub enum BlockProcessType { SingleBlock { id: Id }, SingleBlob { id: Id }, SingleCustodyColumn(Id), + SinglePayloadEnvelope { id: Id }, } impl BlockProcessType { @@ -191,7 +201,8 @@ impl BlockProcessType { match self { BlockProcessType::SingleBlock { id } | BlockProcessType::SingleBlob { id } - | BlockProcessType::SingleCustodyColumn(id) => *id, + | BlockProcessType::SingleCustodyColumn(id) + | BlockProcessType::SinglePayloadEnvelope { id } => *id, } } } @@ -505,6 +516,9 @@ impl SyncManager { SyncRequestId::DataColumnsByRange(req_id) => { self.on_data_columns_by_range_response(req_id, peer_id, RpcEvent::RPCError(error)) } + SyncRequestId::SinglePayloadEnvelope { id } => { + self.on_single_envelope_response(id, peer_id, RpcEvent::RPCError(error)) + } } } @@ -839,6 +853,17 @@ impl SyncManager { } => { self.rpc_data_column_received(sync_request_id, peer_id, data_column, seen_timestamp) } + SyncMessage::RpcPayloadEnvelope { + sync_request_id, + peer_id, + envelope, + seen_timestamp, + } => self.rpc_payload_envelope_received( + sync_request_id, + peer_id, + envelope, + seen_timestamp, + ), SyncMessage::UnknownParentBlock(peer_id, block, block_root) => { let block_slot = block.slot(); let parent_root = block.parent_root(); @@ -1200,6 +1225,57 @@ impl SyncManager { } } + fn rpc_payload_envelope_received( + &mut self, + sync_request_id: SyncRequestId, + peer_id: PeerId, + envelope: Option>>, + seen_timestamp: Duration, + ) { + match sync_request_id { + SyncRequestId::SinglePayloadEnvelope { id } => self.on_single_envelope_response( + id, + peer_id, + RpcEvent::from_chunk(envelope, seen_timestamp), + ), + _ => { + crit!(%peer_id, "bad request id for payload envelope"); + } + } + } + + fn on_single_envelope_response( + &mut self, + id: SingleLookupReqId, + peer_id: PeerId, + rpc_event: RpcEvent>>, + ) { + if let Some(resp) = self + .network + .on_single_envelope_response(id, peer_id, rpc_event) + { + match resp { + Ok((envelope, seen_timestamp)) => { + debug!( + block_root = ?envelope.beacon_block_root(), + %id, + "Downloaded payload envelope, sending for processing" + ); + if let Err(e) = self.network.send_envelope_for_processing( + id.req_id, + envelope, + seen_timestamp, + ) { + error!(error = ?e, "Failed to send envelope for processing"); + } + } + Err(e) => { + debug!(error = ?e, %id, "Payload envelope download failed"); + } + } + } + } + fn on_single_blob_response( &mut self, id: SingleLookupReqId, diff --git a/beacon_node/network/src/sync/network_context.rs b/beacon_node/network/src/sync/network_context.rs index 7e2c0d9a94..8c89098687 100644 --- a/beacon_node/network/src/sync/network_context.rs +++ b/beacon_node/network/src/sync/network_context.rs @@ -36,6 +36,7 @@ pub use requests::LookupVerifyError; use requests::{ ActiveRequests, BlobsByRangeRequestItems, BlobsByRootRequestItems, BlocksByRangeRequestItems, BlocksByRootRequestItems, DataColumnsByRangeRequestItems, DataColumnsByRootRequestItems, + PayloadEnvelopesByRootRequestItems, PayloadEnvelopesByRootSingleRequest, }; #[cfg(test)] use slot_clock::SlotClock; @@ -51,7 +52,7 @@ use tracing::{Span, debug, debug_span, error, warn}; use types::data::FixedBlobSidecarList; use types::{ BlobSidecar, BlockImportSource, ColumnIndex, DataColumnSidecar, DataColumnSidecarList, EthSpec, - ForkContext, Hash256, SignedBeaconBlock, Slot, + ForkContext, Hash256, SignedBeaconBlock, SignedExecutionPayloadEnvelope, Slot, }; pub mod custody; @@ -212,6 +213,9 @@ pub struct SyncNetworkContext { /// A mapping of active DataColumnsByRange requests data_columns_by_range_requests: ActiveRequests>, + /// A mapping of active PayloadEnvelopesByRoot requests + payload_envelopes_by_root_requests: + ActiveRequests>, /// Mapping of active custody column requests for a block root custody_by_root_requests: FnvHashMap>, @@ -297,6 +301,7 @@ impl SyncNetworkContext { blocks_by_range_requests: ActiveRequests::new("blocks_by_range"), blobs_by_range_requests: ActiveRequests::new("blobs_by_range"), data_columns_by_range_requests: ActiveRequests::new("data_columns_by_range"), + payload_envelopes_by_root_requests: ActiveRequests::new("payload_envelopes_by_root"), custody_by_root_requests: <_>::default(), components_by_range_requests: FnvHashMap::default(), custody_backfill_data_column_batch_requests: FnvHashMap::default(), @@ -325,6 +330,7 @@ impl SyncNetworkContext { blocks_by_range_requests, blobs_by_range_requests, data_columns_by_range_requests, + payload_envelopes_by_root_requests, // custody_by_root_requests is a meta request of data_columns_by_root_requests custody_by_root_requests: _, // components_by_range_requests is a meta request of various _by_range requests @@ -360,12 +366,17 @@ impl SyncNetworkContext { .active_requests_of_peer(peer_id) .into_iter() .map(|req_id| SyncRequestId::DataColumnsByRange(*req_id)); + let envelope_by_root_ids = payload_envelopes_by_root_requests + .active_requests_of_peer(peer_id) + .into_iter() + .map(|id| SyncRequestId::SinglePayloadEnvelope { id: *id }); blocks_by_root_ids .chain(blobs_by_root_ids) .chain(data_column_by_root_ids) .chain(blocks_by_range_ids) .chain(blobs_by_range_ids) .chain(data_column_by_range_ids) + .chain(envelope_by_root_ids) .collect() } @@ -422,6 +433,7 @@ impl SyncNetworkContext { blocks_by_range_requests, blobs_by_range_requests, data_columns_by_range_requests, + payload_envelopes_by_root_requests, // custody_by_root_requests is a meta request of data_columns_by_root_requests custody_by_root_requests: _, // components_by_range_requests is a meta request of various _by_range requests @@ -444,6 +456,7 @@ impl SyncNetworkContext { .chain(blocks_by_range_requests.iter_request_peers()) .chain(blobs_by_range_requests.iter_request_peers()) .chain(data_columns_by_range_requests.iter_request_peers()) + .chain(payload_envelopes_by_root_requests.iter_request_peers()) { *active_request_count_by_peer.entry(peer_id).or_default() += 1; } @@ -926,6 +939,57 @@ impl SyncNetworkContext { Ok(LookupRequestResult::RequestSent(id.req_id)) } + /// Request a payload envelope for `block_root` from a peer. + pub fn envelope_lookup_request( + &mut self, + lookup_id: SingleLookupId, + peer_id: PeerId, + block_root: Hash256, + ) -> Result { + let id = SingleLookupReqId { + lookup_id, + req_id: self.next_id(), + }; + + let request = PayloadEnvelopesByRootSingleRequest(block_root); + + let network_request = RequestType::PayloadEnvelopesByRoot( + request + .into_request(&self.fork_context) + .map_err(RpcRequestSendError::InternalError)?, + ); + self.network_send + .send(NetworkMessage::SendRequest { + peer_id, + request: network_request, + app_request_id: AppRequestId::Sync(SyncRequestId::SinglePayloadEnvelope { id }), + }) + .map_err(|_| RpcRequestSendError::InternalError("network send error".to_owned()))?; + + debug!( + method = "PayloadEnvelopesByRoot", + ?block_root, + peer = %peer_id, + %id, + "Sync RPC request sent" + ); + + let request_span = debug_span!( + parent: Span::current(), + "lh_outgoing_envelope_by_root_request", + %block_root, + ); + self.payload_envelopes_by_root_requests.insert( + id, + peer_id, + true, + PayloadEnvelopesByRootRequestItems::new(request), + request_span, + ); + + Ok(id.req_id) + } + /// Request necessary blobs for `block_root`. Requests only the necessary blobs by checking: /// - If we have a downloaded but not yet processed block /// - If the da_checker has a pending block @@ -1434,6 +1498,27 @@ impl SyncNetworkContext { self.on_rpc_response_result(resp, peer_id) } + pub(crate) fn on_single_envelope_response( + &mut self, + id: SingleLookupReqId, + peer_id: PeerId, + rpc_event: RpcEvent>>, + ) -> Option>>> { + let resp = self + .payload_envelopes_by_root_requests + .on_response(id, rpc_event); + let resp = resp.map(|res| { + res.and_then(|(mut envelopes, seen_timestamp)| { + match envelopes.pop() { + Some(envelope) => Ok((envelope, seen_timestamp)), + // Should never happen, request items enforces at least 1 chunk. + None => Err(LookupVerifyError::NotEnoughResponsesReturned { actual: 0 }.into()), + } + }) + }); + self.on_rpc_response_result(resp, peer_id) + } + pub(crate) fn on_single_blob_response( &mut self, id: SingleLookupReqId, @@ -1615,6 +1700,33 @@ impl SyncNetworkContext { }) } + pub fn send_envelope_for_processing( + &self, + id: Id, + envelope: Arc>, + seen_timestamp: Duration, + ) -> Result<(), SendErrorProcessor> { + let beacon_processor = self + .beacon_processor_if_enabled() + .ok_or(SendErrorProcessor::ProcessorNotAvailable)?; + + let block_root = envelope.beacon_block_root(); + debug!(?block_root, ?id, "Sending payload envelope for processing"); + beacon_processor + .send_rpc_payload_envelope( + envelope, + seen_timestamp, + BlockProcessType::SinglePayloadEnvelope { id }, + ) + .map_err(|e| { + error!( + error = ?e, + "Failed to send sync envelope to processor" + ); + SendErrorProcessor::SendError + }) + } + pub fn send_blobs_for_processing( &self, id: Id, diff --git a/beacon_node/network/src/sync/network_context/requests.rs b/beacon_node/network/src/sync/network_context/requests.rs index 8f9540693e..5b5e779d9b 100644 --- a/beacon_node/network/src/sync/network_context/requests.rs +++ b/beacon_node/network/src/sync/network_context/requests.rs @@ -16,6 +16,9 @@ pub use data_columns_by_range::DataColumnsByRangeRequestItems; pub use data_columns_by_root::{ DataColumnsByRootRequestItems, DataColumnsByRootSingleBlockRequest, }; +pub use payload_envelopes_by_root::{ + PayloadEnvelopesByRootRequestItems, PayloadEnvelopesByRootSingleRequest, +}; use crate::metrics; @@ -27,6 +30,7 @@ mod blocks_by_range; mod blocks_by_root; mod data_columns_by_range; mod data_columns_by_root; +mod payload_envelopes_by_root; #[derive(Debug, PartialEq, Eq, IntoStaticStr)] pub enum LookupVerifyError { diff --git a/beacon_node/network/src/sync/network_context/requests/payload_envelopes_by_root.rs b/beacon_node/network/src/sync/network_context/requests/payload_envelopes_by_root.rs new file mode 100644 index 0000000000..7f7097971d --- /dev/null +++ b/beacon_node/network/src/sync/network_context/requests/payload_envelopes_by_root.rs @@ -0,0 +1,53 @@ +use lighthouse_network::rpc::methods::PayloadEnvelopesByRootRequest; +use std::sync::Arc; +use types::{EthSpec, ForkContext, Hash256, SignedExecutionPayloadEnvelope}; + +use super::{ActiveRequestItems, LookupVerifyError}; + +#[derive(Debug, Copy, Clone)] +pub struct PayloadEnvelopesByRootSingleRequest(pub Hash256); + +impl PayloadEnvelopesByRootSingleRequest { + pub fn into_request( + self, + fork_context: &ForkContext, + ) -> Result { + PayloadEnvelopesByRootRequest::new(vec![self.0], fork_context) + } +} + +pub struct PayloadEnvelopesByRootRequestItems { + request: PayloadEnvelopesByRootSingleRequest, + items: Vec>>, +} + +impl PayloadEnvelopesByRootRequestItems { + pub fn new(request: PayloadEnvelopesByRootSingleRequest) -> Self { + Self { + request, + items: vec![], + } + } +} + +impl ActiveRequestItems for PayloadEnvelopesByRootRequestItems { + type Item = Arc>; + + /// Append a response to the single chunk request. If the chunk is valid, the request is + /// resolved immediately. + /// The active request SHOULD be dropped after `add_response` returns an error + fn add(&mut self, envelope: Self::Item) -> Result { + let beacon_block_root = envelope.beacon_block_root(); + if self.request.0 != beacon_block_root { + return Err(LookupVerifyError::UnrequestedBlockRoot(beacon_block_root)); + } + + self.items.push(envelope); + // Always returns true, payload envelopes by root expects a single response + Ok(true) + } + + fn consume(&mut self) -> Vec { + std::mem::take(&mut self.items) + } +}