Merge branch 'develop' into fix/checkpointing-metrics

Author: Nicolás Tallar

src/main/scala/io/iohk/ethereum/faucet/FaucetConfig.scala

@@ -1,6 +1,6 @@
 package io.iohk.ethereum.faucet
 
-import com.typesafe.config.{ConfigFactory, Config}
+import com.typesafe.config.{Config, ConfigFactory}
 import io.iohk.ethereum.domain.Address
 
 import scala.concurrent.duration.{FiniteDuration, _}
@@ -11,36 +11,50 @@ trait FaucetConfigBuilder {
   lazy val faucetConfig: FaucetConfig = FaucetConfig(rawConfig)
 }
 
+case class RpcClientConfig(
+    address: String,
+    timeout: FiniteDuration
+)
+
+object RpcClientConfig {
+  def apply(rpcClientConfig: Config): RpcClientConfig = {
+
+    RpcClientConfig(
+      address = rpcClientConfig.getString("rpc-address"),
+      timeout = rpcClientConfig.getDuration("timeout").toMillis.millis
+    )
+  }
+}
+
 case class FaucetConfig(
     walletAddress: Address,
     walletPassword: String,
     txGasPrice: BigInt,
     txGasLimit: BigInt,
     txValue: BigInt,
-    rpcAddress: String,
+    rpcClient: RpcClientConfig,
     keyStoreDir: String,
     minRequestInterval: FiniteDuration,
     handlerTimeout: FiniteDuration,
-    responseTimeout: FiniteDuration,
+    actorCommunicationMargin: FiniteDuration,
     supervisor: SupervisorConfig,
     shutdownTimeout: FiniteDuration
 )
 
 object FaucetConfig {
   def apply(typesafeConfig: Config): FaucetConfig = {
     val faucetConfig = typesafeConfig.getConfig("faucet")
-
     FaucetConfig(
       walletAddress = Address(faucetConfig.getString("wallet-address")),
       walletPassword = faucetConfig.getString("wallet-password"),
       txGasPrice = faucetConfig.getLong("tx-gas-price"),
       txGasLimit = faucetConfig.getLong("tx-gas-limit"),
       txValue = faucetConfig.getLong("tx-value"),
-      rpcAddress = faucetConfig.getString("rpc-client.rpc-address"),
+      rpcClient = RpcClientConfig(faucetConfig.getConfig("rpc-client")),
       keyStoreDir = faucetConfig.getString("keystore-dir"),
       minRequestInterval = faucetConfig.getDuration("min-request-interval").toMillis.millis,
       handlerTimeout = faucetConfig.getDuration("handler-timeout").toMillis.millis,
-      responseTimeout = faucetConfig.getDuration("response-timeout").toMillis.millis,
+      actorCommunicationMargin = faucetConfig.getDuration("actor-communication-margin").toMillis.millis,
       supervisor = SupervisorConfig(faucetConfig),
       shutdownTimeout = faucetConfig.getDuration("shutdown-timeout").toMillis.millis
     )

src/main/scala/io/iohk/ethereum/faucet/SupervisorConfig.scala

@@ -17,7 +17,7 @@ object SupervisorConfig {
 
     SupervisorConfig(
       supervisorConfig.getDuration("min-backoff").toMillis.millis,
-      supervisorConfig.getDuration("man-backoff").toMillis.millis,
+      supervisorConfig.getDuration("max-backoff").toMillis.millis,
       supervisorConfig.getDouble("random-factor"),
       supervisorConfig.getDuration("auto-reset").toMillis.millis,
       supervisorConfig.getInt("attempts"),

src/main/scala/io/iohk/ethereum/faucet/jsonrpc/FaucetBuilder.scala

@@ -37,7 +37,11 @@ trait FaucetRpcServiceBuilder {
     )
 
   val walletRpcClient: WalletRpcClient =
-    new WalletRpcClient(faucetConfig.rpcAddress, () => sslContext("faucet.rpc-client"))
+    new WalletRpcClient(
+      faucetConfig.rpcClient.address,
+      faucetConfig.rpcClient.timeout,
+      () => sslContext("faucet.rpc-client")
+    )
   val walletService = new WalletService(walletRpcClient, keyStore, faucetConfig)
   val faucetSupervisor: FaucetSupervisor = new FaucetSupervisor(walletService, faucetConfig, shutdown)(system)
   val faucetRpcService = new FaucetRpcService(faucetConfig)

src/main/scala/io/iohk/ethereum/faucet/jsonrpc/FaucetRpcService.scala

@@ -16,7 +16,7 @@ class FaucetRpcService(config: FaucetConfig)(implicit system: ActorSystem)
     with FaucetHandlerSelector
     with Logger {
 
-  implicit lazy val actorTimeout: Timeout = Timeout(config.responseTimeout)
+  implicit lazy val actorTimeout: Timeout = Timeout(config.actorCommunicationMargin + config.rpcClient.timeout)
 
   def sendFunds(sendFundsRequest: SendFundsRequest): ServiceResponse[SendFundsResponse] =
     selectFaucetHandler()

src/main/scala/io/iohk/ethereum/faucet/jsonrpc/WalletRpcClient.scala

@@ -13,11 +13,12 @@ import javax.net.ssl.SSLContext
 import monix.eval.Task
 
 import scala.concurrent.ExecutionContext
+import scala.concurrent.duration.Duration
 
-class WalletRpcClient(node: Uri, getSSLContext: () => Either[SSLError, SSLContext])(implicit
+class WalletRpcClient(node: Uri, timeout: Duration, getSSLContext: () => Either[SSLError, SSLContext])(implicit
     system: ActorSystem,
     ec: ExecutionContext
-) extends RpcClient(node, getSSLContext)
+) extends RpcClient(node, timeout, getSSLContext)
     with Logger {
   import io.iohk.ethereum.jsonrpc.client.CommonJsonCodecs._
 

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

@@ -2,22 +2,31 @@ package io.iohk.ethereum.jsonrpc
 
 import io.iohk.ethereum.consensus.Protocol
 import io.iohk.ethereum.jsonrpc.serialization.JsonEncoder
-import org.json4s.{JInt, JObject, JString, JValue}
+import org.json4s.{JLong, JInt, JObject, JString, JValue}
 
 case class JsonRpcError(code: Int, message: String, data: Option[JValue])
 
 // scalastyle:off magic.number
 // scalastyle:off public.methods.have.type
-object JsonRpcError {
+object JsonRpcError extends JsonMethodsImplicits {
+
   def apply[T: JsonEncoder](code: Int, message: String, data: T): JsonRpcError =
     JsonRpcError(code, message, Some(JsonEncoder[T].encodeJson(data)))
 
+  implicit val rateLimitInformation: JsonEncoder[RateLimitInformation] = (rateLimit: RateLimitInformation) =>
+    JObject(
+      "backoff_seconds" -> JLong(rateLimit.backoffSeconds)
+    )
+
   implicit val jsonRpcErrorEncoder: JsonEncoder[JsonRpcError] = err =>
     JObject(
       List("code" -> JsonEncoder.encode(err.code), "message" -> JsonEncoder.encode(err.message)) ++
         err.data.map("data" -> _)
     )
 
+  case class RateLimitInformation(backoffSeconds: Long)
+  def RateLimitError(backoffSeconds: Long) =
+    JsonRpcError(-32005, "request rate exceeded", RateLimitInformation(backoffSeconds))
   val ParseError = JsonRpcError(-32700, "An error occurred on the server while parsing the JSON text", None)
   val InvalidRequest = JsonRpcError(-32600, "The JSON sent is not a valid Request object", None)
   val MethodNotFound = JsonRpcError(-32601, "The method does not exist / is not available", None)

src/main/scala/io/iohk/ethereum/jsonrpc/client/RpcClient.scala

@@ -5,8 +5,11 @@ import java.util.UUID
 
 import akka.actor.ActorSystem
 import akka.http.scaladsl.model._
+import akka.http.scaladsl.settings.{ClientConnectionSettings, ConnectionPoolSettings}
 import akka.http.scaladsl.unmarshalling.Unmarshal
 import akka.http.scaladsl.{ConnectionContext, Http, HttpsConnectionContext}
+import akka.stream.StreamTcpException
+import akka.stream.scaladsl.TcpIdleTimeoutException
 import io.circe.generic.auto._
 import io.circe.parser.parse
 import io.circe.syntax._
@@ -18,8 +21,9 @@ import javax.net.ssl.SSLContext
 import monix.eval.Task
 
 import scala.concurrent.ExecutionContext
+import scala.concurrent.duration._
 
-abstract class RpcClient(node: Uri, getSSLContext: () => Either[SSLError, SSLContext])(implicit
+abstract class RpcClient(node: Uri, timeout: Duration, getSSLContext: () => Either[SSLError, SSLContext])(implicit
     system: ActorSystem,
     ec: ExecutionContext
 ) extends Logger {
@@ -32,6 +36,12 @@ abstract class RpcClient(node: Uri, getSSLContext: () => Either[SSLError, SSLCon
     Http().defaultClientHttpsContext
   }
 
+  lazy val connectionPoolSettings: ConnectionPoolSettings = ConnectionPoolSettings(system)
+    .withConnectionSettings(
+      ClientConnectionSettings(system)
+        .withIdleTimeout(timeout)
+    )
+
   protected def doRequest[T: Decoder](method: String, args: Seq[Json]): RpcResponse[T] = {
     doJsonRequest(method, args).map(_.flatMap(getResult[T]))
   }
@@ -60,11 +70,21 @@ abstract class RpcClient(node: Uri, getSSLContext: () => Either[SSLError, SSLCon
 
     Task
       .deferFuture(for {
-        response <- Http().singleRequest(request, connectionContext)
+        response <- Http().singleRequest(request, connectionContext, connectionPoolSettings)
         data <- Unmarshal(response.entity).to[String]
-      } yield parse(data).left.map(e => RpcClientError(e.message)))
+      } yield parse(data).left.map(e => ParserError(e.message)))
       .onErrorHandle { ex: Throwable =>
-        Left(RpcClientError(s"RPC request failed: ${exceptionToString(ex)}"))
+        ex match {
+          case _: TcpIdleTimeoutException =>
+            log.error("RPC request", ex)
+            Left(ConnectionError(s"RPC request timeout"))
+          case _: StreamTcpException =>
+            log.error("Connection not established", ex)
+            Left(ConnectionError(s"Connection not established"))
+          case _ =>
+            log.error("RPC request failed", ex)
+            Left(RpcClientError("RPC request failed"))
+        }
       }
   }
 
@@ -97,5 +117,7 @@ object RpcClient {
 
   case class ParserError(msg: String) extends RpcError
 
+  case class ConnectionError(msg: String) extends RpcError
+
   case class RpcClientError(msg: String) extends RpcError
 }

src/main/scala/io/iohk/ethereum/jsonrpc/server/http/JsonRpcHttpServer.scala

@@ -10,19 +10,19 @@ import ch.megard.akka.http.cors.javadsl.CorsRejection
 import ch.megard.akka.http.cors.scaladsl.CorsDirectives._
 import ch.megard.akka.http.cors.scaladsl.model.HttpOriginMatcher
 import ch.megard.akka.http.cors.scaladsl.settings.CorsSettings
+import com.typesafe.config.{Config => TypesafeConfig}
 import de.heikoseeberger.akkahttpjson4s.Json4sSupport
 import io.iohk.ethereum.faucet.jsonrpc.FaucetJsonRpcController
 import io.iohk.ethereum.jsonrpc._
-import io.iohk.ethereum.security.SSLError
 import io.iohk.ethereum.jsonrpc.serialization.JsonSerializers
 import io.iohk.ethereum.jsonrpc.server.controllers.JsonRpcBaseController
 import io.iohk.ethereum.jsonrpc.server.http.JsonRpcHttpServer.JsonRpcHttpServerConfig
+import io.iohk.ethereum.security.SSLError
 import io.iohk.ethereum.utils.{ConfigUtils, Logger}
 import javax.net.ssl.SSLContext
 import monix.eval.Task
 import monix.execution.Scheduler.Implicits.global
 import org.json4s.{DefaultFormats, JInt, native}
-import com.typesafe.config.{Config => TypesafeConfig}
 
 import scala.concurrent.duration.{FiniteDuration, _}
 
@@ -73,10 +73,14 @@ trait JsonRpcHttpServer extends Json4sSupport with RateLimit with Logger {
   }
 
   def handleRateLimitedRequest(clientAddress: RemoteAddress, request: JsonRpcRequest): StandardRoute = {
-    if (isBelowRateLimit(clientAddress)) {
-      log.warn(s"Request limit exceeded for ip ${clientAddress.toIP.getOrElse("unknown")}")
+    if (isBelowRateLimit(clientAddress))
       complete(jsonRpcController.handleRequest(request).runToFuture)
-    } else complete(StatusCodes.TooManyRequests)
+    else {
+      log.warn(s"Request limit exceeded for ip ${clientAddress.toIP.getOrElse("unknown")}")
+      complete(
+        (StatusCodes.TooManyRequests, JsonRpcError.RateLimitError(config.rateLimit.minRequestInterval.toSeconds))
+      )
+    }
   }
 
   /**
@@ -113,7 +117,7 @@ trait JsonRpcHttpServer extends Json4sSupport with RateLimit with Logger {
           .traverse(requests)(request => jsonRpcController.handleRequest(request))
           .runToFuture
       }
-    } else complete(StatusCodes.MethodNotAllowed)
+    } else complete(StatusCodes.MethodNotAllowed, JsonRpcError.MethodNotFound)
   }
 }
 

src/test/resources/application.conf

@@ -156,17 +156,18 @@ faucet {
   rpc-client {
     rpc-address = "http://127.0.0.1:8546/"
     certificate = null
+    timeout = 2.seconds
   }
 
   min-request-interval = 1.minute
 
   handler-timeout = 2.seconds
 
-  response-timeout = 2.seconds
+  actor-communication-margin = 1.seconds
 
   supervisor {
     min-backoff = 3.seconds
-    man-backoff = 30.seconds
+    max-backoff = 30.seconds
     random-factor = 0.2
     auto-reset = 10.seconds
     attempts = 4

src/test/scala/io/iohk/ethereum/faucet/jsonrpc/FaucetRpcServiceSpec.scala

@@ -13,7 +13,7 @@ import io.iohk.ethereum.faucet.FaucetHandler.FaucetHandlerResponse.{
 }
 import io.iohk.ethereum.faucet.FaucetStatus.WalletAvailable
 import io.iohk.ethereum.faucet.jsonrpc.FaucetDomain.{SendFundsRequest, StatusRequest}
-import io.iohk.ethereum.faucet.{FaucetConfig, SupervisorConfig}
+import io.iohk.ethereum.faucet.{FaucetConfig, RpcClientConfig, SupervisorConfig}
 import io.iohk.ethereum.jsonrpc.JsonRpcError
 import io.iohk.ethereum.testing.ActorsTesting.simpleAutoPilot
 import io.iohk.ethereum.{NormalPatience, WithActorSystemShutDown}
@@ -133,11 +133,11 @@ class FaucetRpcServiceSpec
       txGasPrice = 10,
       txGasLimit = 20,
       txValue = 1,
-      rpcAddress = "",
+      rpcClient = RpcClientConfig(address = "", timeout = 10.seconds),
       keyStoreDir = "",
       minRequestInterval = 10.seconds,
       handlerTimeout = 10.seconds,
-      responseTimeout = 10.seconds,
+      actorCommunicationMargin = 10.seconds,
       supervisor = mock[SupervisorConfig],
       shutdownTimeout = 15.seconds
     )

src/test/scala/io/iohk/ethereum/faucet/jsonrpc/WalletServiceSpec.scala

@@ -5,7 +5,8 @@ import java.security.SecureRandom
 import akka.util.ByteString
 import io.iohk.ethereum.crypto._
 import io.iohk.ethereum.domain.{Address, Transaction}
-import io.iohk.ethereum.faucet.{FaucetConfig, SupervisorConfig}
+import io.iohk.ethereum.faucet.{FaucetConfig, RpcClientConfig, SupervisorConfig}
+import io.iohk.ethereum.jsonrpc.client.RpcClient.ConnectionError
 import io.iohk.ethereum.keystore.KeyStore.DecryptionFailed
 import io.iohk.ethereum.keystore.{KeyStore, Wallet}
 import io.iohk.ethereum.network.p2p.messages.CommonMessages.SignedTransactions.SignedTransactionEnc
@@ -21,7 +22,7 @@ import scala.concurrent.duration._
 
 class WalletServiceSpec extends AnyFlatSpec with Matchers with MockFactory {
 
-  "Wallet Service" should "send a transaction" in new TestSetup {
+  "Wallet Service" should "send a transaction successfully when getNonce and sendTransaction successfully" in new TestSetup {
 
     val receivingAddress = Address("0x99")
     val currentNonce = 2
@@ -44,6 +45,17 @@ class WalletServiceSpec extends AnyFlatSpec with Matchers with MockFactory {
 
   }
 
+  it should "failure the transaction when get timeout of getNonce" in new TestSetup {
+
+    val timeout = ConnectionError("timeout")
+    (walletRpcClient.getNonce _).expects(config.walletAddress).returning(Task.pure(Left(timeout)))
+
+    val res = walletService.sendFunds(wallet, Address("0x99")).runSyncUnsafe()
+
+    res shouldEqual Left(timeout)
+
+  }
+
   it should "get wallet successful" in new TestSetup {
     (mockKeyStore.unlockAccount _).expects(config.walletAddress, config.walletPassword).returning(Right(wallet))
 
@@ -76,11 +88,11 @@ class WalletServiceSpec extends AnyFlatSpec with Matchers with MockFactory {
         txGasPrice = 10,
         txGasLimit = 20,
         txValue = 1,
-        rpcAddress = "",
+        rpcClient = RpcClientConfig("", timeout = 10.seconds),
         keyStoreDir = "",
         minRequestInterval = 10.seconds,
         handlerTimeout = 10.seconds,
-        responseTimeout = 10.seconds,
+        actorCommunicationMargin = 10.seconds,
         supervisor = mock[SupervisorConfig],
         shutdownTimeout = 15.seconds
       )

src/universal/conf/faucet.conf

@@ -40,6 +40,9 @@ faucet {
     # null value indicates HTTPS is not being used
     #  password-file = "tls/password"
     #}
+
+    # Response time-out from rpc client resolve
+    timeout = 3.seconds
   }
 
   # How often can a single IP address send a request
@@ -49,12 +52,12 @@ faucet {
   handler-timeout = 1.seconds
 
   # Response time-out from actor resolve
-  response-timeout = 3.seconds
+  actor-communication-margin = 1.seconds
 
   # Supervisor with BackoffSupervisor pattern
   supervisor {
     min-backoff = 3.seconds
-    man-backoff = 30.seconds
+    max-backoff = 30.seconds
     random-factor = 0.2
     auto-reset = 10.seconds
     attempts = 4