[ECTM-104] PR fixes

Author: Michal Mrozek

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

@@ -68,10 +68,10 @@ class FastSync(
       log.info(s"FastSync interrupted during pivot block update, choosing new pivot block")
       val syncingHandler = new SyncingHandler(syncState)
       val pivotBlockSelector = context.actorOf(
-        FastSyncPivotBlockSelector.props(etcPeerManager, peerEventBus, syncConfig, scheduler),
+        PivotBlockSelector.props(etcPeerManager, peerEventBus, syncConfig, scheduler, context.self),
         "pivot-block-selector"
       )
-      pivotBlockSelector ! FastSyncPivotBlockSelector.ChoosePivotBlock
+      pivotBlockSelector ! PivotBlockSelector.SelectPivotBlock
       context become syncingHandler.waitingForPivotBlockUpdate(ImportedLastBlock)
     } else {
       log.info(
@@ -86,28 +86,27 @@ class FastSync(
 
   def startFromScratch(): Unit = {
     val pivotBlockSelector = context.actorOf(
-      FastSyncPivotBlockSelector.props(etcPeerManager, peerEventBus, syncConfig, scheduler),
+      PivotBlockSelector.props(etcPeerManager, peerEventBus, syncConfig, scheduler, context.self),
       "pivot-block-selector"
     )
-    pivotBlockSelector ! FastSyncPivotBlockSelector.ChoosePivotBlock
+    pivotBlockSelector ! PivotBlockSelector.SelectPivotBlock
     context become waitingForPivotBlock
   }
 
-  def waitingForPivotBlock: Receive = handleCommonMessages orElse {
-    case FastSyncPivotBlockSelector.Result(pivotBlockHeader) =>
-      if (pivotBlockHeader.number < 1) {
-        log.info("Unable to start block synchronization in fast mode: pivot block is less than 1")
-        appStateStorage.fastSyncDone().commit()
-        context become idle
-        syncController ! Done
-      } else {
-        val initialSyncState =
-          SyncState(
-            pivotBlockHeader,
-            safeDownloadTarget = pivotBlockHeader.number + syncConfig.fastSyncBlockValidationX
-          )
-        startWithState(initialSyncState)
-      }
+  def waitingForPivotBlock: Receive = handleCommonMessages orElse { case PivotBlockSelector.Result(pivotBlockHeader) =>
+    if (pivotBlockHeader.number < 1) {
+      log.info("Unable to start block synchronization in fast mode: pivot block is less than 1")
+      appStateStorage.fastSyncDone().commit()
+      context become idle
+      syncController ! Done
+    } else {
+      val initialSyncState =
+        SyncState(
+          pivotBlockHeader,
+          safeDownloadTarget = pivotBlockHeader.number + syncConfig.fastSyncBlockValidationX
+        )
+      startWithState(initialSyncState)
+    }
   }
 
   // scalastyle:off number.of.methods
@@ -189,7 +188,7 @@ class FastSync(
     }
 
     def waitingForPivotBlockUpdate(processState: FinalBlockProcessingResult): Receive = handleCommonMessages orElse {
-      case FastSyncPivotBlockSelector.Result(pivotBlockHeader) =>
+      case PivotBlockSelector.Result(pivotBlockHeader) =>
         log.info(s"New pivot block with number ${pivotBlockHeader.number} received")
         if (pivotBlockHeader.number >= syncState.pivotBlock.number) {
           updatePivotSyncState(processState, pivotBlockHeader)
@@ -215,8 +214,10 @@ class FastSync(
         } else {
           log.info("Asking for new pivot block")
           val pivotBlockSelector =
-            context.actorOf(FastSyncPivotBlockSelector.props(etcPeerManager, peerEventBus, syncConfig, scheduler))
-          pivotBlockSelector ! FastSyncPivotBlockSelector.ChoosePivotBlock
+            context.actorOf(
+              PivotBlockSelector.props(etcPeerManager, peerEventBus, syncConfig, scheduler, context.self)
+            )
+          pivotBlockSelector ! PivotBlockSelector.SelectPivotBlock
           context become waitingForPivotBlockUpdate(state)
         }
       } else {

src/main/scala/io/iohk/ethereum/blockchain/sync/FastSyncPivotBlockSelector.scala renamed to src/main/scala/io/iohk/ethereum/blockchain/sync/PivotBlockSelector.scala

@@ -8,60 +8,48 @@ import io.iohk.ethereum.network.PeerEventBusActor.PeerEvent.MessageFromPeer
 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.network.{EtcPeerManagerActor, Peer, PeerId}
 import io.iohk.ethereum.utils.Config.SyncConfig
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.duration.FiniteDuration
 
-class FastSyncPivotBlockSelector(
+class PivotBlockSelector(
     val etcPeerManager: ActorRef,
     val peerEventBus: ActorRef,
     val syncConfig: SyncConfig,
-    val scheduler: Scheduler
+    val scheduler: Scheduler,
+    fastSync: ActorRef
 ) extends Actor
     with ActorLogging
     with PeerListSupport
     with BlacklistSupport {
 
-  import FastSyncPivotBlockSelector._
+  import PivotBlockSelector._
   import syncConfig._
 
-  val fastSync: ActorRef = context.parent
-
   def handleCommonMessages: Receive = handlePeerListMessages orElse handleBlacklistMessages
 
   override def receive: Receive = idle
 
-  def idle: Receive = handleCommonMessages orElse { case ChoosePivotBlock =>
-    val peersUsedToChooseTarget = peersToDownloadFrom.collect { case (peer, PeerInfo(_, _, true, maxBlockNumber, _)) =>
-      (peer, maxBlockNumber)
-    }
-
-    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 correctPeers = peersSortedByBestNumber
-      .takeWhile { case (_, number) => number >= expectedPivotBlock }
-      .map { case (peer, _) => peer }
+  def idle: Receive = handleCommonMessages orElse { case SelectPivotBlock =>
+    val election @ ElectionDetails(correctPeers, expectedPivotBlock) = collectVoters
 
-    if (correctPeers.size >= minPeersToChoosePivotBlock) {
+    if (election.isEnoughVoters(minPeersToChoosePivotBlock)) {
 
       val (peersToAsk, waitingPeers) = correctPeers.splitAt(minPeersToChoosePivotBlock + peersToChoosePivotBlockMargin)
 
       log.info(
         "Trying to choose fast sync pivot block using {} peers ({} correct ones). Ask {} peers for block nr {}",
-        peersUsedToChooseTarget.size,
+        peersToDownloadFrom.size,
         correctPeers.size,
         peersToAsk.size,
         expectedPivotBlock
       )
 
       peersToAsk.foreach(peer => obtainBlockHeaderFromPeer(peer.id, expectedPivotBlock))
 
-      val timeout = scheduler.scheduleOnce(peerResponseTimeout, self, PivotBlockTimeout)
-      context become waitingForPivotBlock(
+      val timeout = scheduler.scheduleOnce(peerResponseTimeout, self, ElectionPivotBlockTimeout)
+      context become runningPivotBlockElection(
         peersToAsk.map(_.id).toSet,
         waitingPeers.map(_.id),
         expectedPivotBlock,
@@ -73,14 +61,14 @@ class FastSyncPivotBlockSelector(
         "Cannot pick pivot block. Need at least {} peers, but there are only {} which meet the criteria ({} all available at the moment). Retrying in {}",
         minPeersToChoosePivotBlock,
         correctPeers.size,
-        peersUsedToChooseTarget.size,
+        peersToDownloadFrom.size,
         startRetryInterval
       )
       scheduleRetry(startRetryInterval)
     }
   }
 
-  def waitingForPivotBlock(
+  def runningPivotBlockElection(
       peersToAsk: Set[PeerId],
       waitingPeers: List[PeerId],
       pivotBlockNumber: BigInt,
@@ -105,7 +93,7 @@ class FastSyncPivotBlockSelector(
             blacklist(peerId, blacklistDuration, "Did not respond with pivot block header, blacklisting")
             votingProcess(updatedPeersToAsk, waitingPeers, pivotBlockNumber, timeout, headers)
         }
-      case PivotBlockTimeout =>
+      case ElectionPivotBlockTimeout =>
         peersToAsk.foreach { peerId =>
           blacklist(peerId, blacklistDuration, "Did not respond with pivot block header (timeout), blacklisting")
         }
@@ -130,12 +118,12 @@ class FastSyncPivotBlockSelector(
     } else if (!isPossibleToReachConsensus(peersToAsk.size, updatedVotes)) {
       timeout.cancel()
       if (waitingPeers.nonEmpty) { // There are more peers to ask
-        val newTimeout = scheduler.scheduleOnce(peerResponseTimeout, self, PivotBlockTimeout)
+        val newTimeout = scheduler.scheduleOnce(peerResponseTimeout, self, ElectionPivotBlockTimeout)
         val additionalPeer :: newWaitingPeers = waitingPeers
 
         obtainBlockHeaderFromPeer(additionalPeer, pivotBlockNumber)
 
-        context become waitingForPivotBlock(
+        context become runningPivotBlockElection(
           peersToAsk + additionalPeer,
           newWaitingPeers,
           pivotBlockNumber,
@@ -149,7 +137,7 @@ class FastSyncPivotBlockSelector(
       }
       // Continue voting
     } else {
-      context become waitingForPivotBlock(
+      context become runningPivotBlockElection(
         peersToAsk,
         waitingPeers,
         pivotBlockNumber,
@@ -163,7 +151,7 @@ class FastSyncPivotBlockSelector(
     peersLeft + bestHeaderVotes >= minPeersToChoosePivotBlock
 
   def scheduleRetry(interval: FiniteDuration): Unit = {
-    scheduler.scheduleOnce(interval, self, ChoosePivotBlock)
+    scheduler.scheduleOnce(interval, self, SelectPivotBlock)
     context become idle
   }
 
@@ -181,16 +169,39 @@ class FastSyncPivotBlockSelector(
       peer
     )
   }
-}
 
-object FastSyncPivotBlockSelector {
-  def props(etcPeerManager: ActorRef, peerEventBus: ActorRef, syncConfig: SyncConfig, scheduler: Scheduler): Props =
-    Props(new FastSyncPivotBlockSelector(etcPeerManager: ActorRef, peerEventBus, syncConfig, scheduler))
+  private def collectVoters: ElectionDetails = {
+    val peersUsedToChooseTarget = peersToDownloadFrom.collect { case (peer, PeerInfo(_, _, true, maxBlockNumber, _)) =>
+      (peer, maxBlockNumber)
+    }
 
-  case object ChoosePivotBlock
+    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 correctPeers = peersSortedByBestNumber
+      .takeWhile { case (_, number) => number >= expectedPivotBlock }
+      .map { case (peer, _) => peer }
+
+    ElectionDetails(correctPeers, expectedPivotBlock)
+  }
+}
+
+object PivotBlockSelector {
+  def props(
+      etcPeerManager: ActorRef,
+      peerEventBus: ActorRef,
+      syncConfig: SyncConfig,
+      scheduler: Scheduler,
+      fastSync: ActorRef
+  ): Props =
+    Props(new PivotBlockSelector(etcPeerManager: ActorRef, peerEventBus, syncConfig, scheduler, fastSync))
+
+  case object SelectPivotBlock
   case class Result(targetBlockHeader: BlockHeader)
 
-  case object PivotBlockTimeout
+  case object ElectionPivotBlockTimeout
 
   case class BlockHeaderWithVotes(header: BlockHeader, votes: Int = 1) {
     def vote: BlockHeaderWithVotes = copy(votes = votes + 1)
@@ -201,4 +212,8 @@ object FastSyncPivotBlockSelector {
       headerWithVotes.votes
     }._2
   }
+
+  case class ElectionDetails(participants: List[Peer], expectedPivotBlock: BigInt) {
+    def isEnoughVoters(minNumberOfVoters: Int): Boolean = participants.size >= minNumberOfVoters
+  }
 }