From ff9bcefe25900483bc584246b2743693dbb35398 Mon Sep 17 00:00:00 2001
From: Igor Artamonov <igor@artamonov.ru>
Date: Wed, 16 Oct 2019 20:58:57 -0400
Subject: [PATCH] solution: better handle for minor upstream errors

---
 .../kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt |  5 +++++
 .../upstream/ethereum/DirectEthereumApi.kt       | 16 +++++++++++-----
 2 files changed, 16 insertions(+), 5 deletions(-)

diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt
index ce24a5f7..7e788fcd 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackTx.kt
@@ -25,6 +25,7 @@ import io.emeraldpay.grpc.Chain
 import io.infinitape.etherjar.domain.BlockHash
 import io.infinitape.etherjar.domain.TransactionId
 import io.infinitape.etherjar.rpc.Commands
+import io.infinitape.etherjar.rpc.RpcException
 import io.infinitape.etherjar.rpc.json.BlockJson
 import io.infinitape.etherjar.rpc.json.TransactionJson
 import io.infinitape.etherjar.rpc.json.TransactionRefJson
@@ -252,6 +253,10 @@ class TrackTx(
         val execution = upstream.getApi(Selector.empty)
                 .flatMap { api -> api.executeAndConvert(Commands.eth().getTransaction(tx.txid)) }
         return execution
+                .onErrorResume(RpcException::class.java) { t ->
+                    log.warn("Upstream error, ignoring. {}", t.rpcMessage)
+                    Mono.empty<TransactionJson>()
+                }
                 .flatMap { updateFromBlock(upstream, tx, it) }
                 .doOnError { t ->
                     log.error("Failed to load tx block", t)
diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt
index 08485de1..52faa582 100644
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/DirectEthereumApi.kt
@@ -18,10 +18,9 @@ package io.emeraldpay.dshackle.upstream.ethereum
 import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.Defaults
 import io.emeraldpay.dshackle.upstream.CallMethods
-import io.infinitape.etherjar.rpc.ReactorBatch
-import io.infinitape.etherjar.rpc.ReactorRpcClient
-import io.infinitape.etherjar.rpc.RpcCall
-import io.infinitape.etherjar.rpc.RpcException
+import io.grpc.Status
+import io.grpc.StatusRuntimeException
+import io.infinitape.etherjar.rpc.*
 import io.infinitape.etherjar.rpc.json.ResponseJson
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
@@ -52,11 +51,18 @@ open class DirectEthereumApi(
                     resp.result = it
                     objectMapper.writer().writeValueAsBytes(resp)
                 }
+                .onErrorResume(StatusRuntimeException::class.java) { t ->
+                    if (t.status.code == Status.Code.CANCELLED) {
+                        Mono.empty<ByteArray>()
+                    } else {
+                        Mono.error(RpcException(RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR, "gRPC error ${t.status}"))
+                    }
+                }
                 .onErrorMap { t ->
                     if (RpcException::class.java.isAssignableFrom(t.javaClass)) {
                         t
                     } else {
-                        log.warn("Convert to RPC error. Exception: ${t.message}")
+                        log.warn("Convert to RPC error. Exception ${t.javaClass}:${t.message}", t)
                         RpcException(-32020, "Error reading from upstream", null, t)
                     }
                 }