Forward sync columns by root

beacon_node/lighthouse_network/src/peer_manager/peerdb.rs

@@ -326,6 +326,36 @@ impl<E: EthSpec> PeerDB<E> {
             .map(|(peer_id, _)| peer_id)
     }
 
+    /// Returns an iterator of all good gossipsub peers that are supposed to be custodying
+    /// the given subnet id and have the epoch according to their status messages.
+    pub fn good_custody_subnet_peer_range_sync(
+        &self,
+        subnet: DataColumnSubnetId,
+        epoch: Epoch,
+    ) -> impl Iterator<Item = &PeerId> {
+        self.peers
+            .iter()
+            .filter(move |(_, info)| {
+                // The custody_subnets hashset can be populated via enr or metadata
+                let is_custody_subnet_peer = info.is_assigned_to_custody_subnet(&subnet);
+
+                info.is_connected()
+                    && is_custody_subnet_peer
+                    && match info.sync_status() {
+                        SyncStatus::Synced { info } => {
+                            info.has_slot(epoch.end_slot(E::slots_per_epoch()))
+                        }
+                        SyncStatus::Advanced { info } => {
+                            info.has_slot(epoch.end_slot(E::slots_per_epoch()))
+                        }
+                        SyncStatus::IrrelevantPeer
+                        | SyncStatus::Behind { .. }
+                        | SyncStatus::Unknown => false,
+                    }
+            })
+            .map(|(peer_id, _)| peer_id)
+    }
+
     /// Checks if there is at least one good peer for each specified custody subnet for the given epoch.
     /// A "good" peer is one that is both connected and synced (or advanced) for the specified epoch.
     pub fn has_good_custody_range_sync_peer(

beacon_node/lighthouse_network/src/service/api_types.rs

@@ -9,6 +9,12 @@ use types::{
 
 pub type Id = u32;
 
+#[derive(Debug, Hash, PartialEq, Eq, Clone, Copy)]
+pub enum RangeRequestType {
+    ForwardSync,
+    BackfillSync,
+}
+
 #[derive(Debug, Hash, PartialEq, Eq, Clone, Copy)]
 pub struct SingleLookupReqId {
     pub lookup_id: Id,
@@ -38,6 +44,7 @@ pub enum SyncRequestId {
 pub struct DataColumnsByRootRequestId {
     pub id: Id,
     pub requester: DataColumnsByRootRequester,
+    pub peer: PeerId,
 }
 
 #[derive(Debug, Hash, PartialEq, Eq, Clone, Copy)]
@@ -46,6 +53,20 @@ pub struct BlocksByRangeRequestId {
     pub id: Id,
     /// The Id of the overall By Range request for block components.
     pub parent_request_id: ComponentsByRangeRequestId,
+    /// The peer that we made this request to
+    pub peer_id: PeerId,
+}
+
+impl BlocksByRangeRequestId {
+    pub fn batch_id(&self) -> Epoch {
+        match self.parent_request_id.requester {
+            RangeRequestId::BackfillSync { batch_id } => batch_id,
+            RangeRequestId::RangeSync {
+                chain_id: _,
+                batch_id,
+            } => batch_id,
+        }
+    }
 }
 
 #[derive(Debug, Hash, PartialEq, Eq, Clone, Copy)]
@@ -86,12 +107,31 @@ pub enum RangeRequestId {
     RangeSync { chain_id: Id, batch_id: Epoch },
     BackfillSync { batch_id: Epoch },
 }
+impl RangeRequestId {
+    pub fn batch_id(&self) -> Epoch {
+        match &self {
+            RangeRequestId::BackfillSync { batch_id } => *batch_id,
+            RangeRequestId::RangeSync {
+                chain_id: _,
+                batch_id,
+            } => *batch_id,
+        }
+    }
+
+    pub fn batch_type(&self) -> RangeRequestType {
+        match &self {
+            RangeRequestId::BackfillSync { .. } => RangeRequestType::BackfillSync,
+            RangeRequestId::RangeSync { .. } => RangeRequestType::ForwardSync,
+        }
+    }
+}
 
 // TODO(das) refactor in a separate PR. We might be able to remove this and replace
 // [`DataColumnsByRootRequestId`] with a [`SingleLookupReqId`].
 #[derive(Debug, Hash, PartialEq, Eq, Clone, Copy)]
 pub enum DataColumnsByRootRequester {
     Custody(CustodyId),
+    RangeSync { parent: ComponentsByRangeRequestId },
 }
 
 #[derive(Debug, Hash, PartialEq, Eq, Clone, Copy)]
@@ -222,6 +262,7 @@ impl Display for DataColumnsByRootRequester {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         match self {
             Self::Custody(id) => write!(f, "Custody/{id}"),
+            Self::RangeSync { parent } => write!(f, "Range/{parent}"),
         }
     }
 }
@@ -255,6 +296,7 @@ mod tests {
                     lookup_id: 101,
                 }),
             }),
+            peer: PeerId::random(),
         };
         assert_eq!(format!("{id}"), "123/Custody/121/Lookup/101");
     }

beacon_node/network/src/metrics.rs

@@ -484,6 +484,12 @@ pub static SYNC_ACTIVE_NETWORK_REQUESTS: LazyLock<Result<IntGaugeVec>> = LazyLoc
         &["type"],
     )
 });
+pub static SYNC_PENDING_ROOT_RANGE_REQUESTS: LazyLock<Result<IntGauge>> = LazyLock::new(|| {
+    try_create_int_gauge(
+        "sync_pending_root_range_requests",
+        "Current count of pending columns by root requests waiting for peers",
+    )
+});
 pub static SYNC_UNKNOWN_NETWORK_REQUESTS: LazyLock<Result<IntCounterVec>> = LazyLock::new(|| {
     try_create_int_counter_vec(
         "sync_unknwon_network_request",