ETCM-732: Handle missing state node in regular sync using Either instead of Task.

Author: milenkara

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

@@ -6,16 +6,16 @@ import cats.data.NonEmptyList
 import io.iohk.ethereum.blockchain.sync.regular.BlockImporter.NewCheckpoint
 import io.iohk.ethereum.blockchain.sync.regular.{BlockFetcher, BlockImporter}
 import io.iohk.ethereum.checkpointing.CheckpointingTestHelpers
-import io.iohk.ethereum.consensus.{GetBlockHeaderByHash, GetNBlocksBack}
 import io.iohk.ethereum.consensus.blocks.CheckpointBlockGenerator
 import io.iohk.ethereum.consensus.ethash.validators.{OmmersValidator, StdOmmersValidator}
 import io.iohk.ethereum.consensus.validators.Validators
+import io.iohk.ethereum.consensus.{GetBlockHeaderByHash, GetNBlocksBack}
 import io.iohk.ethereum.domain._
+import io.iohk.ethereum.ledger.Ledger.BlockResult
 import io.iohk.ethereum.mpt.MerklePatriciaTrie
-import io.iohk.ethereum.utils.Config.SyncConfig
 import io.iohk.ethereum.utils.Config
+import io.iohk.ethereum.utils.Config.SyncConfig
 import io.iohk.ethereum.{Fixtures, Mocks, ObjectGenerators, crypto}
-import io.iohk.ethereum.ledger.Ledger.BlockResult
 import monix.execution.Scheduler
 import org.scalamock.scalatest.MockFactory
 import org.scalatest.BeforeAndAfterAll
@@ -240,4 +240,39 @@ class BlockImporterItSpec
       Thread.sleep(1000); blockchain.getLatestCheckpointBlockNumber() shouldEqual newBlock5.header.number + 1
     }
   }
+
+  it should "ask BlockFetcher to resolve missing node" in {
+    val parent = blockchain.getBestBlock().get
+    val newBlock: Block = getBlock(genesisBlock.number + 5, difficulty = 104, parent = parent.header.hash)
+    val invalidBlock = newBlock.copy(header = newBlock.header.copy(beneficiary = Address(111).bytes))
+
+    val ledger = new TestLedgerImpl(successValidators)
+    val blockImporter = system.actorOf(
+      BlockImporter.props(
+        fetcherProbe.ref,
+        ledger,
+        blockchain,
+        syncConfig,
+        ommersPoolProbe.ref,
+        broadcasterProbe.ref,
+        pendingTransactionsManagerProbe.ref,
+        supervisor.ref
+      )
+    )
+
+    blockImporter ! BlockImporter.Start
+    blockImporter ! BlockFetcher.PickedBlocks(NonEmptyList.fromListUnsafe(List(invalidBlock)))
+
+    eventually {
+      Thread.sleep(1000)
+
+      val msg = fetcherProbe.fishForMessage() {
+        case BlockFetcher.FetchStateNode(_) => true
+        case _ => false
+      }.asInstanceOf[BlockFetcher.FetchStateNode]
+
+      msg.hash.length should be > 0
+    }
+
+  }
 }

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

@@ -206,14 +206,16 @@ class BlockImporter(
           case DuplicateBlock | BlockEnqueued =>
             tryImportBlocks(restOfBlocks, importedBlocks)
 
+          case BlockImportFailedDueToMissingNode(missingNodeException) if syncConfig.redownloadMissingStateNodes =>
+            Task.now((importedBlocks, Some(missingNodeException)))
+
+          case BlockImportFailedDueToMissingNode(missingNodeException) =>
+            Task.raiseError(missingNodeException)
+
           case err @ (UnknownParent | BlockImportFailed(_)) =>
             log.error("Block {} import failed", blocks.head.number)
             Task.now((importedBlocks, Some(err)))
         }
-        .onErrorHandle {
-          case missingNodeEx: MissingNodeException if syncConfig.redownloadMissingStateNodes =>
-            (importedBlocks, Some(missingNodeEx))
-        }
     }
 
   private def importBlock(
@@ -246,18 +248,18 @@ class BlockImporter(
                   supervisor ! ProgressProtocol.ImportedBlock(newBlock.number, block.hasCheckpoint, internally)
                 case None => ()
               }
+            case BlockImportFailedDueToMissingNode(missingNodeException) if syncConfig.redownloadMissingStateNodes =>
+              // state node re-download will be handled when downloading headers
+              doLog(importMessages.missingStateNode(missingNodeException))
+              Running
+            case BlockImportFailedDueToMissingNode(missingNodeException) =>
+              Task.raiseError(missingNodeException)
             case BlockImportFailed(error) =>
               if (informFetcherOnFail) {
                 fetcher ! BlockFetcher.BlockImportFailed(block.number, error)
               }
           }
           .map(_ => Running)
