Add da router, and initial logic

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

@@ -172,7 +172,7 @@ impl<T: BeaconChainTypes> RequestState<T> for CustodyRequestState<T::EthSpec> {
         _: usize,
         cx: &mut SyncNetworkContext<T>,
     ) -> Result<LookupRequestResult, LookupRequestError> {
-        cx.custody_lookup_request(id, self.block_root, lookup_peers)
+        cx.custody_lookup_request(id, self.slot, self.block_root, lookup_peers)
             .map_err(LookupRequestError::SendFailedNetwork)
     }
 

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

@@ -233,7 +233,7 @@ impl<T: BeaconChainTypes> SingleBlockLookup<T> {
                     );
                 } else if cx.chain.should_fetch_custody_columns(block_epoch) {
                     self.component_requests = ComponentRequests::ActiveCustodyRequest(
-                        CustodyRequestState::new(self.block_root),
+                        CustodyRequestState::new(block.slot(), self.block_root),
                     );
                 } else {
                     self.component_requests = ComponentRequests::NotNeeded("outside da window");
@@ -391,13 +391,15 @@ impl<E: EthSpec> BlobRequestState<E> {
 pub struct CustodyRequestState<E: EthSpec> {
     #[educe(Debug(ignore))]
     pub block_root: Hash256,
+    pub slot: Slot,
     pub state: SingleLookupRequestState<DataColumnSidecarList<E>>,
 }
 
 impl<E: EthSpec> CustodyRequestState<E> {
-    pub fn new(block_root: Hash256) -> Self {
+    pub fn new(slot: Slot, block_root: Hash256) -> Self {
         Self {
             block_root,
+            slot,
             state: SingleLookupRequestState::new(),
         }
     }

beacon_node/network/src/sync/network_context.rs

@@ -964,6 +964,7 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
         let imported_blob_indexes = self
             .chain
             .data_availability_checker
+            .v1()
             .cached_blob_indexes(&block_root)
             .unwrap_or_default();
         // Include only the blob indexes not yet imported (received through gossip)
@@ -1078,13 +1079,14 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
     pub fn custody_lookup_request(
         &mut self,
         lookup_id: SingleLookupId,
+        slot: Slot,
         block_root: Hash256,
         lookup_peers: Arc<RwLock<HashSet<PeerId>>>,
     ) -> Result<LookupRequestResult, RpcRequestSendError> {
         let custody_indexes_imported = self
             .chain
             .data_availability_checker
-            .cached_data_column_indexes(&block_root)
+            .cached_data_column_indexes(&block_root, slot)
             .unwrap_or_default();
 
         let current_epoch = self.chain.epoch().map_err(|e| {
@@ -1366,12 +1368,14 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
         if self
             .chain
             .data_availability_checker
+            .v1()
             .data_columns_required_for_epoch(epoch)
         {
             ByRangeRequestType::BlocksAndColumns
         } else if self
             .chain
             .data_availability_checker
+            .v1()
             .blobs_required_for_epoch(epoch)
         {
             ByRangeRequestType::BlocksAndBlobs

beacon_node/network/src/sync/network_context/requests/data_columns_by_root.rs

@@ -26,6 +26,7 @@ impl DataColumnsByRootSingleBlockRequest {
                 block_root: self.block_root,
                 columns,
             }],
+            fork_name,
             spec.max_request_blocks(fork_name),
         )
     }

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

@@ -1079,6 +1079,7 @@ impl TestRig {
             .harness
             .chain
             .data_availability_checker
+            .v1()
             .put_executed_block(executed_block)
             .unwrap()
         {
@@ -1094,6 +1095,7 @@ impl TestRig {
             .harness
             .chain
             .data_availability_checker
+            .v1()
             .put_gossip_verified_blobs(
                 blob.block_root(),
                 std::iter::once(GossipVerifiedBlob::<_, Observe>::__assumed_valid(
@@ -1113,6 +1115,7 @@ impl TestRig {
         self.harness
             .chain
             .data_availability_checker
+            .v1()
             .put_pre_execution_block(block.canonical_root(), block, BlockImportSource::Gossip)
             .unwrap();
     }
@@ -1121,6 +1124,7 @@ impl TestRig {
         self.harness
             .chain
             .data_availability_checker
+            .v1()
             .remove_block_on_execution_error(&block_root);
 
         self.send_sync_message(SyncMessage::GossipBlockProcessResult {