problem: no response for requests that produce very large results which causes the upstream to break the WS connection

src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt

@@ -212,20 +212,22 @@ open class ConfiguredUpstreams(
         }
 
         log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}")
+
+        val directApi: Reader<JsonRpcRequest, JsonRpcResponse>? = buildHttpClient(config)
+        if (directApi == null) {
+            log.warn("Upstream doesn't have API configuration")
+            return
+        }
+
         val ethereumUpstream = if (wsFactoryApi != null && !conn.preferHttp) {
             EthereumWsUpstream(
                 config.id!!,
-                chain, wsFactoryApi,
+                chain, directApi, wsFactoryApi,
                 options, config.role,
                 QuorumForLabels.QuorumItem(1, config.labels),
                 methods
             )
         } else {
-            val directApi: Reader<JsonRpcRequest, JsonRpcResponse>? = buildHttpClient(config)
-            if (directApi == null) {
-                log.warn("Upstream doesn't have API configuration")
-                return
-            }
             EthereumRpcUpstream(
                 config.id!!,
                 chain, directApi, wsFactoryApi,

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt

@@ -24,6 +24,7 @@ import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcSwitchClient
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcWsClient
 import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics
 import io.emeraldpay.grpc.Chain
@@ -38,6 +39,7 @@ import reactor.core.Disposable
 class EthereumWsUpstream(
     id: String,
     val chain: Chain,
+    httpConnection: Reader<JsonRpcRequest, JsonRpcResponse>,
     ethereumWsFactory: EthereumWsFactory,
     options: UpstreamsConfig.Options,
     role: UpstreamsConfig.UpstreamRole,
@@ -51,7 +53,7 @@ class EthereumWsUpstream(
 
     private val head: EthereumWsHead
     private val connection: WsConnection
-    private val api: JsonRpcWsClient
+    private val api: Reader<JsonRpcRequest, JsonRpcResponse>
 
     private var validatorSubscription: Disposable? = null
     private val validator: EthereumUpstreamValidator
@@ -78,7 +80,12 @@ class EthereumWsUpstream(
 
         connection = ethereumWsFactory.create(this, validator, metrics)
         head = EthereumWsHead(connection)
-        api = JsonRpcWsClient(connection)
+        // Sometimes the server may close the WebSocket connection during the execution of a call, for example if the response
+        // is too large for WebSockets Frame (and Geth is unable to split messages into separate frames)
+        // In this case the failed request must be rerouted to the HTTP connection, because otherwise it would always fail
+        api = JsonRpcSwitchClient(
+            JsonRpcWsClient(connection), httpConnection
+        )
     }
 
     override fun getHead(): Head {

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt

@@ -23,6 +23,7 @@ import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.upstream.DefaultUpstream
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.dshackle.upstream.rpcclient.ResponseWSParser
@@ -35,7 +36,6 @@ import io.netty.buffer.ByteBufInputStream
 import io.netty.buffer.Unpooled
 import io.netty.handler.codec.http.HttpHeaderNames
 import io.netty.resolver.DefaultAddressResolverGroup
-import io.netty.resolver.DefaultNameResolver
 import org.reactivestreams.Publisher
 import org.slf4j.LoggerFactory
 import org.springframework.util.backoff.BackOff
@@ -55,6 +55,7 @@ import reactor.retry.Repeat
 import reactor.util.function.Tuples
 import java.net.URI
 import java.time.Duration
+import java.time.Instant
 import java.util.Base64
 import java.util.concurrent.Executors
 import java.util.concurrent.TimeUnit
@@ -62,7 +63,7 @@ import java.util.concurrent.TimeoutException
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicInteger
 
-class WsConnection(
+open class WsConnection(
     private val uri: URI,
     private val origin: URI,
     private val basicAuth: AuthConfig.ClientBasicAuth?,
@@ -113,12 +114,19 @@ class WsConnection(
         .many()
         .multicast()
         .directBestEffort<JsonRpcResponse>()
+    private val disconnects = Sinks
+        .many()
+        .multicast()
+        .directBestEffort<Instant>()
     private val sendIdSeq = AtomicInteger(IDS_START)
     private val sendExecutor = Executors.newSingleThreadExecutor()
     private var keepConnection = true
     private var connection: Disposable? = null
     private val reconnecting = AtomicBoolean(false)
 
+    open val isConnected: Boolean
+        get() = connection != null && !reconnecting.get()
+
     fun setReconnectIntervalSeconds(value: Long) {
         reconnectBackoff = FixedBackOff(value * 1000, FixedBackOff.UNLIMITED_ATTEMPTS)
         currentBackOff = reconnectBackoff.start()
@@ -165,6 +173,7 @@ class WsConnection(
         connection = HttpClient.create()
             .resolver(DefaultAddressResolverGroup.INSTANCE)
             .doOnDisconnected {
+                disconnects.tryEmitNext(Instant.now())
                 log.info("Disconnected from $uri")
                 // mark upstream as UNAVAIL
                 upstream?.setStatus(UpstreamAvailability.UNAVAILABLE)
@@ -374,25 +383,39 @@ class WsConnection(
 
     fun waitForResponse(request: JsonRpcRequest, originalId: Int, startTime: Long): Mono<JsonRpcResponse> {
         val expectedId = request.id.toLong()
-        val failResponse = JsonRpcResponse(
-            null,
+        val noResponse = JsonRpcException(
+            JsonRpcResponse.Id.from(originalId),
             JsonRpcError(
                 RpcResponseError.CODE_INTERNAL_ERROR,
                 "Response not received from WebSocket"
-            ),
-            JsonRpcResponse.Id.from(originalId), null
+            )
         )
 
-        return Flux.from(rpcReceive.asFlux())
+        val response = Flux.from(rpcReceive.asFlux())
             .doOnSubscribe { sendRpc(request) }
             .filter { resp -> resp.id.asNumber() == expectedId }
             .take(Defaults.timeout)
             .take(1)
             .singleOrEmpty()
+
+        val failOnDisconnect = Mono.from(disconnects.asFlux())
+            .flatMap {
+                Mono.error<JsonRpcResponse>(
+                    JsonRpcException(
+                        JsonRpcResponse.Id.from(originalId),
+                        JsonRpcError(
+                            RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR,
+                            "Disconnected from WebSocket"
+                        )
+                    )
+                )
+            }
+
+        return response.or(failOnDisconnect)
             .doOnNext { rpcMetrics?.timer?.record(System.nanoTime() - startTime, TimeUnit.NANOSECONDS) }
             .doOnError { rpcMetrics?.fails?.increment() }
             .map { it.copyWithId(JsonRpcResponse.Id.from(originalId)) }
-            .defaultIfEmpty(failResponse)
+            .switchIfEmpty(Mono.error(noResponse))
     }
 
     fun getBlocksFlux(): Flux<BlockContainer> {

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

@@ -112,6 +112,7 @@ class JsonRpcHttpClient(
         return Mono.just(key)
             .map(JsonRpcRequest::toJson)
             .doOnNext {
+                println("rpc connection ${key.method}")
                 startTime = System.nanoTime()
             }
             .flatMap(this@JsonRpcHttpClient::execute)

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

@@ -0,0 +1,27 @@
+package io.emeraldpay.dshackle.upstream.rpcclient
+
+import io.emeraldpay.dshackle.reader.Reader
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Mono
+
+/**
+ * An aggregating JSON RPC Client that wraps two actual readers, a Primary and a Secondary.
+ * It always calls the Primary reader, and if it fails or produces an empty result, then it calls the Secondary reader.
+ */
+class JsonRpcSwitchClient(
+    private val primary: Reader<JsonRpcRequest, JsonRpcResponse>,
+    private val secondary: Reader<JsonRpcRequest, JsonRpcResponse>,
+) : Reader<JsonRpcRequest, JsonRpcResponse> {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(JsonRpcSwitchClient::class.java)
+    }
+
+    override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
+        return primary.read(key)
+            .switchIfEmpty(Mono.error(IllegalStateException("No response from Primary Connection")))
+            .onErrorResume {
+                secondary.read(key)
+            }
+    }
+}

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

@@ -17,6 +17,7 @@ package io.emeraldpay.dshackle.upstream.rpcclient
 
 import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.upstream.ethereum.WsConnection
+import io.emeraldpay.etherjar.rpc.RpcResponseError
 import reactor.core.publisher.Mono
 
 class JsonRpcWsClient(
@@ -24,6 +25,17 @@ class JsonRpcWsClient(
 ) : Reader<JsonRpcRequest, JsonRpcResponse> {
 
     override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
+        if (!ws.isConnected) {
+            return Mono.error(
+                JsonRpcException(
+                    JsonRpcResponse.NumberId(key.id),
+                    JsonRpcError(
+                        RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR,
+                        "WebSocket is not connected"
+                    )
+                )
+            )
+        }
         return ws.call(key)
     }
 }