[ETCM-1069] Create ConsensusAdapter(#1096)

* make adapter pre-validate block
* enqueue blocks in ConsensusAdapter
* remove blockQueue from Consensus

Author: Aurélien Richez

src/it/scala/io/iohk/ethereum/ledger/BlockImporterItSpec.scala

@@ -26,7 +26,6 @@ import io.iohk.ethereum.blockchain.sync.regular.BlockFetcher
 import io.iohk.ethereum.blockchain.sync.regular.BlockImporter
 import io.iohk.ethereum.blockchain.sync.regular.BlockImporter.NewCheckpoint
 import io.iohk.ethereum.checkpointing.CheckpointingTestHelpers
-import io.iohk.ethereum.consensus.Consensus
 import io.iohk.ethereum.consensus.ConsensusAdapter
 import io.iohk.ethereum.consensus.blocks.CheckpointBlockGenerator
 import io.iohk.ethereum.consensus.pow.validators.OmmersValidator

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

@@ -124,23 +124,25 @@ class RegularSyncItSpec extends FreeSpecBase with Matchers with BeforeAndAfterAl
   "peers should chose the branch with a checkpoint discarding blocks that come after the checkpoint" in customTestCaseResourceM(
     FakePeer.start2FakePeersRes()
   ) { case (peer1, peer2) =>
-    val length = 26
+    val checkpointBlockNumber = 26
     for {
       _ <- peer1.importBlocksUntil(20)(IdentityUpdate)
       _ <- peer2.importBlocksUntil(30)(IdentityUpdate)
       _ <- peer1.startRegularSync()
       _ <- peer2.startRegularSync()
-      _ <- peer2.addCheckpointedBlock(
+      _ <- peer2.addCheckpointedBlock( // checkpointing block 26
         peer2.blockchainReader.getBlockByNumber(peer2.blockchainReader.getBestBranch(), 25).get
       )
-      _ <- peer2.waitForRegularSyncLoadLastBlock(length)
+      _ <- peer2.waitForRegularSyncLoadLastBlock(checkpointBlockNumber)
       _ <- peer1.connectToPeers(Set(peer2.node))
-      _ <- peer1.waitForRegularSyncLoadLastBlock(length)
+      _ <- peer1.waitForRegularSyncLoadLastBlock(checkpointBlockNumber)
     } yield {
       assert(peer1.blockchainReader.getBestBlock().get.hash == peer2.blockchainReader.getBestBlock().get.hash)
       val peer1BestBlockNumber = peer1.blockchainReader.getBestBlock().get.number
       val peer2BestBlockNumber = peer2.blockchainReader.getBestBlock().get.number
-      assert(peer1BestBlockNumber == peer2BestBlockNumber && peer1BestBlockNumber == length)
+
+      assert(peer1BestBlockNumber == peer2BestBlockNumber && peer1BestBlockNumber == checkpointBlockNumber)
+      assert(peer1.blockchainReader.getLatestCheckpointBlockNumber() == checkpointBlockNumber)
       assert(
         peer1.blockchainReader.getLatestCheckpointBlockNumber() == peer2.blockchainReader
           .getLatestCheckpointBlockNumber()

src/it/scala/io/iohk/ethereum/sync/util/RegularSyncItSpecUtils.scala

@@ -111,7 +111,6 @@ object RegularSyncItSpecUtils {
         bl,
         blockchainReader,
         blockchainWriter,
-        blockQueue,
         blockExecution
       )
     lazy val consensusAdapter = new ConsensusAdapter(

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

@@ -9,7 +9,6 @@ import akka.actor.Scheduler
 
 import io.iohk.ethereum.blockchain.sync.fast.FastSync
 import io.iohk.ethereum.blockchain.sync.regular.RegularSync
-import io.iohk.ethereum.consensus.Consensus
 import io.iohk.ethereum.consensus.ConsensusAdapter
 import io.iohk.ethereum.consensus.validators.Validators
 import io.iohk.ethereum.db.storage.AppStateStorage

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

@@ -20,7 +20,6 @@ import io.iohk.ethereum.blockchain.sync.Blacklist.BlacklistReason
 import io.iohk.ethereum.blockchain.sync.regular.BlockBroadcast.BlockToBroadcast
 import io.iohk.ethereum.blockchain.sync.regular.BlockBroadcasterActor.BroadcastBlocks
 import io.iohk.ethereum.blockchain.sync.regular.RegularSync.ProgressProtocol
-import io.iohk.ethereum.consensus.Consensus
 import io.iohk.ethereum.consensus.ConsensusAdapter
 import io.iohk.ethereum.crypto.kec256
 import io.iohk.ethereum.db.storage.StateStorage

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

@@ -19,7 +19,6 @@ import io.iohk.ethereum.blockchain.sync.regular.BlockFetcher.InternalLastBlockIm
 import io.iohk.ethereum.blockchain.sync.regular.RegularSync.NewCheckpoint
 import io.iohk.ethereum.blockchain.sync.regular.RegularSync.ProgressProtocol
 import io.iohk.ethereum.blockchain.sync.regular.RegularSync.ProgressState
-import io.iohk.ethereum.consensus.Consensus
 import io.iohk.ethereum.consensus.ConsensusAdapter
 import io.iohk.ethereum.consensus.validators.BlockValidator
 import io.iohk.ethereum.db.storage.StateStorage

src/main/scala/io/iohk/ethereum/consensus/Consensus.scala

@@ -1,10 +1,17 @@
 package io.iohk.ethereum.consensus
 
+import akka.util.ByteString
+
+import cats.data.NonEmptyList
+
 import monix.eval.Task
 import monix.execution.Scheduler
 
-import io.iohk.ethereum.blockchain.sync.regular.BlockImportResult
+import io.iohk.ethereum.consensus.Consensus.ConsensusResult
 import io.iohk.ethereum.domain.Block
+import io.iohk.ethereum.domain.ChainWeight
+import io.iohk.ethereum.ledger.BlockData
+import io.iohk.ethereum.mpt.MerklePatriciaTrie.MissingNodeException
 import io.iohk.ethereum.utils.BlockchainConfig
 
 /** This file documents the original interface that was designed at ETCM-1018
@@ -13,8 +20,8 @@ import io.iohk.ethereum.utils.BlockchainConfig
   */
 trait Consensus {
   def evaluateBranch(
-      block: Seq[Block]
-  )(implicit blockExecutionScheduler: Scheduler, blockchainConfig: BlockchainConfig): Task[BlockImportResult]
+      block: NonEmptyList[Block]
+  )(implicit blockExecutionScheduler: Scheduler, blockchainConfig: BlockchainConfig): Task[ConsensusResult]
 
   /** Original interface from ETCM-1018, for temporary documentation purposes
     */
@@ -67,3 +74,44 @@ trait Consensus {
 //  private def attemptToSetNewBestBranch(branch: UpdatedBranch): Either[BlockExecutionError, Boolean] = ???
 
 }
+
+object Consensus {
+  /* This return type for consensus is probably overcomplicated for now because some information is needed
+   * to keep the compatibility with the current code (particularly for the block queue handling), and be able
+   * to translate the values to BlockImportResult.
+   * In particular:
+   *  - `blockToEnqueue` fields won't be needed if the block are already stored in memory
+   *  - The distinction between ExtendedCurrentBestBranch and SelectedNewBestBranch won't really be useful
+   *  because there will be no need to put back the old branch into the block queue in case of reorganisation
+   *  - `ConsensusErrorDueToMissingNode` and  `ConsensusError` would mean that the application is in an
+   *  inconsistent state. Unless there is a reason to think that mantis would self heal when that happens, I
+   *  don't think there is a reason to add them here.
+   */
+
+  sealed trait ConsensusResult
+
+  /** The new branch was selected and it extended the best branch. */
+  case class ExtendedCurrentBestBranch(blockImportData: List[BlockData]) extends ConsensusResult
+
+  /** The new branch was selected and it extended the best branch, but it did not execute completely. */
+  case class ExtendedCurrentBestBranchPartially(blockImportData: List[BlockData], failureBranch: BranchExecutionFailure)
+      extends ConsensusResult
+
+  /** The new branch was selected but was not an extension of the best branch. */
+  case class SelectedNewBestBranch(oldBranch: List[Block], newBranch: List[Block], weights: List[ChainWeight])
+      extends ConsensusResult
+
+  /** The proposed new branch was not better than the current best one. */
+  case object KeptCurrentBestBranch extends ConsensusResult
+
+  /** A block in the branch cannot be executed. */
+  case class BranchExecutionFailure(blockToEnqueue: List[Block], failingBlockHash: ByteString, error: String)
+      extends ConsensusResult
+
+  /** An error external the the blocks in the branch occured, which prevents the branch from being executed.
+    * Usually this is due to an inconsistency in the database.
+    */
+  case class ConsensusError(blockToEnqueue: List[Block], err: String) extends ConsensusResult
+  case class ConsensusErrorDueToMissingNode(blockToEnqueue: List[Block], reason: MissingNodeException)
+      extends ConsensusResult
+}

src/main/scala/io/iohk/ethereum/consensus/ConsensusAdapter.scala

@@ -1,11 +1,24 @@
 package io.iohk.ethereum.consensus
 
+import cats.data.NonEmptyList
+
 import monix.eval.Task
 import monix.execution.Scheduler
 
+import io.iohk.ethereum.blockchain.sync.regular.BlockEnqueued
 import io.iohk.ethereum.blockchain.sync.regular.BlockImportFailed
+import io.iohk.ethereum.blockchain.sync.regular.BlockImportFailedDueToMissingNode
 import io.iohk.ethereum.blockchain.sync.regular.BlockImportResult
+import io.iohk.ethereum.blockchain.sync.regular.BlockImportedToTop
+import io.iohk.ethereum.blockchain.sync.regular.ChainReorganised
 import io.iohk.ethereum.blockchain.sync.regular.DuplicateBlock
+import io.iohk.ethereum.consensus.Consensus.BranchExecutionFailure
+import io.iohk.ethereum.consensus.Consensus.ConsensusError
+import io.iohk.ethereum.consensus.Consensus.ConsensusErrorDueToMissingNode
+import io.iohk.ethereum.consensus.Consensus.ExtendedCurrentBestBranch
+import io.iohk.ethereum.consensus.Consensus.ExtendedCurrentBestBranchPartially
+import io.iohk.ethereum.consensus.Consensus.KeptCurrentBestBranch
+import io.iohk.ethereum.consensus.Consensus.SelectedNewBestBranch
 import io.iohk.ethereum.domain.Block
 import io.iohk.ethereum.domain.BlockHeader
 import io.iohk.ethereum.domain.BlockchainReader
@@ -36,19 +49,59 @@ class ConsensusAdapter(
         if (isBlockADuplicate(block.header, bestBlock.header.number)) {
           log.debug("Ignoring duplicated block: {}", block.idTag)
           Task.now(DuplicateBlock)
+        } else if (blockchainReader.getChainWeightByHash(bestBlock.header.hash).isEmpty) {
+          // This part is not really needed except for compatibility as a missing chain weight
+          // would indicate an inconsistent database
+          returnNoTotalDifficulty(bestBlock)
         } else {
           doBlockPreValidation(block).flatMap {
             case Left(error) =>
               Task.now(BlockImportFailed(error.reason.toString))
             case Right(BlockExecutionSuccess) =>
-              consensus.evaluateBranch(Seq(block))
+              enqueueAndGetBranch(block, bestBlock.number)
+                .map(forwardAndTranslateConsensusResult) // a new branch was created so we give it to consensus
+                .getOrElse(Task.now(BlockEnqueued)) // the block was not rooted so it was simply enqueued
           }
         }
       case None =>
         log.error("Couldn't find the current best block")
         Task.now(BlockImportFailed("Couldn't find the current best block"))
     }
 
+  private def forwardAndTranslateConsensusResult(
+      newBranch: NonEmptyList[Block]
+  )(implicit blockExecutionScheduler: Scheduler, blockchainConfig: BlockchainConfig) =
+    consensus
+      .evaluateBranch(newBranch)
+      .map {
+        case SelectedNewBestBranch(oldBranch, newBranch, weights) =>
+          oldBranch.foreach(blockQueue.enqueueBlock(_))
+          ChainReorganised(oldBranch, newBranch, weights)
+        case ExtendedCurrentBestBranch(blockImportData) =>
+          BlockImportedToTop(blockImportData)
+        case ExtendedCurrentBestBranchPartially(
+              blockImportData,
+              BranchExecutionFailure(blocksToEnqueue, failingBlockHash, error)
+            ) =>
+          blocksToEnqueue.foreach(blockQueue.enqueueBlock(_))
+          blockQueue.removeSubtree(failingBlockHash)
+          log.warn("extended best branch partially because of error: {}", error)
+          BlockImportedToTop(blockImportData)
+        case KeptCurrentBestBranch =>
+          newBranch.toList.foreach(blockQueue.enqueueBlock(_))
+          BlockEnqueued
+        case BranchExecutionFailure(blocksToEnqueue, failingBlockHash, error) =>
+          blocksToEnqueue.foreach(blockQueue.enqueueBlock(_))
+          blockQueue.removeSubtree(failingBlockHash)
+          BlockImportFailed(error)
+        case ConsensusError(blocksToEnqueue, error) =>
+          blocksToEnqueue.foreach(blockQueue.enqueueBlock(_))
+          BlockImportFailed(error)
+        case ConsensusErrorDueToMissingNode(blocksToEnqueue, reason) =>
+          blocksToEnqueue.foreach(blockQueue.enqueueBlock(_))
+          BlockImportFailedDueToMissingNode(reason)
+      }
+
   private def doBlockPreValidation(block: Block)(implicit
       blockchainConfig: BlockchainConfig
   ): Task[Either[ValidationBeforeExecError, BlockExecutionSuccess]] =
@@ -71,4 +124,22 @@ class ConsensusAdapter(
     block.number <= currentBestBlockNumber ||
     blockQueue.isQueued(hash)
   }
+
+  private def enqueueAndGetBranch(block: Block, bestBlockNumber: BigInt): Option[NonEmptyList[Block]] =
+    blockQueue
+      .enqueueBlock(block, bestBlockNumber)
+      .map(topBlock => blockQueue.getBranch(topBlock.hash, dequeue = true))
+      .flatMap(NonEmptyList.fromList)
+
+  private def returnNoTotalDifficulty(bestBlock: Block): Task[BlockImportFailed] = {
+    log.error(
+      "Getting total difficulty for current best block with hash: {} failed",
+      bestBlock.header.hashAsHexString
+    )
+    Task.now(
+      BlockImportFailed(
+        s"Couldn't get total difficulty for current best block with hash: ${bestBlock.header.hashAsHexString}"
+      )
+    )
+  }
 }