From e9c5fbf148d011bd96fa77f7ce6b1ec6dd225a2a Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Thu, 28 Jul 2022 23:36:11 -0400 Subject: [PATCH 1/5] problem: WS connection doesn't resolve local domain names rel: https://github.com/netty/netty/issues/11142 rel: https://github.com/reactor/reactor-netty/issues/1431 --- .../io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt index f5fa0159..858491aa 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt @@ -34,6 +34,8 @@ 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 io.netty.resolver.DefaultNameResolver import org.reactivestreams.Publisher import org.slf4j.LoggerFactory import org.springframework.util.backoff.BackOff @@ -161,6 +163,7 @@ class WsConnection( log.info("Connecting to WebSocket: $uri") connection?.dispose() connection = HttpClient.create() + .resolver(DefaultAddressResolverGroup.INSTANCE) .doOnDisconnected { log.info("Disconnected from $uri") // mark upstream as UNAVAIL From 067a9c10ad37358d06a9c52766d27b70fe00e078 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Sun, 31 Jul 2022 21:14:55 -0400 Subject: [PATCH 2/5] problem: doesn't resolve local domain names rel: [e9c5fbf1] --- .../io/emeraldpay/dshackle/upstream/bitcoin/EsploraClient.kt | 2 ++ .../emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt | 2 ++ 2 files changed, 4 insertions(+) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/EsploraClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/EsploraClient.kt index aa2b623d..996d9a31 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/EsploraClient.kt +++ b/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") diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt index 8d3e7b76..fb425ebf 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt +++ b/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") From eacb769c62b7d079e923031eda56208f23eca52c Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Mon, 1 Aug 2022 19:56:04 -0400 Subject: [PATCH 3/5] problem: no response for requests that produce very large results which causes the upstream to break the WS connection --- .../dshackle/startup/ConfiguredUpstreams.kt | 14 ++-- .../upstream/ethereum/EthereumWsUpstream.kt | 11 ++- .../upstream/ethereum/WsConnection.kt | 39 ++++++++--- .../upstream/rpcclient/JsonRpcHttpClient.kt | 1 + .../upstream/rpcclient/JsonRpcSwitchClient.kt | 27 +++++++ .../upstream/rpcclient/JsonRpcWsClient.kt | 12 ++++ .../ethereum/WsConnectionRealSpec.groovy | 14 ++++ .../rpcclient/JsonRpcSwitchClientSpec.groovy | 70 +++++++++++++++++++ .../rpcclient/JsonRpcWsClientSpec.groovy | 23 ++++++ 9 files changed, 195 insertions(+), 16 deletions(-) create mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClient.kt create mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClientSpec.groovy create mode 100644 src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClientSpec.groovy diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index 264294be..33e05ebc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/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? = 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? = buildHttpClient(config) - if (directApi == null) { - log.warn("Upstream doesn't have API configuration") - return - } EthereumRpcUpstream( config.id!!, chain, directApi, wsFactoryApi, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt index c9ddb32b..f2d85e74 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt +++ b/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, 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 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 { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt index 858491aa..d09cda9e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/WsConnection.kt +++ b/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() + private val disconnects = Sinks + .many() + .multicast() + .directBestEffort() 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 { 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( + 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 { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt index fb425ebf..bc56fc41 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt +++ b/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) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClient.kt new file mode 100644 index 00000000..82ab68cd --- /dev/null +++ b/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, + private val secondary: Reader, +) : Reader { + + companion object { + private val log = LoggerFactory.getLogger(JsonRpcSwitchClient::class.java) + } + + override fun read(key: JsonRpcRequest): Mono { + return primary.read(key) + .switchIfEmpty(Mono.error(IllegalStateException("No response from Primary Connection"))) + .onErrorResume { + secondary.read(key) + } + } +} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt index 0a10a868..6f65d0a2 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClient.kt +++ b/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 { override fun read(key: JsonRpcRequest): Mono { + 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) } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionRealSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionRealSpec.groovy index a84e6e00..9f16a8a2 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionRealSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionRealSpec.groovy @@ -93,6 +93,20 @@ 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) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClientSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcSwitchClientSpec.groovy new file mode 100644 index 00000000..883e400b --- /dev/null +++ b/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) { + 1 * read(request) >> Mono.fromCallable { + primaryCalled = true + response + } + } + def secondary = Mock(Reader) { + _ * 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) { + 1 * read(request) >> Mono.fromCallable { + primaryCalled = true + throw new IllegalStateException("Primary Fail") + } + } + def secondary = Mock(Reader) { + 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 + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClientSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcWsClientSpec.groovy new file mode 100644 index 00000000..e0ac8a1b --- /dev/null +++ b/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 + } +} From a26db98376f4113738f7b177b22659cb0aa19eaa Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Mon, 1 Aug 2022 21:38:41 -0400 Subject: [PATCH 4/5] problem: dev logging --- .../emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt | 1 - 1 file changed, 1 deletion(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt index bc56fc41..fb425ebf 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt @@ -112,7 +112,6 @@ class JsonRpcHttpClient( return Mono.just(key) .map(JsonRpcRequest::toJson) .doOnNext { - println("rpc connection ${key.method}") startTime = System.nanoTime() } .flatMap(this@JsonRpcHttpClient::execute) From a34da8b4a364236f5bad1d3b858965cf63e155e2 Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Fri, 5 Aug 2022 21:26:35 -0400 Subject: [PATCH 5/5] problem: head check doesn't use upstream WS metrics --- .../dshackle/startup/ConfiguredUpstreams.kt | 1 + .../upstream/ethereum/EthereumRpcUpstream.kt | 2 +- .../upstream/ethereum/EthereumWsFactory.kt | 37 +++++++++++++++++-- .../upstream/ethereum/EthereumWsUpstream.kt | 25 +------------ .../ethereum/WsConnectionRealSpec.groovy | 11 ++++-- .../upstream/ethereum/WsConnectionSpec.groovy | 17 +++++---- 6 files changed, 53 insertions(+), 40 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index 33e05ebc..e76a486f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/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"), ) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt index 51bed8d7..771820b0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt +++ b/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 { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt index da7aa412..00def0bf 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt +++ b/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 { - return WsConnection(uri, origin, basicAuth, rpcMetrics, upstream, validator).also { ws -> + // metrics are shared between all connections to the same 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 } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt index f2d85e74..c721e3e0 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt @@ -26,12 +26,7 @@ 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 @@ -59,26 +54,8 @@ class EthereumWsUpstream( 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, metrics) + connection = ethereumWsFactory.create(this, validator) head = EthereumWsHead(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) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionRealSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionRealSpec.groovy index 9f16a8a2..eb3f22c0 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionRealSpec.groovy +++ b/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() { @@ -109,8 +110,10 @@ class WsConnectionRealSpec extends Specification { def "Gets UNAVAIL status right after disconnect"() { setup: - def up = Mock(DefaultUpstream) - conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(up, null, null) + def up = Mock(DefaultUpstream) { + _ * 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 @@ -125,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) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionSpec.groovy index 3b4b5edb..b8996287 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/WsConnectionSpec.groovy +++ b/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() 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"))