Fix merge issues

Author: dapplion

beacon_node/lighthouse_network/src/service/api_types.rs

@@ -85,7 +85,7 @@ pub struct ComponentsByRangeRequestId {
 /// Range sync chain or backfill batch
 #[derive(Debug, Hash, PartialEq, Eq, Clone, Copy)]
 pub enum RangeRequestId {
-    RangeSync { chain_id: u64, batch_id: Epoch },
+    RangeSync { chain_id: Id, batch_id: Epoch },
     BackfillSync { batch_id: Epoch },
 }
 

beacon_node/network/src/sync/network_context.rs

@@ -263,6 +263,7 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
             execution_engine_state: _,
             network_beacon_processor: _,
             chain: _,
+            fork_context: _,
             log: _,
         } = self;
 
@@ -382,26 +383,15 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
             if matches!(batch_type, ByRangeRequestType::BlocksAndColumns) {
                 let column_indexes = self.network_globals().sampling_columns.clone();
 
-                for (peer_id, columns_by_range_request) in
-                    self.make_columns_by_range_requests(request, &column_indexes)?
-                {
-                    requested_peers.push(peer_id);
-
-                    debug!(
-                        self.log,
-                        "Sending DataColumnsByRange requests";
-                        "method" => "DataColumnsByRange",
-                        "count" => columns_by_range_request.count,
-                        "epoch" => epoch,
-                        "columns" => ?columns_by_range_request.columns,
-                        "peer" => %peer_id,
-                        "id" => id,
-                    );
-
-                    self.send_network_msg(NetworkMessage::SendRequest {
-                        peer_id,
-                        request: RequestType::DataColumnsByRange(columns_by_range_request),
-                        request_id: AppRequestId::Sync(SyncRequestId::RangeBlockAndBlobs { id }),
+                let data_column_requests = self
+                    .make_columns_by_range_requests(request, &column_indexes)?
+                    .into_iter()
+                    .map(|(peer_id, columns_by_range_request)| {
+                        self.send_data_columns_by_range_request(
+                            peer_id,
+                            columns_by_range_request,
+                            id,
+                        )
                     })
                     .collect::<Result<Vec<_>, _>>()?;
 

beacon_node/network/src/sync/tests/range.rs

@@ -12,7 +12,8 @@ use lighthouse_network::rpc::methods::{
 };
 use lighthouse_network::rpc::{RequestType, StatusMessage};
 use lighthouse_network::service::api_types::{
-    AppRequestId, BlobsByRangeRequestId, BlocksByRangeRequestId, SyncRequestId,
+    AppRequestId, BlobsByRangeRequestId, BlocksByRangeRequestId, DataColumnsByRangeRequestId,
+    SyncRequestId,
 };
 use lighthouse_network::{PeerId, SyncInfo};
 use std::time::Duration;
@@ -30,8 +31,8 @@ pub(crate) enum DataSidecars<E: EthSpec> {
 
 enum ByRangeDataRequestIds {
     PreDeneb,
-    PrePeerDAS(Id, PeerId),
-    PostPeerDAS(Vec<(Id, PeerId)>),
+    PrePeerDAS(BlobsByRangeRequestId, PeerId),
+    PostPeerDAS(Vec<(DataColumnsByRangeRequestId, PeerId)>),
 }
 
 /// Sync tests are usually written in the form:
@@ -152,10 +153,8 @@ impl TestRig {
 
     fn find_blocks_by_range_request(
         &mut self,
-        target_peer_id: &PeerId,
-    ) -> (BlocksByRangeRequestId, Option<BlobsByRangeRequestId>) {
         request_filter: RequestFilter,
-    ) -> ((Id, PeerId), ByRangeDataRequestIds) {
+    ) -> ((BlocksByRangeRequestId, PeerId), ByRangeDataRequestIds) {
         let filter_f = |peer: PeerId, start_slot: u64| {
             if let Some(expected_epoch) = request_filter.epoch {
                 let epoch = Slot::new(start_slot).epoch(E::slots_per_epoch()).as_u64();
@@ -194,7 +193,7 @@ impl TestRig {
                         RequestType::DataColumnsByRange(DataColumnsByRangeRequest {
                             start_slot, ..
                         }),
-                    request_id: AppRequestId::Sync(SyncRequestId::RangeBlockAndBlobs { id }),
+                    request_id: AppRequestId::Sync(SyncRequestId::DataColumnsByRange(id)),
                 } if filter_f(*peer_id, *start_slot) => Some((*id, *peer_id)),
                 _ => None,
             }) {
@@ -243,10 +242,10 @@ impl TestRig {
             ByRangeDataRequestIds::PrePeerDAS(id, peer_id) => {
                 // Complete the request with a single stream termination
                 self.log(&format!(
-                    "Completing BlobsByRange request {id} with empty stream"
+                    "Completing BlobsByRange request {id:?} with empty stream"
                 ));
                 self.send_sync_message(SyncMessage::RpcBlob {
-                request_id: SyncRequestId::BlobsByRange(blobs_req_id),
+                    request_id: SyncRequestId::BlobsByRange(id),
                     peer_id,
                     blob_sidecar: None,
                     seen_timestamp: D,
@@ -256,10 +255,10 @@ impl TestRig {
                 // Complete the request with a single stream termination
                 for (id, peer_id) in data_column_req_ids {
                     self.log(&format!(
-                        "Completing DataColumnsByRange request {id} with empty stream"
+                        "Completing DataColumnsByRange request {id:?} with empty stream"
                     ));
                     self.send_sync_message(SyncMessage::RpcDataColumn {
-                        request_id: SyncRequestId::DataColumnsByRange( id ),
+                        request_id: SyncRequestId::DataColumnsByRange(id),
                         peer_id,
                         data_column: None,
                         seen_timestamp: D,