Merge pull request #86 from p2p-org/rpc-clients-fixes

Rpc clients fixes

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt

@@ -25,6 +25,8 @@ import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
 import io.emeraldpay.etherjar.rpc.RpcException
 import io.emeraldpay.etherjar.rpc.RpcResponseError
+import io.grpc.StatusRuntimeException
+import org.apache.commons.lang3.time.StopWatch
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
 import java.util.concurrent.TimeUnit
@@ -32,7 +34,7 @@ import java.util.concurrent.TimeUnit
 class JsonRpcGrpcClient(
     private val stub: ReactorBlockchainGrpc.ReactorBlockchainStub,
     private val chain: Chain,
-    private val metrics: RpcMetrics
+    private val metrics: RpcMetrics?,
 ) {
 
     companion object {
@@ -46,11 +48,11 @@ class JsonRpcGrpcClient(
     class Executor(
         private val stub: ReactorBlockchainGrpc.ReactorBlockchainStub,
         private val chain: Chain,
-        private val metrics: RpcMetrics
+        private val metrics: RpcMetrics?
     ) : Reader<JsonRpcRequest, JsonRpcResponse> {
 
         override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
-            var startTime: Long = 0
+            val timer = StopWatch()
             val req = BlockchainOuterClass.NativeCallRequest.newBuilder()
                 .setChainValue(chain.id)
 
@@ -66,39 +68,58 @@ class JsonRpcGrpcClient(
             req.addItems(reqItem.build())
 
             return Mono.just(key)
-                .doOnNext {
+                .doOnNext { timer.start() }
-                    startTime = System.nanoTime()
+                .flatMap {
-                }.flatMap {
                     stub.nativeCall(req.build())
                         .single()
-                        .flatMap { resp ->
+                        .onErrorResume(::handleError)
-                            if (resp.succeed) {
+                        .flatMap(::handleResponse)
-                                val bytes = resp.payload.toByteArray()
-                                val signature = if (resp.hasSignature()) {
-                                    extractSignature(resp.signature)
-                                } else {
-                                    null
-                                }
-                                Mono.just(JsonRpcResponse(bytes, null, JsonRpcResponse.NumberId(0), signature))
-                            } else {
-                                metrics.fails.increment()
-                                Mono.error(
-                                    RpcException(
-                                        RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR,
-                                        resp.errorMessage
-                                    )
-                                )
-                            }
-                        }
                 }
                 .doOnNext {
-                    if (startTime > 0) {
+                    if (timer.isStarted) {
-                        val now = System.nanoTime()
+                        metrics?.timer?.record(timer.getTime(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS)
-                        metrics.timer.record(now - startTime, TimeUnit.NANOSECONDS)
                     }
                 }
         }
 
+        fun handleResponse(resp: BlockchainOuterClass.NativeCallReplyItem): Mono<JsonRpcResponse> =
+            if (resp.succeed) {
+                val bytes = resp.payload.toByteArray()
+                val signature = if (resp.hasSignature()) {
+                    extractSignature(resp.signature)
+                } else {
+                    null
+                }
+                Mono.just(JsonRpcResponse(bytes, null, JsonRpcResponse.NumberId(0), signature))
+            } else {
+                metrics?.fails?.increment()
+                Mono.error(
+                    RpcException(
+                        RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR,
+                        resp.errorMessage
+                    )
+                )
+            }
+
+        fun handleError(t: Throwable): Mono<BlockchainOuterClass.NativeCallReplyItem> {
+            metrics?.fails?.increment()
+            return when (t) {
+                is StatusRuntimeException -> Mono.error(
+                    RpcException(
+                        RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR,
+                        "Remote status code: ${t.status.code.name}"
+                    )
+                )
+
+                else -> Mono.error(
+                    RpcException(
+                        RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR,
+                        "Other connection error"
+                    )
+                )
+            }
+        }
+
         fun extractSignature(resp: NativeCallReplySignature?): ResponseSigner.Signature? {
             if (resp == null || resp.signature == null || resp.signature.isEmpty || resp.upstreamId == null || resp.upstreamId.isEmpty()) {
                 return null

src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt

@@ -24,10 +24,13 @@ import io.netty.handler.codec.http.HttpHeaderNames
 import io.netty.handler.codec.http.HttpHeaders
 import io.netty.handler.ssl.SslContextBuilder
 import io.netty.resolver.DefaultAddressResolverGroup
+import org.apache.commons.lang3.time.StopWatch
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
 import reactor.netty.http.client.HttpClient
 import reactor.netty.resources.ConnectionProvider
+import reactor.util.function.Tuple2
+import reactor.util.function.Tuples
 import java.io.ByteArrayInputStream
 import java.security.KeyStore
 import java.security.cert.CertificateFactory
@@ -35,6 +38,7 @@ import java.security.cert.X509Certificate
 import java.util.Base64
 import java.util.concurrent.TimeUnit
 import java.util.function.Consumer
+import java.util.function.Function
 
 /**
  * JSON RPC client
@@ -89,52 +93,95 @@ class JsonRpcHttpClient(
         this.httpClient = build
     }
 
-    fun execute(request: ByteArray): Mono<ByteArray> {
+    fun execute(request: ByteArray): Mono<Tuple2<Int, ByteArray>> {
         val response = httpClient
             .post()
             .uri(target)
             .send(Mono.just(request).map { Unpooled.wrappedBuffer(it) })
 
         return response.response { header, bytes ->
-            if (header.status().code() != 200) {
+            val statusCode = header.status().code()
-                Mono.error(
+            bytes.aggregate().asByteArray().map {
-                    JsonRpcException(
+                Tuples.of(statusCode, it)
-                        JsonRpcResponse.NumberId(-2),
-                        JsonRpcError(
-                            RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE,
-                            "HTTP Code: ${header.status().code()}"
-                        )
-                    )
-                )
-            } else {
-                bytes.aggregate().asByteArray()
             }
         }.single()
     }
 
     override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
-        var startTime: Long = 0
+        val startTime = StopWatch()
         return Mono.just(key)
             .map(JsonRpcRequest::toJson)
-            .doOnNext {
+            .doOnNext { startTime.start() }
-                startTime = System.nanoTime()
-            }
             .flatMap(this@JsonRpcHttpClient::execute)
             .doOnNext {
-                if (startTime > 0) {
+                if (startTime.isStarted) {
-                    val now = System.nanoTime()
+                    metrics.timer.record(startTime.nanoTime, TimeUnit.NANOSECONDS)
-                    metrics.timer.record(now - startTime, TimeUnit.NANOSECONDS)
                 }
             }
-            .map(parser::parse)
+            .transform(asJsonRpcResponse(key))
-            .onErrorResume { t ->
+            .transform(convertErrors(key))
+            .transform(throwIfError())
+    }
+
+    /**
+     * The subscribers expect to catch an exception if the response contains JSON RPC Error. Convert it here to JsonRpcException
+     */
+    private fun throwIfError(): Function<Mono<JsonRpcResponse>, Mono<JsonRpcResponse>> {
+        return Function { resp ->
+            resp.flatMap {
+                if (it.hasError()) {
+                    Mono.error(JsonRpcException(it.id, it.error!!))
+                } else {
+                    Mono.just(it)
+                }
+            }
+        }
+    }
+
+    /**
+     * Convert internal exceptions to standard JsonRpcException
+     */
+    private fun convertErrors(key: JsonRpcRequest): Function<Mono<JsonRpcResponse>, Mono<JsonRpcResponse>> {
+        return Function { resp ->
+            resp.onErrorResume { t ->
                 val err = when (t) {
-                    is RpcException -> JsonRpcResponse.error(t.code, t.rpcMessage)
+                    is RpcException -> JsonRpcException.from(t)
-                    is JsonRpcException -> JsonRpcResponse.error(t.error, JsonRpcResponse.NumberId(1))
+                    is JsonRpcException -> t
-                    else -> JsonRpcResponse.error(1, t.message ?: t.javaClass.name)
+                    else -> JsonRpcException(key.id, t.message ?: t.javaClass.name)
                 }
+                // here we're measure the internal errors, not upstream errors
                 metrics.fails.increment()
-                Mono.just(err)
+                Mono.error(err)
             }
+        }
+    }
+
+    /**
+     * Process response from the upstream and convert it to JsonRpcResponse.
+     * The input is a pair of (Http Status Code, Http Response Body)
+     */
+    private fun asJsonRpcResponse(key: JsonRpcRequest): Function<Mono<Tuple2<Int, ByteArray>>, Mono<JsonRpcResponse>> {
+        return Function { resp ->
+            resp.map {
+                val parsed = parser.parse(it.t2)
+                val statusCode = it.t1
+                if (statusCode != 200) {
+                    if (parsed.hasError() && parsed.error!!.code != RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE) {
+                        // extracted the error details from the HTTP Body
+                        parsed
+                    } else {
+                        // here we got a valid response with ERROR as HTTP Status Code. We assume that HTTP Status has
+                        // a higher priority so return an error here anyway
+                        JsonRpcResponse.error(
+                            RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE,
+                            "HTTP Code: $statusCode",
+                            JsonRpcResponse.NumberId(key.id)
+                        )
+                    }
+                } else {
+                    parsed
+                }
+            }
+        }
     }
 }

src/test/groovy/io/emeraldpay/dshackle/test/MockGrpcServer.groovy

@@ -26,14 +26,6 @@ class MockGrpcServer {
 
     GrpcCleanupRule grpcCleanup = new GrpcCleanupRule()
 
-    ReactorBlockchainGrpc.ReactorBlockchainStub clientForServer(ReactorBlockchainGrpc.BlockchainImplBase impl) {
-        String serverName = InProcessServerBuilder.generateName()
-        grpcCleanup.register(InProcessServerBuilder
-                .forName(serverName).directExecutor().addService(impl).build().start());
-        def channel = grpcCleanup.register(InProcessChannelBuilder.forName(serverName).directExecutor().build())
-        return ReactorBlockchainGrpc.newReactorStub(channel)
-    }
-
     ReactorBlockchainGrpc.ReactorBlockchainStub clientForServer(BlockchainGrpc.BlockchainImplBase impl){
         String serverName = InProcessServerBuilder.generateName()
         grpcCleanup.register(InProcessServerBuilder

src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClientSpec.groovy

@@ -0,0 +1,88 @@
+package io.emeraldpay.dshackle.upstream.rpcclient
+
+import com.google.protobuf.ByteString
+import io.emeraldpay.api.proto.BlockchainGrpc
+import io.emeraldpay.api.proto.BlockchainOuterClass
+import io.emeraldpay.api.proto.Common
+import io.emeraldpay.dshackle.test.MockGrpcServer
+import io.emeraldpay.dshackle.upstream.Selector
+import io.emeraldpay.etherjar.rpc.RpcException
+import io.emeraldpay.etherjar.rpc.RpcResponseError
+import io.emeraldpay.dshackle.Chain
+import io.grpc.stub.StreamObserver
+import spock.lang.Specification
+
+import java.time.Duration
+import java.util.concurrent.atomic.AtomicReference
+
+class JsonRpcGrpcClientSpec extends Specification {
+
+    def "Makes a request"() {
+        setup:
+        def mockGrpc = new MockGrpcServer()
+        def requested = new AtomicReference<BlockchainOuterClass.NativeCallRequest>()
+
+        def grpc = mockGrpc.clientForServer(new BlockchainGrpc.BlockchainImplBase() {
+            @Override
+            void nativeCall(BlockchainOuterClass.NativeCallRequest request, StreamObserver<BlockchainOuterClass.NativeCallReplyItem> responseObserver) {
+                requested.set(request)
+                responseObserver.onNext(BlockchainOuterClass.NativeCallReplyItem.newBuilder()
+                        .setId(1)
+                        .setSucceed(true)
+                        .setPayload(ByteString.copyFromUtf8("\"hello world!\""))
+                        .build())
+                responseObserver.onCompleted()
+            }
+        })
+        def client = new JsonRpcGrpcClient(
+                grpc, Chain.BITCOIN, null
+        ).getReader()
+
+        when:
+        def act = client.read(
+                new JsonRpcRequest("test", [])
+        ).block(Duration.ofSeconds(1))
+
+        then:
+        !act.hasError()
+        act.resultAsProcessedString == "hello world!"
+
+        requested.get() == BlockchainOuterClass.NativeCallRequest.newBuilder()
+                .setChain(Common.ChainRef.CHAIN_BITCOIN)
+                .addAllItems([
+                        BlockchainOuterClass.NativeCallItem.newBuilder()
+                                .setId(1)
+                                .setMethod("test")
+                                .setPayload(ByteString.copyFromUtf8("[]"))
+                                .build()
+                ])
+                .build()
+    }
+
+    def "Return error on HTTP error"() {
+        setup:
+        def mockGrpc = new MockGrpcServer()
+        def grpc = mockGrpc.clientForServer(new BlockchainGrpc.BlockchainImplBase() {
+            @Override
+            void nativeCall(BlockchainOuterClass.NativeCallRequest request, StreamObserver<BlockchainOuterClass.NativeCallReplyItem> responseObserver) {
+                responseObserver.onError(new IllegalStateException("fail"))
+            }
+        })
+        def client = new JsonRpcGrpcClient(
+                grpc, Chain.BITCOIN, null
+        ).getReader()
+
+        when:
+        client.read(
+                new JsonRpcRequest("test", [])
+        ).block(Duration.ofSeconds(1))
+
+        then:
+        def t = thrown(RpcException)
+        with(t.error) {
+            message == "Remote status code: UNKNOWN"
+            it.code == RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR
+        }
+
+    }
+}

src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClientSpec.groovy

@@ -26,6 +26,7 @@ import org.mockserver.model.HttpRequest
 import org.mockserver.model.HttpResponse
 import org.mockserver.model.MediaType
 import org.springframework.util.SocketUtils
+import reactor.core.Exceptions
 import reactor.test.StepVerifier
 import spock.lang.Specification
 
@@ -84,7 +85,7 @@ class JsonRpcHttpClientSpec extends Specification {
                         .withBody("pong")
         )
         when:
-        def act = client.execute("ping".bytes).map { new String(it) }
+        def act = client.execute("ping".bytes).map { new String(it.t2) }
         then:
         StepVerifier.create(act)
                 .expectNext("pong")
@@ -111,13 +112,44 @@ class JsonRpcHttpClientSpec extends Specification {
                         .withBody("pong")
         )
         when:
-        def act = client.execute("ping".bytes).map { new String(it) }
+        def act = client.read(
+                new JsonRpcRequest("ping", [])
+        ).block(Duration.ofSeconds(1))
         then:
-        StepVerifier.create(act)
+        def t = thrown(RuntimeException) // reactor.core.Exceptions$ReactiveException
-                .expectErrorMatches { t ->
+        t.cause instanceof JsonRpcException
-                    t instanceof JsonRpcException && t.error.code == RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE
+        with(((JsonRpcException)t.cause).error) {
-                }
+            code == RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE
-                .verify(Duration.ofSeconds(1))
+            message == "HTTP Code: 500"
+        }
+    }
+
+    def "Tries to extract message if HTTP error if it still contains a JSON RPC message"() {
+        setup:
+        def client = new JsonRpcHttpClient("localhost:${port}", metrics, null, null)
+
+        mockServer.when(
+                HttpRequest.request()
+        ).respond(
+                HttpResponse.response()
+                        .withStatusCode(500)
+                        .withBody('{' +
+                                '"jsonrpc": "2.0", ' +
+                                '"id": 1, ' +
+                                '"error": {"code": -32603, "message": "Something happened"}' +
+                                '}')
+        )
+        when:
+        def act = client.read(
+                new JsonRpcRequest("ping", [])
+        ).block(Duration.ofSeconds(1))
+        then:
+        def t = thrown(RuntimeException) // reactor.core.Exceptions$ReactiveException
+        t.cause instanceof JsonRpcException
+        with(((JsonRpcException)t.cause).error) {
+            code == -32603
+            message == "Something happened"
+        }
     }
 
 }