-          .recover {
-            case missingNodeEx: MissingNodeException if syncConfig.redownloadMissingStateNodes =>
-              // state node re-download will be handled when downloading headers
-              doLog(importMessages.missingStateNode(missingNodeEx))
-              Running
-          }
       },
       blockImportType
     )

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

@@ -30,6 +30,7 @@ sealed abstract class ImportMessages(block: Block) {
       case UnknownParent => orphaned()
       case ChainReorganised(_, newBranch, _) => reorganisedChain(newBranch)
       case BlockImportFailed(error) => importFailed(error)
+      case BlockImportFailedDueToMissingNode(reason) => missingStateNode(reason)
     }
 }
 

src/main/scala/io/iohk/ethereum/ledger/BlockImport.scala

@@ -3,8 +3,9 @@ package io.iohk.ethereum.ledger
 import akka.util.ByteString
 import io.iohk.ethereum.consensus.validators.BlockHeaderError.HeaderParentNotFoundError
 import io.iohk.ethereum.domain._
-import io.iohk.ethereum.ledger.BlockExecutionError.ValidationBeforeExecError
+import io.iohk.ethereum.ledger.BlockExecutionError.{MPTError, ValidationBeforeExecError}
 import io.iohk.ethereum.ledger.BlockQueue.Leaf
+import io.iohk.ethereum.mpt.MerklePatriciaTrie.MissingNodeException
 import io.iohk.ethereum.utils.{ByteStringUtils, Logger}
 import monix.eval.Task
 import monix.execution.Scheduler
@@ -54,6 +55,9 @@ class BlockImport(
           case None =>
             BlockImportedToTop(importedBlocks)
 
+          case Some(MPTError(reason)) if reason.isInstanceOf[MissingNodeException] =>
+            BlockImportFailedDueToMissingNode(reason.asInstanceOf[MissingNodeException])
+
           case Some(error) if importedBlocks.isEmpty =>
             blockQueue.removeSubtree(block.header.hash)
             BlockImportFailed(error.toString)
@@ -161,7 +165,11 @@ class BlockImport(
     reorgResult match {
       case Some(execResult) =>
         execResult.fold(
-          err => BlockImportFailed(s"Error while trying to reorganise chain: $err"),
+          {
+            case MPTError(reason) if reason.isInstanceOf[MissingNodeException] =>
+              BlockImportFailedDueToMissingNode(reason.asInstanceOf[MissingNodeException])
+            case err => BlockImportFailed(s"Error while trying to reorganise chain: $err")
+          },
           ChainReorganised.tupled
         )
 
@@ -180,6 +188,9 @@ class BlockImport(
       case None =>
         Right(oldBlocksData.map(_.block), executedBlocks.map(_.block), executedBlocks.map(_.weight))
 
+      case Some(MPTError(reason)) if reason.isInstanceOf[MissingNodeException] =>
+        Left(MPTError(reason))
+
       case Some(error) =>
         revertChainReorganisation(newBranch, oldBlocksData, executedBlocks)
         Left(error)
@@ -271,4 +282,6 @@ case class ChainReorganised(
 
 case class BlockImportFailed(error: String) extends BlockImportResult
 
+case class BlockImportFailedDueToMissingNode(reason: MissingNodeException) extends BlockImportResult
+
 case object UnknownParent extends BlockImportResult

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

@@ -295,7 +295,7 @@ class RegularSyncSpec
     "fetching state node" should {
       abstract class MissingStateNodeFixture(system: ActorSystem) extends Fixture(system) {
         val failingBlock: Block = testBlocksChunked.head.head
-        ledger.setImportResult(failingBlock, Task.raiseError(new MissingNodeException(failingBlock.hash)))
+        ledger.setImportResult(failingBlock, Task.now(BlockImportFailedDueToMissingNode(new MissingNodeException(failingBlock.hash))))
       }
 
       "blacklist peer which returns empty response" in sync(new MissingStateNodeFixture(testSystem) {
@@ -366,7 +366,7 @@ class RegularSyncSpec
         (ledger
           .importBlock(_: Block)(_: Scheduler))
           .when(*, *)
-          .returns(Task.raiseError(new MissingNodeException(failingBlock.hash)))
+          .returns(Task.now(BlockImportFailedDueToMissingNode(new MissingNodeException(failingBlock.hash))))
 
         var saveNodeWasCalled: Boolean = false
         val nodeData = List(ByteString(failingBlock.header.toBytes: Array[Byte]))