Merge branch 'develop' into ectm-104-target-block-selection

Author: Michał Mrożek

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

@@ -68,14 +68,17 @@ class BlockFetcher(
   private def handleCommands(state: BlockFetcherState): Receive = {
     case PickBlocks(amount) => state.pickBlocks(amount) |> handlePickedBlocks(state) |> fetchBlocks
     case StrictPickBlocks(from, atLeastWith) =>
-      val minBlock = from.min(atLeastWith).max(1)
-      log.debug("Strict Pick blocks from {} to {}", from, atLeastWith)
+      // FIXME: Consider having StrictPickBlocks calls guaranteeing this
+      // from parameter could be negative or 0 so we should cap it to 1 if that's the case
+      val fromCapped = from.max(1)
+      val minBlock = fromCapped.min(atLeastWith).max(1)
+      log.debug("Strict Pick blocks from {} to {}", fromCapped, atLeastWith)
       log.debug("Lowest available block is {}", state.lowestBlock)
 
       val newState = if (minBlock < state.lowestBlock) {
         state.invalidateBlocksFrom(minBlock, None)._2
       } else {
-        state.strictPickBlocks(from, atLeastWith) |> handlePickedBlocks(state)
+        state.strictPickBlocks(fromCapped, atLeastWith) |> handlePickedBlocks(state)
       }
 
       fetchBlocks(newState)

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

@@ -107,9 +107,15 @@ case class BlockFetcherState(
       None
     }
 
+  /**
+    * Returns all the ready blocks but only if it includes blocks with number:
+    * - lower = min(from, atLeastWith)
+    * - upper = max(from, atLeastWith)
+    */
   def strictPickBlocks(from: BigInt, atLeastWith: BigInt): Option[(NonEmptyList[Block], BlockFetcherState)] = {
     val lower = from.min(atLeastWith)
     val upper = from.max(atLeastWith)
+
     readyBlocks.some
       .filter(_.headOption.exists(block => block.number <= lower))
       .filter(_.lastOption.exists(block => block.number >= upper))

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

@@ -45,8 +45,8 @@ class BlockImporter(
     start()
   }
 
-  private def idle: Receive = {
-    case Start => start()
+  private def idle: Receive = { case Start =>
+    start()
   }
 
   private def handleTopMessages(state: ImporterState, currentBehavior: Behavior): Receive = {
@@ -63,8 +63,6 @@ class BlockImporter(
     case MinedBlock(block) =>
       if (!state.importing) {
         importMinedBlock(block, state)
-      } else {
-        ommersPool ! AddOmmers(block.header)
       }
     case ImportNewBlock(block, peerId) if state.isOnTop && !state.importing => importNewBlock(block, peerId, state)
     case ImportDone(newBehavior) =>
@@ -96,16 +94,21 @@ class BlockImporter(
   }
 
   private def pickBlocks(state: ImporterState): Unit = {
-    val msg = state.resolvingBranchFrom.fold[BlockFetcher.FetchMsg](
-      BlockFetcher.PickBlocks(syncConfig.blocksBatchSize))(
-      from => BlockFetcher.StrictPickBlocks(from, startingBlockNumber))
+    val msg =
+      state.resolvingBranchFrom.fold[BlockFetcher.FetchMsg](BlockFetcher.PickBlocks(syncConfig.blocksBatchSize))(from =>
+        BlockFetcher.StrictPickBlocks(from, startingBlockNumber)
+      )
 
     fetcher ! msg
   }
 
   private def importBlocks(blocks: NonEmptyList[Block]): ImportFn =
     importWith {
-      log.debug("Attempting to import blocks starting from {}", blocks.head.number)
+      log.debug(
+        "Attempting to import blocks starting from {} and ending with {}",
+        blocks.head.number,
+        blocks.last.number
+      )
       Future
         .successful(resolveBranch(blocks))
         .flatMap {
@@ -142,8 +145,9 @@ class BlockImporter(
         }
       }
 
-  private def tryImportBlocks(blocks: List[Block], importedBlocks: List[Block] = Nil)(
-      implicit ec: ExecutionContext): Future[(List[Block], Option[Any])] =
+  private def tryImportBlocks(blocks: List[Block], importedBlocks: List[Block] = Nil)(implicit
+      ec: ExecutionContext
+  ): Future[(List[Block], Option[Any])] =
     if (blocks.isEmpty) {
       Future.successful((importedBlocks, None))
     } else {
@@ -290,9 +294,11 @@ object BlockImporter {
       syncConfig: SyncConfig,
       ommersPool: ActorRef,
       broadcaster: ActorRef,
-      pendingTransactionsManager: ActorRef): Props =
+      pendingTransactionsManager: ActorRef
+  ): Props =
     Props(
-      new BlockImporter(fetcher, ledger, blockchain, syncConfig, ommersPool, broadcaster, pendingTransactionsManager))
+      new BlockImporter(fetcher, ledger, blockchain, syncConfig, ommersPool, broadcaster, pendingTransactionsManager)
+    )
 
   type Behavior = ImporterState => Receive
   type ImportFn = ImporterState => Unit

src/main/scala/io/iohk/ethereum/consensus/ethash/EthashBlockCreator.scala

@@ -2,7 +2,7 @@ package io.iohk.ethereum.consensus.ethash
 
 import akka.actor.ActorRef
 import akka.pattern.ask
-import akka.util.Timeout
+import akka.util.{Timeout, ByteString}
 import io.iohk.ethereum.consensus.blocks.PendingBlock
 import io.iohk.ethereum.consensus.ethash.blocks.EthashBlockGenerator
 import io.iohk.ethereum.domain.{Address, Block}
@@ -28,23 +28,21 @@ class EthashBlockCreator(
   def getBlockForMining(parentBlock: Block, withTransactions: Boolean = true): Future[PendingBlock] = {
     val transactions =
       if (withTransactions) getTransactionsFromPool else Future.successful(PendingTransactionsResponse(Nil))
-    getOmmersFromPool(parentBlock.header.number + 1).zip(transactions).flatMap {
-      case (ommers, pendingTxs) =>
-        blockGenerator
-          .generateBlock(parentBlock, pendingTxs.pendingTransactions.map(_.stx.tx), coinbase, ommers.headers) match {
-          case Right(pb) => Future.successful(pb)
-          case Left(err) => Future.failed(new RuntimeException(s"Error while generating block for mining: $err"))
-        }
+    getOmmersFromPool(parentBlock.hash).zip(transactions).flatMap { case (ommers, pendingTxs) =>
+      blockGenerator
+        .generateBlock(parentBlock, pendingTxs.pendingTransactions.map(_.stx.tx), coinbase, ommers.headers) match {
+        case Right(pb) => Future.successful(pb)
+        case Left(err) => Future.failed(new RuntimeException(s"Error while generating block for mining: $err"))
+      }
     }
   }
 
-  private def getOmmersFromPool(blockNumber: BigInt): Future[OmmersPool.Ommers] = {
-    (ommersPool ? OmmersPool.GetOmmers(blockNumber))(Timeout(miningConfig.ommerPoolQueryTimeout))
+  private def getOmmersFromPool(parentBlockHash: ByteString): Future[OmmersPool.Ommers] = {
+    (ommersPool ? OmmersPool.GetOmmers(parentBlockHash))(Timeout(miningConfig.ommerPoolQueryTimeout))
       .mapTo[OmmersPool.Ommers]
-      .recover {
-        case ex =>
-          log.error("Failed to get ommers, mining block with empty ommers list", ex)
-          OmmersPool.Ommers(Nil)
+      .recover { case ex =>
+        log.error("Failed to get ommers, mining block with empty ommers list", ex)
+        OmmersPool.Ommers(Nil)
       }
   }
 

src/main/scala/io/iohk/ethereum/domain/BlockHeader.scala

@@ -27,6 +27,7 @@ case class BlockHeader(
 
   override def toString: String = {
     s"""BlockHeader {
+       |hash: $hashAsHexString
        |parentHash: ${Hex.toHexString(parentHash.toArray[Byte])}
        |ommersHash: ${Hex.toHexString(ommersHash.toArray[Byte])}
        |beneficiary: ${Hex.toHexString(beneficiary.toArray[Byte])}

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

@@ -507,7 +507,7 @@ class EthService(
       import io.iohk.ethereum.consensus.ethash.EthashUtils.{epoch, seed}
 
       val bestBlock = blockchain.getBestBlock()
-      getOmmersFromPool(bestBlock.header.number + 1).zip(getTransactionsFromPool).map { case (ommers, pendingTxs) =>
+      getOmmersFromPool(bestBlock.hash).zip(getTransactionsFromPool).map { case (ommers, pendingTxs) =>
         val blockGenerator = ethash.blockGenerator
         blockGenerator.generateBlock(
           bestBlock,
@@ -530,12 +530,12 @@ class EthService(
       }
     })(Future.successful(Left(JsonRpcErrors.ConsensusIsNotEthash)))
 
-  private def getOmmersFromPool(blockNumber: BigInt): Future[OmmersPool.Ommers] =
+  private def getOmmersFromPool(parentBlockHash: ByteString): Future[OmmersPool.Ommers] =
     consensus.ifEthash(ethash => {
       val miningConfig = ethash.config.specific
       implicit val timeout: Timeout = Timeout(miningConfig.ommerPoolQueryTimeout)
 
-      (ommersPool ? OmmersPool.GetOmmers(blockNumber))
+      (ommersPool ? OmmersPool.GetOmmers(parentBlockHash))
         .mapTo[OmmersPool.Ommers]
         .recover { case ex =>
           log.error("failed to get ommer, mining block with empty ommers list", ex)

src/main/scala/io/iohk/ethereum/ommers/OmmersPool.scala

@@ -1,37 +1,79 @@
 package io.iohk.ethereum.ommers
 
-import akka.actor.{Actor, Props}
+import akka.util.ByteString
+import akka.actor.{Actor, ActorLogging, Props}
+import org.bouncycastle.util.encoders.Hex
 import io.iohk.ethereum.domain.{BlockHeader, Blockchain}
 import io.iohk.ethereum.ommers.OmmersPool.{AddOmmers, GetOmmers, RemoveOmmers}
+import scala.annotation.tailrec
 
-class OmmersPool(blockchain: Blockchain, ommersPoolSize: Int) extends Actor {
+class OmmersPool(blockchain: Blockchain, ommersPoolSize: Int, ommerGenerationLimit: Int, returnedOmmersSizeLimit: Int)
+    extends Actor
+    with ActorLogging {
 
   var ommersPool: Seq[BlockHeader] = Nil
 
-  val ommerGenerationLimit: Int = 6 //Stated on section 11.1, eq. (143) of the YP
-  val ommerSizeLimit: Int = 2
-
   override def receive: Receive = {
     case AddOmmers(ommers) =>
       ommersPool = (ommers ++ ommersPool).take(ommersPoolSize).distinct
+      logStatus(event = "Ommers after add", ommers = ommersPool)
 
     case RemoveOmmers(ommers) =>
       val toDelete = ommers.map(_.hash).toSet
       ommersPool = ommersPool.filter(b => !toDelete.contains(b.hash))
+      logStatus(event = "Ommers after remove", ommers = ommersPool)
 
-    case GetOmmers(blockNumber) =>
-      val ommers = ommersPool.filter { b =>
-        val generationDifference = blockNumber - b.number
-        generationDifference > 0 && generationDifference <= ommerGenerationLimit
-      }.filter { b =>
-        blockchain.getBlockHeaderByHash(b.parentHash).isDefined
-      }.take(ommerSizeLimit)
+    case GetOmmers(parentBlockHash) =>
+      val ancestors = collectAncestors(parentBlockHash, ommerGenerationLimit)
+      val ommers = ommersPool
+        .filter { b =>
+          val notAncestor = ancestors.find(_.hash == b.hash).isEmpty
+          ancestors.find(_.hash == b.parentHash).isDefined && notAncestor
+        }
+        .take(returnedOmmersSizeLimit)
+      logStatus(event = s"Ommers given parent block ${Hex.toHexString(parentBlockHash.toArray)}", ommers)
       sender() ! OmmersPool.Ommers(ommers)
   }
+
+  private def collectAncestors(parentHash: ByteString, generationLimit: Int): List[BlockHeader] = {
+    @tailrec
+    def rec(hash: ByteString, limit: Int, acc: List[BlockHeader]): List[BlockHeader] = {
+      if (limit > 0) {
+        blockchain.getBlockHeaderByHash(hash) match {
+          case Some(bh) => rec(bh.parentHash, limit - 1, acc :+ bh)
+          case None => acc
+        }
+      } else {
+        acc
+      }
+    }
+    rec(parentHash, generationLimit, List.empty)
+  }
+
+  private def logStatus(event: String, ommers: Seq[BlockHeader]): Unit = {
+    lazy val ommersAsString: Seq[String] = ommers.map { bh => s"[number = ${bh.number}, hash = ${bh.hashAsHexString}]" }
+    log.debug(s"$event ${ommersAsString}")
+  }
 }
 
 object OmmersPool {
-  def props(blockchain: Blockchain, ommersPoolSize: Int): Props = Props(new OmmersPool(blockchain, ommersPoolSize))
+
+  /**
+    * As is stated on section 11.1, eq. (143) of the YP
+    *
+    * @param ommerGenerationLimit should be === 6
+    * @param returnedOmmersSizeLimit should be === 2
+    *
+    * ^ Probably not worthy but those params could be placed in consensus config.
+    */
+  def props(
+      blockchain: Blockchain,
+      ommersPoolSize: Int,
+      ommerGenerationLimit: Int = 6,
+      returnedOmmersSizeLimit: Int = 2
+  ): Props = Props(
+    new OmmersPool(blockchain, ommersPoolSize, ommerGenerationLimit, returnedOmmersSizeLimit)
+  )
 
   case class AddOmmers(ommers: List[BlockHeader])
 
@@ -45,7 +87,7 @@ object OmmersPool {
     def apply(b: BlockHeader*): RemoveOmmers = RemoveOmmers(b.toList)
   }
 
-  case class GetOmmers(blockNumber: BigInt)
+  case class GetOmmers(parentBlockHash: ByteString)
 
   case class Ommers(headers: Seq[BlockHeader])
 }

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

@@ -33,6 +33,5 @@ class BlockFetcherStateSpec extends TestKit(ActorSystem()) with AnyWordSpecLike
         newState.knownTop shouldEqual newBestBlock.number
       }
     }
-
   }
 }