Rework internal rpc protocol handling (#4290)

## Issue Addressed

Resolves #3980. Builds on work by @GeemoCandama in #4084 

## Proposed Changes

Extends the `SupportedProtocol` abstraction added in Geemo's PR and attempts to fix internal versioning of requests that are mentioned in this comment https://github.com/sigp/lighthouse/pull/4084#issuecomment-1496380033 

Co-authored-by: geemo <geemo@tutanota.com>

beacon_node/network/src/beacon_processor/worker/rpc_methods.rs

@@ -131,10 +131,10 @@ impl<T: BeaconChainTypes> Worker<T> {
         request_id: PeerRequestId,
         request: BlocksByRootRequest,
     ) {
-        let requested_blocks = request.block_roots.len();
+        let requested_blocks = request.block_roots().len();
         let mut block_stream = match self
             .chain
-            .get_blocks_checking_early_attester_cache(request.block_roots.into(), &executor)
+            .get_blocks_checking_early_attester_cache(request.block_roots().to_vec(), &executor)
         {
             Ok(block_stream) => block_stream,
             Err(e) => return error!(self.log, "Error getting block stream"; "error" => ?e),
@@ -292,18 +292,18 @@ impl<T: BeaconChainTypes> Worker<T> {
     ) {
         debug!(self.log, "Received BlocksByRange Request";
             "peer_id" => %peer_id,
-            "count" => req.count,
-            "start_slot" => req.start_slot,
+            "count" => req.count(),
+            "start_slot" => req.start_slot(),
         );
 
         // Should not send more than max request blocks
-        if req.count > MAX_REQUEST_BLOCKS {
-            req.count = MAX_REQUEST_BLOCKS;
+        if *req.count() > MAX_REQUEST_BLOCKS {
+            *req.count_mut() = MAX_REQUEST_BLOCKS;
         }
 
         let forwards_block_root_iter = match self
             .chain
-            .forwards_iter_block_roots(Slot::from(req.start_slot))
+            .forwards_iter_block_roots(Slot::from(*req.start_slot()))
         {
             Ok(iter) => iter,
             Err(BeaconChainError::HistoricalBlockError(
@@ -326,18 +326,20 @@ impl<T: BeaconChainTypes> Worker<T> {
         // Pick out the required blocks, ignoring skip-slots.
         let mut last_block_root = None;
         let maybe_block_roots = process_results(forwards_block_root_iter, |iter| {
-            iter.take_while(|(_, slot)| slot.as_u64() < req.start_slot.saturating_add(req.count))
-                // map skip slots to None
-                .map(|(root, _)| {
-                    let result = if Some(root) == last_block_root {
-                        None
-                    } else {
-                        Some(root)
-                    };
-                    last_block_root = Some(root);
-                    result
-                })
-                .collect::<Vec<Option<Hash256>>>()
+            iter.take_while(|(_, slot)| {
+                slot.as_u64() < req.start_slot().saturating_add(*req.count())
+            })
+            // map skip slots to None
+            .map(|(root, _)| {
+                let result = if Some(root) == last_block_root {
+                    None
+                } else {
+                    Some(root)
+                };
+                last_block_root = Some(root);
+                result
+            })
+            .collect::<Vec<Option<Hash256>>>()
         });
 
         let block_roots = match maybe_block_roots {
@@ -364,8 +366,8 @@ impl<T: BeaconChainTypes> Worker<T> {
                         Ok(Some(block)) => {
                             // Due to skip slots, blocks could be out of the range, we ensure they
                             // are in the range before sending
-                            if block.slot() >= req.start_slot
-                                && block.slot() < req.start_slot + req.count
+                            if block.slot() >= *req.start_slot()
+                                && block.slot() < req.start_slot() + req.count()
                             {
                                 blocks_sent += 1;
                                 self.send_network_message(NetworkMessage::SendResponse {
@@ -440,15 +442,15 @@ impl<T: BeaconChainTypes> Worker<T> {
                     .slot()
                     .unwrap_or_else(|_| self.chain.slot_clock.genesis_slot());
 
-                if blocks_sent < (req.count as usize) {
+                if blocks_sent < (*req.count() as usize) {
                     debug!(
                         self.log,
                         "BlocksByRange outgoing response processed";
                         "peer" => %peer_id,
                         "msg" => "Failed to return all requested blocks",
-                        "start_slot" => req.start_slot,
+                        "start_slot" => req.start_slot(),
                         "current_slot" => current_slot,
-                        "requested" => req.count,
+                        "requested" => req.count(),
                         "returned" => blocks_sent
                     );
                 } else {
@@ -456,9 +458,9 @@ impl<T: BeaconChainTypes> Worker<T> {
                         self.log,
                         "BlocksByRange outgoing response processed";
                         "peer" => %peer_id,
-                        "start_slot" => req.start_slot,
+                        "start_slot" => req.start_slot(),
                         "current_slot" => current_slot,
-                        "requested" => req.count,
+                        "requested" => req.count(),
                         "returned" => blocks_sent
                     );
                 }

beacon_node/network/src/sync/block_lookups/single_block_lookup.rs

@@ -156,9 +156,7 @@ impl<const MAX_ATTEMPTS: u8> SingleBlockRequest<MAX_ATTEMPTS> {
                 cannot_process: self.failed_processing >= self.failed_downloading,
             })
         } else if let Some(&peer_id) = self.available_peers.iter().choose(&mut rand::thread_rng()) {
-            let request = BlocksByRootRequest {
-                block_roots: VariableList::from(vec![self.hash]),
-            };
+            let request = BlocksByRootRequest::new(VariableList::from(vec![self.hash]));
             self.state = State::Downloading { peer_id };
             self.used_peers.insert(peer_id);
             Ok((peer_id, request))

beacon_node/network/src/sync/network_context.rs

@@ -112,7 +112,7 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
             self.log,
             "Sending BlocksByRange Request";
             "method" => "BlocksByRange",
-            "count" => request.count,
+            "count" => request.count(),
             "peer" => %peer_id,
         );
         let request = Request::BlocksByRange(request);
@@ -138,7 +138,7 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
             self.log,
             "Sending backfill BlocksByRange Request";
             "method" => "BlocksByRange",
-            "count" => request.count,
+            "count" => request.count(),
             "peer" => %peer_id,
         );
         let request = Request::BlocksByRange(request);
@@ -185,7 +185,7 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
             self.log,
             "Sending BlocksByRoot Request";
             "method" => "BlocksByRoot",
-            "count" => request.block_roots.len(),
+            "count" => request.block_roots().len(),
             "peer" => %peer_id
         );
         let request = Request::BlocksByRoot(request);
@@ -209,7 +209,7 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
             self.log,
             "Sending BlocksByRoot Request";
             "method" => "BlocksByRoot",
-            "count" => request.block_roots.len(),
+            "count" => request.block_roots().len(),
             "peer" => %peer_id
         );
         let request = Request::BlocksByRoot(request);

beacon_node/network/src/sync/range_sync/batch.rs

@@ -202,10 +202,10 @@ impl<T: EthSpec, B: BatchConfig> BatchInfo<T, B> {
 
     /// Returns a BlocksByRange request associated with the batch.
     pub fn to_blocks_by_range_request(&self) -> BlocksByRangeRequest {
-        BlocksByRangeRequest {
-            start_slot: self.start_slot.into(),
-            count: self.end_slot.sub(self.start_slot).into(),
-        }
+        BlocksByRangeRequest::new(
+            self.start_slot.into(),
+            self.end_slot.sub(self.start_slot).into(),
+        )
     }
 
     /// After different operations over a batch, this could be in a state that allows it to