Wire blocks by range coupling to new request types

dapplion · dapplion · commit 2f2dd9aef03b · 2024-10-17T00:25:18.000+03:00
diff --git a/beacon_node/lighthouse_network/src/rpc/self_limiter.rs b/beacon_node/lighthouse_network/src/rpc/self_limiter.rs
@@ -212,7 +212,7 @@ mod tests {
     use crate::rpc::rate_limiter::Quota;
     use crate::rpc::self_limiter::SelfRateLimiter;
     use crate::rpc::{Ping, Protocol, RequestType};
-    use crate::service::api_types::{AppRequestId, RequestId, SyncRequestId};
+    use crate::service::api_types::{AppRequestId, RequestId, SingleLookupReqId, SyncRequestId};
     use libp2p::PeerId;
     use std::time::Duration;
     use types::MainnetEthSpec;
@@ -228,12 +228,16 @@ mod tests {
         let mut limiter: SelfRateLimiter<RequestId, MainnetEthSpec> =
             SelfRateLimiter::new(config, log).unwrap();
         let peer_id = PeerId::random();
+        let lookup_id = 0;
 
         for i in 1..=5u32 {
             let _ = limiter.allows(
                 peer_id,
-                RequestId::Application(AppRequestId::Sync(SyncRequestId::RangeBlockAndBlobs {
-                    id: i,
+                RequestId::Application(AppRequestId::Sync(SyncRequestId::SingleBlock {
+                    id: SingleLookupReqId {
+                        lookup_id,
+                        req_id: i,
+                    },
                 })),
                 RequestType::Ping(Ping { data: i as u64 }),
             );
@@ -251,9 +255,9 @@ mod tests {
             for i in 2..=5u32 {
                 assert!(matches!(
                     iter.next().unwrap().request_id,
-                    RequestId::Application(AppRequestId::Sync(SyncRequestId::RangeBlockAndBlobs {
-                        id,
-                    })) if id == i
+                    RequestId::Application(AppRequestId::Sync(SyncRequestId::SingleBlock {
+                        id: SingleLookupReqId { req_id, .. },
+                    })) if req_id == i,
                 ));
             }
 
@@ -276,9 +280,9 @@ mod tests {
             for i in 3..=5 {
                 assert!(matches!(
                     iter.next().unwrap().request_id,
-                    RequestId::Application(AppRequestId::Sync(SyncRequestId::RangeBlockAndBlobs {
-                        id
-                    })) if id == i
+                    RequestId::Application(AppRequestId::Sync(SyncRequestId::SingleBlock {
+                        id: SingleLookupReqId { req_id, .. },
+                    })) if req_id == i,
                 ));
             }
 
diff --git a/beacon_node/lighthouse_network/src/service/api_types.rs b/beacon_node/lighthouse_network/src/service/api_types.rs
@@ -31,8 +31,6 @@ pub enum SyncRequestId {
     SingleBlob { id: SingleLookupReqId },
     /// Request searching for a set of data columns given a hash and list of column indices.
     DataColumnsByRoot(DataColumnsByRootRequestId),
-    /// Range request that is composed by both a block range request and a blob range request.
-    RangeBlockAndBlobs { id: Id },
     /// Blocks by range request
     BlocksByRange(BlocksByRangeRequestId),
     /// Blobs by range request
@@ -52,17 +50,25 @@ pub struct DataColumnsByRootRequestId {
 #[derive(Debug, Hash, PartialEq, Eq, Clone, Copy)]
 pub struct BlocksByRangeRequestId {
     pub id: Id,
-    pub requester: RangeRequestId,
+    pub requester: ComponentsByRangeRequestId,
 }
 
 #[derive(Debug, Hash, PartialEq, Eq, Clone, Copy)]
 pub struct BlobsByRangeRequestId {
     pub id: Id,
-    pub requester: RangeRequestId,
+    pub requester: ComponentsByRangeRequestId,
 }
 
 #[derive(Debug, Hash, PartialEq, Eq, Clone, Copy)]
 pub struct DataColumnsByRangeRequestId {
+    pub id: Id,
+    pub requester: ComponentsByRangeRequestId,
+}
+
+/// Block components by range request for range sync. Includes an ID for downstream consumers to
+/// handle retries and tie all their sub requests together.
+#[derive(Debug, Hash, PartialEq, Eq, Clone, Copy)]
+pub struct ComponentsByRangeRequestId {
     pub id: Id,
     pub requester: RangeRequestId,
 }
diff --git a/beacon_node/network/src/router.rs b/beacon_node/network/src/router.rs
@@ -600,7 +600,7 @@ impl<T: BeaconChainTypes> Router<T> {
     ) {
         let request_id = match request_id {
             AppRequestId::Sync(sync_id) => match sync_id {
-                id @ SyncRequestId::RangeBlockAndBlobs { .. } => id,
+                id @ SyncRequestId::BlocksByRange { .. } => id,
                 other => {
                     crit!(self.log, "BlocksByRange response on incorrect request"; "request" => ?other);
                     return;
diff --git a/beacon_node/network/src/sync/block_sidecar_coupling.rs b/beacon_node/network/src/sync/block_sidecar_coupling.rs
@@ -1,7 +1,6 @@
 use beacon_chain::{
     block_verification_types::RpcBlock, data_column_verification::CustodyDataColumn, get_block_root,
 };
-use lighthouse_network::PeerId;
 use ssz_types::VariableList;
 use std::{
     collections::{HashMap, VecDeque},
@@ -29,16 +28,13 @@ pub struct RangeBlockComponentsRequest<E: EthSpec> {
     /// Used to determine if the number of data columns stream termination this accumulator should
     /// wait for. This may be less than the number of `expects_custody_columns` due to request batching.
     num_custody_column_requests: Option<usize>,
-    /// The peers the request was made to.
-    pub(crate) peer_ids: Vec<PeerId>,
 }
 
 impl<E: EthSpec> RangeBlockComponentsRequest<E> {
     pub fn new(
         expects_blobs: bool,
         expects_custody_columns: Option<Vec<ColumnIndex>>,
         num_custody_column_requests: Option<usize>,
-        peer_ids: Vec<PeerId>,
     ) -> Self {
         Self {
             blocks: <_>::default(),
@@ -50,38 +46,31 @@ impl<E: EthSpec> RangeBlockComponentsRequest<E> {
             expects_blobs,
             expects_custody_columns,
             num_custody_column_requests,
-            peer_ids,
         }
     }
 
-    // TODO: This function should be deprecated when simplying the retry mechanism of this range
-    // requests.
-    pub fn get_requirements(&self) -> (bool, Option<Vec<ColumnIndex>>) {
-        (self.expects_blobs, self.expects_custody_columns.clone())
-    }
-
-    pub fn add_block_response(&mut self, block_opt: Option<Arc<SignedBeaconBlock<E>>>) {
-        match block_opt {
-            Some(block) => self.blocks.push_back(block),
-            None => self.is_blocks_stream_terminated = true,
+    pub fn add_blocks(&mut self, blocks: Vec<Arc<SignedBeaconBlock<E>>>) {
+        for block in blocks {
+            self.blocks.push_back(block);
         }
+        self.is_blocks_stream_terminated = true;
     }
 
-    pub fn add_sidecar_response(&mut self, sidecar_opt: Option<Arc<BlobSidecar<E>>>) {
-        match sidecar_opt {
-            Some(sidecar) => self.blobs.push_back(sidecar),
-            None => self.is_sidecars_stream_terminated = true,
+    pub fn add_blobs(&mut self, blobs: Vec<Arc<BlobSidecar<E>>>) {
+        for blob in blobs {
+            self.blobs.push_back(blob);
         }
+        self.is_sidecars_stream_terminated = true;
     }
 
-    pub fn add_data_column(&mut self, column_opt: Option<Arc<DataColumnSidecar<E>>>) {
-        match column_opt {
-            Some(column) => self.data_columns.push_back(column),
-            // TODO(das): this mechanism is dangerous, if somehow there are two requests for the
-            // same column index it can terminate early. This struct should track that all requests
-            // for all custody columns terminate.
-            None => self.custody_columns_streams_terminated += 1,
+    pub fn add_custody_columns(&mut self, columns: Vec<Arc<DataColumnSidecar<E>>>) {
+        for column in columns {
+            self.data_columns.push_back(column);
         }
+        // TODO(das): this mechanism is dangerous, if somehow there are two requests for the
+        // same column index it can terminate early. This struct should track that all requests
+        // for all custody columns terminate.
+        self.custody_columns_streams_terminated += 1;
     }
 
     pub fn into_responses(self, spec: &ChainSpec) -> Result<Vec<RpcBlock<E>>, String> {
@@ -249,14 +238,15 @@ mod tests {
         let mut info = RangeBlockComponentsRequest::<E>::new(false, None, None, vec![peer_id]);
         let mut rng = XorShiftRng::from_seed([42; 16]);
         let blocks = (0..4)
-            .map(|_| generate_rand_block_and_blobs::<E>(ForkName::Base, NumBlobs::None, &mut rng).0)
+            .map(|_| {
+                generate_rand_block_and_blobs::<E>(ForkName::Base, NumBlobs::None, &mut rng)
+                    .0
+                    .into()
+            })
             .collect::<Vec<_>>();
 
         // Send blocks and complete terminate response
-        for block in blocks {
-            info.add_block_response(Some(block.into()));
-        }
-        info.add_block_response(None);
+        info.add_blocks(blocks);
 
         // Assert response is finished and RpcBlocks can be constructed
         assert!(info.is_finished());
@@ -271,17 +261,16 @@ mod tests {
         let blocks = (0..4)
             .map(|_| {
                 // Always generate some blobs.
-                generate_rand_block_and_blobs::<E>(ForkName::Deneb, NumBlobs::Number(3), &mut rng).0
+                generate_rand_block_and_blobs::<E>(ForkName::Deneb, NumBlobs::Number(3), &mut rng)
+                    .0
+                    .into()
             })
             .collect::<Vec<_>>();
 
         // Send blocks and complete terminate response
-        for block in blocks {
-            info.add_block_response(Some(block.into()));
-        }
-        info.add_block_response(None);
+        info.add_blocks(blocks);
         // Expect no blobs returned
-        info.add_sidecar_response(None);
+        info.add_blobs(vec![]);
 
         // Assert response is finished and RpcBlocks can be constructed, even if blobs weren't returned.
         // This makes sure we don't expect blobs here when they have expired. Checking this logic should
@@ -294,10 +283,11 @@ mod tests {
     fn rpc_block_with_custody_columns() {
         let spec = test_spec::<E>();
         let expects_custody_columns = vec![1, 2, 3, 4];
+        let custody_column_request_ids = vec![0, 1, 2, 3];
         let mut info = RangeBlockComponentsRequest::<E>::new(
             false,
             Some(expects_custody_columns.clone()),
-            Some(expects_custody_columns.len()),
+            Some(custody_column_request_ids.len()),
             vec![PeerId::random()],
         );
         let mut rng = XorShiftRng::from_seed([42; 16]);
@@ -313,25 +303,18 @@ mod tests {
             .collect::<Vec<_>>();
 
         // Send blocks and complete terminate response
-        for block in &blocks {
-            info.add_block_response(Some(block.0.clone().into()));
-        }
-        info.add_block_response(None);
+        info.add_blocks(blocks.iter().map(|b| b.0.clone().into()).collect());
         // Assert response is not finished
         assert!(!info.is_finished());
 
-        // Send data columns interleaved
-        for block in &blocks {
-            for column in &block.1 {
-                if expects_custody_columns.contains(&column.index) {
-                    info.add_data_column(Some(column.clone()));
-                }
-            }
-        }
-
-        // Terminate the requests
-        for (i, _column_index) in expects_custody_columns.iter().enumerate() {
-            info.add_data_column(None);
+        // Send data columns
+        for (i, &column_index) in expects_custody_columns.iter().enumerate() {
+            info.add_custody_columns(
+                blocks
+                    .iter()
+                    .flat_map(|b| b.1.iter().filter(|d| d.index == column_index).cloned())
+                    .collect(),
+            );
 
             if i < expects_custody_columns.len() - 1 {
                 assert!(
@@ -353,12 +336,19 @@ mod tests {
     #[test]
     fn rpc_block_with_custody_columns_batched() {
         let spec = test_spec::<E>();
-        let expects_custody_columns = vec![1, 2, 3, 4];
-        let num_of_data_column_requests = 2;
+        let batched_column_requests = vec![vec![1_u64, 2], vec![3, 4]];
+        let expects_custody_columns = batched_column_requests
+            .iter()
+            .cloned()
+            .flatten()
+            .collect::<Vec<_>>();
+        let custody_column_request_ids =
+            (0..batched_column_requests.len() as u32).collect::<Vec<_>>();
+        let num_of_data_column_requests = custody_column_request_ids.len();
         let mut info = RangeBlockComponentsRequest::<E>::new(
             false,
             Some(expects_custody_columns.clone()),
-            Some(num_of_data_column_requests),
+            Some(custody_column_request_ids.len()),
             vec![PeerId::random()],
         );
         let mut rng = XorShiftRng::from_seed([42; 16]);
@@ -374,25 +364,23 @@ mod tests {
             .collect::<Vec<_>>();
 
         // Send blocks and complete terminate response
-        for block in &blocks {
-            info.add_block_response(Some(block.0.clone().into()));
-        }
-        info.add_block_response(None);
+        info.add_blocks(blocks.iter().map(|b| b.0.clone().into()).collect());
         // Assert response is not finished
         assert!(!info.is_finished());
 
-        // Send data columns interleaved
-        for block in &blocks {
-            for column in &block.1 {
-                if expects_custody_columns.contains(&column.index) {
-                    info.add_data_column(Some(column.clone()));
-                }
-            }
-        }
+        for (i, column_indices) in batched_column_requests.iter().enumerate() {
+            // Send the set of columns in the same batch request
+            info.add_custody_columns(
+                blocks
+                    .iter()
+                    .flat_map(|b| {
+                        b.1.iter()
+                            .filter(|d| column_indices.contains(&d.index))
+                            .cloned()
+                    })
+                    .collect::<Vec<_>>(),
+            );
 
-        // Terminate the requests
-        for i in 0..num_of_data_column_requests {
-            info.add_data_column(None);
             if i < num_of_data_column_requests - 1 {
                 assert!(
                     !info.is_finished(),
diff --git a/beacon_node/network/src/sync/manager.rs b/beacon_node/network/src/sync/manager.rs
diff --git a/beacon_node/network/src/sync/network_context.rs b/beacon_node/network/src/sync/network_context.rs
