[ECTM-104] Fix SyncControllerSpec

Author: Michal Mrozek

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

@@ -69,7 +69,7 @@ class FastSync(
       val syncingHandler = new SyncingHandler(syncState)
       val targetBlockSelector = context.actorOf(
         FastSyncPivotBlockSelector.props(etcPeerManager, peerEventBus, syncConfig, scheduler),
-        "target-block-selector"
+        "pivot-block-selector"
       )
       targetBlockSelector ! FastSyncPivotBlockSelector.ChoosePivotBlock
       context become syncingHandler.waitingForTargetBlockUpdate(ImportedLastBlock)
@@ -87,24 +87,24 @@ class FastSync(
   def startFromScratch(): Unit = {
     val targetBlockSelector = context.actorOf(
       FastSyncPivotBlockSelector.props(etcPeerManager, peerEventBus, syncConfig, scheduler),
-      "target-block-selector"
+      "pivot-block-selector"
     )
     targetBlockSelector ! FastSyncPivotBlockSelector.ChoosePivotBlock
     context become waitingForTargetBlock
   }
 
   def waitingForTargetBlock: Receive = handleCommonMessages orElse {
-    case FastSyncPivotBlockSelector.Result(targetBlockHeader) =>
-      if (targetBlockHeader.number < 1) {
+    case FastSyncPivotBlockSelector.Result(pivotBlockHeader) =>
+      if (pivotBlockHeader.number < 1) {
         log.info("Unable to start block synchronization in fast mode: target block is less than 1")
         appStateStorage.fastSyncDone().commit()
         context become idle
         syncController ! Done
       } else {
         val initialSyncState =
           SyncState(
-            targetBlockHeader,
-            safeDownloadTarget = targetBlockHeader.number + syncConfig.fastSyncBlockValidationX
+            pivotBlockHeader,
+            safeDownloadTarget = pivotBlockHeader.number + syncConfig.fastSyncBlockValidationX
           )
         startWithState(initialSyncState)
       }

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

@@ -1,16 +1,17 @@
 package io.iohk.ethereum.blockchain.sync
 
 import akka.actor.{Actor, ActorLogging, ActorRef, Cancellable, Props, Scheduler}
+import akka.util.ByteString
 import io.iohk.ethereum.domain.BlockHeader
-import io.iohk.ethereum.network.{EtcPeerManagerActor, Peer, PeerId}
 import io.iohk.ethereum.network.EtcPeerManagerActor.PeerInfo
 import io.iohk.ethereum.network.PeerEventBusActor.PeerEvent.MessageFromPeer
-import io.iohk.ethereum.network.PeerEventBusActor.{PeerSelector, Subscribe, Unsubscribe}
 import io.iohk.ethereum.network.PeerEventBusActor.SubscriptionClassifier.MessageClassifier
+import io.iohk.ethereum.network.PeerEventBusActor.{PeerSelector, Subscribe, Unsubscribe}
 import io.iohk.ethereum.network.p2p.messages.PV62.{BlockHeaders, GetBlockHeaders}
+import io.iohk.ethereum.network.{EtcPeerManagerActor, PeerId}
 import io.iohk.ethereum.utils.Config.SyncConfig
-import scala.concurrent.duration.FiniteDuration
 import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.duration.FiniteDuration
 
 class FastSyncPivotBlockSelector(
     val etcPeerManager: ActorRef,
@@ -36,79 +37,88 @@ class FastSyncPivotBlockSelector(
       (peer, maxBlockNumber)
     }
 
-    if (peersUsedToChooseTarget.size >= minPeersToChoosePivotBlock) {
-      tryChooseTargetBlock(peersUsedToChooseTarget)
+    val peersSortedByBestNumber = peersUsedToChooseTarget.toList.sortBy { case (_, number) => -number }
+    val bestPeerBestBlockNumber = peersSortedByBestNumber.headOption
+      .map { case (_, bestPeerBestBlockNumber) => bestPeerBestBlockNumber }
+      .getOrElse(BigInt(0))
+    val expectedPivotBlock = (bestPeerBestBlockNumber - syncConfig.pivotBlockOffset).max(0)
+    val peersToAsk = peersSortedByBestNumber
+      .takeWhile { case (_, number) => number >= expectedPivotBlock }
+      .map { case (peer, _) => peer }
+
+    if (peersToAsk.size >= minPeersToChoosePivotBlock) {
+      log.info(
+        "Trying to choose fast sync pivot block using {} peers. The best block is {}. Ask {} peers for block nr {}",
+        peersUsedToChooseTarget.size,
+        bestPeerBestBlockNumber,
+        peersToAsk.size,
+        expectedPivotBlock
+      )
+
+      peersToAsk.foreach { peer =>
+        peerEventBus ! Subscribe(MessageClassifier(Set(BlockHeaders.code), PeerSelector.WithId(peer.id)))
+        etcPeerManager ! EtcPeerManagerActor.SendMessage(
+          GetBlockHeaders(Left(expectedPivotBlock), 1, 0, reverse = false),
+          peer.id
+        )
+      }
+
+      val timeout = scheduler.scheduleOnce(peerResponseTimeout, self, PivotBlockTimeout)
+      context become waitingForPivotBlock(peersToAsk.map(_.id).toSet, expectedPivotBlock, timeout, Map.empty)
     } else {
       log.info(
-        "Cannot pick pivot block. Need at least {} peers, but there are only {} available at the moment. Retrying in {}",
+        "Cannot pick pivot block. Need at least {} peers, but there are only {} which meet the criteria ({} all available at the moment). Retrying in {}",
         minPeersToChoosePivotBlock,
+        peersToAsk.size,
         peersUsedToChooseTarget.size,
         startRetryInterval
       )
       scheduleRetry(startRetryInterval)
-      context become idle
     }
   }
 
-  def tryChooseTargetBlock(peersWithBestBlockNumbers: Map[Peer, BigInt]): Unit = {
-    val peersSortedByBestNumber = peersWithBestBlockNumbers.toList.sortBy(-_._2)
-    val bestPeerBestBlockNumber = peersSortedByBestNumber.head._2
-    val expectedPivotBlock = (bestPeerBestBlockNumber - syncConfig.pivotBlockOffset).max(0)
-    val peersToAsk = peersSortedByBestNumber.takeWhile(_._2 >= expectedPivotBlock).map(_._1)
-
-    log.info(
-      "Trying to choose fast sync pivot block using {} peers. The best block is {}. Ask {} peers for block nr {}",
-      peersWithBestBlockNumbers.size,
-      bestPeerBestBlockNumber,
-      peersToAsk.size,
-      expectedPivotBlock
-    )
-
-    peersToAsk.foreach { peer =>
-      peerEventBus ! Subscribe(MessageClassifier(Set(BlockHeaders.code), PeerSelector.WithId(peer.id)))
-      etcPeerManager ! EtcPeerManagerActor.SendMessage(
-        GetBlockHeaders(Left(expectedPivotBlock), 1, 0, reverse = false),
-        peer.id
-      )
-    }
-
-    val timeout = scheduler.scheduleOnce(peerResponseTimeout, self, PivotBlockTimeout)
-    context become waitingForPivotBlock(peersToAsk.map(_.id).toSet, expectedPivotBlock, timeout, Map.empty)
-  }
-
   def waitingForPivotBlock(
       peersToAsk: Set[PeerId],
       targetBlockNumber: BigInt,
       timeout: Cancellable,
-      headers: Map[BlockHeader, Int]
+      headers: Map[ByteString, BlockHeaderWithVotes]
   ): Receive =
     handleCommonMessages orElse {
       case MessageFromPeer(blockHeaders: BlockHeaders, peerId) =>
         peerEventBus ! Unsubscribe(MessageClassifier(Set(BlockHeaders.code), PeerSelector.WithId(peerId)))
         val updatedPeersToAsk = peersToAsk - peerId
-        val targetBlockHeaderOpt = blockHeaders.headers.find(header => header.number == targetBlockNumber)
-
+        val targetBlockHeaderOpt =
+          if (blockHeaders.headers.size != 1) None
+          else
+            blockHeaders.headers.find(header => header.number == targetBlockNumber)
         targetBlockHeaderOpt match {
           case Some(targetBlockHeader) =>
             log.info("Received vote for {} from {}", targetBlockHeader.hashAsHexString, peerId.value)
-            val newValue = headers.find(_._1 == targetBlockHeader).map(_._2 + 1).getOrElse(1)
-            val updatedHeaders = headers.updated(targetBlockHeader, newValue)
-            val (mostPopularBlockHeader, votes) = updatedHeaders.maxBy(_._2)
-            if (votes >= minPeersToChoosePivotBlock) {
+            val newValue =
+              headers.get(targetBlockHeader.hash).map(_.vote).getOrElse(BlockHeaderWithVotes(targetBlockHeader))
+            val updatedHeaders = headers.updated(targetBlockHeader.hash, newValue)
+            val BlockHeaderWithVotes(mostPopularBlockHeader, updatedVotes) = updatedHeaders.mostVotedHeader
+            if (updatedVotes >= minPeersToChoosePivotBlock) {
               timeout.cancel()
               sendResponseAndCleanup(mostPopularBlockHeader)
-            } else if (updatedPeersToAsk.size + votes < minPeersToChoosePivotBlock) {
+            } else if (!isPossibleToReachConsensus(updatedPeersToAsk.size, updatedVotes)) {
               timeout.cancel()
               peerEventBus ! Unsubscribe()
               log.info("Not enough votes for pivot block. Retrying in {}", startRetryInterval)
               scheduleRetry(startRetryInterval)
-              context become idle
             } else {
               context become waitingForPivotBlock(updatedPeersToAsk, targetBlockNumber, timeout, updatedHeaders)
             }
           case None =>
             blacklist(peerId, blacklistDuration, "Did not respond with pivot block header, blacklisting")
-            context become waitingForPivotBlock(updatedPeersToAsk, targetBlockNumber, timeout, headers)
+            val BlockHeaderWithVotes(_, votes) = headers.mostVotedHeader
+            if (!isPossibleToReachConsensus(updatedPeersToAsk.size, votes)) {
+              timeout.cancel()
+              log.info("Not enough votes for pivot block. Retrying in {}", startRetryInterval)
+              peerEventBus ! Unsubscribe()
+              scheduleRetry(startRetryInterval)
+            } else
+              context become waitingForPivotBlock(updatedPeersToAsk, targetBlockNumber, timeout, headers)
         }
       case PivotBlockTimeout =>
         peersToAsk.foreach { peerId =>
@@ -117,11 +127,14 @@ class FastSyncPivotBlockSelector(
         peerEventBus ! Unsubscribe()
         log.info("Pivot block header receive timeout. Retrying in {}", startRetryInterval)
         scheduleRetry(startRetryInterval)
-        context become idle
     }
 
+  private def isPossibleToReachConsensus(peersLeft: Int, bestHeaderVotes: Int): Boolean =
+    peersLeft + bestHeaderVotes >= minPeersToChoosePivotBlock
+
   def scheduleRetry(interval: FiniteDuration): Unit = {
     scheduler.scheduleOnce(interval, self, ChoosePivotBlock)
+    context become idle
   }
 
   def sendResponseAndCleanup(pivotBlockHeader: BlockHeader): Unit = {
@@ -140,5 +153,15 @@ object FastSyncPivotBlockSelector {
   case object ChoosePivotBlock
   case class Result(targetBlockHeader: BlockHeader)
 
-  private case object PivotBlockTimeout
+  case object PivotBlockTimeout
+
+  case class BlockHeaderWithVotes(header: BlockHeader, votes: Int = 1) {
+    def vote: BlockHeaderWithVotes = copy(votes = votes + 1)
+  }
+
+  implicit class SortableHeadersMap(headers: Map[ByteString, BlockHeaderWithVotes]) {
+    def mostVotedHeader: BlockHeaderWithVotes = headers.maxBy { case (_, headerWithVotes) =>
+      headerWithVotes.votes
+    }._2
+  }
 }

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

@@ -22,7 +22,12 @@ trait PeerListSupport {
 
   var handshakedPeers: PeersMap = Map.empty
 
-  scheduler.scheduleWithFixedDelay(0.seconds, syncConfig.peersScanInterval, etcPeerManager, EtcPeerManagerActor.GetHandshakedPeers)(global, context.self)
+  scheduler.scheduleWithFixedDelay(
+    0.seconds,
+    syncConfig.peersScanInterval,
+    etcPeerManager,
+    EtcPeerManagerActor.GetHandshakedPeers
+  )(global, context.self)
 
   def removePeer(peerId: PeerId): Unit = {
     peerEventBus ! Unsubscribe(PeerDisconnectedClassifier(PeerSelector.WithId(peerId)))

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

@@ -67,12 +67,12 @@ class FastSyncPivotBlockSelectorSpec extends AnyFlatSpec with Matchers with Befo
     peerMessageBus.expectMsg(Unsubscribe())
   }
 
-  it should "ask for the block number 0 if [bestPeerBestBlockNumber < syncConfig.targetBlockOffset]" in new TestSetup {
+  it should "ask for the block number 0 if [bestPeerBestBlockNumber < syncConfig.pivotBlockOffset]" in new TestSetup {
     val highestNumber = syncConfig.pivotBlockOffset - 1
 
     updateHandshakedPeers(
       HandshakedPeers(
-        allPeers
+        threeAcceptedPeers
           .updated(peer1, threeAcceptedPeers(peer1).copy(maxBlockNumber = highestNumber))
           .updated(peer2, threeAcceptedPeers(peer2).copy(maxBlockNumber = highestNumber / 2))
           .updated(peer3, threeAcceptedPeers(peer3).copy(maxBlockNumber = highestNumber / 5))
@@ -229,7 +229,7 @@ class FastSyncPivotBlockSelectorSpec extends AnyFlatSpec with Matchers with Befo
     peerMessageBus.expectMsg(Unsubscribe())
   }
 
-  it should "check only peers with the highest block at least equal to [bestPeerBestBlockNumber - syncConfig.targetBlockOffset]" in new TestSetup {
+  it should "check only peers with the highest block at least equal to [bestPeerBestBlockNumber - syncConfig.pivotBlockOffset]" in new TestSetup {
     updateHandshakedPeers(
       HandshakedPeers(allPeers.updated(peer1, allPeers(peer1).copy(maxBlockNumber = expectedPivotBlock - 1)))
     )
@@ -244,6 +244,40 @@ class FastSyncPivotBlockSelectorSpec extends AnyFlatSpec with Matchers with Befo
     peerMessageBus.expectNoMessage() // Peer 1 will be skipped
   }
 
+  it should "only use only peers from the correct network to choose pivot block" in new TestSetup() {
+    updateHandshakedPeers(HandshakedPeers(peersFromDifferentNetworks))
+
+    pivotBlockSelector ! ChoosePivotBlock
+
+    peerMessageBus.expectMsgAllOf(
+      Subscribe(MessageClassifier(Set(BlockHeaders.code), PeerSelector.WithId(peer1.id))),
+      // Peer 2 is skipped
+      Subscribe(MessageClassifier(Set(BlockHeaders.code), PeerSelector.WithId(peer3.id))),
+      Subscribe(MessageClassifier(Set(BlockHeaders.code), PeerSelector.WithId(peer4.id)))
+    )
+    peerMessageBus.expectNoMessage()
+
+    etcPeerManager.expectMsgAllOf(
+      EtcPeerManagerActor.SendMessage(GetBlockHeaders(Left(expectedPivotBlock), 1, 0, reverse = false), peer1.id),
+      EtcPeerManagerActor.SendMessage(GetBlockHeaders(Left(expectedPivotBlock), 1, 0, reverse = false), peer3.id),
+      EtcPeerManagerActor.SendMessage(GetBlockHeaders(Left(expectedPivotBlock), 1, 0, reverse = false), peer4.id)
+    )
+
+    // Collecting pivot block (for voting)
+    pivotBlockSelector ! MessageFromPeer(BlockHeaders(Seq(pivotBlockHeader)), peer1.id)
+    pivotBlockSelector ! MessageFromPeer(BlockHeaders(Seq(pivotBlockHeader)), peer3.id)
+    pivotBlockSelector ! MessageFromPeer(BlockHeaders(Seq(pivotBlockHeader)), peer4.id)
+
+    peerMessageBus.expectMsgAllOf(
+      Unsubscribe(MessageClassifier(Set(BlockHeaders.code), PeerSelector.WithId(peer1.id))),
+      Unsubscribe(MessageClassifier(Set(BlockHeaders.code), PeerSelector.WithId(peer3.id))),
+      Unsubscribe(MessageClassifier(Set(BlockHeaders.code), PeerSelector.WithId(peer4.id)))
+    )
+
+    fastSync.expectMsg(Result(pivotBlockHeader))
+    peerMessageBus.expectMsg(Unsubscribe())
+  }
+
   class TestSetup extends TestSyncConfig {
 
     private def isNewBlock(msg: Message): Boolean = msg match {
@@ -377,6 +411,37 @@ class FastSyncPivotBlockSelectorSpec extends AnyFlatSpec with Matchers with Befo
       )
     )
 
+    val peersFromDifferentNetworks = Map(
+      peer1 -> PeerInfo(
+        peer1Status,
+        forkAccepted = true,
+        totalDifficulty = peer1Status.totalDifficulty,
+        maxBlockNumber = bestBlock,
+        bestBlockHash = peer1Status.bestHash
+      ),
+      peer2 -> PeerInfo(
+        peer2Status,
+        forkAccepted = false,
+        totalDifficulty = peer1Status.totalDifficulty,
+        maxBlockNumber = bestBlock,
+        bestBlockHash = peer2Status.bestHash
+      ),
+      peer3 -> PeerInfo(
+        peer3Status,
+        forkAccepted = true,
+        totalDifficulty = peer1Status.totalDifficulty,
+        maxBlockNumber = bestBlock,
+        bestBlockHash = peer3Status.bestHash
+      ),
+      peer4 -> PeerInfo(
+        peer4Status,
+        forkAccepted = true,
+        totalDifficulty = peer1Status.totalDifficulty,
+        maxBlockNumber = bestBlock,
+        bestBlockHash = peer4Status.bestHash
+      )
+    )
+
     def updateHandshakedPeers(handshakedPeers: HandshakedPeers): Unit = pivotBlockSelector ! handshakedPeers
   }
 }