problem: single invalid response from upstream breaks whole batch

solution: handle http and json errors and convert to RPC errors

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

@@ -15,9 +15,10 @@
  */
 package io.emeraldpay.dshackle.upstream.rpcclient
 
-import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.config.AuthConfig
 import io.emeraldpay.dshackle.reader.Reader
+import io.infinitape.etherjar.rpc.RpcException
+import io.infinitape.etherjar.rpc.RpcResponseError
 import io.netty.buffer.Unpooled
 import io.netty.handler.codec.http.HttpHeaderNames
 import io.netty.handler.codec.http.HttpHeaders
@@ -25,7 +26,6 @@ import io.netty.handler.ssl.SslContextBuilder
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
 import reactor.netty.http.client.HttpClient
-import reactor.netty.tcp.SslProvider
 import java.io.ByteArrayInputStream
 import java.security.KeyStore
 import java.security.cert.CertificateFactory
@@ -86,9 +86,13 @@ class JsonRpcHttpClient(
                 .uri(target)
                 .send(Mono.just(request).map { Unpooled.wrappedBuffer(it) })
 
-        return response.responseContent()
+        return response.response { header, bytes ->
-                .aggregate()
+            if (header.status().code() != 200) {
-                .asByteArray()
+                Mono.error(RpcException(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "HTTP Code: ${header.status().code()}"))
+            } else {
+                bytes.aggregate().asByteArray()
+            }
+        }.single()
     }
 
     override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
@@ -96,5 +100,13 @@ class JsonRpcHttpClient(
                 .map(JsonRpcRequest::toJson)
                 .flatMap(this@JsonRpcHttpClient::execute)
                 .map(parser::parse)
+                .onErrorResume { t ->
+                    val err = if (t is RpcException) {
+                        JsonRpcResponse.error(t.code, t.rpcMessage)
+                    } else {
+                        JsonRpcResponse.error(1, t.message ?: t.javaClass.name)
+                    }
+                    Mono.just(err)
+                }
     }
 }

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

@@ -16,6 +16,7 @@
 package io.emeraldpay.dshackle.upstream.rpcclient
 
 import com.fasterxml.jackson.core.JsonFactory
+import com.fasterxml.jackson.core.JsonParseException
 import com.fasterxml.jackson.core.JsonParser
 import com.fasterxml.jackson.core.JsonToken
 import io.infinitape.etherjar.rpc.RpcResponseError
