ETCM-670: handling checkpoint older than local best block

Author: pslaski

src/main/scala/io/iohk/ethereum/blockchain/sync/regular/BlockFetcher.scala

@@ -8,8 +8,11 @@ import cats.data.NonEmptyList
 import cats.instances.option._
 import cats.syntax.either._
 import io.iohk.ethereum.blockchain.sync.PeersClient._
-import io.iohk.ethereum.blockchain.sync.regular.BlockFetcherState.{AwaitingBodiesToBeIgnored, AwaitingHeadersToBeIgnored}
-import io.iohk.ethereum.blockchain.sync.regular.BlockImporter.{ImportNewBlock, NewCheckpointBlock, NotOnTop, OnTop}
+import io.iohk.ethereum.blockchain.sync.regular.BlockFetcherState.{
+  AwaitingBodiesToBeIgnored,
+  AwaitingHeadersToBeIgnored
+}
+import io.iohk.ethereum.blockchain.sync.regular.BlockImporter.{ImportNewBlock, NotOnTop, OnTop}
 import io.iohk.ethereum.blockchain.sync.regular.RegularSync.ProgressProtocol
 import io.iohk.ethereum.consensus.validators.BlockValidator
 import io.iohk.ethereum.crypto.kec256
@@ -238,12 +241,16 @@ class BlockFetcher(
     state.tryInsertBlock(block, peerId) match {
       case Left(_) if block.number <= state.lastBlock =>
         log.debug(
-          s"Checkpoint block ${ByteStringUtils.hash2string(blockHash)} is older than current last block ${state.lastBlock}"
+          s"Checkpoint block ${ByteStringUtils.hash2string(blockHash)} is older than current last block ${state.lastBlock}" +
+            s" - clearing the queues and putting checkpoint to ready blocks queue"
         )
-        state.importer ! NewCheckpointBlock(block, peerId)
+        val newState = state
+          .clearQueues()
+          .enqueueReadyBlock(block, peerId)
+        fetchBlocks(newState)
       case Left(_) if block.number <= state.knownTop =>
         log.debug(
-          s"Checkpoint block ${ByteStringUtils.hash2string(blockHash)} not fit into queues - clearing the queues and setting new top"
+          s"Checkpoint block ${ByteStringUtils.hash2string(blockHash)} not fit into queues - clearing the queues and setting possible new top"
         )
         val newState = state
           .clearQueues()

src/main/scala/io/iohk/ethereum/blockchain/sync/regular/BlockFetcherState.scala

@@ -190,18 +190,21 @@ case class BlockFetcherState(
   def enqueueRequestedBlock(block: Block, fromPeer: PeerId): BlockFetcherState =
     waitingHeaders.dequeueOption
       .map { case (waitingHeader, waitingHeadersTail) =>
-        if (waitingHeader.hash == block.hash)
-          withPeerForBlocks(fromPeer, Seq(block.number))
+        if (waitingHeader.hash == block.hash) {
+          enqueueReadyBlock(block, fromPeer)
             .withPossibleNewTopAt(block.number)
             .copy(
-              readyBlocks = readyBlocks.enqueue(block),
               waitingHeaders = waitingHeadersTail
             )
-        else
+        } else
           this
       }
       .getOrElse(this)
 
+  def enqueueReadyBlock(block: Block, fromPeer: PeerId): BlockFetcherState =
+    withPeerForBlocks(fromPeer, Seq(block.number))
+      .copy(readyBlocks = readyBlocks.enqueue(block))
+
   def pickBlocks(amount: Int): Option[(NonEmptyList[Block], BlockFetcherState)] =
     if (readyBlocks.nonEmpty) {
       val (picked, rest) = readyBlocks.splitAt(amount)

src/main/scala/io/iohk/ethereum/blockchain/sync/regular/BlockImporter.scala

@@ -91,8 +91,6 @@ class BlockImporter(
         }
       }
 
-    case NewCheckpointBlock(block, peerId) => importNewBlock(block, peerId, state)
-
     case ImportNewBlock(block, peerId) if state.isOnTop && !state.importing => importNewBlock(block, peerId, state)
 
     case ImportDone(newBehavior, importType) =>
@@ -389,7 +387,6 @@ object BlockImporter {
   case object NotOnTop extends ImporterMsg
   case class MinedBlock(block: Block) extends ImporterMsg
   case class NewCheckpoint(parentHash: ByteString, signatures: Seq[ECDSASignature]) extends ImporterMsg
-  case class NewCheckpointBlock(block: Block, peerId: PeerId) extends ImporterMsg
   case class ImportNewBlock(block: Block, peerId: PeerId) extends ImporterMsg
   case class ImportDone(newBehavior: NewBehavior, blockImportType: BlockImportType) extends ImporterMsg
   case object PickBlocks extends ImporterMsg

src/test/scala/io/iohk/ethereum/blockchain/sync/regular/BlockFetcherSpec.scala

@@ -434,7 +434,7 @@ class BlockFetcherSpec
       }
     }
 
-    "should inform importer when checkpoint block is older than last block" in new TestSetup {
+    "should put checkpoint to ready blocks when checkpoint block is older than last block" in new TestSetup {
       startFetcher()
 
       triggerFetching(10)
@@ -469,7 +469,17 @@ class BlockFetcherSpec
 
       importer.expectMsg(BlockImporter.OnTop)
 
-      importer.expectMsg(BlockImporter.NewCheckpointBlock(checkpointBlock, fakePeer.id))
+      // We need to wait a while in order to allow fetcher to process all the blocks
+      system.scheduler.scheduleOnce(Timeouts.shortTimeout) {
+        importer.send(blockFetcher, PickBlocks(syncConfig.blocksBatchSize))
+      }
+
+      importer.expectMsgPF() { case BlockFetcher.PickedBlocks(blocks) =>
+        val headers = blocks.map(_.header).toList
+
+        assert(HeadersSeq.areChain(headers))
+        assert(headers.contains(checkpointBlock.header))
+      }
     }
 
     "should properly handle a request timeout" in new TestSetup {