Etcm 1055 Adjust PeersClient.nextBestBlock() (#1101)

* ETCM-1063: Add tests for FetcherService and PeersClient

* ETCM-1055: Make PeersClient.nextBestPeer fetch a peer with a different chain

Author: AnastasiiaL

src/main/scala/io/iohk/ethereum/blockchain/sync/PeerComparator.scala

@@ -0,0 +1,9 @@
+package io.iohk.ethereum.blockchain
+
+import io.iohk.ethereum.network.EtcPeerManagerActor.PeerInfo
+
+object PeerComparator {
+
+  def doPeersHaveSameBestBlock(peerInfo1: PeerInfo, peerInfo2: PeerInfo): Boolean =
+    peerInfo1.bestBlockHash == peerInfo2.bestBlockHash
+}

src/main/scala/io/iohk/ethereum/blockchain/sync/PeersClient.scala

@@ -14,6 +14,7 @@ import scala.concurrent.ExecutionContext
 import scala.jdk.CollectionConverters._
 import scala.reflect.ClassTag
 
+import io.iohk.ethereum.blockchain.PeerComparator
 import io.iohk.ethereum.blockchain.sync.Blacklist.BlacklistReason
 import io.iohk.ethereum.blockchain.sync.PeerListSupportNg.PeerWithInfo
 import io.iohk.ethereum.network.EtcPeerManagerActor.PeerInfo
@@ -43,6 +44,10 @@ class PeersClient(
   val statusSchedule: Cancellable =
     scheduler.scheduleWithFixedDelay(syncConfig.printStatusInterval, syncConfig.printStatusInterval, self, PrintStatus)
 
+  val numberOfPeersToFetchFrom = syncConfig.peersToFetchFrom
+
+  val activeFetchingNodes: mutable.Set[PeerWithInfo] = lruSet(numberOfPeersToFetchFrom)
+
   def receive: Receive = running(Map())
 
   override def postStop(): Unit = {
@@ -100,7 +105,7 @@ class PeersClient(
   private def selectPeer(peerSelector: PeerSelector): Option[Peer] =
     peerSelector match {
       case BestPeer     => bestPeer(peersToDownloadFrom)
-      case NextBestPeer => nextBestPeer(peersToDownloadFrom)
+      case NextBestPeer => nextBestPeer(peersToDownloadFrom, activeFetchingNodes.toSet)
     }
 
   private def responseClassTag[RequestMsg <: Message](requestMsg: RequestMsg): ClassTag[_ <: Message] =
@@ -149,11 +154,6 @@ object PeersClient {
 
   type Requesters = Map[ActorRef, ActorRef]
 
-  //TODO: get the value from syncConfig.peersToFetchFrom
-  val numberOfPeersToFetchFrom = 1
-
-  val activeFetchingNodes: mutable.Set[Peer] = lruSet[Peer](numberOfPeersToFetchFrom)
-
   private def lruSet[A](maxEntries: Int): mutable.Set[A] =
     newSetFromMap[A](new java.util.LinkedHashMap[A, java.lang.Boolean]() {
       override def removeEldestEntry(eldest: java.util.Map.Entry[A, java.lang.Boolean]): Boolean = size > maxEntries
@@ -209,18 +209,23 @@ object PeersClient {
   }
 
   //returns the next best peer after the one already returned previously
-  //TODO: make sure the next best peer has a different best block, so we don't fetch in parallel identical branches
   //TODO: whenever this method is called - do activeFetchingNodes.add(_) on the peer returned
-  def nextBestPeer(peersToDownloadFrom: Map[PeerId, PeerWithInfo]): Option[Peer] = {
+  def nextBestPeer(
+      peersToDownloadFrom: Map[PeerId, PeerWithInfo],
+      activeFetchingNodes: Set[PeerWithInfo]
+  ): Option[Peer] = {
     val peersToUse = peersToDownloadFrom.values
-      .collect { case PeerWithInfo(peer, PeerInfo(_, chainWeight, true, _, _)) =>
-        (peer, chainWeight)
+      .collect { case PeerWithInfo(peer, peerInfo @ PeerInfo(_, _, true, _, _)) =>
+        (peer, peerInfo)
       }
 
     val peer =
       peersToUse
-        .filter { case (peer, _) => !activeFetchingNodes.contains(peer) }
-        .maxByOption { case (_, weight) => weight }
+        .filterNot { case (peer, _) => activeFetchingNodes.map(_.peer).contains(peer) }
+        .filterNot { case (_, peerInfo) =>
+          activeFetchingNodes.map(_.peerInfo).exists(PeerComparator.doPeersHaveSameBestBlock(peerInfo, _))
+        }
+        .maxByOption { case (_, peerInfo) => peerInfo.chainWeight }
         .map { case (peer, _) => peer }
     peer
   }

src/test/scala/io/iohk/ethereum/blockchain/sync/PeersClientSpec.scala

@@ -6,6 +6,8 @@ import akka.actor.ActorSystem
 import akka.testkit.TestProbe
 import akka.util.ByteString
 
+import scala.collection.mutable
+
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 import org.scalatest.prop.TableFor3
@@ -74,14 +76,14 @@ class PeersClientSpec extends AnyFlatSpec with Matchers with ScalaCheckPropertyC
     }
   }
 
-  it should "determine the next best peer (same as  bestPeer when lru set is empty)" in {
+  it should "determine the next best peer (same as bestPeer when lru set is empty)" in {
     forAll(table) { (peerInfoMap, expectedPeer, _) =>
-      PeersClient.nextBestPeer(peerInfoMap) shouldEqual expectedPeer
+      PeersClient.nextBestPeer(peerInfoMap, Set.empty) shouldEqual expectedPeer
     }
   }
 
-  it should "determine the next best peer when lru is used" in {
-    val table = Table[Map[PeerId, PeerWithInfo], Option[Peer], Option[Peer], String](
+  it should "determine the next best peer with a different best block each time" in {
+    val table = Table[Map[PeerId, PeerWithInfo], Option[PeerWithInfo], Option[Peer], String](
       ("PeerInfo map", "Used best peer", "Expected best peer", "Scenario info (selected peer)"),
       (
         Map(),
@@ -100,40 +102,31 @@ class PeersClientSpec extends AnyFlatSpec with Matchers with ScalaCheckPropertyC
           peer1.id -> PeerWithInfo(peer1, peerInfo(0, 100, fork = false)),
           peer2.id -> PeerWithInfo(peer2, peerInfo(0, 50, fork = true))
         ),
-        Some(peer2),
+        Some(PeerWithInfo(peer2, peerInfo(0, 50, fork = true))),
         None,
-        "Peer2 with lower TD but following the ETC fork"
+        "Peer2 with lower TD but following the ETC fork, peer2 is already used"
       ),
       (
         Map(peer1.id -> PeerWithInfo(peer1, peerInfo(0, 100)), peer2.id -> PeerWithInfo(peer2, peerInfo(0, 101))),
-        Some(peer2),
-        Some(peer1),
-        "Peer2 with higher TD"
-      ),
-      (
-        Map(
-          peer1.id -> PeerWithInfo(peer1, peerInfo(0, 100)),
-          peer2.id -> PeerWithInfo(peer2, peerInfo(0, 101)),
-          peer3.id -> PeerWithInfo(peer3, peerInfo(1, 50))
-        ),
-        Some(peer3),
-        Some(peer2),
-        "Peer3 with lower TD but higher checkpoint number"
+        Some(PeerWithInfo(peer2, peerInfo(0, 101))),
+        None,
+        "Both peer are used"
       ),
       (
         Map(
           peer1.id -> PeerWithInfo(peer1, peerInfo(0, 100)),
-          peer2.id -> PeerWithInfo(peer2, peerInfo(4, 101)),
-          peer3.id -> PeerWithInfo(peer3, peerInfo(4, 50))
+          peer2.id -> PeerWithInfo(peer2, peerInfo(1, 50)),
+          peer3.id -> PeerWithInfo(peer3, peerInfo(0, 80).copy(bestBlockHash = ByteString("differenthash")))
         ),
-        Some(peer2),
+        Some(PeerWithInfo(peer2, peerInfo(1, 50))),
         Some(peer3),
-        "Peer2 with equal checkpoint number and higher TD"
+        "Peer2 with lower TD but higher checkpoint number, peer 1 and 2 are used"
       )
     )
-    forAll(table) { (peerInfoMap, usedPeer, expectedPeer, _) =>
-      usedPeer.map(PeersClient.activeFetchingNodes.add)
-      PeersClient.nextBestPeer(peerInfoMap) shouldEqual expectedPeer
+    val activeFetchingNodes: mutable.Set[PeerWithInfo] = mutable.Set.empty
+    forAll(table) { (peerInfoMap, usedPeerWithInfo, expectedPeer, _) =>
+      usedPeerWithInfo.map(activeFetchingNodes.add)
+      PeersClient.nextBestPeer(peerInfoMap, activeFetchingNodes.toSet) shouldEqual expectedPeer
     }
   }