merge master

src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt

@@ -106,7 +106,7 @@ class UpstreamsConfigReader(
                     config.upstreams.add(upstream)
                     val connection = UpstreamsConfig.GrpcConnection()
                     upstream.connection = connection
-                    getValueAsInt(connConfigNode, "node-rating")?.let {
+                    getValueAsInt(connConfigNode, "upstream-rating")?.let {
                         connection.upstreamRating = it
                     }
                     getValueAsString(connConfigNode, "host")?.let {
@@ -170,7 +170,7 @@ class UpstreamsConfigReader(
         getMapping(connConfigNode, "execution")?.let {
             connection.execution = readEthereumConnection(it)
         }
-        getValueAsInt(connConfigNode, "node-rating")?.let {
+        getValueAsInt(connConfigNode, "upstream-rating")?.let {
             connection.upstreamRating = it
         }
         return connection

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

@@ -159,7 +159,7 @@ open class ConfiguredUpstreams(
             return null
         }
         val urls = ArrayList<URI>()
-        val connectorFactory = buildEthereumConnectorFactory(execution, chain, urls, NoChoiceWithPriorityForkChoice(conn.upstreamRating))
+        val connectorFactory = buildEthereumConnectorFactory(config.id!!, execution, chain, urls, NoChoiceWithPriorityForkChoice(conn.upstreamRating))
         val methods = buildMethods(config, chain)
         if (connectorFactory == null) {
             return null
@@ -228,7 +228,7 @@ open class ConfiguredUpstreams(
         val urls = ArrayList<URI>()
         val methods = buildMethods(config, chain)
 
-        val connectorFactory = buildEthereumConnectorFactory(conn, chain, urls, MostWorkForkChoice())
+        val connectorFactory = buildEthereumConnectorFactory(config.id!!, conn, chain, urls, MostWorkForkChoice())
         if (connectorFactory == null) {
             return null
         }
@@ -280,9 +280,10 @@ open class ConfiguredUpstreams(
         }
     }
 
-    private fun buildWsFactory(conn: UpstreamsConfig.EthereumConnection, urls: ArrayList<URI>? = null): EthereumWsFactory? {
+    private fun buildWsFactory(id: String, chain: Chain, conn: UpstreamsConfig.EthereumConnection, urls: ArrayList<URI>? = null): EthereumWsFactory? {
         return conn.ws?.let { endpoint ->
             val wsApi = EthereumWsFactory(
+                id, chain,
                 endpoint.url,
                 endpoint.origin ?: URI("http://localhost"),
             )
@@ -295,8 +296,8 @@ open class ConfiguredUpstreams(
         }
     }
 
-    private fun buildEthereumConnectorFactory(conn: UpstreamsConfig.EthereumConnection, chain: Chain, urls: ArrayList<URI>, forkChoice: ForkChoice): EthereumConnectorFactory? {
+    private fun buildEthereumConnectorFactory(id: String, conn: UpstreamsConfig.EthereumConnection, chain: Chain, urls: ArrayList<URI>, forkChoice: ForkChoice): EthereumConnectorFactory? {
-        val wsFactoryApi = buildWsFactory(conn, urls)
+        val wsFactoryApi = buildWsFactory(id, chain, conn, urls)
         val httpFactory = buildHttpFactory(conn, urls)
         log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}")
         val connectorFactory = EthereumConnectorFactory(conn.preferHttp, wsFactoryApi, httpFactory, forkChoice)

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/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/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/ethereum/connectors/EthereumRpcConnector.kt

@@ -32,7 +32,7 @@ class EthereumRpcConnector(
     init {
         if (wsFactory != null) {
             // do not set upstream to the WS, since it doesn't control the RPC upstream
-            conn = wsFactory.create(null, null, null)
+            conn = wsFactory.create(null, null)
             val wsHead = EthereumWsHead(conn, forkChoice)
             // receive bew blocks through WebSockets, but also periodically verify with RPC in case if WS failed
             val rpcHead = EthereumRpcHead(directReader, forkChoice, Duration.ofSeconds(60))

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/connectors/EthereumWsConnector.kt

@@ -30,24 +30,7 @@ class EthereumWsConnector(
     private val head: EthereumWsHead
 
     init {
-        val metricsTags = listOf(
+        conn = wsFactory.create(upstream, validator)
-            Tag.of("upstream", upstream.getId()),
-            // 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)
-        )
-
-        conn = wsFactory.create(upstream, validator, metrics)
         head = EthereumWsHead(conn, forkChoice)
         api = JsonRpcWsClient(conn)
     }

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/config/UpstreamsConfigReaderSpec.groovy

@@ -166,7 +166,7 @@ class UpstreamsConfigReaderSpec extends Specification {
             with((UpstreamsConfig.EthereumPosConnection) connection) {
                 execution.rpc != null
                 execution.rpc.url == new URI("http://34.106.60.110:8545")
-                blockPriority == 100
+                upstreamRating == 100
             }
         }
     }

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

src/test/resources/upstreams-ethereum-pos.yaml

@@ -6,4 +6,4 @@
           execution:
             rpc:
               url: "http://34.106.60.110:8545"
-          block-priority: 100
+          upstream-rating: 100