Merge branch 'master' of https://github.com/emeraldpay/dshackle

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

@@ -200,6 +200,7 @@ open class ConfiguredUpstreams(
 
         val wsFactoryApi: EthereumWsFactory? = conn.ws?.let { endpoint ->
             val wsApi = EthereumWsFactory(
+                config.id!!, chain,
                 endpoint.url,
                 endpoint.origin ?: URI("http://localhost"),
             )
@@ -212,20 +213,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/bitcoin/EsploraClient.kt

@@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.upstream.bitcoin.data.EsploraUnspent
 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.bitcoinj.core.Address
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
@@ -47,6 +48,7 @@ class EsploraClient(
 
     init {
         var build = HttpClient.create()
+            .resolver(DefaultAddressResolverGroup.INSTANCE)
 
         build = build.headers { h ->
             h.add(HttpHeaderNames.CONTENT_TYPE, "application/json")

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

@@ -79,7 +79,7 @@ open class EthereumRpcUpstream(
     open fun createHead(): Head {
         return if (ethereumWsFactory != null) {
             // do not set upstream to the WS, since it doesn't control the RPC upstream
-            val ws = ethereumWsFactory.create(null, null, null).apply {
+            val ws = ethereumWsFactory.create(null, null).apply {
                 connect()
             }
             val wsHead = EthereumWsHead(ws).apply {

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

@@ -20,18 +20,49 @@ import io.emeraldpay.dshackle.config.AuthConfig
 import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.upstream.DefaultUpstream
 import io.emeraldpay.dshackle.upstream.rpcclient.RpcMetrics
+import io.emeraldpay.grpc.Chain
+import io.micrometer.core.instrument.Counter
+import io.micrometer.core.instrument.Metrics
+import io.micrometer.core.instrument.Tag
+import io.micrometer.core.instrument.Timer
 import java.net.URI
 
 class EthereumWsFactory(
+    private val id: String,
+    private val chain: Chain,
     private val uri: URI,
-    private val origin: URI
+    private val origin: URI,
 ) {
 
     var basicAuth: AuthConfig.ClientBasicAuth? = null
     var config: UpstreamsConfig.WsEndpoint? = null
 
-    fun create(upstream: DefaultUpstream?, validator: EthereumUpstreamValidator?, rpcMetrics: RpcMetrics?): WsConnection {
+    // metrics are shared between all connections to the same WS
-        return WsConnection(uri, origin, basicAuth, rpcMetrics, upstream, validator).also { ws ->
+    private val metrics: RpcMetrics = run {
+        val metricsTags = listOf(
+            Tag.of("upstream", id),
+            // UNSPECIFIED shouldn't happen too
+            Tag.of("chain", chain.chainCode)
+        )
+
+        RpcMetrics(
+            Timer.builder("upstream.ws.conn")
+                .description("Request time through a WebSocket JSON RPC connection")
+                .tags(metricsTags)
+                .publishPercentileHistogram()
+                .register(Metrics.globalRegistry),
+            Counter.builder("upstream.ws.fail")
+                .description("Number of failures of WebSocket JSON RPC requests")
+                .tags(metricsTags)
+                .register(Metrics.globalRegistry)
+        )
+    }
+
+    fun create(upstream: DefaultUpstream?, validator: EthereumUpstreamValidator?): WsConnection {
+        require(upstream == null || upstream.getId() == id) {
+            "Creating instance for different upstream. ${upstream?.getId()} != id"
+        }
+        return WsConnection(uri, origin, basicAuth, metrics, upstream, validator).also { ws ->
             config?.frameSize?.let {
                 ws.frameSize = it
             }

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

@@ -24,13 +24,9 @@ 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
-import io.micrometer.core.instrument.Counter
-import io.micrometer.core.instrument.Metrics
-import io.micrometer.core.instrument.Tag
-import io.micrometer.core.instrument.Timer
 import org.slf4j.LoggerFactory
 import org.springframework.context.Lifecycle
 import reactor.core.Disposable
@@ -38,6 +34,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,34 +48,21 @@ 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
 
     init {
-        val metricsTags = listOf(
-            Tag.of("upstream", id),
-            // UNSPECIFIED shouldn't happen too
-            Tag.of("chain", chain.chainCode)
-        )
-        val metrics = RpcMetrics(
-            Timer.builder("upstream.ws.conn")
-                .description("Request time through a WebSocket JSON RPC connection")
-                .tags(metricsTags)
-                .publishPercentileHistogram()
-                .register(Metrics.globalRegistry),
-            Counter.builder("upstream.ws.fail")
-                .description("Number of failures of WebSocket JSON RPC requests")
-                .tags(metricsTags)
-                .register(Metrics.globalRegistry)
-        )
-
         validator = EthereumUpstreamValidator(this, getOptions())
-
+        connection = ethereumWsFactory.create(this, validator)
-        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
@@ -34,6 +35,7 @@ import io.netty.buffer.ByteBuf
 import io.netty.buffer.ByteBufInputStream
 import io.netty.buffer.Unpooled
 import io.netty.handler.codec.http.HttpHeaderNames
+import io.netty.resolver.DefaultAddressResolverGroup
 import org.reactivestreams.Publisher
 import org.slf4j.LoggerFactory
 import org.springframework.util.backoff.BackOff
@@ -53,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
@@ -60,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?,
@@ -111,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()
@@ -161,7 +171,9 @@ class WsConnection(
         log.info("Connecting to WebSocket: $uri")
         connection?.dispose()
         connection = HttpClient.create()
+            .resolver(DefaultAddressResolverGroup.INSTANCE)
             .doOnDisconnected {
+                disconnects.tryEmitNext(Instant.now())
                 log.info("Disconnected from $uri")
                 // mark upstream as UNAVAIL
                 upstream?.setStatus(UpstreamAvailability.UNAVAILABLE)
@@ -371,25 +383,39 @@ class WsConnection(
 
     fun waitForResponse(request: JsonRpcRequest, originalId: Int, startTime: Long): Mono<JsonRpcResponse> {
         val expectedId = request.id.toLong()
-        val failResponse = JsonRpcResponse(
+        val noResponse = JsonRpcException(
-            null,
+            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

@@ -23,6 +23,7 @@ import io.netty.buffer.Unpooled
 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.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
 import reactor.netty.http.client.HttpClient
@@ -53,6 +54,7 @@ class JsonRpcHttpClient(
 
     init {
         var build = HttpClient.create()
+            .resolver(DefaultAddressResolverGroup.INSTANCE)
 
         build = build.headers { h ->
             h.add(HttpHeaderNames.CONTENT_TYPE, "application/json")

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)
     }
 }

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionRealSpec.groovy

@@ -4,6 +4,7 @@ import io.emeraldpay.dshackle.test.MockWSServer
 import io.emeraldpay.dshackle.upstream.DefaultUpstream
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
+import io.emeraldpay.grpc.Chain
 import reactor.test.StepVerifier
 import spock.lang.Shared
 import spock.lang.Specification
@@ -30,7 +31,7 @@ class WsConnectionRealSpec extends Specification {
         server = new MockWSServer(port)
         server.start()
         Thread.sleep(SLEEP)
-        conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null, null, null)
+        conn = new EthereumWsFactory("test", Chain.ETHEREUM, "ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null, null)
     }
 
     def cleanup() {
@@ -93,10 +94,26 @@ class WsConnectionRealSpec extends Specification {
         act[0].value.contains("\"params\":[\"newHeads\"]")
     }
 
+    def "Error on request when server disconnects"() {
+        when:
+        conn.connect()
+        conn.reconnectIntervalSeconds = 2
+
+        def resp = conn.call(new JsonRpcRequest("foo_bar", []))
+
+        then:
+        StepVerifier.create(resp)
+            .then { server.stop() }
+            .expectError()
+            .verify(Duration.ofSeconds(1))
+    }
+
     def "Gets UNAVAIL status right after disconnect"() {
         setup:
-        def up = Mock(DefaultUpstream)
+        def up = Mock(DefaultUpstream) {
-        conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(up, null, null)
+            _ * getId() >> "test"
+        }
+        conn = new EthereumWsFactory("test", Chain.ETHEREUM, "ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(up, null)
         when:
         conn.connect()
         conn.reconnectIntervalSeconds = 10
@@ -111,7 +128,7 @@ class WsConnectionRealSpec extends Specification {
     def "Validates after connect"() {
         setup:
         def validator = Mock(EthereumUpstreamValidator)
-        conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null, validator, null)
+        conn = new EthereumWsFactory("test", Chain.ETHEREUM, "ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null, validator)
         when:
         conn.connect()
         Thread.sleep(100)

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionSpec.groovy

@@ -25,6 +25,7 @@ import io.emeraldpay.etherjar.rpc.RpcResponseError
 import io.emeraldpay.etherjar.rpc.json.BlockJson
 import io.emeraldpay.etherjar.rpc.json.TransactionJson
 import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
+import io.emeraldpay.grpc.Chain
 import reactor.core.publisher.Flux
 import reactor.test.StepVerifier
 import spock.lang.Specification
@@ -37,7 +38,7 @@ class WsConnectionSpec extends Specification {
 
     def "Fetch block"() {
         setup:
-        def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost"))
+        def wsf = new EthereumWsFactory("test", Chain.ETHEREUM, new URI("http://localhost"), new URI("http://localhost"))
 
         def block = new BlockJson<TransactionRefJson>()
         block.number = 100
@@ -53,7 +54,7 @@ class WsConnectionSpec extends Specification {
 
         def apiMock = TestingCommons.api()
         def wsApiMock = apiMock.asWebsocket()
-        def ws = wsf.create(null, null, null)
+        def ws = wsf.create(null, null)
 
         apiMock.answerOnce("eth_getBlockByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200", false], block)
 
@@ -71,10 +72,10 @@ class WsConnectionSpec extends Specification {
 
     def "Makes a RPC call"() {
         setup:
-        def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost"))
+        def wsf = new EthereumWsFactory("test", Chain.ETHEREUM, new URI("http://localhost"), new URI("http://localhost"))
         def apiMock = TestingCommons.api()
         def wsApiMock = apiMock.asWebsocket()
-        def ws = wsf.create(null, null, null)
+        def ws = wsf.create(null, null)
 
         def tx = new TransactionJson().tap {
             hash = TransactionId.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200")
@@ -96,10 +97,10 @@ class WsConnectionSpec extends Specification {
 
     def "Makes a RPC call - return null"() {
         setup:
-        def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost"))
+        def wsf = new EthereumWsFactory("test", Chain.ETHEREUM, new URI("http://localhost"), new URI("http://localhost"))
         def apiMock = TestingCommons.api()
         def wsApiMock = apiMock.asWebsocket()
-        def ws = wsf.create(null, null, null)
+        def ws = wsf.create(null, null)
 
         apiMock.answerOnce("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], null)
 
@@ -119,10 +120,10 @@ class WsConnectionSpec extends Specification {
 
     def "Makes a RPC call - return error"() {
         setup:
-        def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost"))
+        def wsf = new EthereumWsFactory("test", Chain.ETHEREUM, new URI("http://localhost"), new URI("http://localhost"))
         def apiMock = TestingCommons.api()
         def wsApiMock = apiMock.asWebsocket()
-        def ws = wsf.create(null, null, null)
+        def ws = wsf.create(null, null)
 
         apiMock.answerOnce("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"],
                 new RpcResponseError(RpcResponseError.CODE_METHOD_NOT_EXIST, "test"))

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

@@ -0,0 +1,70 @@
+package io.emeraldpay.dshackle.upstream.rpcclient
+
+import io.emeraldpay.dshackle.reader.Reader
+import reactor.core.publisher.Mono
+import spock.lang.Specification
+
+import java.time.Duration
+
+class JsonRpcSwitchClientSpec extends Specification {
+
+    def "Uses primary response if it works"() {
+        setup:
+        def primaryCalled = false
+        def secondaryCalled = false
+        def request = new JsonRpcRequest("eth_test", [])
+        def response = JsonRpcResponse.ok("test".bytes, new JsonRpcResponse.NumberId(100))
+        def primary = Mock(Reader<JsonRpcRequest, JsonRpcResponse>) {
+            1 * read(request) >> Mono.fromCallable {
+                primaryCalled = true
+                response
+            }
+        }
+        def secondary = Mock(Reader<JsonRpcRequest, JsonRpcResponse>) {
+            _ * read(request) >> Mono.fromCallable {
+                secondaryCalled = true
+                response
+            }
+        }
+
+        def client = new JsonRpcSwitchClient(primary, secondary)
+
+        when:
+        def act = client.read(request).block(Duration.ofSeconds(1))
+
+        then:
+        act == response
+        primaryCalled
+        !secondaryCalled
+    }
+
+    def "Uses secondary response if primary fails"() {
+        setup:
+        def primaryCalled = false
+        def secondaryCalled = false
+        def request = new JsonRpcRequest("eth_test", [])
+        def response = JsonRpcResponse.ok("test".bytes, new JsonRpcResponse.NumberId(100))
+        def primary = Mock(Reader<JsonRpcRequest, JsonRpcResponse>) {
+            1 * read(request) >> Mono.fromCallable {
+                primaryCalled = true
+                throw new IllegalStateException("Primary Fail")
+            }
+        }
+        def secondary = Mock(Reader<JsonRpcRequest, JsonRpcResponse>) {
+            1 * read(request) >> Mono.fromCallable {
+                secondaryCalled = true
+                response
+            }
+        }
+
+        def client = new JsonRpcSwitchClient(primary, secondary)
+
+        when:
+        def act = client.read(request).block(Duration.ofSeconds(1))
+
+        then:
+        act == response
+        primaryCalled
+        secondaryCalled
+    }
+}

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

@@ -0,0 +1,23 @@
+package io.emeraldpay.dshackle.upstream.rpcclient
+
+import io.emeraldpay.dshackle.upstream.ethereum.WsConnection
+import reactor.core.Exceptions
+import spock.lang.Specification
+
+import java.time.Duration
+
+class JsonRpcWsClientSpec extends Specification {
+
+    def "Produce error if WS is not connected"() {
+        setup:
+        def ws = Mock(WsConnection)
+        def client = new JsonRpcWsClient(ws)
+        when:
+        client.read(new JsonRpcRequest("foo_bar", [], 1))
+                .block(Duration.ofSeconds(1))
+        then:
+        def t = thrown(Exceptions.ReactiveException)
+        t.cause instanceof JsonRpcException
+        1 * ws.isConnected() >> false
+    }
+}