Merge remote-tracking branch 'origin/gloas-range-sync' into glamsterdam-devnet-2

beacon_node/network/src/network_beacon_processor/mod.rs

@@ -526,15 +526,11 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
         self: &Arc<Self>,
         message_id: MessageId,
         peer_id: PeerId,
-        proposer_preferences: Box<SignedProposerPreferences>,
+        proposer_preferences: Arc<SignedProposerPreferences>,
     ) -> Result<(), Error<T::EthSpec>> {
         let processor = self.clone();
         let process_fn = move || {
-            processor.process_gossip_proposer_preferences(
-                message_id,
-                peer_id,
-                Arc::new(*proposer_preferences),
-            )
+            processor.process_gossip_proposer_preferences(message_id, peer_id, proposer_preferences)
         };
 
         self.try_send(BeaconWorkEvent {

beacon_node/network/src/router.rs

@@ -26,6 +26,7 @@ use tokio_stream::wrappers::UnboundedReceiverStream;
 use tracing::{debug, error, trace, warn};
 use types::{
     BlobSidecar, DataColumnSidecar, EthSpec, ForkContext, PartialDataColumn, SignedBeaconBlock,
+    SignedExecutionPayloadEnvelope,
 };
 
 /// Handles messages from the network and routes them to the appropriate service to be handled.
@@ -341,10 +342,19 @@ impl<T: BeaconChainTypes> Router<T> {
             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(_) => {
-                debug!("Requesting envelopes by root and by range not supported yet");
+            Response::PayloadEnvelopesByRoot(payload_envelope) => {
+                self.on_payload_envelopes_by_root_response(
+                    peer_id,
+                    app_request_id,
+                    payload_envelope,
+                );
+            }
+            Response::PayloadEnvelopesByRange(payload_envelope) => {
+                self.on_payload_envelopes_by_range_response(
+                    peer_id,
+                    app_request_id,
+                    payload_envelope,
+                );
             }
             // Light client responses should not be received
             Response::LightClientBootstrap(_)
@@ -809,6 +819,62 @@ impl<T: BeaconChainTypes> Router<T> {
         }
     }
 
+    pub fn on_payload_envelopes_by_root_response(
+        &mut self,
+        peer_id: PeerId,
+        app_request_id: AppRequestId,
+        payload_envelope: Option<Arc<SignedExecutionPayloadEnvelope<T::EthSpec>>>,
+    ) {
+        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 {
+            sync_request_id,
+            peer_id,
+            payload_envelope,
+            seen_timestamp: self.chain.slot_clock.now_duration().unwrap_or_default(),
+        });
+    }
+
+    pub fn on_payload_envelopes_by_range_response(
+        &mut self,
+        peer_id: PeerId,
+        app_request_id: AppRequestId,
+        payload_envelope: Option<Arc<SignedExecutionPayloadEnvelope<T::EthSpec>>>,
+    ) {
+        trace!(
+            %peer_id,
+            "Received PayloadEnvelopesByRange Response"
+        );
+
+        if let AppRequestId::Sync(sync_request_id) = app_request_id {
+            self.send_to_sync(SyncMessage::RpcPayloadEnvelope {
+                peer_id,
+                sync_request_id,
+                payload_envelope,
+                seen_timestamp: self.chain.slot_clock.now_duration().unwrap_or_default(),
+            });
+        } else {
+            crit!("All payload envelopes by range responses should belong to sync");
+        }
+    }
+
     fn handle_beacon_processor_send_result(
         &mut self,
         result: Result<(), crate::network_beacon_processor::Error<T::EthSpec>>,

beacon_node/network/src/sync/manager.rs

@@ -134,6 +134,14 @@ pub enum SyncMessage<E: EthSpec> {
         seen_timestamp: Duration,
     },
 
+    /// A payload envelope has been received from the RPC.
+    RpcPayloadEnvelope {
+        sync_request_id: SyncRequestId,
+        peer_id: PeerId,
+        payload_envelope: Option<Arc<SignedExecutionPayloadEnvelope<E>>>,
+        seen_timestamp: Duration,
+    },
+
     /// A block with an unknown parent has been received.
     UnknownParentBlock(PeerId, Arc<SignedBeaconBlock<E>>, Hash256),
 
@@ -853,6 +861,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,
+                payload_envelope,
+                seen_timestamp,
+            } => self.rpc_payload_envelope_received(
+                sync_request_id,
+                peer_id,
+                payload_envelope,
+                seen_timestamp,
+            ),
             SyncMessage::UnknownParentBlock(peer_id, block, block_root) => {
                 let block_slot = block.slot();
                 let parent_root = block.parent_root();
@@ -1242,6 +1261,34 @@ impl<T: BeaconChainTypes> SyncManager<T> {
         }
     }
 
+    fn rpc_payload_envelope_received(
+        &mut self,
+        sync_request_id: SyncRequestId,
+        peer_id: PeerId,
+        payload_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(payload_envelope, seen_timestamp),
+                );
+            }
+            SyncRequestId::PayloadEnvelopesByRange(req_id) => {
+                self.on_payload_envelopes_by_range_response(
+                    req_id,
+                    peer_id,
+                    RpcEvent::from_chunk(payload_envelope, seen_timestamp),
+                );
+            }
+            _ => {
+                crit!(%peer_id, "bad request id for payload_envelope");
+            }
+        }
+    }
+
     fn on_single_blob_response(
         &mut self,
         id: SingleLookupReqId,