@@ -30,48 +31,52 @@ class JsonRpcParser() {
     private val jsonFactory = JsonFactory()
 
     fun parse(json: ByteArray): JsonRpcResponse {
-        val parser: JsonParser = jsonFactory.createParser(json)
+        try {
-        parser.nextToken()
+            val parser: JsonParser = jsonFactory.createParser(json)
-        if (parser.currentToken != JsonToken.START_OBJECT) {
+            parser.nextToken()
-            return JsonRpcResponse(null, JsonRpcResponse.ResponseError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON"))
+            if (parser.currentToken != JsonToken.START_OBJECT) {
-        }
+                return JsonRpcResponse(null, JsonRpcResponse.ResponseError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON"))
-        var nullResponse: JsonRpcResponse? = null
+            }
-        while (parser.nextToken() != JsonToken.END_OBJECT) {
+            var nullResponse: JsonRpcResponse? = null
-            val field = parser.currentName
+            while (parser.nextToken() != JsonToken.END_OBJECT) {
-            if (field == "jsonrpc" || field == "id") {
+                val field = parser.currentName
-                if (!parser.nextToken().isScalarValue) {
+                if (field == "jsonrpc" || field == "id") {
-                    return JsonRpcResponse(null, JsonRpcResponse.ResponseError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON (id/type)"))
+                    if (!parser.nextToken().isScalarValue) {
-                }
+                        return JsonRpcResponse(null, JsonRpcResponse.ResponseError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON (id/type)"))
-                // just skip the field
+                    }
-            } else if (field == "result") {
+                    // just skip the field
-                val value = parser.nextToken()
+                } else if (field == "result") {
-                val start = parser.tokenLocation
+                    val value = parser.nextToken()
-                if (value.isScalarValue) {
+                    val start = parser.tokenLocation
-                    val text = parser.text
+                    if (value.isScalarValue) {
-                    if (value == JsonToken.VALUE_STRING) {
+                        val text = parser.text
-                        return JsonRpcResponse(("\"" + text + "\"").toByteArray(), null)
+                        if (value == JsonToken.VALUE_STRING) {
-                    } else if (value == JsonToken.VALUE_NULL) {
+                            return JsonRpcResponse(("\"" + text + "\"").toByteArray(), null)
-                        //if null we should check if error is present
+                        } else if (value == JsonToken.VALUE_NULL) {
-                        nullResponse = JsonRpcResponse(text.toByteArray(), null)
+                            //if null we should check if error is present
-                    } else {
+                            nullResponse = JsonRpcResponse(text.toByteArray(), null)
-                        return JsonRpcResponse(text.toByteArray(), null)
+                        } else {
+                            return JsonRpcResponse(text.toByteArray(), null)
+                        }
+                    } else if (value == JsonToken.START_OBJECT || value == JsonToken.START_ARRAY) {
+                        parser.skipChildren()
+                        val end = parser.currentLocation.byteOffset.toInt()
+                        val copy = ByteArray((end - start.byteOffset).toInt())
+                        System.arraycopy(json, start.byteOffset.toInt(), copy, 0, copy.size)
+                        return JsonRpcResponse(copy, null)
+                    }
+                } else if (field == "error") {
+                    val err = readError(parser)
+                    if (err != null) {
+                        return JsonRpcResponse(null, err)
                     }
-                } else if (value == JsonToken.START_OBJECT || value == JsonToken.START_ARRAY) {
-                    parser.skipChildren()
-                    val end = parser.currentLocation.byteOffset.toInt()
-                    val copy = ByteArray((end - start.byteOffset).toInt())
-                    System.arraycopy(json, start.byteOffset.toInt(), copy, 0, copy.size)
-                    return JsonRpcResponse(copy, null)
-                }
-            } else if (field == "error") {
-                val err = readError(parser)
-                if (err != null) {
-                    return JsonRpcResponse(null, err)
                 }
             }
-        }
+            if (nullResponse != null) {
-        if (nullResponse != null) {
+                return nullResponse
-            return nullResponse
+            }
+        } catch (e: JsonParseException) {
+            log.warn("Failed to parse JSON from upstream: ${e.message}")
         }
         return JsonRpcResponse(null, JsonRpcResponse.ResponseError(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Invalid JSON structure"))
     }

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

@@ -17,6 +17,8 @@ package io.emeraldpay.dshackle.upstream.rpcclient
 
 import io.emeraldpay.dshackle.config.AuthConfig
 import io.emeraldpay.dshackle.test.TestingCommons
+import io.infinitape.etherjar.rpc.RpcException
+import io.infinitape.etherjar.rpc.RpcResponseError
 import org.mockserver.integration.ClientAndServer
 import org.mockserver.model.HttpRequest
 import org.mockserver.model.HttpResponse
@@ -88,4 +90,25 @@ class JsonRpcHttpClientSpec extends Specification {
         )
     }
 
+    def "Produces RPC Exception on error status code"() {
+        setup:
+        def client = new JsonRpcHttpClient("localhost:18332", null, null)
+
+        mockServer.when(
+                HttpRequest.request()
+        ).respond(
+                HttpResponse.response()
+                        .withStatusCode(500)
+                        .withBody("pong")
+        )
+        when:
+        def act = client.execute("ping".bytes).map { new String(it) }
+        then:
+        StepVerifier.create(act)
+                .expectErrorMatches { t ->
+                    t instanceof RpcException && t.code == RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE
+                }
+                .verify(Duration.ofSeconds(1))
+    }
+
 }

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

@@ -15,6 +15,7 @@
  */
 package io.emeraldpay.dshackle.upstream.rpcclient
 
+import io.infinitape.etherjar.rpc.RpcResponseError
 import spock.lang.Specification
 
 class JsonRpcParserSpec extends Specification {
@@ -177,4 +178,16 @@ class JsonRpcParserSpec extends Specification {
         !act.hasResult()
     }
 
+    def "Handle non-json with producing an error response"() {
+        setup:
+        def json = 'NOT JSON'
+        when:
+        def act = parser.parse(json.getBytes())
+        then:
+        act.error != null
+        act.error.code == RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE
+        act.hasError()
+        !act.hasResult()
+    }
+
 }