[ECTM-104] Add more tests and logs

Author: Michal Mrozek

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

@@ -64,8 +64,8 @@ class FastSyncItSpec extends FlatSpecBase with Matchers with BeforeAndAfter {
       _ <- peer1.startFastSync().delayExecution(50.milliseconds)
       _ <- peer1.waitForFastSyncFinish()
     } yield {
-      assert(peer1.bl.getBestBlockNumber() == peer2.bl.getBestBlockNumber() - peer2.testSyncConfig.targetBlockOffset)
-      assert(peer1.bl.getBestBlockNumber() == peer3.bl.getBestBlockNumber() - peer3.testSyncConfig.targetBlockOffset)
+      assert(peer1.bl.getBestBlockNumber() == peer2.bl.getBestBlockNumber() - peer2.testSyncConfig.pivotBlockOffset)
+      assert(peer1.bl.getBestBlockNumber() == peer3.bl.getBestBlockNumber() - peer3.testSyncConfig.pivotBlockOffset)
     }
   }
 
@@ -81,13 +81,13 @@ class FastSyncItSpec extends FlatSpecBase with Matchers with BeforeAndAfter {
         val trie = peer1.getBestBlockTrie()
         // due to the fact that function generating state is deterministic both peer2 and peer3 ends up with exactly same
         // state, so peer1 can get whole trie from both of them.
-        assert(peer1.bl.getBestBlockNumber() == peer2.bl.getBestBlockNumber() - peer2.testSyncConfig.targetBlockOffset)
-        assert(peer1.bl.getBestBlockNumber() == peer3.bl.getBestBlockNumber() - peer3.testSyncConfig.targetBlockOffset)
+        assert(peer1.bl.getBestBlockNumber() == peer2.bl.getBestBlockNumber() - peer2.testSyncConfig.pivotBlockOffset)
+        assert(peer1.bl.getBestBlockNumber() == peer3.bl.getBestBlockNumber() - peer3.testSyncConfig.pivotBlockOffset)
         assert(trie.isDefined)
       }
   }
 
