problem: unhandled exception when a remote grpc gives an http error

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/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
+        }
+
+    }
+}