From 2733cb91609270cdde8c78ce4387dc76336dcb50 Mon Sep 17 00:00:00 2001 From: dapplion <35266934+dapplion@users.noreply.github.com> Date: Thu, 11 Jun 2026 18:14:47 +0200 Subject: [PATCH 1/7] Gate sync peer selection on per-protocol concurrent-request limit Skip peers already at MAX_CONCURRENT_REQUESTS for the protocol being requested and break remaining ties at random, instead of sorting by a peer's total in-flight request count across all protocols. --- beacon_node/lighthouse_network/src/rpc/mod.rs | 2 +- .../network/src/sync/network_context.rs | 119 +++++++----------- .../src/sync/network_context/custody.rs | 21 ++-- 3 files changed, 58 insertions(+), 84 deletions(-) diff --git a/beacon_node/lighthouse_network/src/rpc/mod.rs b/beacon_node/lighthouse_network/src/rpc/mod.rs index 7c43018af83..e9177586cbd 100644 --- a/beacon_node/lighthouse_network/src/rpc/mod.rs +++ b/beacon_node/lighthouse_network/src/rpc/mod.rs @@ -46,7 +46,7 @@ mod response_limiter; mod self_limiter; // Maximum number of concurrent requests per protocol ID that a client may issue. -const MAX_CONCURRENT_REQUESTS: usize = 2; +pub const MAX_CONCURRENT_REQUESTS: usize = 2; /// Composite trait for a request id. pub trait ReqId: Send + 'static + std::fmt::Debug + Copy + Clone {} diff --git a/beacon_node/network/src/sync/network_context.rs b/beacon_node/network/src/sync/network_context.rs index 8e8abd4fa63..0dda140c2d6 100644 --- a/beacon_node/network/src/sync/network_context.rs +++ b/beacon_node/network/src/sync/network_context.rs @@ -27,7 +27,9 @@ use fnv::FnvHashMap; use lighthouse_network::rpc::methods::{ BlobsByRangeRequest, DataColumnsByRangeRequest, PayloadEnvelopesByRangeRequest, }; -use lighthouse_network::rpc::{BlocksByRangeRequest, GoodbyeReason, RPCError, RequestType}; +use lighthouse_network::rpc::{ + BlocksByRangeRequest, GoodbyeReason, MAX_CONCURRENT_REQUESTS, RPCError, RequestType, +}; pub use lighthouse_network::service::api_types::RangeRequestId; use lighthouse_network::service::api_types::{ AppRequestId, BlobsByRangeRequestId, BlocksByRangeRequestId, ComponentsByRangeRequestId, @@ -40,8 +42,8 @@ use lighthouse_network::{Client, NetworkGlobals, PeerAction, PeerId, ReportSourc use parking_lot::RwLock; pub use requests::LookupVerifyError; use requests::{ - ActiveRequests, BlobsByRangeRequestItems, BlocksByRangeRequestItems, BlocksByRootRequestItems, - DataColumnsByRangeRequestItems, DataColumnsByRootRequestItems, + ActiveRequestItems, ActiveRequests, BlobsByRangeRequestItems, BlocksByRangeRequestItems, + BlocksByRootRequestItems, DataColumnsByRangeRequestItems, DataColumnsByRootRequestItems, PayloadEnvelopesByRangeRequestItems, PayloadEnvelopesByRootRequestItems, }; #[cfg(test)] @@ -100,6 +102,30 @@ pub type RpcResponseResult = Result<(T, Duration), RpcResponseError>; pub type CustodyByRootResult = Result>, RpcResponseError>; +/// Per-peer count of active requests for a single protocol, to keep peer selection within +/// `MAX_CONCURRENT_REQUESTS` concurrent requests per protocol ID. +struct ActiveRequestsPerPeer { + count_by_peer: HashMap, +} + +impl ActiveRequestsPerPeer { + fn new(requests: &ActiveRequests) -> Self + where + K: Copy + Eq + std::hash::Hash + std::fmt::Display, + T: ActiveRequestItems, + { + let mut count_by_peer = HashMap::::new(); + for peer_id in requests.iter_request_peers() { + *count_by_peer.entry(peer_id).or_default() += 1; + } + Self { count_by_peer } + } + + fn at_concurrency_limit(&self, peer_id: &PeerId) -> bool { + self.count_by_peer.get(peer_id).copied().unwrap_or(0) >= MAX_CONCURRENT_REQUESTS + } +} + #[derive(Debug)] #[allow(private_interfaces)] pub enum RpcResponseError { @@ -440,47 +466,6 @@ impl SyncNetworkContext { } } - fn active_request_count_by_peer(&self) -> HashMap { - let Self { - network_send: _, - request_id: _, - blocks_by_root_requests, - payload_envelopes_by_root_requests, - data_columns_by_root_requests, - blocks_by_range_requests, - blobs_by_range_requests, - data_columns_by_range_requests, - payload_envelopes_by_range_requests, - // custody_by_root_requests is a meta request of data_columns_by_root_requests - custody_by_root_requests: _, - // components_by_range_requests is a meta request of various _by_range requests - components_by_range_requests: _, - custody_backfill_data_column_batch_requests: _, - execution_engine_state: _, - network_beacon_processor: _, - chain: _, - fork_context: _, - // Don't use a fallback match. We want to be sure that all requests are considered when - // adding new ones - } = self; - - let mut active_request_count_by_peer = HashMap::::new(); - - for peer_id in blocks_by_root_requests - .iter_request_peers() - .chain(payload_envelopes_by_root_requests.iter_request_peers()) - .chain(data_columns_by_root_requests.iter_request_peers()) - .chain(blocks_by_range_requests.iter_request_peers()) - .chain(blobs_by_range_requests.iter_request_peers()) - .chain(data_columns_by_range_requests.iter_request_peers()) - .chain(payload_envelopes_by_range_requests.iter_request_peers()) - { - *active_request_count_by_peer.entry(peer_id).or_default() += 1; - } - - active_request_count_by_peer - } - /// Retries only the specified failed columns by requesting them again. /// /// Note: This function doesn't retry the whole batch, but retries specific requests within @@ -507,8 +492,6 @@ impl SyncNetworkContext { return Err("request id not present".to_string()); }; - let active_request_count_by_peer = self.active_request_count_by_peer(); - debug!( ?failed_columns, ?id, @@ -518,12 +501,7 @@ impl SyncNetworkContext { // Attempt to find all required custody peers to request the failed columns from let columns_by_range_peers_to_request = self - .select_columns_by_range_peers_to_request( - failed_columns, - peers, - active_request_count_by_peer, - peers_to_deprioritize, - ) + .select_columns_by_range_peers_to_request(failed_columns, peers, peers_to_deprioritize) .map_err(|e| format!("{:?}", e))?; // Reuse the id for the request that received partially correct responses @@ -581,16 +559,16 @@ impl SyncNetworkContext { column_peers = column_peers.len() ); let _guard = range_request_span.clone().entered(); - let active_request_count_by_peer = self.active_request_count_by_peer(); + let blocks_by_range_per_peer = ActiveRequestsPerPeer::new(&self.blocks_by_range_requests); let Some(block_peer) = block_peers .iter() .map(|peer| { ( + // Strictly de-prioritize peers already at the per-protocol concurrency limit + blocks_by_range_per_peer.at_concurrency_limit(peer), // If contains -> 1 (order after), not contains -> 0 (order first) peers_to_deprioritize.contains(peer), - // Prefer peers with less overall requests - active_request_count_by_peer.get(peer).copied().unwrap_or(0), // Random factor to break ties, otherwise the PeerID breaks ties rand::random::(), peer, @@ -620,7 +598,6 @@ impl SyncNetworkContext { Some(self.select_columns_by_range_peers_to_request( &column_indexes, column_peers, - active_request_count_by_peer, peers_to_deprioritize, )?) } else { @@ -731,10 +708,11 @@ impl SyncNetworkContext { &self, custody_indexes: &HashSet, peers: &HashSet, - active_request_count_by_peer: HashMap, peers_to_deprioritize: &HashSet, ) -> Result>, RpcRequestSendError> { let mut columns_to_request_by_peer = HashMap::>::new(); + let data_columns_by_range_per_peer = + ActiveRequestsPerPeer::new(&self.data_columns_by_range_requests); for column_index in custody_indexes { // Strictly consider peers that are custodials of this column AND are part of this @@ -748,21 +726,20 @@ impl SyncNetworkContext { }) .map(|peer| { ( + // Strictly de-prioritize peers already at the per-protocol concurrency limit + data_columns_by_range_per_peer.at_concurrency_limit(peer), // If contains -> 1 (order after), not contains -> 0 (order first) peers_to_deprioritize.contains(peer), - // Prefer peers with less overall requests - // Also account for requests that are not yet issued tracked in peer_id_to_request_map - // We batch requests to the same peer, so count existance in the - // `columns_to_request_by_peer` as a single 1 request. - active_request_count_by_peer.get(peer).copied().unwrap_or(0) - + columns_to_request_by_peer.get(peer).map(|_| 1).unwrap_or(0), + // Spread columns within this round: a peer already assigned a column here + // counts as 1 so we prefer peers not yet picked. + columns_to_request_by_peer.get(peer).map(|_| 1).unwrap_or(0), // Random factor to break ties, otherwise the PeerID breaks ties rand::random::(), peer, ) }) .min() - .map(|(_, _, _, peer)| *peer) + .map(|(_, _, _, _, peer)| *peer) else { // TODO(das): this will be pretty bad UX. To improve we should: // - Handle the no peers case gracefully, maybe add some timeout and give a few @@ -881,14 +858,14 @@ impl SyncNetworkContext { lookup_peers: Arc>>, block_root: Hash256, ) -> Result>>, RpcRequestSendError> { - let active_request_count_by_peer = self.active_request_count_by_peer(); + let blocks_by_root_per_peer = ActiveRequestsPerPeer::new(&self.blocks_by_root_requests); let Some(peer_id) = lookup_peers .read() .iter() .map(|peer| { ( - // Prefer peers with less overall requests - active_request_count_by_peer.get(peer).copied().unwrap_or(0), + // Strictly de-prioritize peers already at the per-protocol concurrency limit + blocks_by_root_per_peer.at_concurrency_limit(peer), // Random factor to break ties, otherwise the PeerID breaks ties rand::random::(), peer, @@ -1001,13 +978,15 @@ impl SyncNetworkContext { )); } - let active_request_count_by_peer = self.active_request_count_by_peer(); + let payload_envelopes_by_root_per_peer = + ActiveRequestsPerPeer::new(&self.payload_envelopes_by_root_requests); let Some(peer_id) = lookup_peers .read() .iter() .map(|peer| { ( - active_request_count_by_peer.get(peer).copied().unwrap_or(0), + // Strictly de-prioritize peers already at the per-protocol concurrency limit + payload_envelopes_by_root_per_peer.at_concurrency_limit(peer), rand::random::(), peer, ) @@ -1756,7 +1735,6 @@ impl SyncNetworkContext { peers: &HashSet, peers_to_deprioritize: &HashSet, ) -> Result { - let active_request_count_by_peer = self.active_request_count_by_peer(); // Attempt to find all required custody peers before sending any request or creating an ID let columns_by_range_peers_to_request = { let column_indexes = self @@ -1769,7 +1747,6 @@ impl SyncNetworkContext { self.select_columns_by_range_peers_to_request( &column_indexes, peers, - active_request_count_by_peer, peers_to_deprioritize, )? }; diff --git a/beacon_node/network/src/sync/network_context/custody.rs b/beacon_node/network/src/sync/network_context/custody.rs index b1a4b52867d..0cb020965fb 100644 --- a/beacon_node/network/src/sync/network_context/custody.rs +++ b/beacon_node/network/src/sync/network_context/custody.rs @@ -16,7 +16,9 @@ use tracing::{Span, debug, debug_span, warn}; use types::{DataColumnSidecar, Hash256, data::ColumnIndex}; use types::{DataColumnSidecarList, EthSpec}; -use super::{LookupRequestResult, PeerGroup, RpcResponseResult, SyncNetworkContext}; +use super::{ + ActiveRequestsPerPeer, LookupRequestResult, PeerGroup, RpcResponseResult, SyncNetworkContext, +}; const MAX_STALE_NO_PEERS_DURATION: Duration = Duration::from_secs(30); @@ -234,7 +236,6 @@ impl ActiveCustodyRequest { ))); } - let active_request_count_by_peer = cx.active_request_count_by_peer(); let mut columns_to_request_by_peer = HashMap::>::new(); let mut columns_without_peers = vec![]; let lookup_peers = self.lookup_peers.read(); @@ -250,13 +251,8 @@ impl ActiveCustodyRequest { return Err(Error::TooManyFailures); } - let peer_to_request = self.select_column_peer( - cx, - &active_request_count_by_peer, - &lookup_peers, - *column_index, - &random_state, - ); + let peer_to_request = + self.select_column_peer(cx, &lookup_peers, *column_index, &random_state); if let Some(peer_id) = peer_to_request { columns_to_request_by_peer @@ -357,7 +353,6 @@ impl ActiveCustodyRequest { fn select_column_peer( &self, cx: &mut SyncNetworkContext, - active_request_count_by_peer: &HashMap, lookup_peers: &HashSet, column_index: ColumnIndex, random_state: &RandomState, @@ -365,6 +360,8 @@ impl ActiveCustodyRequest { // We draw from the total set of peers, but prioritize those peers who we have // received an attestation or a block from (`lookup_peers`), as the `lookup_peers` may take // time to build up and we are likely to not find any column peers initially. + let data_columns_by_root_per_peer = + ActiveRequestsPerPeer::new(&cx.data_columns_by_root_requests); let custodial_peers = cx.get_custodial_peers(column_index); let mut prioritized_peers = custodial_peers .iter() @@ -374,12 +371,12 @@ impl ActiveCustodyRequest { }) .map(|peer| { ( + // Strictly de-prioritize peers already at the per-protocol concurrency limit + data_columns_by_root_per_peer.at_concurrency_limit(peer), // Prioritize peers that claim to know have imported this block if lookup_peers.contains(peer) { 0 } else { 1 }, // De-prioritize peers that we have already attempted to download from self.peer_attempts.get(peer).copied().unwrap_or(0), - // Prefer peers with fewer requests to load balance across peers. - active_request_count_by_peer.get(peer).copied().unwrap_or(0), // The hash ensures consistent peer ordering within this request // to avoid fragmentation while varying selection across different requests. random_state.hash_one(peer), From c724dbed56bdef7cef125d5c791b03ecc1d9a58b Mon Sep 17 00:00:00 2001 From: dapplion <35266934+dapplion@users.noreply.github.com> Date: Fri, 12 Jun 2026 00:27:04 +0200 Subject: [PATCH 2/7] Precompute data_columns_by_root per-peer map in select_column_peer Hoist ActiveRequestsPerPeer out of the per-column loop and account for in-round assignments via columns_to_request_by_peer. --- .../src/sync/network_context/custody.rs | 22 ++++++++++++++----- 1 file changed, 17 insertions(+), 5 deletions(-) diff --git a/beacon_node/network/src/sync/network_context/custody.rs b/beacon_node/network/src/sync/network_context/custody.rs index 0cb020965fb..3df988b45f2 100644 --- a/beacon_node/network/src/sync/network_context/custody.rs +++ b/beacon_node/network/src/sync/network_context/custody.rs @@ -236,6 +236,8 @@ impl ActiveCustodyRequest { ))); } + let data_columns_by_root_per_peer = + ActiveRequestsPerPeer::new(&cx.data_columns_by_root_requests); let mut columns_to_request_by_peer = HashMap::>::new(); let mut columns_without_peers = vec![]; let lookup_peers = self.lookup_peers.read(); @@ -251,8 +253,14 @@ impl ActiveCustodyRequest { return Err(Error::TooManyFailures); } - let peer_to_request = - self.select_column_peer(cx, &lookup_peers, *column_index, &random_state); + let peer_to_request = self.select_column_peer( + cx, + &data_columns_by_root_per_peer, + &columns_to_request_by_peer, + &lookup_peers, + *column_index, + &random_state, + ); if let Some(peer_id) = peer_to_request { columns_to_request_by_peer @@ -353,6 +361,8 @@ impl ActiveCustodyRequest { fn select_column_peer( &self, cx: &mut SyncNetworkContext, + data_columns_by_root_per_peer: &ActiveRequestsPerPeer, + columns_to_request_by_peer: &HashMap>, lookup_peers: &HashSet, column_index: ColumnIndex, random_state: &RandomState, @@ -360,8 +370,6 @@ impl ActiveCustodyRequest { // We draw from the total set of peers, but prioritize those peers who we have // received an attestation or a block from (`lookup_peers`), as the `lookup_peers` may take // time to build up and we are likely to not find any column peers initially. - let data_columns_by_root_per_peer = - ActiveRequestsPerPeer::new(&cx.data_columns_by_root_requests); let custodial_peers = cx.get_custodial_peers(column_index); let mut prioritized_peers = custodial_peers .iter() @@ -377,6 +385,10 @@ impl ActiveCustodyRequest { if lookup_peers.contains(peer) { 0 } else { 1 }, // De-prioritize peers that we have already attempted to download from self.peer_attempts.get(peer).copied().unwrap_or(0), + // Spread columns within this round: the precomputed concurrency map is taken + // once before the loop, so count a peer already assigned a column here as 1 to + // prefer peers not yet picked. + columns_to_request_by_peer.get(peer).map(|_| 1).unwrap_or(0), // The hash ensures consistent peer ordering within this request // to avoid fragmentation while varying selection across different requests. random_state.hash_one(peer), @@ -388,7 +400,7 @@ impl ActiveCustodyRequest { prioritized_peers .first() - .map(|(_, _, _, _, peer_id)| *peer_id) + .map(|(_, _, _, _, _, peer_id)| *peer_id) } } From 58826f0996fe73417d1ae49aadfeff583e189fbb Mon Sep 17 00:00:00 2001 From: dapplion <35266934+dapplion@users.noreply.github.com> Date: Thu, 18 Jun 2026 09:51:46 +0200 Subject: [PATCH 3/7] Fix concurrency limit on columns req --- .../network/src/sync/network_context.rs | 27 ++++++++++--------- 1 file changed, 15 insertions(+), 12 deletions(-) diff --git a/beacon_node/network/src/sync/network_context.rs b/beacon_node/network/src/sync/network_context.rs index 0dda140c2d6..a6b09368329 100644 --- a/beacon_node/network/src/sync/network_context.rs +++ b/beacon_node/network/src/sync/network_context.rs @@ -121,8 +121,8 @@ impl ActiveRequestsPerPeer { Self { count_by_peer } } - fn at_concurrency_limit(&self, peer_id: &PeerId) -> bool { - self.count_by_peer.get(peer_id).copied().unwrap_or(0) >= MAX_CONCURRENT_REQUESTS + fn get(&self, peer_id: &PeerId) -> usize { + self.count_by_peer.get(peer_id).copied().unwrap_or(0) } } @@ -565,10 +565,10 @@ impl SyncNetworkContext { .iter() .map(|peer| { ( - // Strictly de-prioritize peers already at the per-protocol concurrency limit - blocks_by_range_per_peer.at_concurrency_limit(peer), // If contains -> 1 (order after), not contains -> 0 (order first) peers_to_deprioritize.contains(peer), + // Strictly de-prioritize peers already at the per-protocol concurrency limit + blocks_by_range_per_peer.get(peer) >= MAX_CONCURRENT_REQUESTS, // Random factor to break ties, otherwise the PeerID breaks ties rand::random::(), peer, @@ -726,20 +726,23 @@ impl SyncNetworkContext { }) .map(|peer| { ( - // Strictly de-prioritize peers already at the per-protocol concurrency limit - data_columns_by_range_per_peer.at_concurrency_limit(peer), // If contains -> 1 (order after), not contains -> 0 (order first) peers_to_deprioritize.contains(peer), - // Spread columns within this round: a peer already assigned a column here - // counts as 1 so we prefer peers not yet picked. - columns_to_request_by_peer.get(peer).map(|_| 1).unwrap_or(0), + // Strictly de-prioritize peers already at the per-protocol concurrency limit + data_columns_by_range_per_peer.get(peer) + + if columns_to_request_by_peer.contains_key(peer) { + 1 + } else { + 0 + } + >= MAX_CONCURRENT_REQUESTS, // Random factor to break ties, otherwise the PeerID breaks ties rand::random::(), peer, ) }) .min() - .map(|(_, _, _, _, peer)| *peer) + .map(|(_, _, _, peer)| *peer) else { // TODO(das): this will be pretty bad UX. To improve we should: // - Handle the no peers case gracefully, maybe add some timeout and give a few @@ -865,7 +868,7 @@ impl SyncNetworkContext { .map(|peer| { ( // Strictly de-prioritize peers already at the per-protocol concurrency limit - blocks_by_root_per_peer.at_concurrency_limit(peer), + blocks_by_root_per_peer.get(peer) >= MAX_CONCURRENT_REQUESTS, // Random factor to break ties, otherwise the PeerID breaks ties rand::random::(), peer, @@ -986,7 +989,7 @@ impl SyncNetworkContext { .map(|peer| { ( // Strictly de-prioritize peers already at the per-protocol concurrency limit - payload_envelopes_by_root_per_peer.at_concurrency_limit(peer), + payload_envelopes_by_root_per_peer.get(peer) >= MAX_CONCURRENT_REQUESTS, rand::random::(), peer, ) From 93a708721bb0b7fec1ed910fdea957f0e5dfd313 Mon Sep 17 00:00:00 2001 From: dapplion <35266934+dapplion@users.noreply.github.com> Date: Thu, 18 Jun 2026 10:02:06 +0200 Subject: [PATCH 4/7] Fix concurrency limit on columns req --- beacon_node/network/src/sync/network_context.rs | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/beacon_node/network/src/sync/network_context.rs b/beacon_node/network/src/sync/network_context.rs index a6b09368329..b8ccfd71c78 100644 --- a/beacon_node/network/src/sync/network_context.rs +++ b/beacon_node/network/src/sync/network_context.rs @@ -729,13 +729,9 @@ impl SyncNetworkContext { // If contains -> 1 (order after), not contains -> 0 (order first) peers_to_deprioritize.contains(peer), // Strictly de-prioritize peers already at the per-protocol concurrency limit - data_columns_by_range_per_peer.get(peer) - + if columns_to_request_by_peer.contains_key(peer) { - 1 - } else { - 0 - } - >= MAX_CONCURRENT_REQUESTS, + // Note: do not account for to-by-sent requests on + // `data_columns_by_range_by_peer` as we always send at most one request + data_columns_by_range_per_peer.get(peer), // Random factor to break ties, otherwise the PeerID breaks ties rand::random::(), peer, From e3192e1c0180174c84258a120c47a5d617c77665 Mon Sep 17 00:00:00 2001 From: dapplion <35266934+dapplion@users.noreply.github.com> Date: Thu, 18 Jun 2026 10:04:38 +0200 Subject: [PATCH 5/7] Typo --- beacon_node/network/src/sync/network_context.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/beacon_node/network/src/sync/network_context.rs b/beacon_node/network/src/sync/network_context.rs index b8ccfd71c78..f2466295fc7 100644 --- a/beacon_node/network/src/sync/network_context.rs +++ b/beacon_node/network/src/sync/network_context.rs @@ -729,7 +729,7 @@ impl SyncNetworkContext { // If contains -> 1 (order after), not contains -> 0 (order first) peers_to_deprioritize.contains(peer), // Strictly de-prioritize peers already at the per-protocol concurrency limit - // Note: do not account for to-by-sent requests on + // Note: do not account for to-be-sent requests on // `data_columns_by_range_by_peer` as we always send at most one request data_columns_by_range_per_peer.get(peer), // Random factor to break ties, otherwise the PeerID breaks ties From 394e89f78c88a6b55e6a00bcbe4033676900d866 Mon Sep 17 00:00:00 2001 From: dapplion <35266934+dapplion@users.noreply.github.com> Date: Thu, 18 Jun 2026 10:21:14 +0200 Subject: [PATCH 6/7] Use at_concurrency_limit --- beacon_node/network/src/sync/network_context.rs | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/beacon_node/network/src/sync/network_context.rs b/beacon_node/network/src/sync/network_context.rs index f2466295fc7..10c0b8287f7 100644 --- a/beacon_node/network/src/sync/network_context.rs +++ b/beacon_node/network/src/sync/network_context.rs @@ -121,8 +121,8 @@ impl ActiveRequestsPerPeer { Self { count_by_peer } } - fn get(&self, peer_id: &PeerId) -> usize { - self.count_by_peer.get(peer_id).copied().unwrap_or(0) + fn at_concurrency_limit(&self, peer_id: &PeerId) -> bool { + self.count_by_peer.get(peer_id).copied().unwrap_or(0) >= MAX_CONCURRENT_REQUESTS } } @@ -568,7 +568,7 @@ impl SyncNetworkContext { // If contains -> 1 (order after), not contains -> 0 (order first) peers_to_deprioritize.contains(peer), // Strictly de-prioritize peers already at the per-protocol concurrency limit - blocks_by_range_per_peer.get(peer) >= MAX_CONCURRENT_REQUESTS, + blocks_by_range_per_peer.at_concurrency_limit(peer), // Random factor to break ties, otherwise the PeerID breaks ties rand::random::(), peer, @@ -669,6 +669,9 @@ impl SyncNetworkContext { let payloads_req_id = if matches!(batch_type, ByRangeRequestType::BlocksAndEnvelopesAndColumns) { Some(self.send_payload_envelopes_by_range_request( + // Peer selection: for a given peer, the count of sent blocks_by_range requests + // equals the count of sent payloads_by_range requests. So we are under the + // concurrency limit for payloads_by_range requests block_peer, PayloadEnvelopesByRangeRequest { start_slot: *request.start_slot(), @@ -731,7 +734,7 @@ impl SyncNetworkContext { // Strictly de-prioritize peers already at the per-protocol concurrency limit // Note: do not account for to-be-sent requests on // `data_columns_by_range_by_peer` as we always send at most one request - data_columns_by_range_per_peer.get(peer), + data_columns_by_range_per_peer.at_concurrency_limit(peer), // Random factor to break ties, otherwise the PeerID breaks ties rand::random::(), peer, @@ -864,7 +867,7 @@ impl SyncNetworkContext { .map(|peer| { ( // Strictly de-prioritize peers already at the per-protocol concurrency limit - blocks_by_root_per_peer.get(peer) >= MAX_CONCURRENT_REQUESTS, + blocks_by_root_per_peer.at_concurrency_limit(peer), // Random factor to break ties, otherwise the PeerID breaks ties rand::random::(), peer, @@ -985,7 +988,7 @@ impl SyncNetworkContext { .map(|peer| { ( // Strictly de-prioritize peers already at the per-protocol concurrency limit - payload_envelopes_by_root_per_peer.get(peer) >= MAX_CONCURRENT_REQUESTS, + payload_envelopes_by_root_per_peer.at_concurrency_limit(peer), rand::random::(), peer, ) From ef3aa2d17e88ba9bd0e9a38999d379ba44fb2cf8 Mon Sep 17 00:00:00 2001 From: dapplion <35266934+dapplion@users.noreply.github.com> Date: Thu, 18 Jun 2026 10:45:08 +0200 Subject: [PATCH 7/7] Drop within-round spread factor from select_column_peer The map(|_| 1) flag only de-prioritized a peer's second column and is moot since the round batches to one request per peer, matching the by-range path. --- beacon_node/network/src/sync/network_context/custody.rs | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/beacon_node/network/src/sync/network_context/custody.rs b/beacon_node/network/src/sync/network_context/custody.rs index 3df988b45f2..4d47e697f11 100644 --- a/beacon_node/network/src/sync/network_context/custody.rs +++ b/beacon_node/network/src/sync/network_context/custody.rs @@ -256,7 +256,6 @@ impl ActiveCustodyRequest { let peer_to_request = self.select_column_peer( cx, &data_columns_by_root_per_peer, - &columns_to_request_by_peer, &lookup_peers, *column_index, &random_state, @@ -362,7 +361,6 @@ impl ActiveCustodyRequest { &self, cx: &mut SyncNetworkContext, data_columns_by_root_per_peer: &ActiveRequestsPerPeer, - columns_to_request_by_peer: &HashMap>, lookup_peers: &HashSet, column_index: ColumnIndex, random_state: &RandomState, @@ -385,10 +383,6 @@ impl ActiveCustodyRequest { if lookup_peers.contains(peer) { 0 } else { 1 }, // De-prioritize peers that we have already attempted to download from self.peer_attempts.get(peer).copied().unwrap_or(0), - // Spread columns within this round: the precomputed concurrency map is taken - // once before the loop, so count a peer already assigned a column here as 1 to - // prefer peers not yet picked. - columns_to_request_by_peer.get(peer).map(|_| 1).unwrap_or(0), // The hash ensures consistent peer ordering within this request // to avoid fragmentation while varying selection across different requests. random_state.hash_one(peer), @@ -400,7 +394,7 @@ impl ActiveCustodyRequest { prioritized_peers .first() - .map(|(_, _, _, _, _, peer_id)| *peer_id) + .map(|(_, _, _, _, peer_id)| *peer_id) } }