-  it should "should update target block" in customTestCaseResourceM(FakePeer.start2FakePeersRes()) {
+  it should "should update pivot block" in customTestCaseResourceM(FakePeer.start2FakePeersRes()) {
     case (peer1, peer2) =>
       for {
         _ <- peer2.importBlocksUntil(1000)(IdentityUpdate)
@@ -96,7 +96,7 @@ class FastSyncItSpec extends FlatSpecBase with Matchers with BeforeAndAfter {
         _ <- peer1.startFastSync().delayExecution(50.milliseconds)
         _ <- peer1.waitForFastSyncFinish()
       } yield {
-        assert(peer1.bl.getBestBlockNumber() == peer2.bl.getBestBlockNumber() - peer2.testSyncConfig.targetBlockOffset)
+        assert(peer1.bl.getBestBlockNumber() == peer2.bl.getBestBlockNumber() - peer2.testSyncConfig.pivotBlockOffset)
       }
   }
 }
@@ -308,7 +308,7 @@ object FastSyncItSpec {
     lazy val validators = new MockValidatorsAlwaysSucceed
 
     val testSyncConfig = syncConfig.copy(
-      minPeersToChooseTargetBlock = 1,
+      minPeersToChoosePivotBlock = 1,
       peersScanInterval = 5.milliseconds,
       blockHeadersPerRequest = 200,
       blockBodiesPerRequest = 50,

src/main/resources/application.conf

@@ -320,12 +320,12 @@ mantis {
     # Requested number of MPT nodes when syncing from other peers
     nodes-per-request = 384
 
-    # Minimum number of peers required to start fast-sync (by determining the target block)
-    min-peers-to-choose-target-block = 2
+    # Minimum number of peers required to start fast-sync (by determining the pivot block)
+    min-peers-to-choose-pivot-block = 3
 
     # During fast-sync when most up to date block is determined from peers, the actual target block number
     # will be decreased by this value
-    target-block-offset = 128
+    pivot-block-offset = 128
 
     # How often to query peers for new blocks after the top of the chain has been reached
     check-for-new-block-interval = 10.seconds

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

@@ -140,7 +140,7 @@ class FastSync(
       scheduler.scheduleWithFixedDelay(syncRetryInterval, syncRetryInterval * 2, self, ProcessSyncing)
 
     def receive: Receive = handleCommonMessages orElse {
-      case UpdateTargetBlock(state) => updateTargetBlock(state)
+      case UpdatePivotBlock(state) => updatePivotBlock(state)
       case ProcessSyncing => processSyncing()
       case PrintStatus => printStatus()
       case PersistSyncState => persistSyncState()
@@ -189,34 +189,34 @@ class FastSync(
     }
 
     def waitingForTargetBlockUpdate(processState: FinalBlockProcessingResult): Receive = handleCommonMessages orElse {
-      case FastSyncPivotBlockSelector.Result(targetBlockHeader) =>
-        log.info(s"new target block with number ${targetBlockHeader.number} received")
-        if (targetBlockHeader.number >= syncState.pivotBlock.number) {
-          updateTargetSyncState(processState, targetBlockHeader)
+      case FastSyncPivotBlockSelector.Result(pivotBlockHeader) =>
+        log.info(s"New pivot block with number ${pivotBlockHeader.number} received")
+        if (pivotBlockHeader.number >= syncState.pivotBlock.number) {
+          updatePivotSyncState(processState, pivotBlockHeader)
           syncState = syncState.copy(updatingTargetBlock = false)
           context become this.receive
           processSyncing()
         } else {
           syncState = syncState.copy(targetBlockUpdateFailures = syncState.targetBlockUpdateFailures + 1)
-          scheduler.scheduleOnce(syncRetryInterval, self, UpdateTargetBlock(processState))
+          scheduler.scheduleOnce(syncRetryInterval, self, UpdatePivotBlock(processState))
         }
 
       case PersistSyncState => persistSyncState()
 
-      case UpdateTargetBlock(state) => updateTargetBlock(state)
+      case UpdatePivotBlock(state) => updatePivotBlock(state)
     }
 
-    private def updateTargetBlock(state: FinalBlockProcessingResult): Unit = {
+    private def updatePivotBlock(state: FinalBlockProcessingResult): Unit = {
       syncState = syncState.copy(updatingTargetBlock = true)
       if (syncState.targetBlockUpdateFailures <= syncConfig.maximumTargetUpdateFailures) {
         if (assignedHandlers.nonEmpty) {
           log.info(s"Still waiting for some responses, rescheduling target block update")
-          scheduler.scheduleOnce(syncRetryInterval, self, UpdateTargetBlock(state))
+          scheduler.scheduleOnce(syncRetryInterval, self, UpdatePivotBlock(state))
         } else {
-          log.info("Asking for new target block")
-          val targetBlockSelector =
+          log.info("Asking for new pivot block")
+          val pivotBlockSelector =
             context.actorOf(FastSyncPivotBlockSelector.props(etcPeerManager, peerEventBus, syncConfig, scheduler))
-          targetBlockSelector ! FastSyncPivotBlockSelector.ChoosePivotBlock
+          pivotBlockSelector ! FastSyncPivotBlockSelector.ChoosePivotBlock
           context become waitingForTargetBlockUpdate(state)
         }
       } else {
@@ -225,7 +225,7 @@ class FastSync(
       }
     }
 
-    private def updateTargetSyncState(state: FinalBlockProcessingResult, targetBlockHeader: BlockHeader): Unit =
+    private def updatePivotSyncState(state: FinalBlockProcessingResult, targetBlockHeader: BlockHeader): Unit =
       state match {
         case ImportedLastBlock =>
           if (targetBlockHeader.number - syncState.pivotBlock.number <= syncConfig.maxTargetDifference) {
@@ -340,7 +340,7 @@ class FastSync(
         discardLastBlocks(header.number, N)
         syncState = syncState.updateDiscardedBlocks(header, N)
         if (header.number >= syncState.pivotBlock.number) {
-          updateTargetBlock(LastBlockValidationFailed)
+          updatePivotBlock(LastBlockValidationFailed)
         } else {
           processSyncing()
         }
@@ -361,7 +361,7 @@ class FastSync(
           case HeadersProcessingFinished =>
             processSyncing()
           case ImportedTargetBlock =>
-            updateTargetBlock(ImportedLastBlock)
+            updatePivotBlock(ImportedLastBlock)
           case ValidationFailed(header, peerToBlackList) =>
             log.warning(s"validation fo header ${header.idTag} failed")
             handleRewind(header, peerToBlackList, syncConfig.fastSyncBlockValidationN)
@@ -839,7 +839,7 @@ object FastSync {
       )
     )
 
-  private case class UpdateTargetBlock(state: FinalBlockProcessingResult)
+  private case class UpdatePivotBlock(state: FinalBlockProcessingResult)
   private case object ProcessSyncing
   private[sync] case object PersistSyncState
   private case object PrintStatus

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

@@ -34,21 +34,21 @@ class FastSyncPivotBlockSelector(
   def idle: Receive = handleCommonMessages orElse { case ChoosePivotBlock =>
     val peersUsedToChooseTarget = peersToDownloadFrom.filter(_._2.forkAccepted)
 
-    if (peersUsedToChooseTarget.size >= minPeersToChooseTargetBlock) {
+    if (peersUsedToChooseTarget.size >= minPeersToChoosePivotBlock) {
       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)
+      log.info("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,
+        minPeersToChoosePivotBlock,
         peersUsedToChooseTarget.size,
         startRetryInterval
       )
@@ -61,6 +61,12 @@ class FastSyncPivotBlockSelector(
     handleCommonMessages orElse {
       case MessageFromPeer(BlockHeaders(Seq(blockHeader)), peerId) =>
         peerEventBus ! Unsubscribe(MessageClassifier(Set(BlockHeaders.code), PeerSelector.WithId(peerId)))
+        log.info(
+          s"Received block header [number: {}, hash: {}] from peer: {}",
+          blockHeader.number,
+          blockHeader.hashAsHexString,
+          peerId.value
+        )
 
         val newWaitingFor = waitingFor.filterNot(_.id == peerId)
 
@@ -98,12 +104,10 @@ class FastSyncPivotBlockSelector(
 
   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) {
-
+    if (receivedHeaders.size >= minPeersToChoosePivotBlock) {
       val peersWithBestHeaders = receivedHeaders.toList.sortBy(-_._2.number)
-
       val bestPeerBestBlockNumber = peersWithBestHeaders.head._2.number
-      val targetBlock = bestPeerBestBlockNumber - syncConfig.targetBlockOffset
+      val targetBlock = (bestPeerBestBlockNumber - syncConfig.pivotBlockOffset).max(0)
 
       val peersToAsk = peersWithBestHeaders.takeWhile(_._2.number >= targetBlock).map(_._1)
 
@@ -116,12 +120,12 @@ class FastSyncPivotBlockSelector(
       }
 
       val timeout = scheduler.scheduleOnce(peerResponseTimeout, self, PivotBlockTimeout)
-      context become waitingForTargetBlock(peersToAsk.map(_.id).toSet, targetBlock, timeout, Map.empty)
+      context become waitingForPivotBlock(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,
+        minPeersToChoosePivotBlock,
         receivedHeaders.size,
         startRetryInterval
       )
@@ -130,7 +134,7 @@ class FastSyncPivotBlockSelector(
     }
   }
 
-  def waitingForTargetBlock(
+  def waitingForPivotBlock(
       peersToAsk: Set[PeerId],
       targetBlockNumber: BigInt,
       timeout: Cancellable,
@@ -144,23 +148,25 @@ class FastSyncPivotBlockSelector(
 
         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 >= minPeersToChooseTargetBlock) {
+            if (votes >= minPeersToChoosePivotBlock) {
               timeout.cancel()
               sendResponseAndCleanup(mostPopularBlockHeader)
-            } else if (updatedPeersToAsk.isEmpty) {
+            } else if (updatedPeersToAsk.size + votes < minPeersToChoosePivotBlock) {
               timeout.cancel()
+              peerEventBus ! Unsubscribe()
               log.info("Not enough votes for pivot block. Retrying in {}", startRetryInterval)
               scheduleRetry(startRetryInterval)
               context become idle
             } else {
-              context become waitingForTargetBlock(updatedPeersToAsk, targetBlockNumber, timeout, updatedHeaders)
+              context become waitingForPivotBlock(updatedPeersToAsk, targetBlockNumber, timeout, updatedHeaders)
             }
           case None =>
             blacklist(peerId, blacklistDuration, "Did not respond with pivot block header, blacklisting")
-            context become waitingForTargetBlock(updatedPeersToAsk, targetBlockNumber, timeout, headers)
+            context become waitingForPivotBlock(updatedPeersToAsk, targetBlockNumber, timeout, headers)
         }
       case PivotBlockTimeout =>
         peersToAsk.foreach { peerId =>