From 634f704c2e483c97d4d0720e35c946de51206e3b Mon Sep 17 00:00:00 2001 From: KirillPamPam Date: Tue, 4 Jul 2023 14:17:54 +0400 Subject: [PATCH] Ignore blocks from syncing nodes (#243) --- .../dshackle/startup/ConfiguredUpstreams.kt | 15 +-- .../dshackle/upstream/AbstractHead.kt | 7 +- .../emeraldpay/dshackle/upstream/EmptyHead.kt | 2 +- .../io/emeraldpay/dshackle/upstream/Head.kt | 2 +- .../upstream/ethereum/ERC20Balance.kt | 4 +- ...Upstream.kt => EthereumLikeRpcUpstream.kt} | 11 +- ...eumUpstream.kt => EthereumLikeUpstream.kt} | 2 +- .../upstream/ethereum/EthereumMultistream.kt | 2 +- .../upstream/ethereum/EthereumRpcHead.kt | 6 + .../ethereum/EthereumUpstreamValidator.kt | 9 +- .../upstream/ethereum/EthereumWsHead.kt | 19 +++- .../ethereum_pos/EthereumPosMultiStream.kt | 2 +- .../ethereum_pos/EthereumPosRpcUpstream.kt | 103 ------------------ .../ethereum_pos/EthereumPosUpstream.kt | 51 --------- .../upstream/grpc/EthereumGrpcUpstream.kt | 4 +- .../upstream/grpc/EthereumPosGrpcUpstream.kt | 4 +- .../dshackle/rpc/StreamHeadSpec.groovy | 6 +- .../dshackle/test/EthereumHeadMock.groovy | 2 +- .../test/EthereumPosRpcUpstreamMock.groovy | 15 +-- .../test/EthereumRpcUpstreamMock.groovy | 14 +-- .../test/MultistreamHolderMock.groovy | 13 +-- .../dshackle/upstream/FilteredApisSpec.groovy | 9 +- .../dshackle/upstream/MultistreamSpec.groovy | 8 +- .../upstream/ethereum/ERC20BalanceSpec.groovy | 4 +- .../EthereumUpstreamValidatorSpec.groovy | 39 ++++++- .../ethereum/EthereumWsHeadSpec.groovy | 65 +++++++++++ 26 files changed, 191 insertions(+), 227 deletions(-) rename src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/{EthereumRpcUpstream.kt => EthereumLikeRpcUpstream.kt} (91%) rename src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/{EthereumUpstream.kt => EthereumLikeUpstream.kt} (97%) delete mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt delete mode 100644 src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosUpstream.kt diff --git a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt index 309f8544..f8c0271a 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt @@ -42,8 +42,7 @@ import io.emeraldpay.dshackle.upstream.bitcoin.ZMQServer import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.ManagedCallMethods import io.emeraldpay.dshackle.upstream.ethereum.EthereumBlockValidator -import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosRpcUpstream -import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcUpstream +import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeRpcUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsConnectionFactory import io.emeraldpay.dshackle.upstream.ethereum.EthereumWsConnectionPoolFactory import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory @@ -227,7 +226,7 @@ open class ConfiguredUpstreams( if (it.connectorMode == RPC_REQUESTS_WITH_MIXED_HEAD.name) it.rpc?.url ?: it.ws?.url else it.ws?.url ?: it.rpc?.url } val hash = getHash(nodeId, hashUrl!!) - val upstream = EthereumPosRpcUpstream( + val upstream = EthereumLikeRpcUpstream( config.id!!, hash, chain, @@ -236,7 +235,8 @@ open class ConfiguredUpstreams( methods, QuorumForLabels.QuorumItem(1, config.labels), connectorFactory, - chainConf + chainConf, + true ) upstream.start() return upstream @@ -291,7 +291,7 @@ open class ConfiguredUpstreams( chain: Chain, options: UpstreamsConfig.Options, chainConf: ChainsConfig.ChainConfig - ): EthereumRpcUpstream? { + ): Upstream? { val conn = config.connection!! val urls = ArrayList() @@ -310,7 +310,7 @@ open class ConfiguredUpstreams( } val hashUrl = if (conn.connectorMode == RPC_REQUESTS_WITH_MIXED_HEAD.name) conn.rpc?.url ?: conn.ws?.url else conn.ws?.url ?: conn.rpc?.url - val upstream = EthereumRpcUpstream( + val upstream = EthereumLikeRpcUpstream( config.id!!, getHash(nodeId, hashUrl!!), chain, @@ -318,7 +318,8 @@ open class ConfiguredUpstreams( methods, QuorumForLabels.QuorumItem(1, config.labels), connectorFactory, - chainConf + chainConf, + false ) upstream.start() return upstream diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt index 64171e72..2c981fcf 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/AbstractHead.kt @@ -149,7 +149,12 @@ abstract class AbstractHead @JvmOverloads constructor( future = null } - override fun onNoHeadUpdates() { + protected open fun onNoHeadUpdates() { + // NOOP + } + + override fun onSyncingNode(isSyncing: Boolean) { + // NOOP } override fun start() { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyHead.kt index 30dad649..a67bdb2e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/EmptyHead.kt @@ -37,6 +37,6 @@ class EmptyHead : Head { override fun stop() { } - override fun onNoHeadUpdates() { + override fun onSyncingNode(isSyncing: Boolean) { } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt index 4a899a85..d3e133b1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Head.kt @@ -42,5 +42,5 @@ interface Head { fun stop() - fun onNoHeadUpdates() + fun onSyncingNode(isSyncing: Boolean) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/ERC20Balance.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/ERC20Balance.kt index d1c0fa2e..cf889279 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/ERC20Balance.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/ERC20Balance.kt @@ -49,7 +49,7 @@ open class ERC20Balance { apis.request(1) return Flux.from(apis) .flatMap { - getBalance(it.cast(EthereumPosRpcUpstream::class.java), token, address) + getBalance(it.cast(EthereumLikeRpcUpstream::class.java), token, address) } .doOnNext { apis.resolve() @@ -57,7 +57,7 @@ open class ERC20Balance { .next() } - open fun getBalance(upstream: EthereumPosRpcUpstream, token: ERC20Token, address: Address): Mono { + open fun getBalance(upstream: EthereumLikeRpcUpstream, token: ERC20Token, address: Address): Mono { return upstream .getIngressReader() .read(prepareEthCall(token, address, upstream.getHead())) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeRpcUpstream.kt similarity index 91% rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeRpcUpstream.kt index b4b7e54f..3a957919 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeRpcUpstream.kt @@ -32,19 +32,20 @@ import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnector import org.springframework.context.Lifecycle import reactor.core.Disposable -open class EthereumRpcUpstream( +open class EthereumLikeRpcUpstream( id: String, hash: Byte, val chain: Chain, options: UpstreamsConfig.Options, role: UpstreamsConfig.UpstreamRole, targets: CallMethods?, - private val node: QuorumForLabels.QuorumItem?, + node: QuorumForLabels.QuorumItem?, connectorFactory: ConnectorFactory, - chainConfig: ChainsConfig.ChainConfig -) : EthereumUpstream(id, hash, options, role, targets, node, chainConfig), Lifecycle, Upstream, CachesEnabled { + chainConfig: ChainsConfig.ChainConfig, + skipEnhance: Boolean +) : EthereumLikeUpstream(id, hash, options, role, targets, node, chainConfig), Lifecycle, Upstream, CachesEnabled { private val validator: EthereumUpstreamValidator = EthereumUpstreamValidator(this, getOptions()) - private val connector: EthereumConnector = connectorFactory.create(this, validator, chain, false) + private val connector: EthereumConnector = connectorFactory.create(this, validator, chain, skipEnhance) private var validatorSubscription: Disposable? = null diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeUpstream.kt similarity index 97% rename from src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt rename to src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeUpstream.kt index ea87aff0..75692409 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLikeUpstream.kt @@ -23,7 +23,7 @@ import io.emeraldpay.dshackle.upstream.Capability import io.emeraldpay.dshackle.upstream.DefaultUpstream import io.emeraldpay.dshackle.upstream.calls.CallMethods -abstract class EthereumUpstream( +abstract class EthereumLikeUpstream( id: String, hash: Byte, options: UpstreamsConfig.Options, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt index e53c3c66..c6c0e610 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt @@ -47,7 +47,7 @@ import reactor.core.scheduler.Scheduler @Suppress("UNCHECKED_CAST") open class EthereumMultistream( chain: Chain, - val upstreams: MutableList, + val upstreams: MutableList, caches: Caches, private val headScheduler: Scheduler, tracer: Tracer, diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt index 193c833b..3eda1d63 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumRpcHead.kt @@ -35,12 +35,14 @@ class EthereumRpcHead( ) : DefaultEthereumHead(upstreamId, forkChoice, blockValidator, headScheduler), Lifecycle { private var refreshSubscription: Disposable? = null + private var isSyncing = false override fun start() { super.start() refreshSubscription?.dispose() val base = Flux.interval(interval) .publishOn(headScheduler) + .filter { !isSyncing } .flatMap { getLatestBlock(api) } @@ -51,6 +53,10 @@ class EthereumRpcHead( return refreshSubscription != null } + override fun onSyncingNode(isSyncing: Boolean) { + this.isSyncing = isSyncing + } + override fun stop() { super.stop() refreshSubscription?.dispose() 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 9f368e41..8c52a530 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidator.kt @@ -65,8 +65,7 @@ open class EthereumUpstreamValidator( if (!options.validateSyncing) { return Mono.just(UpstreamAvailability.OK) } - return upstream - .getIngressReader() + return upstream.getIngressReader() .read(JsonRpcRequest("eth_syncing", listOf())) .flatMap(JsonRpcResponse::requireResult) .map { objectMapper.readValue(it, SyncingJson::class.java) } @@ -75,8 +74,10 @@ open class EthereumUpstreamValidator( Mono.fromCallable { log.warn("No response for eth_syncing from ${upstream.getId()}") } .then(Mono.error(TimeoutException("Validation timeout for Syncing"))) ) - .map { value -> - if (value.isSyncing) { + .map { + val isSyncing = it.isSyncing + upstream.getHead().onSyncingNode(isSyncing) + if (isSyncing) { UpstreamAvailability.SYNCING } else { UpstreamAvailability.OK diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt index 6c3a5e5e..c613aa7e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHead.kt @@ -50,6 +50,7 @@ class EthereumWsHead( private var connectionId: String? = null private var subscribed = false private var connected = false + private var isSyncing = false private var subscription: Disposable? = null private val noHeadUpdatesSink = Sinks.many().multicast().directBestEffort() @@ -78,6 +79,13 @@ class EthereumWsHead( noHeadUpdatesSink.tryEmitNext(true) } + override fun onSyncingNode(isSyncing: Boolean) { + if (isSyncing && !this.isSyncing) { + cancelSub() + } + this.isSyncing = isSyncing + } + fun listenNewHeads(): Flux { return subscribe() .map { @@ -131,8 +139,7 @@ class EthereumWsHead( override fun stop() { super.stop() - subscription?.dispose() - subscription = null + cancelSub() noHeadUpdatesSink.tryEmitComplete() } @@ -168,10 +175,16 @@ class EthereumWsHead( noHeadUpdatesSink.asFlux(), connectionStates, ).subscribeOn(wsConnectionResubscribeScheduler) - .filter { it && !subscribed && connected } + .filter { it && !subscribed && connected && !isSyncing } .subscribe { log.warn("Restart ws head, upstreamId: $upstreamId") start() } } + + private fun cancelSub() { + subscription?.dispose() + subscription = null + subscribed = false + } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt index f90c570a..6bd8e852 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosMultiStream.kt @@ -45,7 +45,7 @@ import reactor.core.scheduler.Scheduler @Suppress("UNCHECKED_CAST") open class EthereumPosMultiStream( chain: Chain, - val upstreams: MutableList, + val upstreams: MutableList, caches: Caches, private val headScheduler: Scheduler, tracer: Tracer diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt deleted file mode 100644 index 9048884d..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt +++ /dev/null @@ -1,103 +0,0 @@ -/** - * Copyright (c) 2020 EmeraldPay, Inc - * Copyright (c) 2019 ETCDEV GmbH - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package io.emeraldpay.dshackle.upstream.ethereum - -import io.emeraldpay.dshackle.Chain -import io.emeraldpay.dshackle.cache.Caches -import io.emeraldpay.dshackle.cache.CachesEnabled -import io.emeraldpay.dshackle.config.ChainsConfig -import io.emeraldpay.dshackle.config.UpstreamsConfig -import io.emeraldpay.dshackle.reader.JsonRpcReader -import io.emeraldpay.dshackle.startup.QuorumForLabels -import io.emeraldpay.dshackle.upstream.Head -import io.emeraldpay.dshackle.upstream.Lifecycle -import io.emeraldpay.dshackle.upstream.Upstream -import io.emeraldpay.dshackle.upstream.UpstreamAvailability -import io.emeraldpay.dshackle.upstream.calls.CallMethods -import io.emeraldpay.dshackle.upstream.ethereum.connectors.ConnectorFactory -import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnector -import reactor.core.Disposable - -open class EthereumPosRpcUpstream( - id: String, - hash: Byte, - val chain: Chain, - options: UpstreamsConfig.Options, - role: UpstreamsConfig.UpstreamRole, - targets: CallMethods?, - node: QuorumForLabels.QuorumItem?, - connectorFactory: ConnectorFactory, - chainConfig: ChainsConfig.ChainConfig -) : EthereumPosUpstream(id, hash, options, role, targets, node, chainConfig), Lifecycle, Upstream, CachesEnabled { - private val validator: EthereumUpstreamValidator = EthereumUpstreamValidator(this, getOptions()) - private val connector: EthereumConnector = connectorFactory.create(this, validator, chain, true) - - private var validatorSubscription: Disposable? = null - - override fun setCaches(caches: Caches) { - if (connector is CachesEnabled) { - connector.setCaches(caches) - } - } - - override fun start() { - log.info("Configured for ${chain.chainName}") - connector.start() - if (getOptions().disableValidation) { - log.warn("Disable validation for upstream ${this.getId()}") - this.setLag(0) - this.setStatus(UpstreamAvailability.OK) - } else { - log.debug("Start validation for upstream ${this.getId()}") - validatorSubscription = validator.start() - .subscribe(this::setStatus) - } - } - override fun getHead(): Head { - return connector.getHead() - } - - override fun stop() { - validatorSubscription?.dispose() - validatorSubscription = null - connector.stop() - } - - override fun isRunning(): Boolean { - return connector.isRunning() - } - - override fun getIngressReader(): JsonRpcReader { - return connector.getIngressReader() - } - - override fun isGrpc(): Boolean { - return false - } - - @Suppress("UNCHECKED_CAST") - override fun cast(selfType: Class): T { - if (!selfType.isAssignableFrom(this.javaClass)) { - throw ClassCastException("Cannot cast ${this.javaClass} to $selfType") - } - return this as T - } - - override fun getIngressSubscription(): EthereumIngressSubscription { - return connector.getIngressSubscription() - } -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosUpstream.kt deleted file mode 100644 index fadb6921..00000000 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosUpstream.kt +++ /dev/null @@ -1,51 +0,0 @@ -/** - * Copyright (c) 2020 EmeraldPay, Inc - * Copyright (c) 2019 ETCDEV GmbH - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package io.emeraldpay.dshackle.upstream.ethereum - -import io.emeraldpay.dshackle.config.ChainsConfig -import io.emeraldpay.dshackle.config.UpstreamsConfig -import io.emeraldpay.dshackle.startup.QuorumForLabels -import io.emeraldpay.dshackle.upstream.Capability -import io.emeraldpay.dshackle.upstream.DefaultUpstream -import io.emeraldpay.dshackle.upstream.calls.CallMethods - -abstract class EthereumPosUpstream( - id: String, - hash: Byte, - options: UpstreamsConfig.Options, - role: UpstreamsConfig.UpstreamRole, - targets: CallMethods?, - private val node: QuorumForLabels.QuorumItem?, - chainConfig: ChainsConfig.ChainConfig -) : DefaultUpstream(id, hash, options, role, targets, node, chainConfig) { - - private val capabilities = if (options.providesBalance != false) { - setOf(Capability.RPC, Capability.BALANCE) - } else { - setOf(Capability.RPC) - } - - override fun getCapabilities(): Set { - return capabilities - } - - override fun getLabels(): Collection { - return node?.let { listOf(it.labels) } ?: emptyList() - } - - abstract fun getIngressSubscription(): EthereumIngressSubscription -} diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt index 1ef105b3..404139bc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstream.kt @@ -34,7 +34,7 @@ import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.UpstreamAvailability import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription -import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream +import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeUpstream import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumDshackleIngressSubscription import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient @@ -62,7 +62,7 @@ open class EthereumGrpcUpstream( overrideLabels: UpstreamsConfig.Labels?, chainConfig: ChainsConfig.ChainConfig, headScheduler: Scheduler, -) : EthereumUpstream( +) : EthereumLikeUpstream( "${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}", hash, UpstreamsConfig.PartialOptions.getDefaults().buildOptions(), diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt index ef1aa8ba..11a958fc 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/EthereumPosGrpcUpstream.kt @@ -33,7 +33,7 @@ import io.emeraldpay.dshackle.upstream.Lifecycle import io.emeraldpay.dshackle.upstream.Upstream import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.ethereum.EthereumIngressSubscription -import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosUpstream +import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeUpstream import io.emeraldpay.dshackle.upstream.ethereum.subscribe.EthereumDshackleIngressSubscription import io.emeraldpay.dshackle.upstream.forkchoice.NoChoiceWithPriorityForkChoice import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcGrpcClient @@ -56,7 +56,7 @@ open class EthereumPosGrpcUpstream( overrideLabels: UpstreamsConfig.Labels?, chainConfig: ChainsConfig.ChainConfig, headScheduler: Scheduler, -) : EthereumPosUpstream( +) : EthereumLikeUpstream( "${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}", hash, UpstreamsConfig.PartialOptions.getDefaults().buildOptions(), diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/StreamHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/StreamHeadSpec.groovy index 0b53f84c..3ca52a22 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/StreamHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/StreamHeadSpec.groovy @@ -26,9 +26,9 @@ import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.test.EthereumPosRpcUpstreamMock import io.emeraldpay.dshackle.test.MultistreamHolderMock import io.emeraldpay.dshackle.test.TestingCommons -import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosRpcUpstream -import io.emeraldpay.etherjar.domain.BlockHash +import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeRpcUpstream import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson +import io.emeraldpay.etherjar.domain.BlockHash import io.emeraldpay.etherjar.rpc.json.TransactionRefJson import reactor.core.publisher.Mono import reactor.test.StepVerifier @@ -43,7 +43,7 @@ class StreamHeadSpec extends Specification { def "Errors on unavailable chain"() { setup: - def upstreams = new MultistreamHolderMock(Chain.ETHEREUM__MAINNET, Stub(EthereumPosRpcUpstream)) + def upstreams = new MultistreamHolderMock(Chain.ETHEREUM__MAINNET, Stub(EthereumLikeRpcUpstream)) def streamHead = new StreamHead(upstreams) when: def flux = streamHead.add( diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy index c41daba3..da2eeffa 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumHeadMock.groovy @@ -79,7 +79,7 @@ class EthereumHeadMock implements Head { } @Override - void onNoHeadUpdates() { + void onSyncingNode(boolean isSyncing) { } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumPosRpcUpstreamMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumPosRpcUpstreamMock.groovy index 6dd3e71a..37a7eeea 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumPosRpcUpstreamMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumPosRpcUpstreamMock.groovy @@ -24,19 +24,14 @@ import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.upstream.UpstreamAvailability -import io.emeraldpay.dshackle.upstream.calls.AggregatedCallMethods -import io.emeraldpay.dshackle.upstream.calls.CallMethods -import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods -import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods -import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods -import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosRpcUpstream +import io.emeraldpay.dshackle.upstream.calls.* +import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeRpcUpstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import org.jetbrains.annotations.NotNull import org.reactivestreams.Publisher - -class EthereumPosRpcUpstreamMock extends EthereumPosRpcUpstream { +class EthereumPosRpcUpstreamMock extends EthereumLikeRpcUpstream { EthereumHeadMock ethereumHeadMock @@ -75,7 +70,9 @@ class EthereumPosRpcUpstreamMock extends EthereumPosRpcUpstream { methods, new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(labels)), new ConnectorFactoryMock(api, new EthereumHeadMock()), - ChainsConfig.ChainConfig.default()) + ChainsConfig.ChainConfig.default(), + true + ) this.ethereumHeadMock = this.getHead() as EthereumHeadMock setLag(0) setStatus(UpstreamAvailability.OK) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumRpcUpstreamMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumRpcUpstreamMock.groovy index f1d7e0d8..da8c4c77 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/EthereumRpcUpstreamMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/EthereumRpcUpstreamMock.groovy @@ -23,18 +23,14 @@ import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.upstream.UpstreamAvailability -import io.emeraldpay.dshackle.upstream.calls.AggregatedCallMethods -import io.emeraldpay.dshackle.upstream.calls.CallMethods -import io.emeraldpay.dshackle.upstream.calls.DefaultBitcoinMethods -import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods -import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods -import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcUpstream +import io.emeraldpay.dshackle.upstream.calls.* +import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeRpcUpstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import org.jetbrains.annotations.NotNull import org.reactivestreams.Publisher -class EthereumRpcUpstreamMock extends EthereumRpcUpstream { +class EthereumRpcUpstreamMock extends EthereumLikeRpcUpstream { EthereumHeadMock ethereumHeadMock @@ -65,7 +61,9 @@ class EthereumRpcUpstreamMock extends EthereumRpcUpstream { methods, new QuorumForLabels.QuorumItem(1, new UpstreamsConfig.Labels()), new ConnectorFactoryMock(api, new EthereumHeadMock()), - ChainsConfig.ChainConfig.default()) + ChainsConfig.ChainConfig.default(), + false + ) this.ethereumHeadMock = this.getHead() as EthereumHeadMock setLag(0) setStatus(UpstreamAvailability.OK) diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy index 3d2d5b13..e85a2e4c 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy @@ -25,10 +25,9 @@ import io.emeraldpay.dshackle.upstream.bitcoin.BitcoinRpcUpstream import io.emeraldpay.dshackle.upstream.calls.CallMethods import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods import io.emeraldpay.dshackle.upstream.ethereum.EthereumCachingReader +import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeRpcUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream -import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosRpcUpstream import org.jetbrains.annotations.NotNull -import org.springframework.cloud.sleuth.Tracer import org.springframework.cloud.sleuth.brave.bridge.BraveTracer import reactor.core.scheduler.Schedulers @@ -46,9 +45,9 @@ class MultistreamHolderMock implements MultistreamHolder { if (BlockchainType.from(chain) == BlockchainType.EVM_POS) { if (up instanceof EthereumPosMultiStream) { upstreams[chain] = up - } else if (up instanceof EthereumPosRpcUpstream) { + } else if (up instanceof EthereumLikeRpcUpstream) { upstreams[chain] = new EthereumPosMultiStream( - chain, [up as EthereumPosRpcUpstream], Caches.default(), + chain, [up as EthereumLikeRpcUpstream], Caches.default(), Schedulers.parallel(), TestingCommons.tracerMock() ) } else { @@ -97,15 +96,15 @@ class MultistreamHolderMock implements MultistreamHolder { CallMethods customMethods = null Head customHead = null - EthereumMultistreamMock(@NotNull Chain chain, @NotNull List upstreams, @NotNull Caches caches) { + EthereumMultistreamMock(@NotNull Chain chain, @NotNull List upstreams, @NotNull Caches caches) { super(chain, upstreams, caches, Schedulers.parallel(), new BraveTracer(null, null, null)) } - EthereumMultistreamMock(@NotNull Chain chain, @NotNull List upstreams) { + EthereumMultistreamMock(@NotNull Chain chain, @NotNull List upstreams) { this(chain, upstreams, Caches.default()) } - EthereumMultistreamMock(@NotNull Chain chain, @NotNull EthereumPosRpcUpstream upstream) { + EthereumMultistreamMock(@NotNull Chain chain, @NotNull EthereumLikeRpcUpstream upstream) { this(chain, [upstream]) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy index 7fd36894..d0bca0fa 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy @@ -23,7 +23,7 @@ import io.emeraldpay.dshackle.startup.QuorumForLabels import io.emeraldpay.dshackle.test.EthereumApiStub import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods -import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcUpstream +import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeRpcUpstream import io.emeraldpay.dshackle.upstream.ethereum.connectors.EthereumConnectorFactory import io.emeraldpay.dshackle.upstream.forkchoice.MostWorkForkChoice import reactor.core.scheduler.Schedulers @@ -42,7 +42,7 @@ class FilteredApisSpec extends Specification { def "Verifies labels"() { setup: def i = 0 - List upstreams = [ + List upstreams = [ [test: "foo"], [test: "bar"], [test: "foo", test2: "baz"], @@ -61,7 +61,7 @@ class FilteredApisSpec extends Specification { Schedulers.parallel(), Schedulers.parallel() ) - new EthereumRpcUpstream( + new EthereumLikeRpcUpstream( "test", (byte) 123, Chain.ETHEREUM__MAINNET, @@ -70,7 +70,8 @@ class FilteredApisSpec extends Specification { ethereumTargets, new QuorumForLabels.QuorumItem(1, UpstreamsConfig.Labels.fromMap(it)), connectorFactory, - ChainsConfig.ChainConfig.default() + ChainsConfig.ChainConfig.default(), + false ) } def matcher = new Selector.LabelMatcher("test", ["foo"]) diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy index ee60377c..6a0113ec 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy @@ -27,8 +27,8 @@ import io.emeraldpay.dshackle.startup.UpstreamChangeEvent import io.emeraldpay.dshackle.test.EthereumPosRpcUpstreamMock import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods +import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeRpcUpstream import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream -import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosUpstream import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson import io.emeraldpay.dshackle.upstream.grpc.EthereumPosGrpcUpstream import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest @@ -252,7 +252,7 @@ class MultistreamSpec extends Specification { setup: def up1 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2", "eth_test3"])) def up2 = new EthereumPosRpcUpstreamMock("test2", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"])) - def ms = new EthereumPosMultiStream(Chain.ETHEREUM__MAINNET, new ArrayList(), Caches.default(), Schedulers.parallel(), TestingCommons.tracerMock()) + def ms = new EthereumPosMultiStream(Chain.ETHEREUM__MAINNET, new ArrayList(), Caches.default(), Schedulers.parallel(), TestingCommons.tracerMock()) when: ms.onUpstreamChange( new UpstreamChangeEvent(Chain.ETHEREUM__MAINNET, up1, UpstreamChangeEvent.ChangeType.ADDED) @@ -305,7 +305,7 @@ class MultistreamSpec extends Specification { setup: def up1 = new EthereumPosRpcUpstreamMock("test1", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2", "eth_test3"])) def up2 = new EthereumPosRpcUpstreamMock("test2", Chain.ETHEREUM__MAINNET, TestingCommons.api(), new DirectCallMethods(["eth_test1", "eth_test2"])) - def ms = new EthereumPosMultiStream(Chain.ETHEREUM__MAINNET, new ArrayList(), Caches.default(), Schedulers.parallel(), TestingCommons.tracerMock()) + def ms = new EthereumPosMultiStream(Chain.ETHEREUM__MAINNET, new ArrayList(), Caches.default(), Schedulers.parallel(), TestingCommons.tracerMock()) def head1 = createBlock(250, "0x0d050c785de17179f935b9b93aca09c442964cc59972c71ae68e74731448401b") def head2 = createBlock(270, "0x0d050c785de17179f935b9b93aca09c442964cc59972c71ae68e74731448402b") def head3 = createBlock(100, "0x0d050c785de17179f935b9b93aca09c442964cc59972c71ae68e74731448412b") @@ -339,7 +339,7 @@ class MultistreamSpec extends Specification { class TestEthereumPosMultistream extends EthereumPosMultiStream { - TestEthereumPosMultistream(@NotNull Chain chain, @NotNull List upstreams, @NotNull Caches caches) { + TestEthereumPosMultistream(@NotNull Chain chain, @NotNull List upstreams, @NotNull Caches caches) { super(chain, upstreams, caches, Schedulers.parallel(), TestingCommons.tracerMock()) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/ERC20BalanceSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/ERC20BalanceSpec.groovy index 55a05ccd..2c684a7d 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/ERC20BalanceSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/ERC20BalanceSpec.groovy @@ -47,7 +47,7 @@ class ERC20BalanceSpec extends Specification { JsonRpcResponse.ok('"0x0000000000000000000000000000000000000000000000000000001f28d72868"') ) - EthereumPosRpcUpstream upstream = new EthereumPosRpcUpstreamMock(Chain.ETHEREUM__MAINNET, api) + EthereumLikeRpcUpstream upstream = new EthereumPosRpcUpstreamMock(Chain.ETHEREUM__MAINNET, api) ERC20Token token = new ERC20Token(Address.from("0x54EedeAC495271d0F6B175474E89094C44Da98b9")) ERC20Balance query = new ERC20Balance() @@ -73,7 +73,7 @@ class ERC20BalanceSpec extends Specification { JsonRpcResponse.ok('"0x0000000000000000000000000000000000000000000000000000001f28d72868"') ) - EthereumPosRpcUpstream upstream = new EthereumPosRpcUpstreamMock(Chain.ETHEREUM__MAINNET, api) + EthereumLikeRpcUpstream upstream = new EthereumPosRpcUpstreamMock(Chain.ETHEREUM__MAINNET, api) ERC20Token token = new ERC20Token(Address.from("0x54EedeAC495271d0F6B175474E89094C44Da98b9")) ERC20Balance query = new ERC20Balance() diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidatorSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidatorSpec.groovy index c65bf2a6..b8158d41 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidatorSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstreamValidatorSpec.groovy @@ -16,9 +16,13 @@ package io.emeraldpay.dshackle.upstream.ethereum import io.emeraldpay.dshackle.config.UpstreamsConfig +import io.emeraldpay.dshackle.reader.Reader import io.emeraldpay.dshackle.test.ApiReaderMock import io.emeraldpay.dshackle.test.TestingCommons +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse import io.emeraldpay.etherjar.rpc.RpcResponseError +import reactor.core.publisher.Mono import reactor.util.function.Tuples import spock.lang.Specification @@ -30,7 +34,7 @@ class EthereumUpstreamValidatorSpec extends Specification { def "Resolve to final availability"() { setup: - def validator = new EthereumUpstreamValidator(Stub(EthereumUpstream), UpstreamsConfig.PartialOptions.getDefaults().buildOptions()) + def validator = new EthereumUpstreamValidator(Stub(EthereumLikeUpstream), UpstreamsConfig.PartialOptions.getDefaults().buildOptions()) expect: validator.resolve(Tuples.of(sync, peers)) == exp where: @@ -51,7 +55,7 @@ class EthereumUpstreamValidatorSpec extends Specification { def options = UpstreamsConfig.PartialOptions.getDefaults().tap { it.validateSyncing = false }.buildOptions() - def up = Mock(EthereumUpstream) + def up = Mock(EthereumLikeUpstream) def validator = new EthereumUpstreamValidator(up, options) when: @@ -79,6 +83,33 @@ class EthereumUpstreamValidatorSpec extends Specification { act == OK } + def "Execute onSyncingNode with result of eth_syncing"() { + setup: + def options = UpstreamsConfig.PartialOptions.getDefaults().tap { + it.validateSyncing = true + }.buildOptions() + def up = Mock(EthereumLikeUpstream) { + 2 * getIngressReader() >> Mock(Reader) { reader -> + 2 * reader.read(_) >>> [ + Mono.just(new JsonRpcResponse('true'.getBytes(), null)), + Mono.just(new JsonRpcResponse('false'.getBytes(), null)) + ] + } + 2 * getHead() >> Mock(Head) {head -> + 1 * head.onSyncingNode(true) + 1 * head.onSyncingNode(false) + } + } + def validator = new EthereumUpstreamValidator(up, options) + + when: + def act = validator.validateSyncing().block(Duration.ofSeconds(1)) + def act2 = validator.validateSyncing().block(Duration.ofSeconds(1)) + then: + act == SYNCING + act2 == OK + } + def "Syncing is SYNCING when state returned from upstream"() { setup: def options = UpstreamsConfig.PartialOptions.getDefaults().tap { @@ -121,7 +152,7 @@ class EthereumUpstreamValidatorSpec extends Specification { it.validatePeers = false it.minPeers = 10 }.buildOptions() - def up = Mock(EthereumUpstream) + def up = Mock(EthereumLikeUpstream) def validator = new EthereumUpstreamValidator(up, options) when: @@ -137,7 +168,7 @@ class EthereumUpstreamValidatorSpec extends Specification { it.validatePeers = true it.minPeers = 0 }.buildOptions() - def up = Mock(EthereumUpstream) + def up = Mock(EthereumLikeUpstream) def validator = new EthereumUpstreamValidator(up, options) when: diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy index 871576da..58a33a40 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumWsHeadSpec.groovy @@ -257,4 +257,69 @@ class EthereumWsHeadSpec extends Specification { .thenCancel() .verify(Duration.ofSeconds(1)) } + + def "Reset current subscription if upstream is syncing and then restore it"() { + setup: + def block = new BlockJson() + block.timestamp = Instant.now().truncatedTo(ChronoUnit.SECONDS) + block.number = 103 + block.parentHash = parent + block.hash = BlockHash.from("0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200") + def secondBlock = new BlockJson() + secondBlock.parentHash = parent + secondBlock.timestamp = Instant.now().truncatedTo(ChronoUnit.SECONDS) + secondBlock.number = 105 + secondBlock.hash = BlockHash.from("0x29229361dc5aa1ec66c323dc7a299e2b61a8c8dd2a3522d41255ec10eca25dd8") + + def firstHeadBlock = block.with { + Global.objectMapper.writeValueAsBytes(it) + } + def secondHeadBlock = secondBlock.with { + Global.objectMapper.writeValueAsBytes(it) + } + + def apiMock = TestingCommons.api() + def connectionInfoSink = Sinks.many().multicast().directBestEffort() + apiMock.answerOnce("eth_getBlockByHash", ["0x3ec2ebf5d0ec474d0ac6bc50d2770d8409ad76e119968e7919f85d5ec8915200", false], null) + apiMock.answerOnce("eth_blockNumber", [], Mono.empty()) + apiMock.answerOnce("eth_getBlockByHash", ["0x29229361dc5aa1ec66c323dc7a299e2b61a8c8dd2a3522d41255ec10eca25dd8", false], null) + apiMock.answerOnce("eth_blockNumber", [], Mono.empty()) + + def ws = Mock(WsSubscriptions) { + 1 * it.connectionInfoFlux() >> connectionInfoSink.asFlux() + 2 * subscribe("newHeads") >>> [ + new WsSubscriptions.SubscribeData(Flux.fromIterable([firstHeadBlock]), "id"), + new WsSubscriptions.SubscribeData(Flux.fromIterable([secondHeadBlock]), "id"), + ] + } + + def head = new EthereumWsHead("fake", new AlwaysForkChoice(), BlockValidator.ALWAYS_VALID, apiMock, ws, true, Schedulers.parallel(), Schedulers.parallel()) + + when: + def act = head.getFlux() + + then: + StepVerifier.create(act) + .then { head.start() } + .expectNext(BlockContainer.from(block)) + .then { + head.onSyncingNode(true) + } + .then { + assert !head.isRunning() + } + .then { + head.onNoHeadUpdates() + } + .then { + assert !head.isRunning() + } + .then { + head.onSyncingNode(false) + head.onNoHeadUpdates() + } + .expectNext(BlockContainer.from(secondBlock)) + .thenCancel() + .verify(Duration.ofSeconds(1)) + } }