From bd8c0f7c19df3580eb25e587c12568e1148e12fa Mon Sep 17 00:00:00 2001 From: Igor Artamonov Date: Wed, 22 Sep 2021 17:08:43 -0400 Subject: [PATCH] solution: mark Upstream status right after WS connect/disconnect --- .../dshackle/upstream/DefaultUpstream.kt | 2 +- .../upstream/ethereum/EthereumRpcUpstream.kt | 3 +- .../ethereum/EthereumUpstreamValidator.kt | 4 +-- .../upstream/ethereum/EthereumWsFactory.kt | 15 +++++++-- .../upstream/ethereum/EthereumWsUpstream.kt | 6 ++-- .../ethereum/EthereumWsFactoryRealSpec.groovy | 31 ++++++++++++++++++- .../ethereum/EthereumWsFactorySpec.groovy | 8 ++--- 7 files changed, 55 insertions(+), 14 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt index 86d54225..6646d85b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt @@ -61,7 +61,7 @@ abstract class DefaultUpstream( return status.get().status } - fun setStatus(avail: UpstreamAvailability) { + open fun setStatus(avail: UpstreamAvailability) { status.updateAndGet { curr -> Status(curr.lag, avail, statusByLag(curr.lag, avail)) } 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 2bae1815..b354abc9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt @@ -74,7 +74,8 @@ open class EthereumRpcUpstream( open fun createHead(): Head { return if (ethereumWsFactory != null) { - val ws = ethereumWsFactory.create(null).apply { + // do not set upstream to the WS, since it doesn't control the RPC upstream + val ws = ethereumWsFactory.create(null, null, null).apply { connect() } val wsHead = EthereumWsHead(ws).apply { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt index 99d326b1..a968753a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt @@ -32,7 +32,7 @@ import reactor.core.scheduler.Schedulers import java.time.Duration import java.util.concurrent.Executors -class EthereumUpstreamValidator( +open class EthereumUpstreamValidator( private val upstream: EthereumUpstream, private val options: UpstreamsConfig.Options ) { @@ -43,7 +43,7 @@ class EthereumUpstreamValidator( private val objectMapper: ObjectMapper = Global.objectMapper - fun validate(): Mono { + open fun validate(): Mono { return upstream .getApi() .read(JsonRpcRequest("eth_syncing", listOf())) 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 19b06dae..60b98efc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactory.kt @@ -21,6 +21,8 @@ import io.emeraldpay.dshackle.Global import io.emeraldpay.dshackle.SilentException import io.emeraldpay.dshackle.config.AuthConfig import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.upstream.DefaultUpstream +import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.dshackle.upstream.rpcclient.ResponseWSParser @@ -64,15 +66,17 @@ class EthereumWsFactory( var basicAuth: AuthConfig.ClientBasicAuth? = null - fun create(rpcMetrics: RpcMetrics?): EthereumWs { - return EthereumWs(uri, origin, basicAuth, rpcMetrics) + fun create(upstream: DefaultUpstream?, validator: EthereumUpstreamValidator?, rpcMetrics: RpcMetrics?): EthereumWs { + return EthereumWs(uri, origin, basicAuth, rpcMetrics, upstream, validator) } class EthereumWs( private val uri: URI, private val origin: URI, private val basicAuth: AuthConfig.ClientBasicAuth?, - private val rpcMetrics: RpcMetrics? + private val rpcMetrics: RpcMetrics?, + private val upstream: DefaultUpstream?, + private val validator: EthereumUpstreamValidator? ) : AutoCloseable { companion object { @@ -154,6 +158,8 @@ class EthereumWsFactory( connection = HttpClient.create() .doOnDisconnected { log.info("Disconnected from $uri") + // mark upstream as UNAVAIL + upstream?.setStatus(UpstreamAvailability.UNAVAILABLE) if (keepConnection) { tryReconnectLater() } @@ -203,6 +209,9 @@ class EthereumWsFactory( //restart backoff after connection currentBackOff = reconnectBackoff.start() + //validate the connection, it can also be UNAVAIL if market as such after disconnect + validator?.validate() + val consumer = inbound // Accept up to 15Mb messages, same config is used by Geth .aggregateFrames(15 * 1024 * 1024) 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 cd1163a8..7046be7b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsUpstream.kt @@ -53,6 +53,7 @@ class EthereumWsUpstream( private val api: JsonRpcWsClient private var validatorSubscription: Disposable? = null + private val validator: EthereumUpstreamValidator init { val metricsTags = listOf( @@ -72,7 +73,9 @@ class EthereumWsUpstream( .register(Metrics.globalRegistry) ) - connection = ethereumWsFactory.create(metrics) + validator = EthereumUpstreamValidator(this, getOptions()) + + connection = ethereumWsFactory.create(this, validator, metrics) head = EthereumWsHead(connection) api = JsonRpcWsClient(connection) } @@ -102,7 +105,6 @@ class EthereumWsUpstream( head.start() log.debug("Start validation for upstream ${this.getId()}") - val validator = EthereumUpstreamValidator(this, getOptions()) validatorSubscription = validator.start() .subscribe(this::setStatus) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy index 8edfc3b2..069c7a05 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactoryRealSpec.groovy @@ -1,6 +1,8 @@ package io.emeraldpay.dshackle.upstream.ethereum 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 reactor.test.StepVerifier import spock.lang.Shared @@ -28,7 +30,7 @@ class EthereumWsFactoryRealSpec extends Specification { server = new MockWSServer(port) server.start() Thread.sleep(SLEEP) - conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null) + conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null, null, null) } def cleanup() { @@ -91,6 +93,33 @@ class EthereumWsFactoryRealSpec extends Specification { act[0].value.contains("\"params\":[\"newHeads\"]") } + 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) + when: + conn.connect() + conn.reconnectIntervalSeconds = 10 + Thread.sleep(SLEEP) + server.stop() + Thread.sleep(100) + + then: + 1 * up.setStatus(UpstreamAvailability.UNAVAILABLE) + } + + def "Validates after connect"() { + setup: + def validator = Mock(EthereumUpstreamValidator) + conn = new EthereumWsFactory("ws://localhost:${port}".toURI(), "http://localhost:${port}".toURI()).create(null, validator, null) + when: + conn.connect() + Thread.sleep(100) + + then: + 1 * validator.validate() + } + def "Try to connects to server until it's available"() { when: server.stop() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactorySpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactorySpec.groovy index f5415999..6da34f11 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactorySpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsFactorySpec.groovy @@ -53,7 +53,7 @@ class EthereumWsFactorySpec extends Specification { def apiMock = TestingCommons.api() def wsApiMock = apiMock.asWebsocket() - def ws = wsf.create() + def ws = wsf.create(null, null, null) apiMock.answerOnce("eth_getBlockByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200", false], block) @@ -74,7 +74,7 @@ class EthereumWsFactorySpec extends Specification { def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost")) def apiMock = TestingCommons.api() def wsApiMock = apiMock.asWebsocket() - def ws = wsf.create() + def ws = wsf.create(null, null, null) def tx = new TransactionJson().tap { hash = TransactionId.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200") @@ -99,7 +99,7 @@ class EthereumWsFactorySpec extends Specification { def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost")) def apiMock = TestingCommons.api() def wsApiMock = apiMock.asWebsocket() - def ws = wsf.create() + def ws = wsf.create(null, null, null) apiMock.answerOnce("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], null) @@ -122,7 +122,7 @@ class EthereumWsFactorySpec extends Specification { def wsf = new EthereumWsFactory(new URI("http://localhost"), new URI("http://localhost")) def apiMock = TestingCommons.api() def wsApiMock = apiMock.asWebsocket() - def ws = wsf.create() + def ws = wsf.create(null, null, null) apiMock.answerOnce("eth_getTransactionByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200"], new RpcResponseError(RpcResponseError.CODE_METHOD_NOT_EXIST, "test"))