[ETCM-177] Fix Ommers pool

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,9 +94,10 @@ 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
   }
@@ -142,8 +141,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 +290,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/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/jsonrpc/EthServiceSpec.scala

@@ -427,7 +427,7 @@ class EthServiceSpec
     pendingTransactionsManager.expectMsg(PendingTransactionsManager.GetPendingTransactions)
     pendingTransactionsManager.reply(PendingTransactionsManager.PendingTransactionsResponse(Nil))
 
-    ommersPool.expectMsg(OmmersPool.GetOmmers(1))
+    ommersPool.expectMsg(OmmersPool.GetOmmers(parentBlock.hash))
     ommersPool.reply(OmmersPool.Ommers(Nil))
 
     response.futureValue shouldEqual Right(GetWorkResponse(powHash, seedHash, target))

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

@@ -224,7 +224,8 @@ class JsonRpcControllerSpec
     val blockToRequest = Block(Fixtures.Blocks.Block3125369.header, Fixtures.Blocks.Block3125369.body)
     val blockTd = blockToRequest.header.difficulty
 
-    blockchain.storeBlock(blockToRequest)
+    blockchain
+      .storeBlock(blockToRequest)
       .and(blockchain.storeTotalDifficulty(blockToRequest.header.hash, blockTd))
       .commit()
 
@@ -250,7 +251,8 @@ class JsonRpcControllerSpec
     val blockToRequest = Block(Fixtures.Blocks.Block3125369.header, Fixtures.Blocks.Block3125369.body)
     val blockTd = blockToRequest.header.difficulty
 
-    blockchain.storeBlock(blockToRequest)
+    blockchain
+      .storeBlock(blockToRequest)
       .and(blockchain.storeTotalDifficulty(blockToRequest.header.hash, blockTd))
       .commit()
 
@@ -628,7 +630,7 @@ class JsonRpcControllerSpec
     pendingTransactionsManager.expectMsg(PendingTransactionsManager.GetPendingTransactions)
     pendingTransactionsManager.reply(PendingTransactionsManager.PendingTransactionsResponse(Nil))
 
-    ommersPool.expectMsg(OmmersPool.GetOmmers(2))
+    ommersPool.expectMsg(OmmersPool.GetOmmers(parentBlock.hash))
     ommersPool.reply(Ommers(Nil))
 
     val response = result.futureValue
@@ -674,7 +676,7 @@ class JsonRpcControllerSpec
     val result: Future[JsonRpcResponse] = jsonRpcController.handleRequest(request)
 
     pendingTransactionsManager.expectMsg(PendingTransactionsManager.GetPendingTransactions)
-    ommersPool.expectMsg(OmmersPool.GetOmmers(2))
+    ommersPool.expectMsg(OmmersPool.GetOmmers(parentBlock.hash))
     //on time out it should respond with empty list
 
     val response = result.futureValue(timeout(Timeouts.longTimeout))
@@ -786,7 +788,9 @@ class JsonRpcControllerSpec
   }
 
   it should "eth_gasPrice" in new TestSetup {
-    blockchain.storeBlock(Block(Fixtures.Blocks.Block3125369.header.copy(number = 42), Fixtures.Blocks.Block3125369.body)).commit()
+    blockchain
+      .storeBlock(Block(Fixtures.Blocks.Block3125369.header.copy(number = 42), Fixtures.Blocks.Block3125369.body))
+      .commit()
     blockchain.saveBestKnownBlock(42)
 
     val request: JsonRpcRequest = JsonRpcRequest(