Etcm 139 incomplete data errors during fast sync (#748)

Author: kapke

build.sbt

@@ -68,8 +68,12 @@ val root = {
       libraryDependencies ++= dep
     )
     .settings(executableScriptName := name.value)
-    .settings(inConfig(Integration)(Defaults.testSettings
-      ++ org.scalafmt.sbt.ScalafmtPlugin.scalafmtConfigSettings :+ (Test / parallelExecution := false)): _*)
+    .settings(
+      inConfig(Integration)(
+        Defaults.testSettings
+          ++ org.scalafmt.sbt.ScalafmtPlugin.scalafmtConfigSettings :+ (Test / parallelExecution := false)
+      ): _*
+    )
     .settings(inConfig(Benchmark)(Defaults.testSettings :+ (Test / parallelExecution := false)): _*)
     .settings(inConfig(Evm)(Defaults.testSettings :+ (Test / parallelExecution := false)): _*)
     .settings(inConfig(Ets)(Defaults.testSettings :+ (Test / parallelExecution := false)): _*)
@@ -161,7 +165,7 @@ addCommandAlias(
     |;scalafmtAll
     |;scalastyle
     |;test:scalastyle
-    |;test
+    |;testQuick
     |;it:test
     |""".stripMargin
 )

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

@@ -3,16 +3,16 @@ package io.iohk.ethereum.jsonrpc
 import akka.util.ByteString
 import io.iohk.ethereum.crypto.ECDSASignature
 import io.iohk.ethereum.jsonrpc.CheckpointingService._
-import io.iohk.ethereum.jsonrpc.JsonRpcController.Codec
-import io.iohk.ethereum.jsonrpc.JsonRpcErrors.InvalidParams
-import io.iohk.ethereum.jsonrpc.JsonSerializers.QuantitiesSerializer
+import io.iohk.ethereum.jsonrpc.JsonRpcError.InvalidParams
+import io.iohk.ethereum.jsonrpc.serialization.JsonMethodCodec
+import io.iohk.ethereum.jsonrpc.serialization.JsonSerializers.QuantitiesSerializer
 import org.json4s.JsonAST._
 import org.json4s.{Extraction, JsonAST}
 
 object CheckpointingJsonMethodsImplicits extends JsonMethodsImplicits {
 
-  implicit val checkpointing_getLatestBlock: Codec[GetLatestBlockRequest, GetLatestBlockResponse] =
-    new Codec[GetLatestBlockRequest, GetLatestBlockResponse] {
+  implicit val checkpointing_getLatestBlock: JsonMethodCodec[GetLatestBlockRequest, GetLatestBlockResponse] =
+    new JsonMethodCodec[GetLatestBlockRequest, GetLatestBlockResponse] {
       override def decodeJson(
           params: Option[JsonAST.JArray]
       ): Either[JsonRpcError, GetLatestBlockRequest] =
@@ -30,8 +30,8 @@ object CheckpointingJsonMethodsImplicits extends JsonMethodsImplicits {
         Extraction.decompose(resp)(formats - QuantitiesSerializer)
     }
 
-  implicit val checkpointing_pushCheckpoint: Codec[PushCheckpointRequest, PushCheckpointResponse] =
-    new Codec[PushCheckpointRequest, PushCheckpointResponse] {
+  implicit val checkpointing_pushCheckpoint: JsonMethodCodec[PushCheckpointRequest, PushCheckpointResponse] =
+    new JsonMethodCodec[PushCheckpointRequest, PushCheckpointResponse] {
       override def decodeJson(
           params: Option[JsonAST.JArray]
       ): Either[JsonRpcError, PushCheckpointRequest] =

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

@@ -1,14 +1,14 @@
 package io.iohk.ethereum.jsonrpc
 
 import io.iohk.ethereum.jsonrpc.DebugService.{ListPeersInfoRequest, ListPeersInfoResponse}
-import io.iohk.ethereum.jsonrpc.JsonRpcController.{Codec, JsonEncoder}
-import io.iohk.ethereum.jsonrpc.JsonRpcController.JsonDecoder.NoParamsDecoder
+import io.iohk.ethereum.jsonrpc.serialization.JsonMethodDecoder.NoParamsMethodDecoder
+import io.iohk.ethereum.jsonrpc.serialization.{JsonEncoder, JsonMethodCodec}
 import org.json4s.JsonAST.{JArray, JString, JValue}
 
 object DebugJsonMethodsImplicits extends JsonMethodsImplicits {
 
-  implicit val debug_listPeersInfo: Codec[ListPeersInfoRequest, ListPeersInfoResponse] =
-    new NoParamsDecoder(ListPeersInfoRequest()) with JsonEncoder[ListPeersInfoResponse] {
+  implicit val debug_listPeersInfo: JsonMethodCodec[ListPeersInfoRequest, ListPeersInfoResponse] =
+    new NoParamsMethodDecoder(ListPeersInfoRequest()) with JsonEncoder[ListPeersInfoResponse] {
       def encodeJson(t: ListPeersInfoResponse): JValue =
         JArray(t.peers.map(a => JString(a.toString)))
     }

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

@@ -19,6 +19,7 @@ import io.iohk.ethereum.jsonrpc.FilterManager.{FilterChanges, FilterLogs, LogFil
 import io.iohk.ethereum.jsonrpc.JsonRpcController.JsonRpcConfig
 import io.iohk.ethereum.keystore.KeyStore
 import io.iohk.ethereum.ledger.{InMemoryWorldStateProxy, Ledger, StxLedger}
+import io.iohk.ethereum.mpt.MerklePatriciaTrie.MissingNodeException
 import io.iohk.ethereum.ommers.OmmersPool
 import io.iohk.ethereum.rlp
 import io.iohk.ethereum.rlp.RLPImplicitConversions._
@@ -234,7 +235,7 @@ class EthService(
 
   private[this] def ifEthash[Req, Res](req: Req)(f: Req => Res): ServiceResponse[Res] = {
     @inline def F[A](x: A): Future[A] = Future.successful(x)
-    consensus.ifEthash[ServiceResponse[Res]](_ => F(Right(f(req))))(F(Left(JsonRpcErrors.ConsensusIsNotEthash)))
+    consensus.ifEthash[ServiceResponse[Res]](_ => F(Right(f(req))))(F(Left(JsonRpcError.ConsensusIsNotEthash)))
   }
 
   def protocolVersion(req: ProtocolVersionRequest): ServiceResponse[ProtocolVersionResponse] =
@@ -565,7 +566,7 @@ class EthService(
           )
         }
       response
-    })(Future.successful(Left(JsonRpcErrors.ConsensusIsNotEthash)))
+    })(Future.successful(Left(JsonRpcError.ConsensusIsNotEthash)))
 
   private def getOmmersFromPool(parentBlockHash: ByteString): Future[OmmersPool.Ommers] =
     consensus.ifEthash(ethash => {
@@ -610,7 +611,7 @@ class EthService(
             Right(SubmitWorkResponse(false))
         }
       }
-    })(Future.successful(Left(JsonRpcErrors.ConsensusIsNotEthash)))
+    })(Future.successful(Left(JsonRpcError.ConsensusIsNotEthash)))
 
   /**
     * Implements the eth_syncing method that returns syncing information if the node is syncing.
@@ -649,10 +650,10 @@ class EthService(
           pendingTransactionsManager ! PendingTransactionsManager.AddOrOverrideTransaction(signedTransaction)
           Future.successful(Right(SendRawTransactionResponse(signedTransaction.hash)))
         } else {
-          Future.successful(Left(JsonRpcErrors.InvalidRequest))
+          Future.successful(Left(JsonRpcError.InvalidRequest))
         }
       case Failure(_) =>
-        Future.successful(Left(JsonRpcErrors.InvalidRequest))
+        Future.successful(Left(JsonRpcError.InvalidRequest))
     }
   }
 
@@ -669,7 +670,7 @@ class EthService(
     val dataEither = (tx.function, tx.contractCode) match {
       case (Some(function), None) => Right(rlp.encode(RLPList(function, args)))
       case (None, Some(contractCode)) => Right(rlp.encode(RLPList(contractCode, args)))
-      case _ => Left(JsonRpcErrors.InvalidParams("Iele transaction should contain either functionName or contractCode"))
+      case _ => Left(JsonRpcError.InvalidParams("Iele transaction should contain either functionName or contractCode"))
     }
 
     dataEither match {
@@ -722,7 +723,7 @@ class EthService(
           Right(GetUncleCountByBlockHashResponse(blockBody.uncleNodesList.size))
         case None =>
           Left(
-            JsonRpcErrors.InvalidParams(s"Block with hash ${Hex.toHexString(req.blockHash.toArray[Byte])} not found")
+            JsonRpcError.InvalidParams(s"Block with hash ${Hex.toHexString(req.blockHash.toArray[Byte])} not found")
           )
       }
     }
@@ -786,31 +787,22 @@ class EthService(
       .flatMap(_ => Right(None))
   }
 
-  def getBalance(req: GetBalanceRequest): ServiceResponse[GetBalanceResponse] = {
-    Future {
-      withAccount(req.address, req.block) { account =>
-        GetBalanceResponse(account.balance)
-      }
+  def getBalance(req: GetBalanceRequest): ServiceResponse[GetBalanceResponse] =
+    withAccount(req.address, req.block) { account =>
+      GetBalanceResponse(account.balance)
     }
-  }
 
-  def getStorageAt(req: GetStorageAtRequest): ServiceResponse[GetStorageAtResponse] = {
-    Future {
-      withAccount(req.address, req.block) { account =>
-        GetStorageAtResponse(
-          blockchain.getAccountStorageAt(account.storageRoot, req.position, blockchainConfig.ethCompatibleStorage)
-        )
-      }
+  def getStorageAt(req: GetStorageAtRequest): ServiceResponse[GetStorageAtResponse] =
+    withAccount(req.address, req.block) { account =>
+      GetStorageAtResponse(
+        blockchain.getAccountStorageAt(account.storageRoot, req.position, blockchainConfig.ethCompatibleStorage)
+      )
     }
-  }
 
-  def getTransactionCount(req: GetTransactionCountRequest): ServiceResponse[GetTransactionCountResponse] = {
-    Future {
-      withAccount(req.address, req.block) { account =>
-        GetTransactionCountResponse(account.nonce)
-      }
+  def getTransactionCount(req: GetTransactionCountRequest): ServiceResponse[GetTransactionCountResponse] =
+    withAccount(req.address, req.block) { account =>
+      GetTransactionCountResponse(account.nonce)
     }
-  }
 
   def newFilter(req: NewFilterRequest): ServiceResponse[NewFilterResponse] = {
     implicit val timeout: Timeout = Timeout(filterConfig.filterManagerQueryTimeout)
@@ -875,20 +867,25 @@ class EthService(
       }
   }
 
-  private def withAccount[T](address: Address, blockParam: BlockParam)(f: Account => T): Either[JsonRpcError, T] = {
-    resolveBlock(blockParam).map { case ResolvedBlock(block, _) =>
-      f(
-        blockchain.getAccount(address, block.header.number).getOrElse(Account.empty(blockchainConfig.accountStartNonce))
-      )
+  private def withAccount[T](address: Address, blockParam: BlockParam)(makeResponse: Account => T): ServiceResponse[T] =
+    Future {
+      resolveBlock(blockParam)
+        .map { case ResolvedBlock(block, _) =>
+          blockchain
+            .getAccount(address, block.header.number)
+            .getOrElse(Account.empty(blockchainConfig.accountStartNonce))
+        }
+        .map(makeResponse)
+    }.recover { case _: MissingNodeException =>
+      Left(JsonRpcError.NodeNotFound)
     }
-  }
 
   private def resolveBlock(blockParam: BlockParam): Either[JsonRpcError, ResolvedBlock] = {
     def getBlock(number: BigInt): Either[JsonRpcError, Block] = {
       blockchain
         .getBlockByNumber(number)
         .map(Right.apply)
-        .getOrElse(Left(JsonRpcErrors.InvalidParams(s"Block $number not found")))
+        .getOrElse(Left(JsonRpcError.InvalidParams(s"Block $number not found")))
     }
 
     blockParam match {
@@ -941,7 +938,7 @@ class EthService(
     if (numBlocksToSearch > jsonRpcConfig.accountTransactionsMaxBlocks) {
       Future.successful(
         Left(
-          JsonRpcErrors.InvalidParams(
+          JsonRpcError.InvalidParams(
             s"""Maximum number of blocks to search is ${jsonRpcConfig.accountTransactionsMaxBlocks}, requested: $numBlocksToSearch.
            |See: 'network.rpc.account-transactions-max-blocks' config.""".stripMargin
           )
@@ -975,13 +972,10 @@ class EthService(
     }
   }
 
-  def getStorageRoot(req: GetStorageRootRequest): ServiceResponse[GetStorageRootResponse] = {
-    Future {
-      withAccount(req.address, req.block) { account =>
-        GetStorageRootResponse(account.storageRoot)
-      }
+  def getStorageRoot(req: GetStorageRootRequest): ServiceResponse[GetStorageRootResponse] =
+    withAccount(req.address, req.block) { account =>
+      GetStorageRootResponse(account.storageRoot)
     }
-  }
 
   /**
     * Returns the transactions that are pending in the transaction pool and have a from address that is one of the accounts this node manages.

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

@@ -2,9 +2,9 @@ package io.iohk.ethereum.jsonrpc
 
 import akka.util.ByteString
 import io.iohk.ethereum.jsonrpc.EthService._
-import io.iohk.ethereum.jsonrpc.JsonRpcController.{JsonDecoder, JsonEncoder}
-import io.iohk.ethereum.jsonrpc.JsonRpcErrors.InvalidParams
+import io.iohk.ethereum.jsonrpc.JsonRpcError.InvalidParams
 import io.iohk.ethereum.jsonrpc.PersonalService.{InvalidAddress, SendIeleTransactionRequest}
+import io.iohk.ethereum.jsonrpc.serialization.{JsonEncoder, JsonMethodDecoder}
 import org.json4s.JsonAST.{JArray, JObject, JString, JValue}
 
 object IeleJsonMethodsImplicits extends JsonMethodsImplicits {
@@ -34,7 +34,7 @@ object IeleJsonMethodsImplicits extends JsonMethodsImplicits {
     )
   }
 
-  implicit val iele_call = new JsonDecoder[IeleCallRequest] with JsonEncoder[IeleCallResponse] {
+  implicit val iele_call = new JsonMethodDecoder[IeleCallRequest] with JsonEncoder[IeleCallResponse] {
     def decodeJson(params: Option[JArray]): Either[JsonRpcError, IeleCallRequest] =
       params match {
         case Some(JArray((txObj: JObject) :: (blockValue: JValue) :: Nil)) =>
@@ -93,7 +93,7 @@ object IeleJsonMethodsImplicits extends JsonMethodsImplicits {
     } yield IeleTransactionRequest(from, to, value, gas, gasPrice, nonce, function, arguments, contractCode)
   }
 
-  implicit val iele_sendTransaction = new JsonDecoder[SendIeleTransactionRequest] {
+  implicit val iele_sendTransaction = new JsonMethodDecoder[SendIeleTransactionRequest] {
     def decodeJson(params: Option[JArray]): Either[JsonRpcError, SendIeleTransactionRequest] =
       params match {
         case Some(JArray(JObject(tx) :: _)) =>