[ETCM-685] Improve pivot block selection (#949)

* [ETCM-685] Add retry logic to PivotBlockSelector

* [ETCM-685] Add config entries

* [ETCM-685] Add more unit test

* [ETCM-685] Fix best block number being pushed back to -ve numbers

Author: robinraju

src/main/resources/conf/base.conf

@@ -473,6 +473,13 @@ mantis {
     # Maximum number of retries performed by fast sync when the master peer sends invalid block headers.
     # On reaching this limit, it will perform branch resolving.
     fast-sync-max-batch-retries = 5
+
+    # If the expected pivot block cannot be confirmed from `min-peers-to-choose-pivot-block`,
+    # the pivot block number is pushed back by the follwing number of blocks and the confirmation process repeats.
+    pivot-block-number-reset-delta = 50
+
+    # Max number of times a pivot block is checked against available best peers before the whole process is restarted.
+    max-pivot-block-failures-count = 5
   }
 
   pruning {

src/main/scala/io/iohk/ethereum/blockchain/sync/fast/PivotBlockSelector.scala

@@ -35,13 +35,19 @@ class PivotBlockSelector(
   import PivotBlockSelector._
   import syncConfig._
 
+  private var pivotBlockRetryCount = 0
+
   override def receive: Receive = idle
 
   private def idle: Receive = handlePeerListMessages orElse { case SelectPivotBlock =>
-    val election @ ElectionDetails(correctPeers, expectedPivotBlock) = collectVoters
+    val electionDetails = collectVoters()
+    startPivotBlockSelection(electionDetails)
+  }
 
-    if (election.isEnoughVoters(minPeersToChoosePivotBlock)) {
+  private def startPivotBlockSelection(election: ElectionDetails): Unit = {
+    val ElectionDetails(correctPeers, currentBestBlockNumber, expectedPivotBlock) = election
 
+    if (election.hasEnoughVoters(minPeersToChoosePivotBlock)) {
       val (peersToAsk, waitingPeers) = correctPeers.splitAt(minPeersToChoosePivotBlock + peersToChoosePivotBlockMargin)
 
       log.info(
@@ -64,17 +70,35 @@ class PivotBlockSelector(
       )
     } else {
       log.info(
-        "Cannot pick pivot block. Need at least {} peers, but there are only {} which meet the criteria ({} all 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).",
         minPeersToChoosePivotBlock,
         correctPeers.size,
         peersToDownloadFrom.size,
+        currentBestBlockNumber
+      )
+      retryPivotBlockSelection(currentBestBlockNumber)
+    }
+  }
+
+  // Voters are collected until minimum peers to choose pivot block is obtained.
+  private def retryPivotBlockSelection(pivotBlockNumber: BigInt): Unit = {
+    pivotBlockRetryCount += 1
+    if (pivotBlockRetryCount <= maxPivotBlockFailuresCount && pivotBlockNumber > 0) {
+      val electionDetails = collectVoters(Some(pivotBlockNumber))
+      startPivotBlockSelection(electionDetails)
+    } else {
+      log.debug(
+        "Cannot pick pivot block. Current best block number [{}]. Retrying in [{}]",
+        pivotBlockNumber,
         startRetryInterval
       )
+      // Restart the whole process.
       scheduleRetry(startRetryInterval)
     }
   }
 
-  def runningPivotBlockElection(
+  private def runningPivotBlockElection(
       peersToAsk: Set[PeerId],
       waitingPeers: List[PeerId],
       pivotBlockNumber: BigInt,
@@ -85,10 +109,7 @@ class PivotBlockSelector(
       case MessageFromPeer(blockHeaders: BlockHeaders, peerId) =>
         peerEventBus ! Unsubscribe(MessageClassifier(Set(Codes.BlockHeadersCode), PeerSelector.WithId(peerId)))
         val updatedPeersToAsk = peersToAsk - peerId
-        val targetBlockHeaderOpt =
-          if (blockHeaders.headers.size != 1) None
-          else
-            blockHeaders.headers.find(header => header.number == pivotBlockNumber)
+        val targetBlockHeaderOpt = blockHeaders.headers.find(header => header.number == pivotBlockNumber)
         targetBlockHeaderOpt match {
           case Some(targetBlockHeader) =>
             val newValue =
@@ -159,6 +180,7 @@ class PivotBlockSelector(
     peersLeft + bestHeaderVotes >= minPeersToChoosePivotBlock
 
   private def scheduleRetry(interval: FiniteDuration): Unit = {
+    pivotBlockRetryCount = 0
     scheduler.scheduleOnce(interval, self, SelectPivotBlock)
     context become idle
   }
@@ -178,7 +200,7 @@ class PivotBlockSelector(
     )
   }
 
-  private def collectVoters: ElectionDetails = {
+  private def collectVoters(previousBestBlockNumber: Option[BigInt] = None): ElectionDetails = {
     val peersUsedToChooseTarget = peersToDownloadFrom.collect {
       case (_, PeerWithInfo(peer, PeerInfo(_, _, true, maxBlockNumber, _))) =>
         (peer, maxBlockNumber)
@@ -188,12 +210,20 @@ class PivotBlockSelector(
     val bestPeerBestBlockNumber = peersSortedByBestNumber.headOption
       .map { case (_, bestPeerBestBlockNumber) => bestPeerBestBlockNumber }
       .getOrElse(BigInt(0))
-    val expectedPivotBlock = (bestPeerBestBlockNumber - syncConfig.pivotBlockOffset).max(0)
+
+    // The current best block number is pushed back by `pivotBlockNumberResetDelta`
+    // if this request is issued by the retry logic.
+    val currentBestBlockNumber: BigInt =
+      previousBestBlockNumber
+        .map(_ - pivotBlockNumberResetDelta.max(0))
+        .getOrElse(bestPeerBestBlockNumber)
+
+    val expectedPivotBlock = (currentBestBlockNumber - syncConfig.pivotBlockOffset).max(0)
     val correctPeers = peersSortedByBestNumber
       .takeWhile { case (_, number) => number >= expectedPivotBlock }
       .map { case (peer, _) => peer }
 
-    ElectionDetails(correctPeers, expectedPivotBlock)
+    ElectionDetails(correctPeers, currentBestBlockNumber, expectedPivotBlock)
   }
 }
 
@@ -209,7 +239,7 @@ object PivotBlockSelector {
     Props(new PivotBlockSelector(etcPeerManager: ActorRef, peerEventBus, syncConfig, scheduler, fastSync, blacklist))
 
   case object SelectPivotBlock
-  case class Result(targetBlockHeader: BlockHeader)
+  final case class Result(targetBlockHeader: BlockHeader)
 
   case object ElectionPivotBlockTimeout
 
@@ -223,7 +253,11 @@ object PivotBlockSelector {
     }
   }
 
-  case class ElectionDetails(participants: List[Peer], expectedPivotBlock: BigInt) {
-    def isEnoughVoters(minNumberOfVoters: Int): Boolean = participants.size >= minNumberOfVoters
+  final case class ElectionDetails(
+      participants: List[Peer],
+      currentBestBlockNumber: BigInt,
+      expectedPivotBlock: BigInt
+  ) {
+    def hasEnoughVoters(minNumberOfVoters: Int): Boolean = participants.size >= minNumberOfVoters
   }
 }

src/main/scala/io/iohk/ethereum/utils/Config.scala

@@ -135,7 +135,9 @@ object Config {
       stateSyncPersistBatchSize: Int,
       pivotBlockReScheduleInterval: FiniteDuration,
       maxPivotBlockAge: Int,
-      fastSyncMaxBatchRetries: Int
+      fastSyncMaxBatchRetries: Int,
+      pivotBlockNumberResetDelta: Int,
+      maxPivotBlockFailuresCount: Int
   )
 
   object SyncConfig {
@@ -179,7 +181,9 @@ object Config {
         stateSyncPersistBatchSize = syncConfig.getInt("state-sync-persist-batch-size"),
         pivotBlockReScheduleInterval = syncConfig.getDuration("pivot-block-reschedule-interval").toMillis.millis,
         maxPivotBlockAge = syncConfig.getInt("max-pivot-block-age"),
-        fastSyncMaxBatchRetries = syncConfig.getInt("fast-sync-max-batch-retries")
+        fastSyncMaxBatchRetries = syncConfig.getInt("fast-sync-max-batch-retries"),
+        pivotBlockNumberResetDelta = syncConfig.getInt("pivot-block-number-reset-delta"),
+        maxPivotBlockFailuresCount = syncConfig.getInt("max-pivot-block-failures-count")
       )
     }
   }

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

@@ -418,6 +418,78 @@ class PivotBlockSelectorSpec
     peerMessageBus.expectMsg(Unsubscribe())
   }
 
+  it should "retry pivot block election with fallback to lower peer numbers" in new TestSetup {
+
+    override val minPeersToChoosePivotBlock = 2
+    override val peersToChoosePivotBlockMargin = 1
+
+    updateHandshakedPeers(
+      HandshakedPeers(
+        allPeers
+          .updated(peer1, allPeers(peer1).copy(maxBlockNumber = 2000))
+          .updated(peer2, allPeers(peer2).copy(maxBlockNumber = 800))
+          .updated(peer3, allPeers(peer3).copy(maxBlockNumber = 900))
+          .updated(peer4, allPeers(peer4).copy(maxBlockNumber = 1400))
+      )
+    )
+
+    pivotBlockSelector ! SelectPivotBlock
+
+    peerMessageBus.expectMsgAllOf(
+      Subscribe(MessageClassifier(Set(Codes.BlockHeadersCode), PeerSelector.WithId(peer1.id))),
+      Subscribe(MessageClassifier(Set(Codes.BlockHeadersCode), PeerSelector.WithId(peer4.id)))
+    )
+
+    etcPeerManager.expectMsgAllOf(
+      EtcPeerManagerActor.SendMessage(GetBlockHeaders(Left(1400), 1, 0, reverse = false), peer1.id),
+      EtcPeerManagerActor.SendMessage(GetBlockHeaders(Left(1400), 1, 0, reverse = false), peer4.id)
+    )
+    etcPeerManager.expectNoMessage()
+
+    // Collecting pivot block (for voting)
+    pivotBlockSelector ! MessageFromPeer(BlockHeaders(Seq(baseBlockHeader.copy(number = 1400))), peer1.id)
+    pivotBlockSelector ! MessageFromPeer(BlockHeaders(Seq(baseBlockHeader.copy(number = 1400))), peer4.id)
+
+    peerMessageBus.expectMsgAllOf(
+      Unsubscribe(MessageClassifier(Set(Codes.BlockHeadersCode), PeerSelector.WithId(peer1.id))),
+      Unsubscribe(MessageClassifier(Set(Codes.BlockHeadersCode), PeerSelector.WithId(peer4.id))),
+      Unsubscribe()
+    )
+    peerMessageBus.expectNoMessage()
+
+    fastSync.expectMsg(Result(baseBlockHeader.copy(number = 1400)))
+  }
+
+  it should "restart pivot block selection after `maxPivotBlockFailuresCount` is reached" in new TestSetup {
+
+    override val minPeersToChoosePivotBlock = 2
+    override val peersToChoosePivotBlockMargin = 1
+
+    updateHandshakedPeers(
+      HandshakedPeers(
+        allPeers
+          .updated(peer1, allPeers(peer1).copy(maxBlockNumber = 2000))
+          .updated(peer2, allPeers(peer2).copy(maxBlockNumber = 800))
+          .updated(peer3, allPeers(peer3).copy(maxBlockNumber = 900))
+          .updated(peer4, allPeers(peer4).copy(maxBlockNumber = 1000))
+      )
+    )
+
+    pivotBlockSelector ! SelectPivotBlock
+
+    peerMessageBus.expectNoMessage()
+
+    updateHandshakedPeers(HandshakedPeers(threeAcceptedPeers))
+
+    time.advance(syncConfig.startRetryInterval)
+
+    peerMessageBus.expectMsgAllOf(
+      Subscribe(MessageClassifier(Set(Codes.BlockHeadersCode), PeerSelector.WithId(peer1.id))),
+      Subscribe(MessageClassifier(Set(Codes.BlockHeadersCode), PeerSelector.WithId(peer2.id))),
+      Subscribe(MessageClassifier(Set(Codes.BlockHeadersCode), PeerSelector.WithId(peer3.id)))
+    )
+  }
+
   class TestSetup extends TestSyncConfig {
 
     val blacklist: Blacklist = CacheBasedBlacklist.empty(100)

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

@@ -44,7 +44,9 @@ trait TestSyncConfig extends SyncConfigBuilder {
     stateSyncPersistBatchSize = 1000,
     pivotBlockReScheduleInterval = 1.second,
     maxPivotBlockAge = 96,
-    fastSyncMaxBatchRetries = 3
+    fastSyncMaxBatchRetries = 3,
+    pivotBlockNumberResetDelta = 50,
+    maxPivotBlockFailuresCount = 3
   )
 
   override lazy val syncConfig: SyncConfig = defaultSyncConfig