[ETCM-198] Update fastsync to request headers only from peers with be…

src/it/scala/io/iohk/ethereum/sync/FastSyncItSpec.scala

@@ -148,6 +148,34 @@ class FastSyncItSpec extends FlatSpecBase with Matchers with BeforeAndAfterAll {
       assert(peer1.bl.getBestBlockNumber() == peer2.bl.getBestBlockNumber() - peer2.testSyncConfig.pivotBlockOffset)
     }
   }
+
+  it should "follow the longest chains" in customTestCaseResourceM(
+    FakePeer.start4FakePeersRes()
+  ) { case (peer1, peer2, peer3, peer4) =>
+    for {
+      _ <- peer2.importBlocksUntil(1000)(IdentityUpdate)
+      _ <- peer3.importBlocksUntil(1000)(IdentityUpdate)
+      _ <- peer4.importBlocksUntil(1000)(IdentityUpdate)
+
+      _ <- peer2.importBlocksUntil(2000)(IdentityUpdate)
+      _ <- peer3.importBlocksUntil(3000)(updateStateAtBlock(1001, endAccount = 3000))
+      _ <- peer4.importBlocksUntil(3000)(updateStateAtBlock(1001, endAccount = 3000))
+
+      _ <- peer1.connectToPeers(Set(peer2.node, peer3.node, peer4.node))
+      _ <- peer1.startFastSync().delayExecution(50.milliseconds)
+      _ <- peer1.waitForFastSyncFinish()
+    } yield {
+      val trie = peer1.getBestBlockTrie()
+      val synchronizingPeerHaveAllData = peer1.containsExpectedDataUpToAccountAtBlock(3000, 1001)
+      // due to the fact that function generating state is deterministic both peer3 and peer4 ends up with exactly same
+      // state, so peer1 can get whole trie from both of them.
+      assert(peer1.bl.getBestBlockNumber() == peer3.bl.getBestBlockNumber() - peer3.testSyncConfig.pivotBlockOffset)
+      assert(peer1.bl.getBestBlockNumber() == peer4.bl.getBestBlockNumber() - peer4.testSyncConfig.pivotBlockOffset)
+      assert(trie.isDefined)
+      assert(synchronizingPeerHaveAllData)
+    }
+  }
+
 }
 
 object FastSyncItSpec {

src/it/scala/io/iohk/ethereum/sync/util/FastSyncItSpecUtils.scala

@@ -144,6 +144,19 @@ object FastSyncItSpecUtils {
         peer3 <- start1FakePeerRes(fakePeerCustomConfig3, "Peer3")
       } yield (peer1, peer2, peer3)
     }
-  }
 
+    def start4FakePeersRes(
+        fakePeerCustomConfig1: FakePeerCustomConfig = defaultConfig,
+        fakePeerCustomConfig2: FakePeerCustomConfig = defaultConfig,
+        fakePeerCustomConfig3: FakePeerCustomConfig = defaultConfig,
+        fakePeerCustomConfig4: FakePeerCustomConfig = defaultConfig
+    ): Resource[Task, (FakePeer, FakePeer, FakePeer, FakePeer)] = {
+      for {
+        peer1 <- start1FakePeerRes(fakePeerCustomConfig1, "Peer1")
+        peer2 <- start1FakePeerRes(fakePeerCustomConfig2, "Peer2")
+        peer3 <- start1FakePeerRes(fakePeerCustomConfig3, "Peer3")
+        peer4 <- start1FakePeerRes(fakePeerCustomConfig4, "Peer3")
+      } yield (peer1, peer2, peer3, peer4)
+    }
+  }
 }

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

@@ -1,7 +1,6 @@
 package io.iohk.ethereum.blockchain.sync
 
 import java.time.Instant
-
 import akka.actor._
 import akka.util.ByteString
 import cats.data.NonEmptyList
@@ -25,7 +24,6 @@ import io.iohk.ethereum.network.p2p.messages.PV63._
 import io.iohk.ethereum.utils.ByteStringUtils
 import io.iohk.ethereum.utils.Config.SyncConfig
 import org.bouncycastle.util.encoders.Hex
-
 import scala.annotation.tailrec
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.duration.{FiniteDuration, _}
@@ -647,24 +645,25 @@ class FastSync(
       } else {
         val now = Instant.now()
         val peers = unassignedPeers
-          .filter(p => peerRequestsTime.get(p).forall(d => d.plusMillis(fastSyncThrottle.toMillis).isBefore(now)))
+          .filter(p => peerRequestsTime.get(p.peer).forall(d => d.plusMillis(fastSyncThrottle.toMillis).isBefore(now)))
         peers
           .take(maxConcurrentRequests - assignedHandlers.size)
-          .toSeq
-          .sortBy(_.ref.toString())
+          .sortBy(_.info.maxBlockNumber)(Ordering[BigInt].reverse)
           .foreach(assignBlockchainWork)
       }
     }
 
-    def assignBlockchainWork(peer: Peer): Unit = {
+    def assignBlockchainWork(peerWithInfo: PeerWithInfo): Unit = {
+      val PeerWithInfo(peer, peerInfo) = peerWithInfo
       if (syncState.receiptsQueue.nonEmpty) {
         requestReceipts(peer)
       } else if (syncState.blockBodiesQueue.nonEmpty) {
         requestBlockBodies(peer)
       } else if (
         requestedHeaders.isEmpty &&
         context.child(BlockHeadersHandlerName).isEmpty &&
-        syncState.bestBlockHeaderNumber < syncState.safeDownloadTarget
+        syncState.bestBlockHeaderNumber < syncState.safeDownloadTarget &&
+        peerInfo.maxBlockNumber >= syncState.pivotBlock.number
       ) {
         requestBlockHeaders(peer)
       }
@@ -737,7 +736,8 @@ class FastSync(
       peerRequestsTime += (peer -> Instant.now())
     }
 
-    def unassignedPeers: Set[Peer] = peersToDownloadFrom.keySet diff assignedHandlers.values.toSet
+    def unassignedPeers: List[PeerWithInfo] =
+      (peersToDownloadFrom -- assignedHandlers.values).map(PeerWithInfo.tupled).toList
 
     def blockchainDataToDownload: Boolean =
       syncState.blockChainWorkQueued || syncState.bestBlockHeaderNumber < syncState.safeDownloadTarget
@@ -768,6 +768,8 @@ class FastSync(
 
 object FastSync {
 
+  case class PeerWithInfo(peer: Peer, info: PeerInfo)
+
   // scalastyle:off parameter.number
   def props(
       fastSyncStateStorage: FastSyncStateStorage,