Range sync: fetch data columns via custody-by-root (#9496)

- Generalize the custody-by-root request to accept a `Vec<Hash256>` of block roots so a whole batch is fetched in one request; `DataColumnsByRootRequestParams { block_roots, indices }` completes at `block_roots.len() * indices.len()`.
- `custody_lookup_request` takes `block_roots: &[Hash256]` + `block_epoch`; `cached_data_column_indexes` takes `block_epoch`.
- `block_sidecar_coupling.rs`: couple a batch's blocks with the columns returned by the single custody-by-root request; drop the per-column request map and retry/faulty-peer machinery.
- Remove the now-unused columns-by-range range-sync path and the serialize-downloads gate.


Co-Authored-By: dapplion <35266934+dapplion@users.noreply.github.com>

beacon_node/lighthouse_network/src/service/api_types.rs

@@ -80,7 +80,6 @@ pub struct DataColumnsByRangeRequestId {
 
 #[derive(Debug, Hash, PartialEq, Eq, Clone, Copy)]
 pub enum DataColumnsByRangeRequester {
-    ComponentsByRange(ComponentsByRangeRequestId),
     CustodyBackfillSync(CustodyBackFillBatchRequestId),
 }
 
@@ -138,10 +137,13 @@ pub struct CustodyId {
     pub requester: CustodyRequester,
 }
 
-/// Downstream components that perform custody by root requests.
-/// Currently, it's only single block lookups, so not using an enum
+/// Downstream components that perform custody by root requests. A range sync request fetches the
+/// custody columns of an entire batch (identified by its `ComponentsByRangeRequestId`) in one go.
 #[derive(Debug, Hash, PartialEq, Eq, Clone, Copy)]
-pub struct CustodyRequester(pub SingleLookupReqId);
+pub enum CustodyRequester {
+    SingleLookup(SingleLookupReqId),
+    RangeSync(ComponentsByRangeRequestId),
+}
 
 /// Application level requests sent to the network.
 #[derive(Debug, Clone, Copy, PartialEq)]
@@ -290,7 +292,10 @@ impl Display for DataColumnsByRootRequester {
 
 impl Display for CustodyRequester {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}", self.0)
+        match self {
+            Self::SingleLookup(id) => write!(f, "{id}"),
+            Self::RangeSync(id) => write!(f, "RangeSync/{id}"),
+        }
     }
 }
 
@@ -306,7 +311,6 @@ impl Display for RangeRequestId {
 impl Display for DataColumnsByRangeRequester {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         match self {
-            Self::ComponentsByRange(id) => write!(f, "ByRange/{id}"),
             Self::CustodyBackfillSync(id) => write!(f, "CustodyBackfill/{id}"),
         }
     }
@@ -321,7 +325,7 @@ mod tests {
         let id = DataColumnsByRootRequestId {
             id: 123,
             requester: DataColumnsByRootRequester::Custody(CustodyId {
-                requester: CustodyRequester(SingleLookupReqId {
+                requester: CustodyRequester::SingleLookup(SingleLookupReqId {
                     req_id: 121,
                     lookup_id: 101,
                 }),
@@ -334,17 +338,17 @@ mod tests {
     fn display_id_data_columns_by_range() {
         let id = DataColumnsByRangeRequestId {
             id: 123,
-            parent_request_id: DataColumnsByRangeRequester::ComponentsByRange(
-                ComponentsByRangeRequestId {
+            parent_request_id: DataColumnsByRangeRequester::CustodyBackfillSync(
+                CustodyBackFillBatchRequestId {
                     id: 122,
-                    requester: RangeRequestId::RangeSync {
-                        chain_id: 54,
-                        batch_id: Epoch::new(0),
+                    batch_id: CustodyBackfillBatchId {
+                        epoch: Epoch::new(0),
+                        run_id: 54,
                     },
                 },
             ),
             peer: PeerId::random(),
         };
-        assert_eq!(format!("{id}"), "123/ByRange/122/RangeSync/0/54");
+        assert_eq!(format!("{id}"), "123/CustodyBackfill/122/0/54");
     }
 }

beacon_node/lighthouse_network/tests/rpc_tests.rs

@@ -9,7 +9,8 @@ use libp2p::PeerId;
 use lighthouse_network::rpc::{RequestType, methods::*};
 use lighthouse_network::service::api_types::{
     AppRequestId, BlobsByRangeRequestId, BlocksByRangeRequestId, ComponentsByRangeRequestId,
-    DataColumnsByRangeRequestId, DataColumnsByRangeRequester, RangeRequestId, SyncRequestId,
+    CustodyBackFillBatchRequestId, CustodyBackfillBatchId, DataColumnsByRangeRequestId,
+    DataColumnsByRangeRequester, RangeRequestId, SyncRequestId,
 };
 use lighthouse_network::{NetworkEvent, ReportSource, Response};
 use ssz::Encode;
@@ -1828,12 +1829,12 @@ fn test_request_too_large_data_columns_by_range() {
         AppRequestId::Sync(SyncRequestId::DataColumnsByRange(
             DataColumnsByRangeRequestId {
                 id: 1,
-                parent_request_id: DataColumnsByRangeRequester::ComponentsByRange(
-                    ComponentsByRangeRequestId {
+                parent_request_id: DataColumnsByRangeRequester::CustodyBackfillSync(
+                    CustodyBackFillBatchRequestId {
                         id: 1,
-                        requester: RangeRequestId::RangeSync {
-                            chain_id: 1,
-                            batch_id: Epoch::new(1),
+                        batch_id: CustodyBackfillBatchId {
+                            epoch: Epoch::new(1),
+                            run_id: 1,
                         },
                     },
                 ),