boiler plate:

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<E: EthSpec> {
         seen_timestamp: Duration,
     },
 
+    /// An execution payload envelope has been received from the RPC.
+    RpcPayloadEnvelope {
+        sync_request_id: SyncRequestId,
+        peer_id: PeerId,
+        envelope: Option<Arc<SignedExecutionPayloadEnvelope<E>>>,
+        seen_timestamp: Duration,
+    },
+
     /// A block with an unknown parent has been received.
     UnknownParentBlock(PeerId, Arc<SignedBeaconBlock<E>>, 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<T: BeaconChainTypes> SyncManager<T> {
             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<T: BeaconChainTypes> SyncManager<T> {
             } => {
                 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<T: BeaconChainTypes> SyncManager<T> {
         }
     }
 
+    fn rpc_payload_envelope_received(
+        &mut self,
+        sync_request_id: SyncRequestId,
+        peer_id: PeerId,
+        envelope: Option<Arc<SignedExecutionPayloadEnvelope<T::EthSpec>>>,
+        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<Arc<SignedExecutionPayloadEnvelope<T::EthSpec>>>,
+    ) {
+        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,