[ECTM-104] Pivot block selection algorithm

Author: Michal Mrozek

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

@@ -0,0 +1,197 @@
+package io.iohk.ethereum.blockchain.sync
+
+import akka.actor.{Actor, ActorLogging, ActorRef, Cancellable, Props, Scheduler}
+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.p2p.messages.PV62.{BlockHeaders, GetBlockHeaders}
+import io.iohk.ethereum.utils.Config.SyncConfig
+import scala.concurrent.duration.FiniteDuration
+import scala.concurrent.ExecutionContext.Implicits.global
+
+class FastSyncPivotBlockSelector(
+    val etcPeerManager: ActorRef,
+    val peerEventBus: ActorRef,
+    val syncConfig: SyncConfig,
+    val scheduler: Scheduler
+) extends Actor
+    with ActorLogging
+    with PeerListSupport
+    with BlacklistSupport {
+
+  import FastSyncPivotBlockSelector._
+  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.filter(_._2.forkAccepted)
+
+    if (peersUsedToChooseTarget.size >= minPeersToChooseTargetBlock) {
+      peersUsedToChooseTarget.foreach { case (peer, PeerInfo(status, _, _, _, _)) =>
+        peerEventBus ! Subscribe(MessageClassifier(Set(BlockHeaders.code), PeerSelector.WithId(peer.id)))
+        etcPeerManager ! EtcPeerManagerActor.SendMessage(
+          GetBlockHeaders(Right(status.bestHash), 1, 0, reverse = false),
+          peer.id
+        )
+      }
+      log.debug("Asking {} peers for block headers", peersUsedToChooseTarget.size)
+      val timeout = scheduler.scheduleOnce(peerResponseTimeout, self, BlockHeadersTimeout)
+      context become waitingForBlockHeaders(peersUsedToChooseTarget.keySet, Map.empty, timeout)
+    } else {
+      log.info(
+        "Cannot pick pivot block. Need at least {} peers, but there are only {} available at the moment. Retrying in {}",
+        minPeersToChooseTargetBlock,
+        peersUsedToChooseTarget.size,
+        startRetryInterval
+      )
+      scheduleRetry(startRetryInterval)
+      context become idle
+    }
+  }
+
+  def waitingForBlockHeaders(waitingFor: Set[Peer], received: Map[Peer, BlockHeader], timeout: Cancellable): Receive =
+    handleCommonMessages orElse {
+      case MessageFromPeer(BlockHeaders(Seq(blockHeader)), peerId) =>
+        peerEventBus ! Unsubscribe(MessageClassifier(Set(BlockHeaders.code), PeerSelector.WithId(peerId)))
+
+        val newWaitingFor = waitingFor.filterNot(_.id == peerId)
+
+        waitingFor.find(_.id == peerId).foreach { peer =>
+          val newReceived = received + (peer -> blockHeader)
+
+          if (newWaitingFor.isEmpty) {
+            timeout.cancel()
+            tryChooseTargetBlock(newReceived)
+          } else context become waitingForBlockHeaders(newWaitingFor, newReceived, timeout)
+        }
+
+      case MessageFromPeer(BlockHeaders(blockHeaders), peerId) =>
+        peerEventBus ! Unsubscribe(MessageClassifier(Set(BlockHeaders.code), PeerSelector.WithId(peerId)))
+        blacklist(
+          peerId,
+          blacklistDuration,
+          s"did not respond with 1 header but with ${blockHeaders.size}, blacklisting for $blacklistDuration"
+        )
+        waitingFor.find(_.id == peerId).foreach { peer =>
+          context become waitingForBlockHeaders(waitingFor - peer, received, timeout)
+        }
+
+      case BlockHeadersTimeout =>
+        waitingFor.foreach { peer =>
+          peerEventBus ! Unsubscribe(MessageClassifier(Set(BlockHeaders.code), PeerSelector.WithId(peer.id)))
+          blacklist(
+            peer.id,
+            blacklistDuration,
+            s"did not respond within required time with block header, blacklisting for $blacklistDuration"
+          )
+        }
+        tryChooseTargetBlock(received)
+    }
+
+  def tryChooseTargetBlock(receivedHeaders: Map[Peer, BlockHeader]): Unit = {
+    log.debug("Trying to choose fast sync pivot block. Received {} block headers", receivedHeaders.size)
+    if (receivedHeaders.size >= minPeersToChooseTargetBlock) {
+
+      val peersWithBestHeaders = receivedHeaders.toList.sortBy(-_._2.number)
+
+      val bestPeerBestBlockNumber = peersWithBestHeaders.head._2.number
+      val targetBlock = bestPeerBestBlockNumber - syncConfig.targetBlockOffset
+
+      val peersToAsk = peersWithBestHeaders.takeWhile(_._2.number >= targetBlock).map(_._1)
+
+      peersToAsk.foreach { peer =>
+        peerEventBus ! Subscribe(MessageClassifier(Set(BlockHeaders.code), PeerSelector.WithId(peer.id)))
+        etcPeerManager ! EtcPeerManagerActor.SendMessage(
+          GetBlockHeaders(Left(targetBlock), 1, 0, reverse = false),
+          peer.id
+        )
+      }
+
+      val timeout = scheduler.scheduleOnce(peerResponseTimeout, self, PivotBlockTimeout)
+      context become waitingForTargetBlock(peersToAsk.map(_.id).toSet, targetBlock, timeout, Map.empty)
+
+    } else {
+      log.info(
+        "Cannot pick pivot block. Need to receive block headers from at least {} peers, but received only from {}. Retrying in {}",
+        minPeersToChooseTargetBlock,
+        receivedHeaders.size,
+        startRetryInterval
+      )
+      scheduleRetry(startRetryInterval)
+      context become idle
+    }
+  }
+
+  def waitingForTargetBlock(
+      peersToAsk: Set[PeerId],
+      targetBlockNumber: BigInt,
+      timeout: Cancellable,
+      headers: Map[BlockHeader, Int]
+  ): 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)
+
+        targetBlockHeaderOpt match {
+          case Some(targetBlockHeader) =>
+            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 >= minPeersToChooseTargetBlock) {
+              timeout.cancel()
+              sendResponseAndCleanup(mostPopularBlockHeader)
+            } else if (updatedPeersToAsk.isEmpty) {
+              timeout.cancel()
+              log.info("Not enough votes for pivot block. Retrying in {}", startRetryInterval)
+              scheduleRetry(startRetryInterval)
+              context become idle
+            } else {
+              context become waitingForTargetBlock(updatedPeersToAsk, targetBlockNumber, timeout, updatedHeaders)
+            }
+          case None =>
+            blacklist(peerId, blacklistDuration, "Did not respond with pivot block header, blacklisting")
+            context become waitingForTargetBlock(updatedPeersToAsk, targetBlockNumber, timeout, headers)
+        }
+      case PivotBlockTimeout =>
+        peersToAsk.foreach { peerId =>
+          blacklist(peerId, blacklistDuration, "Did not respond with pivot block header (timeout), blacklisting")
+        }
+        peerEventBus ! Unsubscribe()
+        log.info("Pivot block header receive timeout. Retrying in {}", startRetryInterval)
+        scheduleRetry(startRetryInterval)
+        context become idle
+    }
+
+  def scheduleRetry(interval: FiniteDuration): Unit = {
+    scheduler.scheduleOnce(interval, self, ChoosePivotBlock)
+  }
+
+  def sendResponseAndCleanup(pivotBlockHeader: BlockHeader): Unit = {
+    log.info("Found pivot block: {} hash: {}", pivotBlockHeader.number, pivotBlockHeader.hashAsHexString)
+    fastSync ! Result(pivotBlockHeader)
+    peerEventBus ! Unsubscribe()
+    context stop self
+  }
+
+}
+
+object FastSyncPivotBlockSelector {
+  def props(etcPeerManager: ActorRef, peerEventBus: ActorRef, syncConfig: SyncConfig, scheduler: Scheduler): Props =
+    Props(new FastSyncPivotBlockSelector(etcPeerManager: ActorRef, peerEventBus, syncConfig, scheduler))
+
+  case object ChoosePivotBlock
+  case class Result(targetBlockHeader: BlockHeader)
+
+  private case object BlockHeadersTimeout
+  private case object PivotBlockTimeout
+}