Pr update

core/network/impl/synchronizer_impl.cpp

@@ -330,29 +330,36 @@ namespace kagome::network {
       return false;
     }
 
-    std::vector<libp2p::peer::PeerId> selected_peers = {peer_id};
-    std::vector<libp2p::peer::PeerId> active_peers;
-    peer_manager_->enumeratePeerState(
-        [&active_peers, &block_info, &peer_id](const PeerId &p_id,
-                                               PeerState &peer_state) {
-          if (peer_state.best_block >= block_info and p_id != peer_id) {
-            active_peers.push_back(p_id);
-          }
-          return true;
-        });
-    std::ranges::shuffle(active_peers, random_gen_);
-    for (const auto &p_id : active_peers) {
-      if (selected_peers.size() >= max_parallel_downloads_) {
-        break;
-      }
-      selected_peers.push_back(p_id);
-    }
     // Block is already enqueued
     if (auto it = known_blocks_.find(block_info.hash);
         it != known_blocks_.end()) {
-      auto &block_in_queue = it->second;
+      std::vector<libp2p::peer::PeerId> selected_peers = {peer_id};
+      std::vector<libp2p::peer::PeerId> active_peers;
+      const auto &peers = it->second.peers;
+      std::for_each(
+          peers.begin(), peers.end(), [&active_peers](const auto &p_id) {
+            active_peers.push_back(p_id);
+          });
+      static const auto peers_to_add_number =
+          max_parallel_downloads_ ? (max_parallel_downloads_ - 1) : 0;
+      if (const auto active_peers_size = active_peers.size();
+          active_peers_size <= peers_to_add_number) {
+        selected_peers.insert(
+            selected_peers.end(), active_peers.begin(), active_peers.end());
+      } else {
+        std::vector<size_t> indexes(active_peers_size);
+        std::iota(indexes.begin(), indexes.end(), 0);
+        std::ranges::shuffle(indexes, random_gen_);
+        for (size_t i = 0; i < peers_to_add_number; ++i) {
+          selected_peers.push_back(active_peers[indexes[i]]);
+        }
+      }
       for (const auto &p_id : selected_peers) {
-        block_in_queue.peers.emplace(p_id);
+        loadBlocks(p_id, block_info, [wp{weak_from_this()}](auto res) {
+          if (auto self = wp.lock()) {
+            SL_TRACE(self->log_, "Block(s) enqueued to apply by announce");
+          }
+        });
       }
       return false;
     }
@@ -375,29 +382,24 @@ namespace kagome::network {
         or block_tree_->has(header.parent_hash);
 
     if (parent_is_known) {
-      for (const auto &p_id : selected_peers) {
-        loadBlocks(p_id, block_info, [wp{weak_from_this()}](auto res) {
-          if (auto self = wp.lock()) {
-            SL_TRACE(self->log_, "Block(s) enqueued to apply by announce");
-          }
-        });
-      }
+      loadBlocks(peer_id, block_info, [wp{weak_from_this()}](auto res) {
+        if (auto self = wp.lock()) {
+          SL_TRACE(self->log_, "Block(s) enqueued to apply by announce");
+        }
+      });
       return true;
     }
 
     // Otherwise, is using base way to enqueue
-    auto res = true;
-    for (const auto &p_id : selected_peers) {
-      res &= syncByBlockInfo(
-          block_info,
-          p_id,
-          [wp{weak_from_this()}](auto res) {
-            if (auto self = wp.lock()) {
-              SL_TRACE(self->log_, "Block(s) enqueued to load by announce");
-            }
-          },
-          false);
-    }
+    const auto res = syncByBlockInfo(
+        block_info,
+        peer_id,
+        [wp{weak_from_this()}](auto res) {
+          if (auto self = wp.lock()) {
+            SL_TRACE(self->log_, "Block(s) enqueued to load by announce");
+          }
+        },
+        false);
     return res;
   }