ETCM-636: None.get on best block tests

Author: AnastasiiaL

project/Dependencies.scala

@@ -59,7 +59,7 @@ object Dependencies {
 
   val testing: Seq[ModuleID] = Seq(
     "org.scalatest" %% "scalatest" % "3.2.2" % "it,test",
-    "org.scalamock" %% "scalamock" % "5.0.0" % "test",
+    "org.scalamock" %% "scalamock" % "5.0.0" % "it,test",
     "org.scalatestplus" %% "scalacheck-1-15" % "3.2.3.0" % "test",
     "org.scalacheck" %% "scalacheck" % "1.15.1" % "it,test",
     "com.softwaremill.diffx" %% "diffx-core" % "0.3.30" % "test",

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

@@ -0,0 +1,108 @@
+package io.iohk.ethereum.ledger
+
+import akka.testkit.TestProbe
+import akka.util.ByteString
+import cats.data.NonEmptyList
+import io.iohk.ethereum.blockchain.sync.regular.{BlockFetcher, BlockImporter}
+import io.iohk.ethereum.consensus.blocks.CheckpointBlockGenerator
+import io.iohk.ethereum.domain._
+import io.iohk.ethereum.mpt.MerklePatriciaTrie
+import io.iohk.ethereum.utils.Config.SyncConfig
+import io.iohk.ethereum.utils.Config
+import io.iohk.ethereum.Fixtures
+import monix.execution.Scheduler
+import org.scalamock.scalatest.MockFactory
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.flatspec.AsyncFlatSpecLike
+import org.scalatest.matchers.should.Matchers
+import scala.concurrent.duration._
+
+class BlockImporterItSpec extends MockFactory with TestSetupWithVmAndValidators with AsyncFlatSpecLike with Matchers with BeforeAndAfterAll {
+
+  implicit val testScheduler = Scheduler.fixedPool("test", 32)
+
+  override def afterAll(): Unit = {
+    testScheduler.shutdown()
+    testScheduler.awaitTermination(60.second)
+  }
+
+  val bl = BlockchainImpl(storagesInstance.storages)
+
+  val blockQueue = BlockQueue(bl, SyncConfig(Config.config))
+
+  val genesis = Block(
+    Fixtures.Blocks.Genesis.header.copy(stateRoot = ByteString(MerklePatriciaTrie.EmptyRootHash)),
+    Fixtures.Blocks.Genesis.body
+  )
+  val genesisWeight = ChainWeight.zero.increase(genesis.header)
+
+  bl.save(genesis, Seq(), genesisWeight, saveAsBestBlock = true)
+
+  lazy val checkpointBlockGenerator: CheckpointBlockGenerator = new CheckpointBlockGenerator
+
+  val fetcherProbe = TestProbe()
+  val ommersPoolProbe = TestProbe()
+  val broadcasterProbe = TestProbe()
+  val pendingTransactionsManagerProbe = TestProbe()
+  val supervisor = TestProbe()
+
+  override lazy val ledger: TestLedgerImpl =  new TestLedgerImpl(validators) {
+    override private[ledger] lazy val blockExecution = mock[BlockExecution]
+  }
+
+  val blockImporter = system.actorOf(
+    BlockImporter.props(
+      fetcherProbe.ref,
+      ledger,
+      bl,
+      syncConfig,
+      ommersPoolProbe.ref,
+      broadcasterProbe.ref,
+      pendingTransactionsManagerProbe.ref,
+      checkpointBlockGenerator,
+      supervisor.ref
+    ))
+
+  "BlockImporter" should "return a correct new best block after reorganising longer chain to a shorter one" in {
+
+    val genesis = bl.getBestBlock()
+    val block1: Block = getBlock(genesis.number + 1, parent = genesis.header.hash)
+    // new chain is shorter but has a higher weight
+    val newBlock2: Block = getBlock(genesis.number + 2, difficulty = 101, parent = block1.header.hash)
+    val newBlock3: Block = getBlock(genesis.number + 3, difficulty = 333, parent = newBlock2.header.hash)
+    val oldBlock2: Block = getBlock(genesis.number + 2, difficulty = 102, parent = block1.header.hash)
+    val oldBlock3: Block = getBlock(genesis.number + 3, difficulty = 103, parent = oldBlock2.header.hash)
+    val oldBlock4: Block = getBlock(genesis.number + 4, difficulty = 104, parent = oldBlock3.header.hash)
+
+    val weight1 = ChainWeight.totalDifficultyOnly(block1.header.difficulty + 999)
+    val newWeight2 = weight1.increase(newBlock2.header)
+    val newWeight3 = newWeight2.increase(newBlock3.header)
+    val oldWeight2 = weight1.increase(oldBlock2.header)
+    val oldWeight3 = oldWeight2.increase(oldBlock3.header)
+    val oldWeight4 = oldWeight3.increase(oldBlock4.header)
+
+    //saving initial main chain
+    bl.save(block1, Nil, weight1, saveAsBestBlock = true)
+    bl.save(oldBlock2, Nil, oldWeight2, saveAsBestBlock = true)
+    bl.save(oldBlock3, Nil, oldWeight3, saveAsBestBlock = true)
+    bl.save(oldBlock4, Nil, oldWeight4, saveAsBestBlock = true)
+
+    val blockData2 = BlockData(newBlock2, Seq.empty[Receipt], newWeight2)
+    val blockData3 = BlockData(newBlock3, Seq.empty[Receipt], newWeight3)
+
+    val newBranch = List(newBlock2, newBlock3)
+
+    blockImporter ! BlockImporter.Start
+    blockImporter ! BlockFetcher.PickedBlocks(NonEmptyList.fromListUnsafe(newBranch))
+
+    (ledger.blockExecution.executeAndValidateBlocks _)
+      .expects(newBranch, *)
+      .returning((List(blockData2, blockData3), None))
+
+    // Saving new blocks, because it's part of executeBlocks method mechanism
+    bl.save(blockData2.block, blockData2.receipts, blockData2.weight, saveAsBestBlock = true)
+    bl.save(blockData3.block, blockData3.receipts, blockData3.weight, saveAsBestBlock = true)
+
+    bl.getBestBlock() shouldEqual newBlock3
+  }
+}

src/main/scala/io/iohk/ethereum/blockchain/sync/fast/FastSync.scala

@@ -823,8 +823,8 @@ class FastSync(
         val bestReceivedBlock = fullBlocks.maxBy(_.number)
         val lastStoredBestBlockNumber = appStateStorage.getBestBlockNumber()
         if (lastStoredBestBlockNumber < bestReceivedBlock.number) {
-          appStateStorage.putBestBlockNumber(bestReceivedBlock.number).commit()
           blockchain.saveBestKnownBlocks(bestReceivedBlock.number)
+          appStateStorage.putBestBlockNumber(bestReceivedBlock.number).commit()
         }
         syncState = syncState.copy(lastFullBlockNumber = bestReceivedBlock.number.max(lastStoredBestBlockNumber))
       }

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

@@ -341,12 +341,6 @@ class BlockchainImpl(
   }
 
   def save(block: Block, receipts: Seq[Receipt], weight: ChainWeight, saveAsBestBlock: Boolean): Unit = {
-    log.debug("Saving new block block {} to database", block.idTag)
-    storeBlock(block)
-      .and(storeReceipts(block.header.hash, receipts))
-      .and(storeChainWeight(block.header.hash, weight))
-      .commit()
-
     if (saveAsBestBlock && block.hasCheckpoint) {
       log.debug(
         "New best known block block number - {}, new best checkpoint number - {}",
@@ -362,6 +356,12 @@ class BlockchainImpl(
       saveBestKnownBlock(block.header.number)
     }
 
+    log.debug("Saving new block block {} to database", block.idTag)
+    storeBlock(block)
+      .and(storeReceipts(block.header.hash, receipts))
+      .and(storeChainWeight(block.header.hash, weight))
+      .commit()
+
     // not transactional part
     // the best blocks data will be persisted only when the cache will be persisted
     stateStorage.onBlockSave(block.header.number, appStateStorage.getBestBlockNumber())(persistBestBlocksData)
@@ -397,20 +397,17 @@ class BlockchainImpl(
     }
   }
 
-  private def saveBestKnownBlock(bestBlockNumber: BigInt): Unit = {
+  private def saveBestKnownBlock(bestBlockNumber: BigInt): Unit =
     bestKnownBlockAndLatestCheckpoint.updateAndGet(_.copy(bestBlockNumber = bestBlockNumber))
-  }
 
-  private def saveBestKnownBlockAndLatestCheckpointNumber(number: BigInt, latestCheckpointNumber: BigInt): Unit = {
+  private def saveBestKnownBlockAndLatestCheckpointNumber(number: BigInt, latestCheckpointNumber: BigInt): Unit =
     bestKnownBlockAndLatestCheckpoint.set(BestBlockLatestCheckpointNumbers(number, latestCheckpointNumber))
-  }
 
   def storeChainWeight(blockhash: ByteString, weight: ChainWeight): DataSourceBatchUpdate =
     chainWeightStorage.put(blockhash, weight)
 
-  def saveNode(nodeHash: NodeHash, nodeEncoded: NodeEncoded, blockNumber: BigInt): Unit = {
+  def saveNode(nodeHash: NodeHash, nodeEncoded: NodeEncoded, blockNumber: BigInt): Unit =
     stateStorage.saveNode(nodeHash, nodeEncoded, blockNumber)
-  }
 
   override protected def getHashByBlockNumber(number: BigInt): Option[ByteString] =
     blockNumberMappingStorage.get(number)

src/test/scala/io/iohk/ethereum/blockchain/sync/ScenarioSetup.scala

@@ -21,8 +21,8 @@ import scala.concurrent.ExecutionContext
   * [[io.iohk.ethereum.nodebuilder.Node Node]].
   */
 trait ScenarioSetup extends StdTestConsensusBuilder with SyncConfigBuilder with StdLedgerBuilder {
-  protected lazy val executionContext = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(4))
-  protected lazy val monixScheduler = Scheduler(executionContext)
+  protected lazy val executionContextExecutor = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(4))
+  protected lazy val monixScheduler = Scheduler(executionContextExecutor)
   protected lazy val successValidators: Validators = Mocks.MockValidatorsAlwaysSucceed
   protected lazy val failureValidators: Validators = Mocks.MockValidatorsAlwaysFail
   protected lazy val ethashValidators: ValidatorsExecutor = ValidatorsExecutor(blockchainConfig, Protocol.Ethash)

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

@@ -102,7 +102,7 @@ class BlockFetcherSpec
       peersClient.expectMsgPF() { case PeersClient.Request(msg, _, _) if msg == firstGetBlockHeadersRequest => () }
     }
 
-    "should not enqueue requested blocks if the received bodies does not match" in new TestSetup {
+    "should not enqueue requested blocks if the received bodies do not match" in new TestSetup {
 
       // Important: Here we are forcing the mismatch between request headers and received bodies
       override lazy val validators = new MockValidatorsFailingOnBlockBodies

src/test/scala/io/iohk/ethereum/ledger/BlockImportSpec.scala

@@ -146,6 +146,57 @@ class BlockImportSpec extends AnyFlatSpec with Matchers with ScalaFutures {
     blockQueue.isQueued(oldBlock3.header.hash) shouldBe true
   }
 
+  it should "fail to get bestblock after reorganisation of the longer chain to a shorter one if desync state happened between cache and db" in new EphemBlockchain {
+    val block1: Block = getBlock(bestNum - 2)
+    // new chain is shorter but has a higher weight
+    val newBlock2: Block = getBlock(bestNum - 1, difficulty = 101, parent = block1.header.hash)
+    val newBlock3: Block = getBlock(bestNum, difficulty = 333, parent = newBlock2.header.hash)
+    val oldBlock2: Block = getBlock(bestNum - 1, difficulty = 102, parent = block1.header.hash)
+    val oldBlock3: Block = getBlock(bestNum, difficulty = 103, parent = oldBlock2.header.hash)
+    val oldBlock4: Block = getBlock(bestNum + 1, difficulty = 104, parent = oldBlock3.header.hash)
+
+    val weight1 = ChainWeight.totalDifficultyOnly(block1.header.difficulty + 999)
+    val newWeight2 = weight1.increase(newBlock2.header)
+    val newWeight3 = newWeight2.increase(newBlock3.header)
+    val oldWeight2 = weight1.increase(oldBlock2.header)
+    val oldWeight3 = oldWeight2.increase(oldBlock3.header)
+    val oldWeight4 = oldWeight3.increase(oldBlock4.header)
+
+    blockchain.save(block1, Nil, weight1, saveAsBestBlock = true)
+    blockchain.save(oldBlock2, receipts, oldWeight2, saveAsBestBlock = true)
+    blockchain.save(oldBlock3, Nil, oldWeight3, saveAsBestBlock = true)
+    blockchain.save(oldBlock4, Nil, oldWeight4, saveAsBestBlock = true)
+
+    val ancestorForValidation: Block = getBlock(0, difficulty = 1)
+    blockchain.save(ancestorForValidation, Nil, ChainWeight.totalDifficultyOnly(1), saveAsBestBlock = false)
+
+    val oldBranch = List(oldBlock2, oldBlock3, oldBlock4)
+    val newBranch = List(newBlock2, newBlock3)
+    val blockData2 = BlockData(newBlock2, Seq.empty[Receipt], newWeight2)
+    val blockData3 = BlockData(newBlock3, Seq.empty[Receipt], newWeight3)
+
+    (ledgerWithMockedBlockExecution.blockExecution.executeAndValidateBlocks _)
+      .expects(newBranch, *)
+      .returning((List(blockData2, blockData3), None))
+
+    whenReady(ledgerWithMockedBlockExecution.importBlock(newBlock3).runToFuture) { _ shouldEqual BlockEnqueued }
+    whenReady(ledgerWithMockedBlockExecution.importBlock(newBlock2).runToFuture) { result =>
+      result shouldEqual ChainReorganised(oldBranch, newBranch, List(newWeight2, newWeight3))
+    }
+
+    // Saving new blocks, because it's part of executeBlocks method mechanism
+    blockchain.save(blockData2.block, blockData2.receipts, blockData2.weight, saveAsBestBlock = true)
+    blockchain.save(blockData3.block, blockData3.receipts, blockData3.weight, saveAsBestBlock = true)
+
+    //saving to cache the value of the best block from the initial chain. This recreates the bug ETCM-626, where (possiby) because of the thread of execution
+    // dying before updating the storage but after updating the cache, inconsistency is created
+    blockchain.saveBestKnownBlocks(oldBlock4.number)
+
+    assertThrows[NoSuchElementException] {
+      blockchain.getBestBlock()
+    }
+  }
+
   it should "handle error when trying to reorganise chain" in new EphemBlockchain {
     val block1: Block = getBlock(bestNum - 2)
     val newBlock2: Block = getBlock(bestNum - 1, difficulty = 101, parent = block1.header.hash)
@@ -166,11 +217,8 @@ class BlockImportSpec extends AnyFlatSpec with Matchers with ScalaFutures {
     val ancestorForValidation: Block = getBlock(0, difficulty = 1)
     blockchain.save(ancestorForValidation, Nil, ChainWeight.totalDifficultyOnly(1), saveAsBestBlock = false)
 
-    //FIXME: unused vals???
-    val oldBranch = List(oldBlock2, oldBlock3)
     val newBranch = List(newBlock2, newBlock3)
     val blockData2 = BlockData(newBlock2, Seq.empty[Receipt], newWeight2)
-    val blockData3 = BlockData(newBlock3, Seq.empty[Receipt], newWeight3)
 
     (ledgerWithMockedBlockExecution.blockExecution.executeAndValidateBlocks _)
       .expects(newBranch, *)