Merge pull request #942 from input-output-hk/feature/ETCM-655

Feature/etcm 655

Author: AnastasiiaL

src/main/resources/application.conf

@@ -410,17 +410,15 @@ mantis {
     # we found a fork. To resolve it, we need to query the same peer for previous headers, to find a common ancestor.
     branch-resolution-request-size = 30
 
-    # TODO investigate proper value to handle ETC reorgs correctly
     # threshold for storing non-main-chain blocks in queue.
     # if: current_best_block_number - block_number > max-queued-block-number-behind
     # then: the block will not be queued (such already queued blocks will be removed)
-    max-queued-block-number-behind = 100
+    max-queued-block-number-behind = 1000
 
-    # TODO investigate proper value to handle ETC reorgs correctly
     # threshold for storing non-main-chain blocks in queue.
     # if: block_number - current_best_block_number > max-queued-block-number-ahead
     # then: the block will not be queued (such already queued blocks will be removed)
-    max-queued-block-number-ahead = 100
+    max-queued-block-number-ahead = 1000
 
     # Maximum number of blocks, after which block hash from NewBlockHashes packet is considered ancient
     # and peer sending it is blacklisted
@@ -623,7 +621,8 @@ akka {
   loglevel = "INFO"
   logging-filter = "akka.event.slf4j.Slf4jLoggingFilter"
   logger-startup-timeout = 30s
-  log-dead-letters = off
+  log-dead-letters-during-shutdown = off
+  log-dead-letters = 5
 
   coordinated-shutdown.phases {
     actor-system-terminate {

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

@@ -1,8 +1,10 @@
 package io.iohk.ethereum.blockchain.sync.regular
 
 import akka.actor.Status.Failure
-import akka.actor.{Actor, ActorLogging, ActorRef, Props}
+import akka.actor.{Actor, ActorLogging, ActorRef, ActorSystem, Props}
 import akka.pattern.{ask, pipe}
+import akka.stream.scaladsl.{Keep, Sink, Source, SourceQueue}
+import akka.stream.{Attributes, DelayOverflowStrategy, OverflowStrategy}
 import akka.util.{ByteString, Timeout}
 import cats.data.NonEmptyList
 import cats.instances.option._
@@ -45,8 +47,24 @@ class BlockFetcher(
   import BlockFetcher._
 
   implicit val ec: MonixScheduler = MonixScheduler(context.dispatcher)
+  implicit val sys: ActorSystem = context.system
   implicit val timeout: Timeout = syncConfig.peerResponseTimeout + 2.second // some margin for actor communication
 
+  private val queue: SourceQueue[BlockFetcherState] = {
+    val cap = 1000
+    val numberOfElements = 1
+    Source
+      .queue[BlockFetcherState](cap, OverflowStrategy.backpressure)
+      .delay(5.seconds, DelayOverflowStrategy.dropHead)
+      .addAttributes(Attributes.inputBuffer(numberOfElements, numberOfElements))
+      .map { s =>
+        log.debug("Resuming fetching with the latest state")
+        fetchBlocks(s.withResumedFetching)
+      }
+      .toMat(Sink.ignore)(Keep.left)
+      .run()
+  }
+
   override def receive: Receive = idle()
 
   override def postStop(): Unit = {
@@ -103,7 +121,7 @@ class BlockFetcher(
   }
 
   private def handleHeadersMessages(state: BlockFetcherState): Receive = {
-    case Response(peer, BlockHeaders(headers)) if state.isFetchingHeaders =>
+    case Response(_, BlockHeaders(headers)) if state.isFetchingHeaders =>
       val newState =
         if (state.fetchingHeadersState == AwaitingHeadersToBeIgnored) {
           log.debug(
@@ -153,7 +171,7 @@ class BlockFetcher(
               state.withBodiesFetchReceived.handleRequestedBlocks(newBlocks, peer.id)
           }
         val waitingHeadersDequeued = state.waitingHeaders.size - newState.waitingHeaders.size
-        log.debug(s"Processed ${waitingHeadersDequeued} new blocks from received block bodies")
+        log.debug(s"Processed $waitingHeadersDequeued new blocks from received block bodies")
         fetchBlocks(newState)
       }
     case RetryBodiesRequest if state.isFetchingBodies =>
@@ -267,20 +285,32 @@ class BlockFetcher(
   }
 
   private def handlePossibleTopUpdate(state: BlockFetcherState): Receive = {
-    //by handling these type of messages, fetcher can received from network, fresh info about blocks on top
+    //by handling these type of messages, fetcher can receive from network fresh info about blocks on top
     //ex. After a successful handshake, fetcher will receive the info about the header of the peer best block
     case MessageFromPeer(BlockHeaders(headers), _) =>
       headers.lastOption.map { bh =>
         log.debug(s"Candidate for new top at block ${bh.number}, current known top ${state.knownTop}")
         val newState = state.withPossibleNewTopAt(bh.number)
         fetchBlocks(newState)
       }
-    //keep fetcher state updated in case new checkpoint block or mined block was imported
+    //keep fetcher state updated in case new mined block was imported
     case InternalLastBlockImport(blockNr) =>
-      log.debug(s"New last block $blockNr imported from the inside")
+      log.debug(s"New mined block $blockNr imported from the inside")
       val newState = state.withLastBlock(blockNr).withPossibleNewTopAt(blockNr)
 
       fetchBlocks(newState)
+
+    //keep fetcher state updated in case new checkpoint block was imported
+    case InternalCheckpointImport(blockNr) =>
+      log.debug(s"New checkpoint block $blockNr imported from the inside")
+
+      val newState = state
+        .clearQueues()
+        .withLastBlock(blockNr)
+        .withPossibleNewTopAt(blockNr)
+        .withPausedFetching
+
+      fetchBlocks(newState)
   }
 
   private def handlePickedBlocks(
@@ -296,9 +326,14 @@ class BlockFetcher(
   private def fetchBlocks(state: BlockFetcherState): Unit = {
     // Remember that tryFetchHeaders and tryFetchBodies can issue a request
     // Nice and clean way to express that would be to use SyncIO from cats-effect
-    val newState = state |> tryFetchHeaders |> tryFetchBodies
 
-    context become started(newState)
+    if (state.pausedFetching) {
+      queue.offer(state)
+      context become started(state)
+    } else {
+      val newState = state |> tryFetchHeaders |> tryFetchBodies
+      context become started(newState)
+    }
   }
 
   private def tryFetchHeaders(fetcherState: BlockFetcherState): BlockFetcherState =
@@ -329,7 +364,7 @@ class BlockFetcher(
       .filter(!_.isFetchingBodies)
       .filter(_.waitingHeaders.nonEmpty)
       .tap(fetchBodies)
-      .map(state => state.withNewBodiesFetch)
+      .map(_.withNewBodiesFetch)
       .getOrElse(fetcherState)
 
   private def fetchBodies(state: BlockFetcherState): Unit = {
@@ -403,13 +438,13 @@ object BlockFetcher {
     Props(new BlockFetcher(peersClient, peerEventBus, supervisor, syncConfig, blockValidator))
 
   sealed trait FetchMsg
-  case class Start(importer: ActorRef, fromBlock: BigInt) extends FetchMsg
-  case class FetchStateNode(hash: ByteString) extends FetchMsg
-  case object RetryFetchStateNode extends FetchMsg
-  case class PickBlocks(amount: Int) extends FetchMsg
-  case class StrictPickBlocks(from: BigInt, atLEastWith: BigInt) extends FetchMsg
-  case object PrintStatus extends FetchMsg
-  case class InvalidateBlocksFrom(fromBlock: BigInt, reason: String, toBlacklist: Option[BigInt]) extends FetchMsg
+  final case class Start(importer: ActorRef, fromBlock: BigInt) extends FetchMsg
+  final case class FetchStateNode(hash: ByteString) extends FetchMsg
+  final case object RetryFetchStateNode extends FetchMsg
+  final case class PickBlocks(amount: Int) extends FetchMsg
+  final case class StrictPickBlocks(from: BigInt, atLEastWith: BigInt) extends FetchMsg
+  final case object PrintStatus extends FetchMsg
+  final case class InvalidateBlocksFrom(fromBlock: BigInt, reason: String, toBlacklist: Option[BigInt]) extends FetchMsg
 
   object InvalidateBlocksFrom {
 
@@ -419,12 +454,13 @@ object BlockFetcher {
     def apply(from: BigInt, reason: String, toBlacklist: Option[BigInt]): InvalidateBlocksFrom =
       new InvalidateBlocksFrom(from, reason, toBlacklist)
   }
-  case class BlockImportFailed(blockNr: BigInt, reason: String) extends FetchMsg
-  case class InternalLastBlockImport(blockNr: BigInt) extends FetchMsg
-  case object RetryBodiesRequest extends FetchMsg
-  case object RetryHeadersRequest extends FetchMsg
+  final case class BlockImportFailed(blockNr: BigInt, reason: String) extends FetchMsg
+  final case class InternalLastBlockImport(blockNr: BigInt) extends FetchMsg
+  final case class InternalCheckpointImport(blockNr: BigInt) extends FetchMsg
+  final case object RetryBodiesRequest extends FetchMsg
+  final case object RetryHeadersRequest extends FetchMsg
 
   sealed trait FetchResponse
-  case class PickedBlocks(blocks: NonEmptyList[Block]) extends FetchResponse
-  case class FetchedStateNode(stateNode: NodeData) extends FetchResponse
+  final case class PickedBlocks(blocks: NonEmptyList[Block]) extends FetchResponse
+  final case class FetchedStateNode(stateNode: NodeData) extends FetchResponse
 }

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

@@ -41,6 +41,7 @@ case class BlockFetcherState(
     waitingHeaders: Queue[BlockHeader],
     fetchingHeadersState: FetchingHeadersState,
     fetchingBodiesState: FetchingBodiesState,
+    pausedFetching: Boolean = false,
     stateNodeFetcher: Option[StateNodeFetcher],
     lastBlock: BigInt,
     knownTop: BigInt,
@@ -295,6 +296,9 @@ case class BlockFetcherState(
   def withNewBodiesFetch: BlockFetcherState = copy(fetchingBodiesState = AwaitingBodies)
   def withBodiesFetchReceived: BlockFetcherState = copy(fetchingBodiesState = NotFetchingBodies)
 
+  def withPausedFetching: BlockFetcherState = copy(pausedFetching = true)
+  def withResumedFetching: BlockFetcherState = copy(pausedFetching = false)
+
   def fetchingStateNode(hash: ByteString, requestor: ActorRef): BlockFetcherState =
     copy(stateNodeFetcher = Some(StateNodeFetcher(hash, requestor)))
 

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

@@ -184,7 +184,8 @@ class BlockImporter(
   ): Task[(List[Block], Option[Any])] =
     if (blocks.isEmpty) {
       importedBlocks.headOption match {
-        case Some(block) => supervisor ! ProgressProtocol.ImportedBlock(block.number, internally = false)
+        case Some(block) =>
+          supervisor ! ProgressProtocol.ImportedBlock(block.number, block.hasCheckpoint, internally = false)
         case None => ()
       }
 
@@ -258,15 +259,16 @@ class BlockImporter(
               val (blocks, weights) = importedBlocksData.map(data => (data.block, data.weight)).unzip
               broadcastBlocks(blocks, weights)
               updateTxPool(importedBlocksData.map(_.block), Seq.empty)
-              supervisor ! ProgressProtocol.ImportedBlock(block.number, internally)
+              supervisor ! ProgressProtocol.ImportedBlock(block.number, block.hasCheckpoint, internally)
             case BlockEnqueued => ()
             case DuplicateBlock => ()
             case UnknownParent => () // This is normal when receiving broadcast blocks
             case ChainReorganised(oldBranch, newBranch, weights) =>
               updateTxPool(newBranch, oldBranch)
               broadcastBlocks(newBranch, weights)
               newBranch.lastOption match {
-                case Some(newBlock) => supervisor ! ProgressProtocol.ImportedBlock(newBlock.number, internally)
+                case Some(newBlock) =>
+                  supervisor ! ProgressProtocol.ImportedBlock(newBlock.number, block.hasCheckpoint, internally)
                 case None => ()
               }
             case BlockImportFailed(error) =>

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

@@ -5,7 +5,7 @@ import akka.util.ByteString
 import io.iohk.ethereum.blockchain.sync.SyncProtocol
 import io.iohk.ethereum.blockchain.sync.SyncProtocol.Status
 import io.iohk.ethereum.blockchain.sync.SyncProtocol.Status.Progress
-import io.iohk.ethereum.blockchain.sync.regular.BlockFetcher.InternalLastBlockImport
+import io.iohk.ethereum.blockchain.sync.regular.BlockFetcher.{InternalCheckpointImport, InternalLastBlockImport}
 import io.iohk.ethereum.blockchain.sync.regular.RegularSync.{NewCheckpoint, ProgressProtocol, ProgressState}
 import io.iohk.ethereum.consensus.blocks.CheckpointBlockGenerator
 import io.iohk.ethereum.consensus.validators.BlockValidator
@@ -100,10 +100,14 @@ class RegularSync(
       log.info(s"Got information about new block [number = $blockNumber]")
       val newState = progressState.copy(bestKnownNetworkBlock = blockNumber)
       context become running(newState)
-    case ProgressProtocol.ImportedBlock(blockNumber, internally) =>
+    case ProgressProtocol.ImportedBlock(blockNumber, isCheckpoint, internally) =>
       log.info(s"Imported new block [number = $blockNumber, internally = $internally]")
-      val newState = progressState.copy(currentBlock = blockNumber)
-      if (internally) {
+      val newState =
+        if (isCheckpoint) progressState.copy(currentBlock = blockNumber, bestKnownNetworkBlock = blockNumber)
+        else progressState.copy(currentBlock = blockNumber)
+      if (internally && isCheckpoint) {
+        fetcher ! InternalCheckpointImport(blockNumber)
+      } else if (internally) {
         fetcher ! InternalLastBlockImport(blockNumber)
       }
       context become running(newState)
@@ -171,6 +175,6 @@ object RegularSync {
     case object StartedFetching extends ProgressProtocol
     case class StartingFrom(blockNumber: BigInt) extends ProgressProtocol
     case class GotNewBlock(blockNumber: BigInt) extends ProgressProtocol
-    case class ImportedBlock(blockNumber: BigInt, internally: Boolean) extends ProgressProtocol
+    case class ImportedBlock(blockNumber: BigInt, isCheckpoint: Boolean, internally: Boolean) extends ProgressProtocol
   }
 }

src/main/scala/io/iohk/ethereum/jsonrpc/CheckpointingJsonMethodsImplicits.scala

@@ -17,9 +17,18 @@ object CheckpointingJsonMethodsImplicits extends JsonMethodsImplicits {
           params: Option[JsonAST.JArray]
       ): Either[JsonRpcError, GetLatestBlockRequest] =
         params match {
-          case Some(JArray(JInt(chkpInterval) :: Nil)) =>
+          case Some(JArray(JInt(chkpInterval) :: second :: Nil)) =>
             if (chkpInterval > 0 && chkpInterval <= Int.MaxValue)
-              Right(GetLatestBlockRequest(chkpInterval.toInt))
+              second match {
+                case JString(blockHash) =>
+                  for {
+                    hash <- extractHash(blockHash)
+                  } yield GetLatestBlockRequest(chkpInterval.toInt, Some(hash))
+                case JNull =>
+                  Right(GetLatestBlockRequest(chkpInterval.toInt, None))
+                case _ =>
+                  Left(InvalidParams("Not supported type for parentCheckpoint"))
+              }
             else
               Left(InvalidParams("Expected positive integer"))
           case _ =>

src/main/scala/io/iohk/ethereum/jsonrpc/CheckpointingService.scala

@@ -18,13 +18,17 @@ class CheckpointingService(
   def getLatestBlock(req: GetLatestBlockRequest): ServiceResponse[GetLatestBlockResponse] = {
     lazy val bestBlockNum = blockchain.getBestBlockNumber()
     lazy val blockToReturnNum = bestBlockNum - bestBlockNum % req.checkpointingInterval
+    lazy val isValidParent = req.parentCheckpoint.forall(blockchain.getBlockHeaderByHash(_).isDefined)
 
     Task {
       blockchain.getBlockByNumber(blockToReturnNum)
     }.flatMap {
-      case Some(b) =>
-        val resp = GetLatestBlockResponse(b.hash, b.number)
-        Task.now(Right(resp))
+      case Some(b) if isValidParent =>
+        Task.now(Right(GetLatestBlockResponse(Some(BlockInfo(b.hash, b.number)))))
+
+      case Some(_) =>
+        log.debug("Parent checkpoint is not found in a local blockchain")
+        Task.now(Right(GetLatestBlockResponse(None)))
 
       case None =>
         log.error(
@@ -42,8 +46,9 @@ class CheckpointingService(
 }
 
 object CheckpointingService {
-  case class GetLatestBlockRequest(checkpointingInterval: Int)
-  case class GetLatestBlockResponse(hash: ByteString, number: BigInt)
+  case class GetLatestBlockRequest(checkpointingInterval: Int, parentCheckpoint: Option[ByteString])
+  case class GetLatestBlockResponse(block: Option[BlockInfo])
+  case class BlockInfo(hash: ByteString, number: BigInt)
 
   case class PushCheckpointRequest(hash: ByteString, signatures: List[ECDSASignature])
   case class PushCheckpointResponse()

src/test/scala/io/iohk/ethereum/jsonrpc/CheckpointingJRCSpec.scala

@@ -31,30 +31,38 @@ class CheckpointingJRCSpec
   import Req._
 
   "CheckpointingJRC" should "getLatestBlock" in new TestSetup {
-    val request = getLatestBlockRequestBuilder(JArray(JInt(4) :: Nil))
-    val servResp = GetLatestBlockResponse(block.hash, block.number)
+    val request = getLatestBlockRequestBuilder(JArray(JInt(4) :: JNull :: Nil))
+    val servResp = GetLatestBlockResponse(Some(BlockInfo(block.hash, block.number)))
     (checkpointingService.getLatestBlock _)
-      .expects(GetLatestBlockRequest(4))
+      .expects(GetLatestBlockRequest(4, None))
       .returning(Task.now(Right(servResp)))
 
     val expectedResult = JObject(
-      "hash" -> JString("0x" + ByteStringUtils.hash2string(block.hash)),
-      "number" -> JInt(block.number)
-    )
+      "block" -> JObject(
+        "hash" -> JString("0x" + ByteStringUtils.hash2string(block.hash)),
+        "number" -> JInt(block.number)
+    ))
 
     val response = jsonRpcController.handleRequest(request).runSyncUnsafe()
     response should haveResult(expectedResult)
   }
 
+  it should "return invalid params when checkpoint parent is of the wrong type" in new TestSetup {
+    val request = getLatestBlockRequestBuilder(JArray(JInt(1) :: JBool(true) :: Nil))
+
+    val response = jsonRpcController.handleRequest(request).runSyncUnsafe()
+    response should haveError(notSupportedTypeError)
+  }
+
   it should "return invalid params when checkpoint interval is not positive (getLatestBlock)" in new TestSetup {
-    val request = getLatestBlockRequestBuilder(JArray(JInt(-1) :: Nil))
+    val request = getLatestBlockRequestBuilder(JArray(JInt(-1) :: JNull :: Nil))
 
     val response = jsonRpcController.handleRequest(request).runSyncUnsafe()
     response should haveError(expectedPositiveIntegerError)
   }
 
   it should "return invalid params when checkpoint interval is too big (getLatestBlock)" in new TestSetup {
-    val request = getLatestBlockRequestBuilder(JArray(JInt(BigInt(Int.MaxValue) + 1) :: Nil))
+    val request = getLatestBlockRequestBuilder(JArray(JInt(BigInt(Int.MaxValue) + 1) :: JNull :: Nil))
 
     val response = jsonRpcController.handleRequest(request).runSyncUnsafe()
     response should haveError(expectedPositiveIntegerError)
@@ -191,6 +199,7 @@ class CheckpointingJRCSpec
     )
 
     val expectedPositiveIntegerError = InvalidParams("Expected positive integer")
+    val notSupportedTypeError = InvalidParams("Not supported type for parentCheckpoint")
 
     def pushCheckpointRequestBuilder(json: JArray) = JsonRpcRequest(
       "2.0",